2142 lines
119 KiB
Python
2142 lines
119 KiB
Python
"""Magazine route-layer regressions: path safety, review gating, script parsing."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
_CONFIG = tempfile.mkdtemp(prefix="ttvc-magazine-routes-test-")
|
|
os.environ.setdefault("CONFIG_DIR", _CONFIG)
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from routes.magazine import _article_id, _doc_dir, _history, _parse_script
|
|
|
|
|
|
async def _run_sync(fn, *args, **kwargs):
|
|
"""Stand-in for asyncio.to_thread that runs the callable inline and
|
|
stays awaitable, for routes that await its result directly rather than
|
|
firing it as a background job."""
|
|
return fn(*args, **kwargs)
|
|
|
|
|
|
class PathSafetyTests(unittest.TestCase):
|
|
def test_traversal_and_junk_ids_are_rejected(self):
|
|
for bad in ("../../etc", "a/b", "", "..", "x" * 41, "id-with-dash", "a.b"):
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
_doc_dir(bad)
|
|
self.assertEqual(ctx.exception.status_code, 400, bad)
|
|
|
|
def test_wellformed_but_missing_id_is_a_404_not_a_400(self):
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
_doc_dir("0123456789abcdef0123456789abcdef")
|
|
self.assertEqual(ctx.exception.status_code, 404)
|
|
|
|
def test_article_ids_reject_separators(self):
|
|
self.assertEqual(_article_id("article-001"), "article-001")
|
|
for bad in ("../x", "a/b", "", "a" * 65, "a b"):
|
|
with self.assertRaises(HTTPException):
|
|
_article_id(bad)
|
|
|
|
|
|
class PodcastScriptParsingTests(unittest.TestCase):
|
|
def test_plain_json_parses(self):
|
|
raw = json.dumps({
|
|
"title": "Batteries",
|
|
"turns": [{"host": "A", "text": "Hello."},
|
|
{"host": "B", "text": "Hi."}],
|
|
})
|
|
got = _parse_script(raw, ["A", "B"])
|
|
self.assertEqual(got["title"], "Batteries")
|
|
self.assertEqual([t["host"] for t in got["turns"]], ["A", "B"])
|
|
|
|
def test_markdown_fence_is_tolerated(self):
|
|
raw = '```json\n{"title":"T","turns":[{"host":"A","text":"Hi."}]}\n```'
|
|
self.assertEqual(_parse_script(raw, ["A", "B"])["turns"][0]["text"], "Hi.")
|
|
|
|
def test_prose_around_the_json_is_stripped(self):
|
|
raw = 'Sure! Here is the script:\n{"title":"T","turns":[{"host":"A","text":"Hi."}]}\nHope that helps.'
|
|
self.assertEqual(_parse_script(raw, ["A", "B"])["title"], "T")
|
|
|
|
def test_unknown_host_label_is_reassigned_not_dropped(self):
|
|
"""An unrecognised label would otherwise leave a turn with no voice."""
|
|
raw = json.dumps({"turns": [
|
|
{"host": "NARRATOR", "text": "One."},
|
|
{"host": "B", "text": "Two."},
|
|
]})
|
|
got = _parse_script(raw, ["A", "B"])
|
|
self.assertEqual(len(got["turns"]), 2)
|
|
self.assertTrue(all(t["host"] in ("A", "B") for t in got["turns"]))
|
|
|
|
def test_empty_turns_are_skipped(self):
|
|
raw = json.dumps({"turns": [
|
|
{"host": "A", "text": " "},
|
|
{"host": "B", "text": "Real line."},
|
|
]})
|
|
got = _parse_script(raw, ["A", "B"])
|
|
self.assertEqual(len(got["turns"]), 1)
|
|
|
|
def test_unusable_output_raises_502_not_500(self):
|
|
for raw in ("not json at all", "{}", json.dumps({"turns": []}),
|
|
json.dumps({"turns": [{"host": "A", "text": ""}]})):
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
_parse_script(raw, ["A", "B"])
|
|
self.assertEqual(ctx.exception.status_code, 502, raw)
|
|
|
|
def test_overlong_turn_text_is_capped(self):
|
|
raw = json.dumps({"turns": [{"host": "A", "text": "x" * 5000}]})
|
|
self.assertEqual(len(_parse_script(raw, ["A"])["turns"][0]["text"]), 2000)
|
|
|
|
|
|
def tearDownModule():
|
|
shutil.rmtree(_CONFIG, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
class ReviewImageTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_image_selection_is_saved_and_unknown_ids_rejected(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_article
|
|
article = {"id": "article-001", "images": []}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_articles', return_value=[article]), \
|
|
patch('routes.magazine.mag.load_layout', return_value={"images": [{"id": "chart"}]}), \
|
|
patch('routes.magazine.mag.save_articles') as save:
|
|
request = AsyncMock()
|
|
request.json.return_value = {"images": ["chart"]}
|
|
result = await magazine_update_article('a' * 32, 'article-001', request)
|
|
self.assertEqual(result['images'], ['chart'])
|
|
save.assert_called_once()
|
|
save.reset_mock()
|
|
request.json.return_value = {"images": ["missing"]}
|
|
with self.assertRaises(HTTPException) as error:
|
|
await magazine_update_article('a' * 32, 'article-001', request)
|
|
self.assertEqual(error.exception.status_code, 400)
|
|
save.assert_not_called()
|
|
|
|
|
|
class MagazinePlaybackProgressTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_completed_audio_progress_keeps_duration_and_resume_position(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_progress
|
|
meta = {"kind": "magazine", "progress": {}}
|
|
request = AsyncMock()
|
|
request.json.return_value = {
|
|
"article": "article-001", "seconds": 73.5,
|
|
"duration": 73.5, "completed": True,
|
|
}
|
|
with patch('routes.magazine._require_magazine', return_value=(Path('/tmp/doc'), meta)), \
|
|
patch('routes.magazine._write_meta') as write:
|
|
result = await magazine_progress('a' * 32, request)
|
|
self.assertEqual(result, {"ok": True})
|
|
self.assertEqual(meta["progress"]["article-001"]["seconds"], 73.5)
|
|
self.assertEqual(meta["progress"]["article-001"]["duration"], 73.5)
|
|
self.assertTrue(meta["progress"]["article-001"]["completed"])
|
|
write.assert_called_once()
|
|
|
|
|
|
class AdvertisementReviewTests(unittest.TestCase):
|
|
def test_llm_marks_only_known_confident_advertising_regions(self):
|
|
from unittest.mock import Mock, patch
|
|
from routes.magazine import _recognise_advertisements
|
|
layout = {'blocks': [
|
|
{'id': 'editorial', 'page': 1, 'type': 'body', 'bbox': [0,0,100,100], 'text': 'Solar technology news'},
|
|
{'id': 'advert', 'page': 1, 'type': 'title', 'bbox': [0,100,100,100], 'text': 'Buy webinar tickets'},
|
|
]}
|
|
response = Mock()
|
|
response.json.return_value = {'choices': [{'message': {'content': json.dumps({'advertisements': [
|
|
{'id': 'advert', 'confidence': 0.99}, {'id': 'editorial', 'confidence': 0.4}, {'id': 'invented', 'confidence': 1}
|
|
]})}}]}
|
|
with patch('routes.conversation._post_llm_chat_completion', return_value=response):
|
|
_recognise_advertisements(layout, {'llm_model': 'test'})
|
|
self.assertEqual(layout['blocks'][0]['type'], 'body')
|
|
self.assertEqual(layout['blocks'][1]['type'], 'ad')
|
|
self.assertEqual(layout['advertisement_review'], 'done')
|
|
|
|
def test_failed_llm_keeps_existing_regions(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _recognise_advertisements
|
|
layout = {'blocks': [{'id': 'b', 'page': 1, 'type': 'body', 'bbox': [0,0,1,1], 'text': 'Article'}]}
|
|
with patch('routes.conversation._post_llm_chat_completion', side_effect=RuntimeError('offline')):
|
|
_recognise_advertisements(layout, {'llm_model': 'test'})
|
|
self.assertEqual(layout['blocks'][0]['type'], 'body')
|
|
self.assertEqual(layout['advertisement_review'], 'partial_failure')
|
|
|
|
def test_semantic_review_collapses_a_full_page_advertisement(self):
|
|
from unittest.mock import Mock, patch
|
|
from routes.magazine import _recognise_semantic_layout
|
|
layout = {
|
|
'pages': [{'page': 33, 'width': 600, 'height': 800}],
|
|
'blocks': [
|
|
{'id': 'offer', 'page': 33, 'type': 'ad', 'bbox': [30,40,500,80], 'text': 'Subscriber discount'},
|
|
{'id': 'order', 'page': 33, 'type': 'ad', 'bbox': [30,500,500,80], 'text': 'Order now'},
|
|
],
|
|
}
|
|
response = Mock()
|
|
response.json.return_value = {'choices': [{'message': {'content': json.dumps({
|
|
'page_kind': 'advertisement', 'confidence': .99,
|
|
'regions': [{'type': 'advertisement', 'block_ids': ['offer', 'order']}],
|
|
})}}]}
|
|
with patch('routes.conversation._post_llm_chat_completion', return_value=response):
|
|
_recognise_semantic_layout(layout, {'llm_model': 'test'}, page=33)
|
|
self.assertEqual(layout['semantic_review'], 'done')
|
|
self.assertEqual(len(layout['blocks']), 1)
|
|
self.assertEqual(layout['blocks'][0]['type'], 'ad')
|
|
self.assertEqual(layout['blocks'][0]['bbox'], [0, 0, 600, 800])
|
|
|
|
|
|
class RefineFiguresJobTests(unittest.TestCase):
|
|
"""Document-wide figure re-snap: reports what moved, keeps text untouched."""
|
|
|
|
def _layout(self):
|
|
return {
|
|
"pages": [{"page": 1, "width": 600, "height": 800},
|
|
{"page": 2, "width": 600, "height": 800}],
|
|
"blocks": [
|
|
{"id": "fig", "page": 1, "type": "figure", "bbox": [10, 10, 50, 50], "text": ""},
|
|
{"id": "copy", "page": 2, "type": "body", "bbox": [0, 0, 100, 100], "text": "Body"},
|
|
],
|
|
}
|
|
|
|
def test_moved_boxes_are_counted_and_saved(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _refine_figures_sync
|
|
layout = self._layout()
|
|
|
|
def refresh(_dir, doc, page=None):
|
|
if page == 1:
|
|
doc["blocks"][0]["bbox"] = [20, 20, 90, 90]
|
|
|
|
with patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.refresh_visual_regions', side_effect=refresh), \
|
|
patch('routes.magazine.mag.save_layout') as save, \
|
|
patch('routes.magazine.background_job_update') as update:
|
|
_refine_figures_sync('a' * 32, 'job')
|
|
save.assert_called_once()
|
|
done = [c.kwargs for c in update.call_args_list if c.kwargs.get('done')]
|
|
self.assertEqual(done[-1]['stage'], 'done')
|
|
self.assertEqual(done[-1]['result'], {'moved': 1})
|
|
self.assertEqual(layout['blocks'][1]['bbox'], [0, 0, 100, 100])
|
|
|
|
def test_nothing_to_correct_still_finishes_cleanly(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _refine_figures_sync
|
|
with patch('routes.magazine.mag.load_layout', return_value=self._layout()), \
|
|
patch('routes.magazine.mag.refresh_visual_regions'), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine.background_job_update') as update:
|
|
_refine_figures_sync('a' * 32, 'job')
|
|
done = [c.kwargs for c in update.call_args_list if c.kwargs.get('done')]
|
|
self.assertEqual(done[-1]['result'], {'moved': 0})
|
|
self.assertNotIn('error', done[-1])
|
|
|
|
def test_page_job_only_refreshes_the_requested_page(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _refine_figures_sync
|
|
layout = self._layout()
|
|
with patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.refresh_visual_regions') as refresh, \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine.background_job_update') as update:
|
|
_refine_figures_sync('a' * 32, 'job', page=2)
|
|
self.assertEqual(refresh.call_count, 1)
|
|
self.assertEqual(refresh.call_args.args[2], 2)
|
|
progress = [c.kwargs.get('detail', '') for c in update.call_args_list]
|
|
self.assertIn('Checking figure boundaries on page 2', progress)
|
|
self.assertFalse(any('of 2' in detail for detail in progress))
|
|
|
|
def test_a_document_with_no_layout_reports_failure(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _refine_figures_sync
|
|
with patch('routes.magazine.mag.load_layout', return_value=None), \
|
|
patch('routes.magazine.mag.save_layout') as save, \
|
|
patch('routes.magazine.background_job_update') as update:
|
|
_refine_figures_sync('a' * 32, 'job')
|
|
save.assert_not_called()
|
|
done = [c.kwargs for c in update.call_args_list if c.kwargs.get('done')]
|
|
self.assertEqual(done[-1]['stage'], 'failed')
|
|
self.assertIn('No layout', done[-1]['error'])
|
|
|
|
|
|
class PageQualityJobTests(unittest.TestCase):
|
|
def test_missing_text_quality_job_saves_additive_result(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _quality_page_sync
|
|
layout = {'pages': [{'page': 2}], 'blocks': []}
|
|
updated = {'pages': [{'page': 2}], 'blocks': [{'id': 'new'}]}
|
|
with patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.recover_missing_text',
|
|
return_value=(updated, 1)) as recover, \
|
|
patch('routes.magazine.mag.save_layout') as save, \
|
|
patch('routes.magazine.background_job_update') as update:
|
|
_quality_page_sync('a' * 32, 'job', 2, 'missing')
|
|
recover.assert_called_once()
|
|
save.assert_called_once()
|
|
self.assertEqual(save.call_args.args[1], updated)
|
|
done = [call.kwargs for call in update.call_args_list if call.kwargs.get('done')]
|
|
self.assertEqual(done[-1]['result'], {'page': 2, 'added': 1})
|
|
|
|
def test_recognition_quality_job_reports_checked_and_corrected(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _quality_page_sync
|
|
layout = {'pages': [{'page': 3}], 'blocks': []}
|
|
with patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.verify_ocr_recognition',
|
|
return_value=(layout, 4, 2)), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine.background_job_update') as update:
|
|
_quality_page_sync('a' * 32, 'job', 3, 'recognition')
|
|
done = [call.kwargs for call in update.call_args_list if call.kwargs.get('done')]
|
|
self.assertEqual(done[-1]['result'], {'page': 3, 'checked': 4, 'corrected': 2})
|
|
|
|
|
|
class PageQualityRouteTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_missing_text_route_starts_page_scoped_job(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_find_missing_text
|
|
layout = {'pages': [{'page': 7}], 'blocks': []}
|
|
with patch('routes.magazine._require_magazine', return_value=(Path('/tmp/doc'), {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.background_job_create') as create, \
|
|
patch('routes.magazine.asyncio.to_thread', new=lambda *a, **k: None), \
|
|
patch('asyncio.create_task') as create_task:
|
|
result = await magazine_find_missing_text('a' * 32, 7)
|
|
self.assertIn('job_id', result)
|
|
create.assert_called_once()
|
|
create_task.assert_called_once()
|
|
|
|
async def test_quality_route_rejects_unknown_page(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_find_missing_text
|
|
with patch('routes.magazine._require_magazine', return_value=(Path('/tmp/doc'), {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value={'pages': []}):
|
|
with self.assertRaises(HTTPException) as error:
|
|
await magazine_find_missing_text('a' * 32, 99)
|
|
self.assertEqual(error.exception.status_code, 404)
|
|
|
|
|
|
class IngestPreviewHtmlTests(unittest.TestCase):
|
|
"""The live ingest preview: a fast, best-effort HTML render of one
|
|
page's just-read blocks, well before classify_blocks or article
|
|
grouping has run."""
|
|
|
|
def _block(self, text, font_size=10.0, type="body", bbox=None):
|
|
from core.magazine import Block
|
|
return Block(id="b", page=1, bbox=bbox or [0, 0, 100, 10],
|
|
type=type, text=text, font_size=font_size)
|
|
|
|
def test_reading_order_follows_vertical_position(self):
|
|
from routes.magazine import _render_ingest_preview_html
|
|
top = self._block("First", bbox=[0, 10, 100, 10])
|
|
bottom = self._block("Second", bbox=[0, 200, 100, 10])
|
|
html = _render_ingest_preview_html([bottom, top], None)
|
|
self.assertLess(html.index("First"), html.index("Second"))
|
|
|
|
def test_large_font_becomes_a_heading(self):
|
|
from routes.magazine import _render_ingest_preview_html
|
|
body = self._block("Normal copy", font_size=10.0)
|
|
heading = self._block("Big Headline", font_size=24.0, bbox=[0, 50, 100, 10])
|
|
html = _render_ingest_preview_html([body, heading], None)
|
|
self.assertIn("<h3>Big Headline</h3>", html)
|
|
self.assertNotIn("<h3>Normal copy</h3>", html)
|
|
|
|
def test_furniture_is_muted_never_a_heading(self):
|
|
from routes.magazine import _render_ingest_preview_html
|
|
from core.magazine import BLOCK_FURNITURE
|
|
furniture = self._block("Page 12", font_size=24.0, type=BLOCK_FURNITURE)
|
|
html = _render_ingest_preview_html([furniture], None)
|
|
self.assertIn('class="mag-ingest-preview-excluded"', html)
|
|
self.assertNotIn("<h3", html)
|
|
|
|
def test_figure_block_text_is_not_duplicated_as_a_paragraph(self):
|
|
from routes.magazine import _render_ingest_preview_html
|
|
from core.magazine import BLOCK_FIGURE
|
|
figure = self._block("Post your new Instagram photos", type=BLOCK_FIGURE)
|
|
html = _render_ingest_preview_html([figure], None)
|
|
self.assertNotIn("Instagram", html)
|
|
|
|
def test_an_image_becomes_a_placeholder_not_the_figure_text(self):
|
|
from routes.magazine import _render_ingest_preview_html
|
|
html = _render_ingest_preview_html([], [{"bbox": [0, 5, 50, 50]}])
|
|
self.assertIn("mag-ingest-preview-figure", html)
|
|
self.assertIn("Picture", html)
|
|
self.assertNotIn("<img", html) # no filename yet -- not cropped, so no thumbnail to show
|
|
|
|
def test_a_cropped_region_shows_its_actual_thumbnail(self):
|
|
from routes.magazine import _render_ingest_preview_html
|
|
chart = {"bbox": [0, 5, 50, 50], "type": "chart", "filename": "p0001-region-b1.png"}
|
|
html = _render_ingest_preview_html([], [chart], doc_id="doc1")
|
|
self.assertIn('<img src="/api/magazine/doc1/images/p0001-region-b1.png"', html)
|
|
self.assertIn("Chart", html)
|
|
self.assertIn("<figcaption>Chart</figcaption>", html)
|
|
|
|
def test_a_bad_filename_falls_back_to_the_placeholder_rather_than_a_bad_src(self):
|
|
from routes.magazine import _render_ingest_preview_html
|
|
bad = {"bbox": [0, 5, 50, 50], "type": "table", "filename": "../../etc/passwd"}
|
|
html = _render_ingest_preview_html([], [bad], doc_id="doc1")
|
|
self.assertNotIn("<img", html)
|
|
self.assertIn("Table", html)
|
|
|
|
def test_text_is_escaped(self):
|
|
from routes.magazine import _render_ingest_preview_html
|
|
block = self._block("<script>alert(1)</script> & co")
|
|
html = _render_ingest_preview_html([block], None)
|
|
self.assertNotIn("<script>", html)
|
|
self.assertIn("<script>", html)
|
|
self.assertIn("& co", html)
|
|
|
|
def test_empty_text_is_skipped(self):
|
|
from routes.magazine import _render_ingest_preview_html
|
|
blank = self._block(" ")
|
|
real = self._block("Real text", bbox=[0, 50, 100, 10])
|
|
html = _render_ingest_preview_html([blank, real], None)
|
|
self.assertEqual(html.count("<div>"), 1)
|
|
|
|
def test_a_flood_of_blocks_is_capped_not_rendered_in_full(self):
|
|
from routes.magazine import _render_ingest_preview_html, _INGEST_PREVIEW_BUDGET
|
|
blocks = [self._block("x" * 200, bbox=[0, i, 100, 1]) for i in range(200)]
|
|
html = _render_ingest_preview_html(blocks, None)
|
|
self.assertLess(len(html), _INGEST_PREVIEW_BUDGET + 500)
|
|
self.assertIn("mag-ingest-preview-more", html)
|
|
|
|
def test_no_blocks_or_images_is_an_empty_string(self):
|
|
from routes.magazine import _render_ingest_preview_html
|
|
self.assertEqual(_render_ingest_preview_html([], None), "")
|
|
self.assertEqual(_render_ingest_preview_html([], []), "")
|
|
|
|
|
|
class DescribeImageJobTests(unittest.TestCase):
|
|
"""Figure descriptions are additive: generating one never touches the
|
|
article's own blocks, only the image record's own description field."""
|
|
|
|
def _layout(self):
|
|
return {
|
|
"images": [
|
|
{"id": "img1", "page": 1, "bbox": [0, 0, 10, 10], "filename": "img1.png", "type": "figure"},
|
|
],
|
|
}
|
|
|
|
def test_successful_description_is_saved(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _describe_image_sync
|
|
layout = self._layout()
|
|
with patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.describe_visual_region_chain',
|
|
return_value='A photo of a mountain.') as chain, \
|
|
patch('routes.magazine.mag.save_layout') as save, \
|
|
patch('routes.magazine.background_job_update') as update, \
|
|
patch('pathlib.Path.is_file', return_value=True):
|
|
_describe_image_sync('a' * 32, 'job', 'img1', 'http://e/v1', 'm', None, None)
|
|
save.assert_called_once()
|
|
self.assertEqual(layout['images'][0]['description'], 'A photo of a mountain.')
|
|
chain.assert_called_once()
|
|
self.assertEqual(chain.call_args.args[1], 'figure') # region type passed through
|
|
done = [c.kwargs for c in update.call_args_list if c.kwargs.get('done')]
|
|
self.assertEqual(done[-1]['stage'], 'done')
|
|
self.assertEqual(done[-1]['result'], {'image_id': 'img1', 'description': 'A photo of a mountain.'})
|
|
|
|
def test_no_layout_reports_failure(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _describe_image_sync
|
|
with patch('routes.magazine.mag.load_layout', return_value=None), \
|
|
patch('routes.magazine.mag.save_layout') as save, \
|
|
patch('routes.magazine.background_job_update') as update:
|
|
_describe_image_sync('a' * 32, 'job', 'img1', 'http://e/v1', 'm', None, None)
|
|
save.assert_not_called()
|
|
done = [c.kwargs for c in update.call_args_list if c.kwargs.get('done')]
|
|
self.assertEqual(done[-1]['stage'], 'failed')
|
|
self.assertIn('No layout', done[-1]['error'])
|
|
|
|
def test_unknown_image_id_reports_failure(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _describe_image_sync
|
|
with patch('routes.magazine.mag.load_layout', return_value=self._layout()), \
|
|
patch('routes.magazine.mag.save_layout') as save, \
|
|
patch('routes.magazine.background_job_update') as update:
|
|
_describe_image_sync('a' * 32, 'job', 'missing', 'http://e/v1', 'm', None, None)
|
|
save.assert_not_called()
|
|
done = [c.kwargs for c in update.call_args_list if c.kwargs.get('done')]
|
|
self.assertEqual(done[-1]['stage'], 'failed')
|
|
self.assertIn('not found', done[-1]['error'].lower())
|
|
|
|
def test_engine_returning_nothing_usable_is_a_failure_not_a_blank_save(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _describe_image_sync
|
|
layout = self._layout()
|
|
with patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.describe_visual_region_chain', return_value=None), \
|
|
patch('routes.magazine.mag.save_layout') as save, \
|
|
patch('routes.magazine.background_job_update') as update, \
|
|
patch('pathlib.Path.is_file', return_value=True):
|
|
_describe_image_sync('a' * 32, 'job', 'img1', 'http://e/v1', 'm', None, None)
|
|
save.assert_not_called()
|
|
self.assertNotIn('description', layout['images'][0])
|
|
done = [c.kwargs for c in update.call_args_list if c.kwargs.get('done')]
|
|
self.assertEqual(done[-1]['stage'], 'failed')
|
|
|
|
|
|
class DescribeImageRouteTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_unconfigured_vision_engine_is_rejected(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_describe_image
|
|
layout = {"images": [{"id": "img1", "page": 1, "bbox": [0, 0, 1, 1]}]}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine._load_settings', return_value={}):
|
|
request = AsyncMock()
|
|
request.json.return_value = {}
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_describe_image('a' * 32, 'img1', request)
|
|
self.assertEqual(err.exception.status_code, 400)
|
|
|
|
async def test_unknown_image_id_is_a_404(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_describe_image
|
|
layout = {"images": [{"id": "img1", "page": 1, "bbox": [0, 0, 1, 1]}]}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout):
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_describe_image('a' * 32, 'missing', None)
|
|
self.assertEqual(err.exception.status_code, 404)
|
|
|
|
async def test_valid_request_starts_a_background_job(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_describe_image
|
|
layout = {"images": [{"id": "img1", "page": 1, "bbox": [0, 0, 1, 1]}]}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine._load_settings',
|
|
return_value={'vision_ocr_url': 'http://e/v1', 'vision_ocr_model': 'm'}), \
|
|
patch('routes.magazine.background_job_create') as create, \
|
|
patch('routes.magazine.asyncio.to_thread', new=lambda *a, **k: None), \
|
|
patch('asyncio.create_task') as create_task:
|
|
request = AsyncMock()
|
|
request.json.return_value = {}
|
|
result = await magazine_describe_image('a' * 32, 'img1', request)
|
|
self.assertIn('job_id', result)
|
|
create.assert_called_once()
|
|
create_task.assert_called_once()
|
|
|
|
|
|
class UpdateImageDescriptionRouteTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_sets_the_description_verbatim(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_image_description
|
|
layout = {"images": [{"id": "img1", "page": 1, "bbox": [0, 0, 1, 1]}]}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.save_layout') as save:
|
|
request = AsyncMock()
|
|
request.json.return_value = {'description': ' A hand-typed description. '}
|
|
result = await magazine_update_image_description('a' * 32, 'img1', request)
|
|
self.assertEqual(result['description'], 'A hand-typed description.')
|
|
self.assertEqual(layout['images'][0]['description'], 'A hand-typed description.')
|
|
save.assert_called_once()
|
|
|
|
async def test_unknown_image_id_is_a_404(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_image_description
|
|
layout = {"images": [{"id": "img1", "page": 1, "bbox": [0, 0, 1, 1]}]}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout):
|
|
request = AsyncMock()
|
|
request.json.return_value = {'description': 'x'}
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_update_image_description('a' * 32, 'missing', request)
|
|
self.assertEqual(err.exception.status_code, 404)
|
|
|
|
async def test_missing_description_field_is_rejected(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_image_description
|
|
layout = {"images": [{"id": "img1", "page": 1, "bbox": [0, 0, 1, 1]}]}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout):
|
|
request = AsyncMock()
|
|
request.json.return_value = {}
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_update_image_description('a' * 32, 'img1', request)
|
|
self.assertEqual(err.exception.status_code, 400)
|
|
|
|
async def test_no_layout_is_a_404(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_image_description
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=None):
|
|
request = AsyncMock()
|
|
request.json.return_value = {'description': 'x'}
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_update_image_description('a' * 32, 'img1', request)
|
|
self.assertEqual(err.exception.status_code, 404)
|
|
|
|
|
|
class UpdateRegionRouteTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_moves_a_block_and_returns_no_text_without_reocr(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_region
|
|
layout = {"pages": [{"page": 1, "width": 600, "height": 800}],
|
|
"blocks": [{"id": "b1", "page": 1, "bbox": [10, 10, 100, 20], "type": "body", "text": "hi"}]}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.save_layout') as save, \
|
|
patch('routes.magazine._load_settings', return_value={}), \
|
|
patch('routes.magazine.asyncio.to_thread', new=_run_sync):
|
|
request = AsyncMock()
|
|
request.json.return_value = {'bbox': [20, 20, 100, 20]}
|
|
result = await magazine_update_region('a' * 32, 1, 'b1', request)
|
|
self.assertTrue(result['ok'])
|
|
self.assertIsNone(result['text'])
|
|
save.assert_called_once()
|
|
|
|
async def test_reocr_replaces_text_when_extraction_succeeds(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_region
|
|
layout = {"pages": [{"page": 1, "width": 600, "height": 800}],
|
|
"blocks": [{"id": "b1", "page": 1, "bbox": [10, 10, 100, 20], "type": "body", "text": "stale"}]}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine.mag.extract_region_text', return_value='fresh text'), \
|
|
patch('routes.magazine._load_settings',
|
|
return_value={'vision_ocr_url': 'http://e/v1', 'vision_ocr_model': 'm'}), \
|
|
patch('routes.magazine.asyncio.to_thread', new=_run_sync):
|
|
request = AsyncMock()
|
|
request.json.return_value = {'bbox': [10, 10, 100, 20], 'reocr': True}
|
|
result = await magazine_update_region('a' * 32, 1, 'b1', request)
|
|
self.assertEqual(result['text'], 'fresh text')
|
|
|
|
async def test_bad_bbox_is_rejected(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_region
|
|
layout = {"pages": [{"page": 1, "width": 600, "height": 800}], "blocks": []}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout):
|
|
request = AsyncMock()
|
|
request.json.return_value = {'bbox': [1, 2, 3]}
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_update_region('a' * 32, 1, 'b1', request)
|
|
self.assertEqual(err.exception.status_code, 400)
|
|
|
|
async def test_unknown_region_is_a_400(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_region
|
|
layout = {"pages": [{"page": 1, "width": 600, "height": 800}], "blocks": [], "images": []}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine._load_settings', return_value={}), \
|
|
patch('routes.magazine.asyncio.to_thread', new=_run_sync):
|
|
request = AsyncMock()
|
|
request.json.return_value = {'bbox': [1, 2, 3, 4]}
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_update_region('a' * 32, 1, 'missing', request)
|
|
self.assertEqual(err.exception.status_code, 400)
|
|
|
|
async def test_no_layout_is_a_404(self):
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_region
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=None):
|
|
request = AsyncMock()
|
|
request.json.return_value = {'bbox': [1, 2, 3, 4]}
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_update_region('a' * 32, 1, 'b1', request)
|
|
self.assertEqual(err.exception.status_code, 404)
|
|
|
|
|
|
class DeleteRegionRouteTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_removes_a_known_block(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_delete_region
|
|
layout = {"blocks": [{"id": "b1", "page": 1}], "images": []}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.save_layout') as save, \
|
|
patch('routes.magazine.mag.load_articles', return_value=[]):
|
|
result = await magazine_delete_region('a' * 32, 1, 'b1')
|
|
self.assertTrue(result['ok'])
|
|
save.assert_called_once()
|
|
|
|
async def test_removes_a_known_image(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_delete_region
|
|
layout = {"blocks": [], "images": [{"id": "img1", "page": 1}]}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine.mag.load_articles', return_value=[]):
|
|
result = await magazine_delete_region('a' * 32, 1, 'img1')
|
|
self.assertTrue(result['ok'])
|
|
|
|
async def test_unknown_id_is_a_404(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_delete_region
|
|
layout = {"blocks": [], "images": []}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout):
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_delete_region('a' * 32, 1, 'missing')
|
|
self.assertEqual(err.exception.status_code, 404)
|
|
|
|
async def test_no_layout_is_a_404(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_delete_region
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=None):
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_delete_region('a' * 32, 1, 'b1')
|
|
self.assertEqual(err.exception.status_code, 404)
|
|
|
|
async def test_a_deleted_region_leaves_every_article_that_listed_it(self):
|
|
"""An article still listing a deleted id made Approve fail with "Unknown block ids"."""
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_delete_region
|
|
layout = {"blocks": [{"id": "b1", "page": 1}, {"id": "b2", "page": 1}], "images": [{"id": "img", "page": 1}]}
|
|
articles = [{"id": "a1", "blocks": ["b1", "b2"], "images": ["img"]}, {"id": "a2", "blocks": ["b2"], "images": []}]
|
|
saved = {}
|
|
for rid, expect in (("b1", [["b2"], ["img"], ["b2"]]), ("img", [["b1", "b2"], [], ["b2"]])):
|
|
with self.subTest(rid=rid), \
|
|
patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine.mag.load_articles', return_value=[dict(a) for a in articles]), \
|
|
patch('routes.magazine.mag.save_articles', side_effect=lambda _d, a: saved.__setitem__('a', a)):
|
|
await magazine_delete_region('a' * 32, 1, rid)
|
|
self.assertEqual([saved['a'][0]['blocks'], saved['a'][0]['images'], saved['a'][1]['blocks']], expect)
|
|
|
|
|
|
class ChangeRegionTypeTests(unittest.TestCase):
|
|
"""Changing what a detected region is, from the review list."""
|
|
|
|
def setUp(self):
|
|
from pathlib import Path
|
|
from PIL import Image
|
|
self.dir = Path(tempfile.mkdtemp(prefix="ttvc-regiontype-"))
|
|
(self.dir / "pages").mkdir()
|
|
Image.new("RGB", (600, 800), "white").save(self.dir / "pages" / "0001.png")
|
|
self.layout = {
|
|
"pages": [{"page": 1, "width": 600, "height": 800}],
|
|
"blocks": [
|
|
{"id": "b1", "page": 1, "bbox": [20, 20, 250, 40], "type": "body", "text": "A headline set as body"},
|
|
{"id": "b2", "page": 1, "bbox": [20, 100, 250, 200], "type": "body", "text": "Running text."},
|
|
{"id": "b3", "page": 1, "bbox": [320, 100, 250, 200], "type": "body", "text": "Actually a photo."},
|
|
],
|
|
"images": [{"id": "img1", "page": 1, "bbox": [320, 400, 250, 150], "filename": "img1.png", "reading_order": 0}],
|
|
}
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
def test_a_text_region_becomes_a_headline_and_keeps_its_text(self):
|
|
from core.magazine import change_region_type
|
|
layout, new_id = change_region_type(self.dir, self.layout, 1, "b1", "title")
|
|
block = next(b for b in layout["blocks"] if b["id"] == "b1")
|
|
self.assertEqual((new_id, block["type"], block["text"]), ("b1", "title", "A headline set as body"))
|
|
|
|
def test_page_number_and_source_are_furniture_of_different_kinds(self):
|
|
from core.magazine import change_region_type
|
|
layout, _ = change_region_type(self.dir, self.layout, 1, "b1", "page_number")
|
|
block = next(b for b in layout["blocks"] if b["id"] == "b1")
|
|
self.assertEqual((block["type"], block["region_label"]), ("furniture", "page_number"))
|
|
layout, _ = change_region_type(self.dir, layout, 1, "b1", "source")
|
|
self.assertEqual(next(b for b in layout["blocks"] if b["id"] == "b1")["region_label"], "source")
|
|
|
|
def test_back_to_body_clears_the_furniture_kind(self):
|
|
from core.magazine import change_region_type
|
|
layout, _ = change_region_type(self.dir, self.layout, 1, "b1", "page_number")
|
|
layout, _ = change_region_type(self.dir, layout, 1, "b1", "body")
|
|
self.assertEqual(next(b for b in layout["blocks"] if b["id"] == "b1")["region_label"], "")
|
|
|
|
def test_a_text_region_becomes_an_image_crop(self):
|
|
from core.magazine import change_region_type
|
|
layout, new_id = change_region_type(self.dir, self.layout, 1, "b3", "image")
|
|
self.assertNotIn("b3", [b["id"] for b in layout["blocks"]])
|
|
image = next(i for i in layout["images"] if i["id"] == new_id)
|
|
self.assertEqual(image["bbox"], [320, 100, 250, 200])
|
|
self.assertTrue((self.dir / "images" / image["filename"]).is_file())
|
|
|
|
def test_an_image_becomes_text_read_by_ocr(self):
|
|
from unittest import mock
|
|
import core.magazine as mag
|
|
with mock.patch.object(mag, "vision_ocr_page_chain", return_value=[{"text": "Read from the box"}]):
|
|
layout, new_id = mag.change_region_type(self.dir, self.layout, 1, "img1", "caption", "http://ocr", "m")
|
|
self.assertNotIn("img1", [i["id"] for i in layout["images"]])
|
|
block = next(b for b in layout["blocks"] if b["id"] == new_id)
|
|
self.assertEqual((block["type"], block["text"]), ("caption", "Read from the box"))
|
|
|
|
def test_an_unknown_type_or_region_is_rejected(self):
|
|
from core.magazine import change_region_type
|
|
with self.assertRaises(ValueError):
|
|
change_region_type(self.dir, self.layout, 1, "b1", "nonsense")
|
|
with self.assertRaises(ValueError):
|
|
change_region_type(self.dir, self.layout, 1, "missing", "body")
|
|
|
|
def test_a_hand_ordered_page_keeps_its_order(self):
|
|
from core.magazine import change_region_type
|
|
self.layout["pages"][0]["order_locked"] = True
|
|
for i, b in enumerate(self.layout["blocks"]):
|
|
b["reading_order"] = [2, 0, 1][i]
|
|
layout, _ = change_region_type(self.dir, self.layout, 1, "b1", "page_number")
|
|
self.assertEqual([b["reading_order"] for b in layout["blocks"]], [2, 0, 1])
|
|
|
|
def test_route_moves_article_membership_when_text_becomes_an_image(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_region
|
|
articles = [{"id": "a1", "blocks": ["b2", "b3"], "images": []}]
|
|
saved = {}
|
|
with patch('routes.magazine._require_magazine', return_value=(self.dir, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=self.layout), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine.mag.load_articles', return_value=articles), \
|
|
patch('routes.magazine.mag.save_articles', side_effect=lambda d, a: saved.setdefault('a', a)), \
|
|
patch('routes.magazine._load_settings', return_value={}), \
|
|
patch('routes.magazine.asyncio.to_thread', new=_run_sync):
|
|
request = AsyncMock()
|
|
request.json.return_value = {'type': 'image'}
|
|
result = asyncio.run(magazine_update_region('a' * 32, 1, 'b3', request))
|
|
self.assertNotEqual(result['id'], 'b3')
|
|
self.assertEqual(saved['a'][0]['blocks'], ['b2'])
|
|
self.assertEqual(saved['a'][0]['images'], [result['id']])
|
|
|
|
def test_route_rejects_an_unknown_type(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_update_region
|
|
with patch('routes.magazine._require_magazine', return_value=(self.dir, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=self.layout):
|
|
request = AsyncMock()
|
|
request.json.return_value = {'type': 'nonsense'}
|
|
with self.assertRaises(HTTPException) as err:
|
|
asyncio.run(magazine_update_region('a' * 32, 1, 'b1', request))
|
|
self.assertEqual(err.exception.status_code, 400)
|
|
|
|
|
|
class RecomputeReadingOrderTests(unittest.TestCase):
|
|
def layout(self, locked=False):
|
|
# Two columns stored in an interleaved (wrong) order.
|
|
rows = [("l1", 50, 100, 1), ("r1", 320, 100, 2), ("l2", 50, 200, 3), ("r2", 320, 200, 4)]
|
|
return {"pages": [{"page": 1, "width": 595.0, "height": 842.0, **({"order_locked": True} if locked else {})}],
|
|
"blocks": [{"id": i, "page": 1, "bbox": [x, y, 200, 80], "type": "body", "text": i,
|
|
"reading_order": ro} for i, x, y, ro in rows]}
|
|
|
|
def order(self, layout):
|
|
return [b["id"] for b in sorted(layout["blocks"], key=lambda b: b["reading_order"])]
|
|
|
|
def test_an_interleaved_page_is_reread_column_by_column(self):
|
|
from core.magazine import recompute_reading_order
|
|
layout, changed = recompute_reading_order(self.layout())
|
|
self.assertEqual(self.order(layout), ["l1", "l2", "r1", "r2"])
|
|
self.assertEqual(changed, [1])
|
|
|
|
def test_a_hand_ordered_page_is_left_alone(self):
|
|
from core.magazine import recompute_reading_order
|
|
layout, changed = recompute_reading_order(self.layout(locked=True))
|
|
self.assertEqual(self.order(layout), ["l1", "r1", "l2", "r2"])
|
|
self.assertEqual(changed, [])
|
|
|
|
def test_text_and_types_are_untouched(self):
|
|
from core.magazine import recompute_reading_order
|
|
before = self.layout()
|
|
after, _ = recompute_reading_order(before)
|
|
strip = lambda bs: sorted((b["id"], b["text"], b["type"], tuple(b["bbox"])) for b in bs)
|
|
self.assertEqual(strip(after["blocks"]), strip(before["blocks"]))
|
|
|
|
|
|
class HandOrderTests(unittest.TestCase):
|
|
"""Ordering a page by hand, and every other path leaving that order alone."""
|
|
|
|
def layout(self):
|
|
return {
|
|
"pages": [{"page": 1, "width": 595.0, "height": 842.0}],
|
|
"blocks": [{"id": i, "page": 1, "bbox": [x, y, 200, 80], "type": "body", "text": i, "reading_order": ro}
|
|
for i, x, y, ro in (("l1", 50, 100, 0), ("l2", 50, 200, 1), ("r1", 320, 100, 2), ("r2", 320, 200, 3))],
|
|
"images": [{"id": "img", "page": 1, "bbox": [320, 400, 200, 100], "filename": "img.png", "reading_order": 4}],
|
|
}
|
|
|
|
def order(self, layout):
|
|
items = [r for r in layout["blocks"] + layout["images"] if r["page"] == 1]
|
|
return [r["id"] for r in sorted(items, key=lambda r: r["reading_order"])]
|
|
|
|
def test_the_listed_order_is_kept_and_the_page_is_locked(self):
|
|
from core.magazine import set_page_order
|
|
layout = set_page_order(self.layout(), 1, ["r1", "img", "l1", "l2", "r2"])
|
|
self.assertEqual(self.order(layout), ["r1", "img", "l1", "l2", "r2"])
|
|
self.assertTrue(layout["pages"][0]["order_locked"])
|
|
|
|
def test_regions_left_out_follow_in_their_old_order(self):
|
|
from core.magazine import set_page_order
|
|
layout = set_page_order(self.layout(), 1, ["r2"])
|
|
self.assertEqual(self.order(layout), ["r2", "l1", "l2", "r1", "img"])
|
|
|
|
def test_an_id_from_another_page_is_rejected(self):
|
|
from core.magazine import set_page_order
|
|
with self.assertRaises(ValueError):
|
|
set_page_order(self.layout(), 1, ["l1", "elsewhere"])
|
|
|
|
def test_fix_reading_order_and_type_changes_leave_a_hand_ordered_page_alone(self):
|
|
from core.magazine import recompute_reading_order, set_page_order
|
|
layout = set_page_order(self.layout(), 1, ["r1", "r2", "l1", "l2"])
|
|
layout, changed = recompute_reading_order(layout)
|
|
self.assertEqual(changed, [])
|
|
self.assertEqual(self.order(layout)[:4], ["r1", "r2", "l1", "l2"])
|
|
|
|
def test_a_new_region_on_a_hand_ordered_page_goes_last(self):
|
|
from pathlib import Path
|
|
from PIL import Image
|
|
from core.magazine import add_manual_region, set_page_order
|
|
d = Path(tempfile.mkdtemp(prefix="ttvc-handorder-"))
|
|
(d / "pages").mkdir()
|
|
Image.new("RGB", (600, 800), "white").save(d / "pages" / "0001.png")
|
|
try:
|
|
layout = set_page_order(self.layout(), 1, ["r1", "r2", "l1", "l2"])
|
|
layout = add_manual_region(d, layout, 1, [50, 20, 400, 40], "text", "New headline")
|
|
order = self.order(layout)
|
|
self.assertEqual(order[:4], ["r1", "r2", "l1", "l2"])
|
|
self.assertTrue(order[-1].startswith("m0001-"))
|
|
finally:
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
def test_clearing_the_hand_order_goes_back_to_columns(self):
|
|
from core.magazine import clear_page_order, set_page_order
|
|
layout = clear_page_order(set_page_order(self.layout(), 1, ["r1", "r2", "l1", "l2"]), 1)
|
|
self.assertNotIn("order_locked", layout["pages"][0])
|
|
blocks = [b["id"] for b in sorted(layout["blocks"], key=lambda b: b["reading_order"])]
|
|
self.assertEqual(blocks, ["l1", "l2", "r1", "r2"])
|
|
|
|
def test_route_validates_the_body(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_set_page_order
|
|
for body in ({}, {"order": []}, {"order": "l1"}, {"order": [1, 2]}):
|
|
with self.subTest(body=body), \
|
|
patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=self.layout()):
|
|
request = AsyncMock(); request.json.return_value = body
|
|
with self.assertRaises(HTTPException) as err:
|
|
asyncio.run(magazine_set_page_order('a' * 32, 1, request))
|
|
self.assertEqual(err.exception.status_code, 400)
|
|
|
|
|
|
class CleanUpBoxesTests(unittest.TestCase):
|
|
"""Each rule, and what each rule must leave alone (sized on a real 204-page issue)."""
|
|
|
|
def base(self, blocks, images=()):
|
|
return {"pages": [{"page": 1, "width": 595.0, "height": 842.0}], "blocks": blocks, "images": list(images)}
|
|
|
|
def blk(self, bid, x, y, w, h, text, btype="body", ro=0):
|
|
return {"id": bid, "page": 1, "bbox": [x, y, w, h], "type": btype, "text": text, "reading_order": ro}
|
|
|
|
def types(self, layout):
|
|
return {b["id"]: (b["type"], b.get("region_label", "")) for b in layout["blocks"]}
|
|
|
|
def test_a_short_label_inside_a_picture_is_not_read_but_long_text_there_is(self):
|
|
from core.magazine import clean_up_regions
|
|
photo = {"id": "img", "page": 1, "bbox": [50, 50, 400, 300], "filename": "img.png"}
|
|
layout = self.base([
|
|
self.blk("label", 100, 100, 120, 20, "Apple Intelligence", "title"),
|
|
self.blk("long", 60, 150, 380, 150, "nutzen, ohne dass du ein Prüfverfahren durchlaufen musst. " * 3),
|
|
self.blk("outside", 50, 400, 400, 100, "Running text under the photo."),
|
|
], [photo])
|
|
types = self.types(clean_up_regions(layout)[0])
|
|
self.assertEqual(types["label"], ("furniture", "in_picture"))
|
|
self.assertEqual(types["long"][0], "body") # real text caught in a loose frame stays
|
|
self.assertEqual(types["outside"][0], "body")
|
|
|
|
def test_one_or_two_characters_are_noise_but_a_phone_number_is_not(self):
|
|
from core.magazine import clean_up_regions
|
|
layout = self.base([
|
|
self.blk("one", 10, 10, 10, 10, "1", "title"),
|
|
self.blk("j3", 30, 10, 20, 10, "J 3", "title"),
|
|
self.blk("hash", 60, 10, 10, 10, "#"),
|
|
self.blk("phone", 80, 10, 90, 10, "0511/592 99 077", "caption"),
|
|
self.blk("cite", 200, 10, 90, 10, "c't 17/2025, S. 136", "caption"),
|
|
])
|
|
types = self.types(clean_up_regions(layout)[0])
|
|
for bid in ("one", "j3", "hash"):
|
|
self.assertEqual(types[bid], ("furniture", "ocr_noise"), bid)
|
|
self.assertEqual(types["phone"][0], "caption")
|
|
self.assertEqual(types["cite"][0], "caption")
|
|
|
|
def test_a_subheading_the_body_box_also_read_is_a_duplicate(self):
|
|
from core.magazine import clean_up_regions
|
|
layout = self.base([
|
|
self.blk("body", 50, 100, 300, 400, "Erster Absatz endet hier. Fazit Am Ende bleibt festzuhalten, dass alles gut ist."),
|
|
self.blk("sub", 50, 200, 60, 14, "Fazit", "heading"),
|
|
])
|
|
self.assertEqual(self.types(clean_up_regions(layout)[0])["sub"], ("furniture", "duplicate"))
|
|
|
|
def test_an_overlapping_box_with_different_text_stays(self):
|
|
"""The large majority of overlaps on the measured issue: hiding one would lose text."""
|
|
from core.magazine import clean_up_regions
|
|
layout = self.base([
|
|
self.blk("big", 50, 100, 300, 400, "One paragraph about something."),
|
|
self.blk("small", 50, 300, 300, 80, "A different paragraph entirely."),
|
|
])
|
|
self.assertEqual(self.types(clean_up_regions(layout)[0])["small"][0], "body")
|
|
|
|
def test_a_runon_author_line_is_split_into_name_and_its_own_paragraph(self):
|
|
from core.magazine import clean_up_regions
|
|
layout = self.base([
|
|
self.blk("by", 29, 223, 200, 30, "Von Andreas Weck Als Microsoft-CTO Kevin Scott vor einem Jahr", "byline", 0),
|
|
self.blk("left", 29, 260, 200, 200, "sagte, dass KI bis 2030 rund 95 Prozent des Codes schreiben wird.", "body", 1),
|
|
self.blk("right", 240, 223, 200, 200, "ihnen gebraucht werde, um Programmieraufgaben zu erledigen.", "body", 2),
|
|
])
|
|
out, report = clean_up_regions(layout)
|
|
by = next(b for b in out["blocks"] if b["id"] == "by")
|
|
self.assertEqual(by["text"], "Von Andreas Weck")
|
|
self.assertIn("Als Microsoft-CTO", by["original_text"])
|
|
new_id = report["byline_split"][0]["new_id"]
|
|
order = [b["id"] for b in sorted(out["blocks"], key=lambda b: b["reading_order"])]
|
|
# The sentence is its own paragraph, right after the name and before the left column's next
|
|
# paragraph -- never glued onto the other column.
|
|
self.assertEqual(order[:3], ["by", new_id, "left"])
|
|
self.assertTrue(next(b for b in out["blocks"] if b["id"] == "right")["text"].startswith("ihnen"))
|
|
|
|
def test_only_the_name_is_kept_when_the_sentence_is_already_on_the_page(self):
|
|
from core.magazine import clean_up_regions
|
|
layout = self.base([
|
|
self.blk("by", 29, 223, 200, 30, "Von Robin Ahrens Kurz vor dem Abendessen sind die Kinder", "byline", 0),
|
|
self.blk("body", 29, 260, 200, 200, "Kurz vor dem Abendessen sind die Kinder gereizt und müde.", "body", 1),
|
|
])
|
|
out, report = clean_up_regions(layout)
|
|
self.assertIsNone(report["byline_split"][0]["new_id"])
|
|
self.assertEqual(len(out["blocks"]), 2)
|
|
|
|
def test_a_plain_two_author_byline_and_a_sentence_starting_with_von_are_untouched(self):
|
|
from core.magazine import clean_up_regions
|
|
layout = self.base([
|
|
self.blk("two", 29, 100, 200, 14, "Von Sabrina Patsch und Morton Swimmer", "byline"),
|
|
self.blk("sentence", 29, 200, 200, 80, "Von Familienmitgliedern vergangener Generationen bleibt oft nicht viel."),
|
|
])
|
|
out, report = clean_up_regions(layout)
|
|
self.assertEqual(report["byline_split"], [])
|
|
self.assertEqual({b["id"]: b["text"] for b in out["blocks"]}["two"], "Von Sabrina Patsch und Morton Swimmer")
|
|
|
|
def test_running_it_twice_changes_nothing_the_second_time(self):
|
|
from core.magazine import clean_up_regions
|
|
photo = {"id": "img", "page": 1, "bbox": [50, 50, 400, 300], "filename": "img.png"}
|
|
layout = self.base([self.blk("label", 100, 100, 120, 20, "Label", "title"), self.blk("n", 10, 10, 5, 5, "V"),
|
|
self.blk("by", 29, 400, 200, 30, "Von Jo Bager KI gilt als Antwort", "byline")], [photo])
|
|
once, _ = clean_up_regions(layout)
|
|
_, report = clean_up_regions(once)
|
|
self.assertEqual({k: len(v) for k, v in report.items()}, {"in_picture": 0, "ocr_noise": 0, "duplicate": 0, "byline_split": 0})
|
|
|
|
def test_route_keeps_a_backup_and_adds_the_new_paragraph_to_the_article(self):
|
|
import asyncio
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_clean_up_boxes
|
|
d = Path(tempfile.mkdtemp(prefix="ttvc-cleanup-"))
|
|
layout = self.base([self.blk("by", 29, 223, 200, 30, "Von Jo Bager KI gilt als Antwort", "byline", 0),
|
|
self.blk("body", 29, 260, 200, 200, "auf den Fachkräftemangel.", "body", 1)])
|
|
articles = [{"id": "a1", "blocks": ["by", "body"], "images": []}]
|
|
saved = {}
|
|
try:
|
|
with patch('routes.magazine._require_magazine', return_value=(d, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine.mag.load_articles', return_value=articles), \
|
|
patch('routes.magazine.mag.save_articles', side_effect=lambda _d, a: saved.setdefault('a', a)), \
|
|
patch('routes.magazine.asyncio.to_thread', new=_run_sync):
|
|
out = asyncio.run(magazine_clean_up_boxes('a' * 32))
|
|
self.assertTrue((d / out["backup"]).is_file())
|
|
self.assertEqual(out["counts"]["byline_split"], 1)
|
|
new_id = out["report"]["byline_split"][0]["new_id"]
|
|
self.assertEqual(saved["a"][0]["blocks"], ["by", new_id, "body"])
|
|
finally:
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
|
|
class ArticleAsidesTests(unittest.TestCase):
|
|
"""Sidebars are shown beside an article's text but are never part of what is narrated."""
|
|
|
|
def layout(self):
|
|
def blk(i, page, typ, text, order):
|
|
return {"id": i, "page": page, "bbox": [0, 0, 10, 10], "type": typ, "text": text, "reading_order": order}
|
|
return {"pages": [{"page": 5}, {"page": 6}, {"page": 7}], "images": [], "blocks": [
|
|
blk("t", 5, "title", "Headline", 0), blk("p1", 5, "body", "First paragraph.", 1),
|
|
blk("s1", 5, "sidebar", "Box: did you know?", 2), blk("p2", 5, "body", "Second paragraph.", 3),
|
|
blk("s2", 6, "sidebar", "Glossary (continued on page 9)", 0),
|
|
blk("ad", 7, "sidebar", "Shared page box", 0),
|
|
]}
|
|
|
|
def article(self, aid="a1", pages=(5, 6), blocks=("t", "p1", "p2")):
|
|
return {"id": aid, "pages": list(pages), "blocks": list(blocks)}
|
|
|
|
def test_sidebars_on_the_articles_own_pages_follow_the_right_block(self):
|
|
from core.magazine import article_asides
|
|
a = self.article()
|
|
out = article_asides(self.layout(), a, [a])
|
|
self.assertEqual([(x["id"], x["after"]) for x in out], [("s1", "p1"), ("s2", None)])
|
|
self.assertEqual(out[1]["text"], "Glossary") # the jump cue is print furniture, not content
|
|
|
|
def test_a_page_shared_with_another_article_shows_no_sidebar(self):
|
|
from core.magazine import article_asides
|
|
a, b = self.article(), self.article("a2", pages=(5,), blocks=())
|
|
self.assertEqual([x["id"] for x in article_asides(self.layout(), a, [a, b])], ["s2"])
|
|
|
|
def test_a_sidebar_the_reader_ticked_in_is_narrated_not_an_aside(self):
|
|
from core.magazine import article_asides
|
|
a = self.article(blocks=("t", "p1", "s1", "p2"))
|
|
self.assertEqual([x["id"] for x in article_asides(self.layout(), a, [a])], ["s2"])
|
|
|
|
def test_advertisement_pages_and_empty_sidebars_are_skipped(self):
|
|
from core.magazine import article_asides
|
|
layout = self.layout()
|
|
layout["pages"][1]["advertisement"] = True
|
|
a = self.article()
|
|
self.assertEqual([x["id"] for x in article_asides(layout, a, [a])], ["s1"])
|
|
layout = self.layout()
|
|
layout["blocks"][2]["text"] = " "
|
|
self.assertEqual([x["id"] for x in article_asides(layout, a, [a])], ["s2"])
|
|
|
|
|
|
class QuickCorrectionTests(unittest.TestCase):
|
|
"""Edit text, merge with the next region, split a region -- from the review list."""
|
|
|
|
def layout(self):
|
|
return {"pages": [{"page": 1, "width": 595.0, "height": 842.0}], "images": [], "blocks": [
|
|
{"id": "a", "page": 1, "bbox": [50, 100, 200, 40], "type": "body", "text": "Der Anfang eines Absat-", "reading_order": 0},
|
|
{"id": "b", "page": 1, "bbox": [50, 140, 200, 40], "type": "body", "text": "zes geht hier weiter.", "reading_order": 1},
|
|
{"id": "c", "page": 1, "bbox": [50, 200, 200, 80], "type": "body", "text": "Erster Satz. Zweiter Absatz beginnt.", "reading_order": 2},
|
|
]}
|
|
|
|
def order(self, layout):
|
|
return [b["id"] for b in sorted(layout["blocks"], key=lambda b: b["reading_order"])]
|
|
|
|
def test_typed_text_replaces_the_ocr_text(self):
|
|
from core.magazine import set_region_text
|
|
out = set_region_text(self.layout(), 1, "c", " Corrected by hand. ")
|
|
self.assertEqual(next(b for b in out["blocks"] if b["id"] == "c")["text"], "Corrected by hand.")
|
|
with self.assertRaises(ValueError):
|
|
set_region_text(self.layout(), 1, "c", " ")
|
|
|
|
def get(self, layout, bid):
|
|
return next(b for b in layout["blocks"] if b["id"] == bid)
|
|
|
|
def test_the_first_edit_keeps_the_recognised_text_and_later_edits_do_not_overwrite_it(self):
|
|
from core.magazine import set_region_text
|
|
once = set_region_text(self.layout(), 1, "c", "Erster Satz. Zweiter Absatz.")
|
|
self.assertEqual(self.get(once, "c")["ocr_text"], "Erster Satz. Zweiter Absatz beginnt.")
|
|
twice = set_region_text(once, 1, "c", "Ganz anders.")
|
|
self.assertEqual(self.get(twice, "c")["ocr_text"], "Erster Satz. Zweiter Absatz beginnt.")
|
|
|
|
def test_typing_the_original_back_clears_the_edit_mark(self):
|
|
from core.magazine import set_region_text
|
|
edited = set_region_text(self.layout(), 1, "c", "Changed.")
|
|
back = set_region_text(edited, 1, "c", "Erster Satz. Zweiter Absatz beginnt.")
|
|
self.assertNotIn("ocr_text", self.get(back, "c"))
|
|
|
|
def test_saving_unchanged_text_marks_nothing(self):
|
|
from core.magazine import set_region_text
|
|
same = set_region_text(self.layout(), 1, "c", "Erster Satz. Zweiter Absatz beginnt.")
|
|
self.assertNotIn("ocr_text", self.get(same, "c"))
|
|
|
|
def test_other_regions_and_the_input_layout_are_untouched(self):
|
|
from core.magazine import set_region_text
|
|
layout = self.layout()
|
|
out = set_region_text(layout, 1, "c", "Changed.")
|
|
self.assertNotIn("ocr_text", self.get(out, "a"))
|
|
self.assertNotIn("ocr_text", self.get(layout, "c"))
|
|
|
|
def test_ocr_text_survives_a_block_round_trip(self):
|
|
from core.magazine import Block
|
|
b = Block(id="x", page=1, bbox=[0, 0, 1, 1], type="body", text="t", ocr_text="old")
|
|
self.assertEqual(b.to_json()["ocr_text"], "old")
|
|
self.assertNotIn("ocr_text", Block(id="y", page=1, bbox=[0, 0, 1, 1], type="body", text="t").to_json())
|
|
|
|
def test_merging_closes_a_hyphenated_word_and_covers_both_boxes(self):
|
|
from core.magazine import merge_with_next
|
|
out, removed = merge_with_next(self.layout(), 1, "a")
|
|
merged = next(b for b in out["blocks"] if b["id"] == "a")
|
|
self.assertEqual(removed, "b")
|
|
self.assertEqual(merged["text"], "Der Anfang eines Absatzes geht hier weiter.")
|
|
self.assertEqual(merged["bbox"], [50, 100, 200, 80])
|
|
self.assertNotIn("b", [x["id"] for x in out["blocks"]])
|
|
|
|
def test_the_last_region_has_nothing_to_merge_with(self):
|
|
from core.magazine import merge_with_next
|
|
with self.assertRaises(ValueError):
|
|
merge_with_next(self.layout(), 1, "c")
|
|
|
|
def test_splitting_makes_a_second_region_read_right_after_the_first(self):
|
|
from core.magazine import split_region
|
|
out, new_id = split_region(self.layout(), 1, "c", len("Erster Satz."))
|
|
texts = {b["id"]: b["text"] for b in out["blocks"]}
|
|
self.assertEqual((texts["c"], texts[new_id]), ("Erster Satz.", "Zweiter Absatz beginnt."))
|
|
self.assertEqual(self.order(out), ["a", "b", "c", new_id])
|
|
first, second = (next(b for b in out["blocks"] if b["id"] == i)["bbox"] for i in ("c", new_id))
|
|
self.assertAlmostEqual(first[1] + first[3], second[1]) # stacked, no gap or overlap
|
|
|
|
def test_a_split_at_the_edge_is_rejected(self):
|
|
from core.magazine import split_region
|
|
for at in (0, 999):
|
|
with self.assertRaises(ValueError):
|
|
split_region(self.layout(), 1, "c", at)
|
|
|
|
def test_merge_route_drops_the_folded_region_from_articles(self):
|
|
import asyncio
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_merge_region
|
|
articles = [{"id": "x", "blocks": ["a", "b", "c"], "images": []}]
|
|
saved = {}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=self.layout()), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine.mag.load_articles', return_value=articles), \
|
|
patch('routes.magazine.mag.save_articles', side_effect=lambda _d, a: saved.setdefault('a', a)):
|
|
asyncio.run(magazine_merge_region('a' * 32, 1, 'a'))
|
|
self.assertEqual(saved['a'][0]['blocks'], ['a', 'c'])
|
|
|
|
def test_split_route_adds_the_new_part_after_the_original_in_articles(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_split_region
|
|
articles = [{"id": "x", "blocks": ["a", "b", "c"], "images": []}]
|
|
saved = {}
|
|
with patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=self.layout()), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine.mag.load_articles', return_value=articles), \
|
|
patch('routes.magazine.mag.save_articles', side_effect=lambda _d, a: saved.setdefault('a', a)):
|
|
request = AsyncMock(); request.json.return_value = {"at": 12}
|
|
out = asyncio.run(magazine_split_region('a' * 32, 1, 'c', request))
|
|
self.assertEqual(saved['a'][0]['blocks'], ['a', 'b', 'c', out['new_id']])
|
|
|
|
def test_split_route_rejects_a_bad_position(self):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
from routes.magazine import magazine_split_region
|
|
for body in ({}, {"at": "3"}, {"at": 0}, {"at": True}):
|
|
with self.subTest(body=body), \
|
|
patch('routes.magazine._require_magazine', return_value=(None, {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=self.layout()):
|
|
request = AsyncMock(); request.json.return_value = body
|
|
with self.assertRaises(HTTPException):
|
|
asyncio.run(magazine_split_region('a' * 32, 1, 'c', request))
|
|
|
|
|
|
class BulkUndoFoldTests(unittest.TestCase):
|
|
"""Folding labels into a new picture, bulk changes, and one-level undo."""
|
|
|
|
def setUp(self):
|
|
from pathlib import Path
|
|
from PIL import Image
|
|
self.dir = Path(tempfile.mkdtemp(prefix="ttvc-bulk-"))
|
|
(self.dir / "pages").mkdir()
|
|
Image.new("RGB", (600, 800), "white").save(self.dir / "pages" / "0001.png")
|
|
self.layout = {"pages": [{"page": 1, "width": 600.0, "height": 800.0}], "images": [], "blocks": [
|
|
{"id": "chart", "page": 1, "bbox": [50, 300, 400, 300], "type": "body", "text": "Chart OCR text", "reading_order": 0},
|
|
{"id": "lg1", "page": 1, "bbox": [60, 320, 80, 12], "type": "caption", "text": "Einsteiger (22-25 Jahre)", "reading_order": 1},
|
|
{"id": "lg2", "page": 1, "bbox": [160, 320, 80, 12], "type": "caption", "text": "Senior (50 Jahre und älter)", "reading_order": 2},
|
|
{"id": "body", "page": 1, "bbox": [50, 50, 400, 200], "type": "body", "text": "Running text above the chart.", "reading_order": 3},
|
|
]}
|
|
self.articles = [{"id": "a1", "blocks": ["chart", "lg1", "lg2", "body"], "images": []}]
|
|
from core import magazine as mag
|
|
mag.save_layout(self.dir, self.layout)
|
|
mag.save_articles(self.dir, self.articles)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
def run_route(self, fn, *args, body=None):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
with patch('routes.magazine._require_magazine', return_value=(self.dir, {})), \
|
|
patch('routes.magazine._load_settings', return_value={}), \
|
|
patch('routes.magazine.asyncio.to_thread', new=_run_sync):
|
|
request = AsyncMock(); request.json.return_value = body or {}
|
|
return asyncio.run(fn('a' * 32, *args, request) if body is not None else fn('a' * 32, *args))
|
|
|
|
def test_labels_inside_a_region_turned_into_a_picture_fold_into_it(self):
|
|
from core.magazine import change_region_type
|
|
layout, _ = change_region_type(self.dir, self.layout, 1, "chart", "image")
|
|
types = {b["id"]: (b["type"], b.get("region_label")) for b in layout["blocks"]}
|
|
self.assertEqual(types["lg1"], ("furniture", "in_picture"))
|
|
self.assertEqual(types["lg2"], ("furniture", "in_picture"))
|
|
self.assertEqual(types["body"][0], "body") # outside the picture: untouched
|
|
|
|
def test_a_drawn_picture_folds_the_labels_inside_it_too(self):
|
|
from core.magazine import add_manual_region
|
|
layout = add_manual_region(self.dir, self.layout, 1, [50, 300, 400, 300], "image")
|
|
self.assertEqual(next(b for b in layout["blocks"] if b["id"] == "lg1")["region_label"], "in_picture")
|
|
|
|
def test_bulk_delete_removes_regions_and_article_entries_and_can_be_undone(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_bulk_regions, magazine_undo
|
|
out = self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg1", "lg2"], "action": "delete"})
|
|
self.assertEqual(out["count"], 2)
|
|
self.assertEqual({b["id"] for b in mag.load_layout(self.dir)["blocks"]}, {"chart", "body"})
|
|
self.assertEqual(mag.load_articles(self.dir)[0]["blocks"], ["chart", "body"])
|
|
self.run_route(magazine_undo, body={"token": out["undo"]})
|
|
self.assertEqual(len(mag.load_layout(self.dir)["blocks"]), 4)
|
|
self.assertEqual(mag.load_articles(self.dir)[0]["blocks"], ["chart", "lg1", "lg2", "body"])
|
|
|
|
def test_bulk_type_change_applies_to_every_selected_region(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_bulk_regions
|
|
self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg1", "lg2"], "action": "type", "type": "ocr_noise"})
|
|
labels = {b["id"]: b.get("region_label") for b in mag.load_layout(self.dir)["blocks"]}
|
|
self.assertEqual((labels["lg1"], labels["lg2"]), ("ocr_noise", "ocr_noise"))
|
|
|
|
def test_bulk_turning_text_into_a_picture_moves_it_to_the_article_images(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_bulk_regions
|
|
out = self.run_route(magazine_bulk_regions, 1, body={"ids": ["chart"], "action": "type", "type": "image"})
|
|
new_id = out["remap"]["chart"]
|
|
article = mag.load_articles(self.dir)[0]
|
|
self.assertNotIn("chart", article["blocks"])
|
|
self.assertEqual(article["images"], [new_id])
|
|
|
|
def test_only_the_latest_change_can_be_undone(self):
|
|
from routes.magazine import magazine_bulk_regions, magazine_undo
|
|
first = self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg1"], "action": "delete"})
|
|
self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg2"], "action": "delete"})
|
|
with self.assertRaises(HTTPException) as err:
|
|
self.run_route(magazine_undo, body={"token": first["undo"]})
|
|
self.assertEqual(err.exception.status_code, 409)
|
|
|
|
def test_bulk_rejects_bad_requests_before_changing_anything(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_bulk_regions
|
|
for body in ({"ids": [], "action": "delete"}, {"ids": ["lg1"], "action": "nuke"},
|
|
{"ids": ["lg1"], "action": "type", "type": "nonsense"}, {"ids": ["elsewhere"], "action": "delete"}):
|
|
with self.subTest(body=body):
|
|
with self.assertRaises(HTTPException) as err:
|
|
self.run_route(magazine_bulk_regions, 1, body=body)
|
|
self.assertEqual(err.exception.status_code, 400)
|
|
self.assertEqual(len(mag.load_layout(self.dir)["blocks"]), 4)
|
|
self.assertEqual(_history(self.dir)["undo"], [])
|
|
|
|
|
|
class UndoSafetyAndExtraFieldTests(BulkUndoFoldTests):
|
|
"""Undo never rolls back a later change; a split author line's extra field survives every path."""
|
|
|
|
def test_undo_is_refused_once_anything_else_changed(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_bulk_regions, magazine_undo
|
|
out = self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg1"], "action": "delete"})
|
|
# A change that records no undo point of its own, e.g. drawing a region or fixing the order.
|
|
layout = mag.load_layout(self.dir)
|
|
layout["blocks"][0]["text"] = "edited later"
|
|
mag.save_layout(self.dir, layout)
|
|
with self.assertRaises(HTTPException) as err:
|
|
self.run_route(magazine_undo, body={"token": out["undo"]})
|
|
self.assertEqual(err.exception.status_code, 409)
|
|
self.assertEqual(mag.load_layout(self.dir)["blocks"][0]["text"], "edited later")
|
|
|
|
def test_an_unsealed_snapshot_is_refused(self):
|
|
from routes.magazine import magazine_undo, _snapshot_for_undo
|
|
_snapshot_for_undo(self.dir, "x") # taken but never sealed: its aftermath is unknown
|
|
with self.assertRaises(HTTPException) as err:
|
|
self.run_route(magazine_undo, body={})
|
|
self.assertEqual(err.exception.status_code, 409)
|
|
|
|
def _ids(self):
|
|
from core import magazine as mag
|
|
return sorted(b["id"] for b in mag.load_layout(self.dir)["blocks"])
|
|
|
|
def test_undo_then_redo_puts_the_change_back_and_the_history_says_what_is_possible(self):
|
|
from routes.magazine import magazine_bulk_regions, magazine_history, magazine_redo, magazine_undo
|
|
before = self._ids()
|
|
self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg1"], "action": "delete"})
|
|
after = self._ids()
|
|
state = self.run_route(magazine_history)
|
|
self.assertEqual((state["can_undo"], state["can_redo"]), (True, False))
|
|
self.assertIn("Deleted", state["undo_label"])
|
|
undone = self.run_route(magazine_undo, body={}) # no token: the latest change
|
|
self.assertEqual((self._ids(), undone["can_undo"], undone["can_redo"]), (before, False, True))
|
|
redone = self.run_route(magazine_redo)
|
|
self.assertEqual((self._ids(), redone["can_undo"], redone["can_redo"]), (after, True, False))
|
|
self.run_route(magazine_undo, body={})
|
|
self.assertEqual(self._ids(), before) # and it can be undone again
|
|
|
|
def test_several_changes_are_undone_one_by_one_and_redone_in_order(self):
|
|
from routes.magazine import magazine_bulk_regions, magazine_redo, magazine_undo
|
|
start = self._ids()
|
|
self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg1"], "action": "delete"})
|
|
one = self._ids()
|
|
self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg2"], "action": "delete"})
|
|
two = self._ids()
|
|
self.run_route(magazine_undo, body={}); self.assertEqual(self._ids(), one)
|
|
self.run_route(magazine_undo, body={}); self.assertEqual(self._ids(), start)
|
|
self.run_route(magazine_redo); self.assertEqual(self._ids(), one)
|
|
self.run_route(magazine_redo); self.assertEqual(self._ids(), two)
|
|
|
|
def test_a_new_change_ends_the_redo_history(self):
|
|
from routes.magazine import magazine_bulk_regions, magazine_history, magazine_redo, magazine_undo
|
|
self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg1"], "action": "delete"})
|
|
self.run_route(magazine_undo, body={})
|
|
self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg2"], "action": "delete"})
|
|
self.assertFalse(self.run_route(magazine_history)["can_redo"])
|
|
with self.assertRaises(HTTPException) as err:
|
|
self.run_route(magazine_redo)
|
|
self.assertEqual(err.exception.status_code, 409)
|
|
|
|
def test_redo_is_refused_once_anything_else_changed(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_bulk_regions, magazine_history, magazine_redo, magazine_undo
|
|
self.run_route(magazine_bulk_regions, 1, body={"ids": ["lg1"], "action": "delete"})
|
|
self.run_route(magazine_undo, body={})
|
|
layout = mag.load_layout(self.dir)
|
|
layout["blocks"][0]["text"] = "edited later"
|
|
mag.save_layout(self.dir, layout)
|
|
self.assertFalse(self.run_route(magazine_history)["can_redo"])
|
|
with self.assertRaises(HTTPException) as err:
|
|
self.run_route(magazine_redo)
|
|
self.assertEqual(err.exception.status_code, 409)
|
|
self.assertEqual(mag.load_layout(self.dir)["blocks"][0]["text"], "edited later")
|
|
|
|
def test_only_the_latest_twenty_changes_are_kept(self):
|
|
from routes.magazine import _UNDO_DEPTH, _history, _seal_undo, _snapshot_for_undo
|
|
for i in range(_UNDO_DEPTH + 5):
|
|
_seal_undo(self.dir, _snapshot_for_undo(self.dir, f"change {i}"))
|
|
kept = _history(self.dir)["undo"]
|
|
self.assertEqual((len(kept), kept[0]["label"], kept[-1]["label"]), (_UNDO_DEPTH, "change 5", f"change {_UNDO_DEPTH + 4}"))
|
|
self.assertEqual(len(list((self.dir / "undo").glob("*.json.gz"))), _UNDO_DEPTH)
|
|
|
|
def test_nothing_to_undo_or_redo_is_a_plain_409(self):
|
|
from routes.magazine import magazine_history, magazine_redo, magazine_undo
|
|
for fn, kw in ((magazine_undo, {"body": {}}), (magazine_redo, {})):
|
|
with self.assertRaises(HTTPException) as err:
|
|
self.run_route(fn, **kw)
|
|
self.assertEqual(err.exception.status_code, 409)
|
|
self.assertEqual(self.run_route(magazine_history)["can_undo"], False)
|
|
|
|
def test_drawing_a_region_on_a_layout_with_a_split_author_line_works(self):
|
|
""""Could not add region: Block.__init__() got an unexpected keyword argument 'original_text'"."""
|
|
from core.magazine import add_manual_region, clean_up_regions
|
|
self.layout["blocks"].append({"id": "by", "page": 1, "bbox": [50, 20, 300, 30], "type": "byline",
|
|
"text": "Von Jo Bager KI gilt als Antwort", "reading_order": 9})
|
|
cleaned, report = clean_up_regions(self.layout)
|
|
self.assertEqual(len(report["byline_split"]), 1)
|
|
after = add_manual_region(self.dir, cleaned, 1, [400, 700, 100, 40], "text", "New text")
|
|
byline = next(b for b in after["blocks"] if b["id"] == "by")
|
|
self.assertIn("KI gilt als Antwort", byline["original_text"])
|
|
self.assertTrue(any(b["text"] == "New text" for b in after["blocks"]))
|
|
|
|
def test_blocks_without_the_field_do_not_gain_it(self):
|
|
from core.magazine import Block
|
|
self.assertNotIn("original_text", Block(id="x", page=1, bbox=[0, 0, 1, 1], type="body", text="t").to_json())
|
|
|
|
|
|
class LivePreviewBoxTests(unittest.TestCase):
|
|
"""The ingest preview carries the page's boxes, so they can be drawn as they are made."""
|
|
|
|
def test_boxes_are_in_percent_of_the_page_and_numbered_in_reading_order(self):
|
|
from core.magazine import Block
|
|
from routes.magazine import _ingest_preview_boxes
|
|
blocks = [Block(id="b", page=1, bbox=[300, 400, 150, 100], type="body", text="t", reading_order=1),
|
|
Block(id="a", page=1, bbox=[0, 0, 600, 80], type="title", text="t", reading_order=0),
|
|
Block(id="n", page=1, bbox=[280, 780, 40, 10], type="furniture", text="12", reading_order=2,
|
|
region_label="page_number")]
|
|
boxes = _ingest_preview_boxes(blocks, [{"bbox": [60, 100, 240, 300]}], (600, 800))
|
|
self.assertEqual([(b["type"], b["n"]) for b in boxes], [("title", 1), ("body", 2), ("page_number", 3), ("image", None)])
|
|
self.assertEqual({k: boxes[1][k] for k in ("x", "y", "w", "h")}, {"x": 50.0, "y": 50.0, "w": 25.0, "h": 12.5})
|
|
|
|
def test_the_page_scan_is_not_drawn_as_a_picture(self):
|
|
from routes.magazine import _ingest_preview_boxes
|
|
figure = {"id": "f", "bbox": [50, 400, 300, 200], "type": "figure", "reading_order": 0}
|
|
scan = {"bbox": [0, 0, 600, 800]}
|
|
boxes = _ingest_preview_boxes([figure], [scan], (600, 800))
|
|
self.assertEqual([b["type"] for b in boxes], ["figure"])
|
|
|
|
def test_a_crop_of_a_drawn_region_is_not_drawn_twice(self):
|
|
from routes.magazine import _ingest_preview_boxes
|
|
chart = {"id": "c", "bbox": [50, 400, 300, 200], "type": "chart", "reading_order": 0}
|
|
crop = {"bbox": [50, 400, 300, 200], "source_block": "c", "type": "chart"}
|
|
other = {"bbox": [400, 50, 100, 100]}
|
|
boxes = _ingest_preview_boxes([chart], [crop, other], (600, 800))
|
|
self.assertEqual([b["type"] for b in boxes], ["chart", "image"])
|
|
|
|
def test_dict_blocks_from_the_labelling_pass_work_too(self):
|
|
from routes.magazine import _ingest_preview_boxes
|
|
boxes = _ingest_preview_boxes([{"bbox": [0, 0, 100, 100], "type": "ad", "reading_order": 0}], [], (200, 200))
|
|
self.assertEqual(boxes[0]["type"], "ad")
|
|
|
|
def test_no_page_size_means_no_boxes_rather_than_an_error(self):
|
|
from routes.magazine import _ingest_preview_boxes
|
|
self.assertEqual(_ingest_preview_boxes([{"bbox": [0, 0, 1, 1], "type": "body"}], [], None), [])
|
|
|
|
def test_the_preview_file_holds_page_stage_text_and_boxes(self):
|
|
import json
|
|
from pathlib import Path
|
|
from routes.magazine import _write_ingest_preview
|
|
d = Path(tempfile.mkdtemp(prefix="ttvc-preview-"))
|
|
try:
|
|
_write_ingest_preview(d, 7, [{"bbox": [0, 0, 50, 50], "type": "body", "text": "Hallo", "reading_order": 0}],
|
|
[], (100, 100), "labelling")
|
|
data = json.loads((d / "ingest_preview.json").read_text("utf-8"))
|
|
self.assertEqual((data["page"], data["stage"], len(data["boxes"])), (7, "labelling", 1))
|
|
self.assertIn("Hallo", data["text"])
|
|
finally:
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
def test_the_preview_file_drops_the_page_scan_and_can_show_a_real_thumbnail(self):
|
|
import json
|
|
from pathlib import Path
|
|
from routes.magazine import _write_ingest_preview
|
|
d = Path(tempfile.mkdtemp(prefix="ttvc-preview-"))
|
|
try:
|
|
_write_ingest_preview(d, 3, [], [
|
|
{"bbox": [0, 0, 100, 100]}, # the page scan -- must not appear
|
|
{"bbox": [10, 10, 40, 40], "type": "figure", "filename": "p0003-region-b1.png"},
|
|
], (100, 100), "labelling")
|
|
data = json.loads((d / "ingest_preview.json").read_text("utf-8"))
|
|
self.assertEqual(len(data["boxes"]), 1)
|
|
self.assertIn(f"/api/magazine/{d.name}/images/p0003-region-b1.png", data["text"])
|
|
finally:
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
def test_the_labelling_pass_reports_every_page_even_when_its_model_call_fails(self):
|
|
from unittest import mock
|
|
import routes.magazine as rm
|
|
layout = {"pages": [{"page": 1, "width": 1, "height": 1}, {"page": 2, "width": 1, "height": 1}],
|
|
"blocks": [{"id": "a", "page": 1, "bbox": [0, 0, 1, 1], "type": "body", "text": "x"},
|
|
{"id": "b", "page": 2, "bbox": [0, 0, 1, 1], "type": "body", "text": "y"}], "images": []}
|
|
seen = []
|
|
with mock.patch("routes.conversation._post_llm_chat_completion", side_effect=RuntimeError("down")):
|
|
rm._recognise_advertisements(layout, {"llm_model": "m", "llm_url": "http://x"}, on_page=seen.append)
|
|
self.assertEqual(seen, [1, 2])
|
|
|
|
|
|
class IngestSyncVisualRefreshTests(unittest.TestCase):
|
|
"""Pictures, charts and tables are snapped and cropped to real files per page, as each
|
|
page is labelled -- not only once at the very end -- so the live preview is accurate
|
|
and can show an actual thumbnail while the import is still running."""
|
|
|
|
def test_refresh_is_called_per_page_during_semantic_labelling(self):
|
|
from unittest.mock import patch, call
|
|
import routes.magazine as rm
|
|
with patch.object(rm, "_LIB", Path(tempfile.mkdtemp(prefix="ttvc-lib-"))) as lib:
|
|
d = lib / "doc1"
|
|
d.mkdir(parents=True)
|
|
(d / "source.pdf").write_bytes(b"x")
|
|
layout = {"pages": [{"page": 1, "width": 100, "height": 100},
|
|
{"page": 2, "width": 100, "height": 100}], "blocks": [], "images": []}
|
|
|
|
def semantic(_layout, _settings, page=None, doc_dir=None, on_page=None):
|
|
on_page(1)
|
|
on_page(2)
|
|
|
|
with patch.object(rm.mag, "extract_pages", return_value=layout), \
|
|
patch.object(rm, "_recognise_advertisements"), \
|
|
patch.object(rm, "_recognise_semantic_layout", side_effect=semantic), \
|
|
patch.object(rm.mag, "refresh_visual_regions") as refresh, \
|
|
patch.object(rm.mag, "clean_up_regions", return_value=(layout, {})), \
|
|
patch.object(rm.mag, "save_layout"), \
|
|
patch.object(rm.mag, "build_articles", return_value=[]), \
|
|
patch.object(rm.mag, "save_articles"), \
|
|
patch.object(rm, "_quality_run_sync"), \
|
|
patch.object(rm, "background_job_update"), \
|
|
patch.object(rm, "_write_meta"):
|
|
rm._ingest_sync("doc1", "job1", "eng", 150, None, None)
|
|
# Once per page during labelling, plus the final whole-document safety-net call.
|
|
self.assertEqual(refresh.call_args_list,
|
|
[call(d, layout, 1), call(d, layout, 2), call(d, layout)])
|
|
shutil.rmtree(lib, ignore_errors=True)
|
|
|
|
def test_a_refresh_failure_on_one_page_does_not_abort_ingestion(self):
|
|
from unittest.mock import patch, call
|
|
import routes.magazine as rm
|
|
with patch.object(rm, "_LIB", Path(tempfile.mkdtemp(prefix="ttvc-lib-"))) as lib:
|
|
d = lib / "doc1"
|
|
d.mkdir(parents=True)
|
|
(d / "source.pdf").write_bytes(b"x")
|
|
layout = {"pages": [{"page": 1, "width": 100, "height": 100}], "blocks": [], "images": []}
|
|
|
|
def semantic(_layout, _settings, page=None, doc_dir=None, on_page=None):
|
|
on_page(1)
|
|
|
|
written = {}
|
|
with patch.object(rm.mag, "extract_pages", return_value=layout), \
|
|
patch.object(rm, "_recognise_advertisements"), \
|
|
patch.object(rm, "_recognise_semantic_layout", side_effect=semantic), \
|
|
patch.object(rm.mag, "refresh_visual_regions", side_effect=RuntimeError("boom")), \
|
|
patch.object(rm.mag, "clean_up_regions", return_value=(layout, {})), \
|
|
patch.object(rm.mag, "save_layout") as save_layout, \
|
|
patch.object(rm.mag, "build_articles", return_value=[]), \
|
|
patch.object(rm.mag, "save_articles"), \
|
|
patch.object(rm, "_quality_run_sync"), \
|
|
patch.object(rm, "background_job_update"), \
|
|
patch.object(rm, "_write_meta", side_effect=lambda d, m: written.update(m)):
|
|
rm._ingest_sync("doc1", "job1", "eng", 150, None, None)
|
|
save_layout.assert_called_once()
|
|
self.assertEqual(written["ingest"]["status"], "done") # never surfaced as an ingest failure
|
|
shutil.rmtree(lib, ignore_errors=True)
|
|
|
|
|
|
class IngestSyncDuplicateRecheckTests(unittest.TestCase):
|
|
"""The per-page geometry loop (seam mending, region growth) can merge a wider box out of
|
|
fragments clean_up_regions's one earlier call had no reason to flag yet -- confirmed live
|
|
on c't KI Wissen 2026 page 152, where an infographic caption fragment was left narrated
|
|
twice this way. Confirms clean_up_regions runs again after that loop, not only before it."""
|
|
|
|
def test_clean_up_regions_runs_again_after_the_geometry_loop(self):
|
|
from unittest.mock import patch, call
|
|
import routes.magazine as rm
|
|
with patch.object(rm, "_LIB", Path(tempfile.mkdtemp(prefix="ttvc-lib-"))) as lib:
|
|
d = lib / "doc1"
|
|
d.mkdir(parents=True)
|
|
(d / "source.pdf").write_bytes(b"x")
|
|
layout = {"pages": [{"page": 1, "width": 100, "height": 100}], "blocks": [], "images": []}
|
|
|
|
def semantic(_layout, _settings, page=None, doc_dir=None, on_page=None):
|
|
on_page(1)
|
|
|
|
with patch.object(rm.mag, "extract_pages", return_value=layout), \
|
|
patch.object(rm, "_recognise_advertisements"), \
|
|
patch.object(rm, "_recognise_semantic_layout", side_effect=semantic), \
|
|
patch.object(rm.mag, "refresh_visual_regions"), \
|
|
patch.object(rm.mag, "clean_up_regions", return_value=(layout, {})) as cleanup, \
|
|
patch.object(rm, "_fix_page_geometry", return_value=(layout, {}, False)) as geometry, \
|
|
patch.object(rm.mag, "save_layout"), \
|
|
patch.object(rm.mag, "build_articles", return_value=[]), \
|
|
patch.object(rm.mag, "save_articles"), \
|
|
patch.object(rm, "_quality_run_sync"), \
|
|
patch.object(rm, "background_job_update"), \
|
|
patch.object(rm, "_write_meta"):
|
|
rm._ingest_sync("doc1", "job1", "eng", 150, None, None)
|
|
# Once before the geometry loop, once after it -- geometry itself ran in between.
|
|
self.assertEqual(cleanup.call_args_list, [call(layout), call(layout)])
|
|
geometry.assert_called_once()
|
|
shutil.rmtree(lib, ignore_errors=True)
|
|
|
|
|
|
class IngestSyncAutoQualityRunTests(unittest.TestCase):
|
|
"""A fresh import used to stop once laid out, leaving every page's missed text,
|
|
suspicious OCR and provable misspellings for a separate, manual "Auto-fix all pages"
|
|
click. It now runs that same pass itself, under the same job, so the reviewer opens an
|
|
already-worked magazine."""
|
|
|
|
def _ingest(self, quality_run):
|
|
from unittest.mock import patch, call
|
|
import routes.magazine as rm
|
|
with patch.object(rm, "_LIB", Path(tempfile.mkdtemp(prefix="ttvc-lib-"))) as lib:
|
|
d = lib / "doc1"
|
|
d.mkdir(parents=True)
|
|
(d / "source.pdf").write_bytes(b"x")
|
|
layout = {"pages": [{"page": 1, "width": 100, "height": 100}], "blocks": [], "images": []}
|
|
|
|
def semantic(_layout, _settings, page=None, doc_dir=None, on_page=None):
|
|
on_page(1)
|
|
|
|
written = {}
|
|
with patch.object(rm.mag, "extract_pages", return_value=layout), \
|
|
patch.object(rm, "_recognise_advertisements"), \
|
|
patch.object(rm, "_recognise_semantic_layout", side_effect=semantic), \
|
|
patch.object(rm.mag, "refresh_visual_regions"), \
|
|
patch.object(rm.mag, "clean_up_regions", return_value=(layout, {})), \
|
|
patch.object(rm, "_fix_page_geometry", return_value=(layout, {}, False)), \
|
|
patch.object(rm.mag, "save_layout"), \
|
|
patch.object(rm.mag, "build_articles", return_value=[{"id": "a1"}]), \
|
|
patch.object(rm.mag, "save_articles"), \
|
|
patch.object(rm, "_quality_run_sync", side_effect=quality_run) as qrun, \
|
|
patch.object(rm, "background_job_update"), \
|
|
patch.object(rm, "_write_meta", side_effect=lambda d, m: written.update(m)):
|
|
rm._ingest_sync("doc1", "job1", "eng", 150, None, None)
|
|
shutil.rmtree(lib, ignore_errors=True)
|
|
return written, qrun
|
|
|
|
def test_the_quality_run_is_called_with_the_ingest_jobs_own_id(self):
|
|
written, qrun = self._ingest(lambda doc_id, job_id, settings: None)
|
|
qrun.assert_called_once()
|
|
self.assertEqual(qrun.call_args.args[:2], ("doc1", "job1"))
|
|
|
|
def test_success_is_reported_as_ingest_done(self):
|
|
import routes.magazine as rm
|
|
from unittest.mock import patch
|
|
with patch.object(rm, "background_job_get", return_value={"stage": "done", "done": True}):
|
|
written, _ = self._ingest(lambda doc_id, job_id, settings: None)
|
|
self.assertEqual(written["ingest"]["status"], "done")
|
|
self.assertIn("finished", written["ingest"])
|
|
|
|
def test_a_quality_run_failure_is_reported_without_losing_the_ingest(self):
|
|
"""_quality_run_sync catches its own errors rather than raising -- the ingest itself
|
|
(articles already saved) must not be reported as failed too, but "done" would be a
|
|
lie, so the ingest status becomes "failed" with the Auto-fix pass's own error."""
|
|
import routes.magazine as rm
|
|
from unittest.mock import patch
|
|
with patch.object(rm, "background_job_get",
|
|
return_value={"stage": "failed", "done": True, "error": "vision engine unreachable"}):
|
|
written, _ = self._ingest(lambda doc_id, job_id, settings: None)
|
|
self.assertEqual(written["ingest"]["status"], "failed")
|
|
self.assertIn("vision engine unreachable", written["ingest"]["error"])
|
|
self.assertEqual(written["ingest"]["articles"], 1) # the ingest's own result is kept
|
|
|
|
|
|
class InterruptedIngestTests(unittest.TestCase):
|
|
def test_an_ingest_killed_by_a_restart_is_reported_as_failed_not_running_for_ever(self):
|
|
import asyncio
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_ingest_status
|
|
meta = {"kind": "magazine", "ingest": {"status": "running", "job_id": "j1"}}
|
|
written = {}
|
|
with patch('routes.magazine._require_magazine', return_value=(Path(tempfile.mkdtemp()), meta)), \
|
|
patch('routes.magazine.background_job_get', return_value={"done": True, "stage": "error",
|
|
"error": "Background job interrupted by server restart"}), \
|
|
patch('routes.magazine._write_meta', side_effect=lambda d, m: written.update(m)):
|
|
out = asyncio.run(magazine_ingest_status('a' * 32))
|
|
self.assertEqual(out["ingest"]["status"], "failed")
|
|
self.assertIn("server restart", out["ingest"]["error"])
|
|
self.assertEqual(written["ingest"]["status"], "failed")
|
|
|
|
def test_a_job_still_running_stays_running(self):
|
|
import asyncio
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_ingest_status
|
|
meta = {"kind": "magazine", "ingest": {"status": "running", "job_id": "j1"}}
|
|
with patch('routes.magazine._require_magazine', return_value=(Path(tempfile.mkdtemp()), meta)), \
|
|
patch('routes.magazine.background_job_get', return_value={"done": False, "stage": "extract"}), \
|
|
patch('routes.magazine._write_meta') as write:
|
|
out = asyncio.run(magazine_ingest_status('a' * 32))
|
|
self.assertEqual(out["ingest"]["status"], "running")
|
|
write.assert_not_called()
|
|
|
|
|
|
class QualityRunJobTests(unittest.TestCase):
|
|
def _run(self, settings, audit_row):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _quality_run_sync
|
|
layout = {'pages': [{'page': 1}, {'page': 2}], 'blocks': []}
|
|
lib = Path(tempfile.mkdtemp(prefix='ttvc-qrun-'))
|
|
self.addCleanup(shutil.rmtree, lib, True)
|
|
(lib / ('a' * 32)).mkdir()
|
|
with patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.recover_missing_text', return_value=(layout, 1)), \
|
|
patch('routes.magazine.mag.verify_ocr_recognition', return_value=(layout, 3, 2)) as verify, \
|
|
patch('routes.magazine.mag.audit_page_coverage', side_effect=audit_row), \
|
|
patch('routes.magazine.mag.add_uncovered_pictures', side_effect=lambda d, layout, page: (layout, [])), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine._LIB', lib), \
|
|
patch('routes.magazine.background_job_update') as update:
|
|
_quality_run_sync('a' * 32, 'job', settings)
|
|
done = [c.kwargs for c in update.call_args_list if c.kwargs.get('done')]
|
|
return done[-1], verify
|
|
|
|
@staticmethod
|
|
def _row(page, **kw):
|
|
row = {'page': page, 'uncovered': [], 'uncovered_ratio': 0.0, 'suspect_blocks': [],
|
|
'overlaps': [], 'clipped': [], 'measured': True}
|
|
row.update(kw)
|
|
return row
|
|
|
|
def test_flags_pages_and_summarises(self):
|
|
rows = iter([self._row(1), self._row(2, overlaps=[['a', 'b']])])
|
|
done, verify = self._run({'vision_ocr_url': 'http://x', 'vision_ocr_model': 'm'},
|
|
lambda *a, **k: next(rows))
|
|
self.assertEqual(done['result']['pages_needing_review'], [2])
|
|
self.assertEqual(done['result']['regions_added'], 2)
|
|
self.assertEqual(done['result']['regions_corrected'], 4)
|
|
self.assertEqual(verify.call_count, 2)
|
|
|
|
def test_skips_verification_without_vision_engine(self):
|
|
rows = iter([self._row(1), self._row(2)])
|
|
done, verify = self._run({}, lambda *a, **k: next(rows))
|
|
verify.assert_not_called()
|
|
self.assertFalse(done['result']['verified_with_vision'])
|
|
self.assertIn('skipped', done['detail'])
|
|
|
|
def test_a_duplicate_created_by_recovering_missing_text_is_still_caught(self):
|
|
"""clean_up_regions runs at the start of _fix_page_geometry -- but a later step, in the
|
|
same function or in recover_missing_text right after it, can still grow or add a box
|
|
whose text now duplicates another's (confirmed live on c't KI Wissen 2026 page 152: an
|
|
infographic caption fragment was left narrated twice this way). _fix_page_geometry is
|
|
stubbed out here so the test isolates the specific gap the fix closes: a duplicate that
|
|
appears only once recover_missing_text has run, after _fix_page_geometry's own check."""
|
|
from unittest.mock import patch
|
|
from routes.magazine import _quality_run_sync
|
|
text = "Die Plattformen bilden das Zentrum der Kreisläufe."
|
|
big = {'id': 'wide', 'page': 1, 'type': 'body', 'bbox': [0, 0, 500, 100],
|
|
'text': "Intro. More text after it.", 'reading_order': 0} # not yet a duplicate
|
|
small = {'id': 'frag', 'page': 1, 'type': 'body', 'bbox': [10, 10, 200, 20],
|
|
'text': text, 'reading_order': 1}
|
|
before = {'pages': [{'page': 1}], 'blocks': [big, small]}
|
|
grown = {'pages': [{'page': 1}], 'blocks': [
|
|
{**big, 'text': f"Intro. {text} More text after it."}, small]} # now contains it
|
|
seen_types: list[str] = []
|
|
|
|
def fake_audit(d, layout, page):
|
|
seen_types.append(next(b['type'] for b in layout['blocks'] if b['id'] == 'frag'))
|
|
return self._row(page)
|
|
|
|
lib = Path(tempfile.mkdtemp(prefix='ttvc-qrun-dup-'))
|
|
self.addCleanup(shutil.rmtree, lib, True)
|
|
(lib / ('a' * 32)).mkdir()
|
|
with patch('routes.magazine.mag.load_layout', return_value=before), \
|
|
patch('routes.magazine._fix_page_geometry', return_value=(before, self._geom_counts(), False)), \
|
|
patch('routes.magazine.mag.recover_missing_text', return_value=(grown, 1)), \
|
|
patch('routes.magazine.mag.audit_page_coverage', side_effect=fake_audit), \
|
|
patch('routes.magazine.mag.save_layout'), \
|
|
patch('routes.magazine._LIB', lib), \
|
|
patch('routes.magazine.background_job_update'):
|
|
_quality_run_sync('a' * 32, 'job', {})
|
|
self.assertEqual(seen_types, ['furniture'])
|
|
|
|
@staticmethod
|
|
def _geom_counts():
|
|
return {k: 0 for k in ('boxes_fitted', 'paragraphs_cut', 'pullquotes_found',
|
|
'pictures_added', 'overlaps_fixed', 'seams_mended')}
|
|
|
|
|
|
class PageAdvertisementRouteTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_mark_and_unmark_round_trip_and_404(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import (magazine_mark_page_advertisement,
|
|
magazine_unmark_page_advertisement)
|
|
layout = {'pages': [{'page': 1}, {'page': 2}], 'blocks': []}
|
|
with patch('routes.magazine._require_magazine', return_value=(Path('/nowhere'), {})), \
|
|
patch('routes.magazine.mag.load_layout', return_value=layout), \
|
|
patch('routes.magazine.mag.save_layout') as save:
|
|
await magazine_mark_page_advertisement('a' * 32, 2)
|
|
self.assertTrue(save.call_args.args[1]['pages'][1]['advertisement'])
|
|
await magazine_unmark_page_advertisement('a' * 32, 2)
|
|
self.assertNotIn('advertisement', save.call_args.args[1]['pages'][1])
|
|
with self.assertRaises(HTTPException):
|
|
await magazine_mark_page_advertisement('a' * 32, 9)
|
|
|
|
|
|
class ArticleAcronymGlossaryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from core import magazine as mag
|
|
self.dir = Path(tempfile.mkdtemp(prefix="ttvc-acronyms-"))
|
|
mag.save_layout(self.dir, {"pages": [{"page": 1}], "images": [], "blocks": [
|
|
{"id": "b1", "page": 1, "type": "body", "reading_order": 0, "bbox": [0, 0, 1, 1],
|
|
"text": "Der NAS speichert die Daten und die KI wertet sie aus. Ein NAS ist schnell."},
|
|
]})
|
|
mag.save_articles(self.dir, [{"id": "a1", "blocks": ["b1"], "images": []}])
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
def call(self, fn, *args, body=None, settings=None, llm=None):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
request = AsyncMock(); request.json.return_value = body
|
|
with patch('routes.magazine._require_magazine', return_value=(self.dir, {})), \
|
|
patch('routes.magazine._load_settings', return_value=settings or {}), \
|
|
patch('routes.magazine.asyncio.to_thread', new=_run_sync), \
|
|
patch('routes.conversation._post_llm_chat_completion') as post:
|
|
if llm is not None:
|
|
reply = MagicMock(); reply.raise_for_status.return_value = None
|
|
reply.json.return_value = {"choices": [{"message": {"content": llm}}]}
|
|
post.return_value = reply
|
|
args = (*args, request) if body is not None else args
|
|
return asyncio.run(fn('a' * 32, 'a1', *args))
|
|
|
|
def test_unknown_acronyms_are_listed_with_context_and_known_ones_are_not(self):
|
|
from routes.magazine import magazine_acronyms_get
|
|
state = self.call(magazine_acronyms_get)
|
|
self.assertEqual(state["language"], "de")
|
|
self.assertEqual([e["acronym"] for e in state["unknown"]], ["NAS"])
|
|
|
|
def test_llm_decisions_are_saved_and_change_the_spoken_text(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_acronyms_resolve
|
|
llm = json.dumps({"acronyms": [
|
|
{"acronym": "NAS", "kind": "expansion", "spoken": "Netzwerkspeicher"},
|
|
{"acronym": "XYZ", "kind": "expansion", "spoken": "never asked about"}]})
|
|
state = self.call(magazine_acronyms_resolve, settings={"llm_model": "m"}, llm=llm)
|
|
self.assertEqual(state["glossary"], {"NAS": "Netzwerkspeicher"})
|
|
self.assertEqual(state["unknown"], [])
|
|
spoken = mag.article_text(mag.load_layout(self.dir), mag.load_articles(self.dir)[0],
|
|
fuse=True, glossary=state["glossary"])[0]["spoken_text"]
|
|
self.assertIn("Der Netzwerkspeicher speichert", spoken)
|
|
|
|
def test_letters_and_word_decisions_are_made_by_the_server_not_the_model(self):
|
|
from routes.magazine import magazine_acronyms_resolve
|
|
llm = json.dumps({"acronyms": [{"acronym": "NAS", "kind": "letters", "spoken": "ignore me"}]})
|
|
self.assertEqual(self.call(magazine_acronyms_resolve, settings={"llm_model": "m"}, llm=llm)["glossary"],
|
|
{"NAS": "N A S"})
|
|
|
|
def test_a_failing_llm_saves_nothing(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_acronyms_resolve
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
self.call(magazine_acronyms_resolve, settings={"llm_model": "m"}, llm="not json")
|
|
self.assertEqual(ctx.exception.status_code, 502)
|
|
self.assertNotIn("glossary", mag.load_articles(self.dir)[0])
|
|
|
|
def test_no_model_configured_is_a_clear_400(self):
|
|
from routes.magazine import magazine_acronyms_resolve
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
self.call(magazine_acronyms_resolve)
|
|
self.assertEqual(ctx.exception.status_code, 400)
|
|
|
|
def test_reviewer_edits_are_validated_and_stored(self):
|
|
from routes.magazine import magazine_acronyms_put
|
|
state = self.call(magazine_acronyms_put, body={"glossary": {"NAS": " Netz speicher ", "": "x", "BAD": 5, "LONG": "y" * 200}})
|
|
self.assertEqual(state["glossary"], {"NAS": "Netz speicher"})
|
|
with self.assertRaises(HTTPException):
|
|
self.call(magazine_acronyms_put, body={"glossary": ["x"]})
|
|
|
|
|
|
class SpellingRouteTests(unittest.TestCase):
|
|
DE = {"der", "ist", "gut", "und", "test", "tests", "die", "daten", "bank", "datenbank", "wächst", "nicht", "hilft"}
|
|
|
|
def setUp(self):
|
|
from core import magazine as mag
|
|
self.dir = Path(tempfile.mkdtemp(prefix="ttvc-spelling-"))
|
|
mag.save_layout(self.dir, {"pages": [{"page": 1}], "images": [], "blocks": [
|
|
{"id": "b1", "page": 1, "type": "body", "reading_order": 0, "bbox": [0, 0, 1, 1],
|
|
"text": "Der Tesst ist gut und die Datenbank wächst nicht. Zorbex hilft."},
|
|
{"id": "ad", "page": 1, "type": "ad", "reading_order": 1, "bbox": [0, 0, 1, 1], "text": "Kaufen Sie Blorp."},
|
|
]})
|
|
mag.save_articles(self.dir, [{"id": "a1", "blocks": ["b1"], "images": []}])
|
|
words = self.DE
|
|
|
|
class Checker:
|
|
def __contains__(self, w): return w.casefold() in words
|
|
def candidates(self, w): return {x for x in words if x != w and abs(len(x) - len(w)) <= 1 and sum(a != b for a, b in zip(x, w)) <= 1}
|
|
word_usage_frequency = None
|
|
from unittest.mock import patch
|
|
self._patches = [patch('core.pronunciations._german_spellchecker', return_value=Checker()),
|
|
patch('core.pronunciations._english_spellchecker', return_value=Checker())]
|
|
for p in self._patches:
|
|
p.start()
|
|
|
|
def tearDown(self):
|
|
for p in self._patches:
|
|
p.stop()
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
def call(self, fn, *args, body=None, **kw):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
request = AsyncMock(); request.json.return_value = body
|
|
with patch('routes.magazine._require_magazine', return_value=(self.dir, {})):
|
|
return asyncio.run(fn('a' * 32, *args, *( [request] if body is not None else []), **kw))
|
|
|
|
def test_findings_are_scoped_to_narratable_text_and_marked_with_suggestions(self):
|
|
from routes.magazine import magazine_spelling
|
|
out = self.call(magazine_spelling, page=1)
|
|
self.assertEqual([i["token"] for i in out["issues"]], ["Tesst", "Zorbex"])
|
|
self.assertEqual(out["issues"][0]["block_id"], "b1")
|
|
self.assertIn("Test", out["issues"][0]["suggestions"])
|
|
|
|
def test_a_reviewer_fix_edits_the_text_and_can_be_undone(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_spelling, magazine_spelling_fix, magazine_undo
|
|
issue = self.call(magazine_spelling, article="a1")["issues"][0]
|
|
out = self.call(magazine_spelling_fix, body={**{k: issue[k] for k in ("block_id", "start", "end", "token")},
|
|
"replacement": "Test"})
|
|
self.assertIn("Der Test ist gut", mag.load_layout(self.dir)["blocks"][0]["text"])
|
|
self.call(magazine_undo, body={"token": out["undo"]})
|
|
self.assertIn("Der Tesst ist gut", mag.load_layout(self.dir)["blocks"][0]["text"])
|
|
|
|
def test_a_stale_position_is_refused_and_markup_or_no_change_is_rejected(self):
|
|
from routes.magazine import magazine_spelling, magazine_spelling_fix
|
|
issue = self.call(magazine_spelling)["issues"][0]
|
|
base = {k: issue[k] for k in ("block_id", "start", "end", "token")}
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
self.call(magazine_spelling_fix, body={**base, "start": issue["start"] + 1, "end": issue["end"] + 1, "replacement": "Test"})
|
|
self.assertEqual(ctx.exception.status_code, 409)
|
|
for bad in ("<b>Test</b>", "Zeile\nzwei", "x" * 121, " ", issue["token"]):
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
self.call(magazine_spelling_fix, body={**base, "replacement": bad})
|
|
self.assertEqual(ctx.exception.status_code, 400, repr(bad))
|
|
|
|
def test_the_reviewer_can_type_their_own_correction_even_several_words(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_spelling, magazine_spelling_fix
|
|
issue = self.call(magazine_spelling)["issues"][0]
|
|
base = {k: issue[k] for k in ("block_id", "start", "end", "token")}
|
|
self.call(magazine_spelling_fix, body={**base, "replacement": "der Test"})
|
|
self.assertIn("Der der Test ist gut", mag.load_layout(self.dir)["blocks"][0]["text"])
|
|
|
|
def test_dictionary_entries_silence_a_finding_and_can_be_removed(self):
|
|
from routes.magazine import magazine_spelling, magazine_spelling_dictionary
|
|
self.call(magazine_spelling_dictionary, body={"word": "Zorbex"})
|
|
self.assertEqual([i["token"] for i in self.call(magazine_spelling)["issues"]], ["Tesst"])
|
|
self.call(magazine_spelling_dictionary, body={"word": "Zorbex", "remove": True})
|
|
self.assertEqual(len(self.call(magazine_spelling)["issues"]), 2)
|
|
with self.assertRaises(HTTPException):
|
|
self.call(magazine_spelling_dictionary, body={"word": "two words"})
|
|
|
|
def test_the_agent_fixes_what_passes_the_guards_and_leaves_the_rest_marked(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import spelling_agent_pass, _load_dictionary
|
|
|
|
def check(settings, language, text, issues):
|
|
return [{"token": "Tesst", "action": "replace", "replacement": "Test"},
|
|
{"token": "Zorbex", "action": "keep"}]
|
|
result = spelling_agent_pass(self.dir, mag.load_layout(self.dir), {}, check=check)
|
|
self.assertEqual((result["fixed"], result["kept"], result["remaining"]), (1, 1, 0))
|
|
self.assertIn("Der Test ist gut", result["layout"]["blocks"][0]["text"])
|
|
self.assertEqual(_load_dictionary(self.dir), {"zorbex": "agent"})
|
|
|
|
def test_an_agent_that_rewrites_or_fails_changes_nothing_and_findings_stay_marked(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import spelling_agent_pass
|
|
layout = mag.load_layout(self.dir)
|
|
rewrite = lambda *a: [{"token": "Tesst", "action": "replace", "replacement": "völlig anderer Satz"}]
|
|
result = spelling_agent_pass(self.dir, layout, {}, check=rewrite)
|
|
self.assertEqual((result["fixed"], result["remaining"]), (0, 2))
|
|
self.assertEqual(result["layout"]["blocks"][0]["text"], layout["blocks"][0]["text"])
|
|
|
|
def boom(*a):
|
|
raise RuntimeError("llm down")
|
|
result = spelling_agent_pass(self.dir, layout, {}, check=boom)
|
|
self.assertEqual((result["failed"], result["fixed"], result["remaining"]), (1, 0, 2))
|
|
|
|
def test_starting_the_agent_without_a_model_is_a_clear_400(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import magazine_spelling_agent
|
|
with patch('routes.magazine._load_settings', return_value={}), self.assertRaises(HTTPException) as ctx:
|
|
self.call(magazine_spelling_agent, body={})
|
|
self.assertEqual(ctx.exception.status_code, 400)
|
|
|
|
|
|
class PageAuditAndAutofixTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from PIL import Image, ImageDraw
|
|
from core import magazine as mag
|
|
self.dir = Path(tempfile.mkdtemp(prefix="ttvc-autofix-"))
|
|
(self.dir / "pages").mkdir()
|
|
im = Image.new("RGB", (600, 800), (248, 242, 244))
|
|
draw = ImageDraw.Draw(im)
|
|
for y in range(100, 200, 10): # text x 100..300
|
|
draw.rectangle([100, y, 300, y + 5], fill=(93, 91, 92))
|
|
im.save(self.dir / "pages" / "0001.png")
|
|
# the box stops 20 px short of the text's right end: a clipped edge
|
|
mag.save_layout(self.dir, {"pages": [{"page": 1, "width": 600, "height": 800}], "images": [], "blocks": [
|
|
{"id": "b1", "page": 1, "type": "body", "reading_order": 0, "bbox": [98, 98, 182, 108],
|
|
"text": "Ein Absatz mit genug Text, damit er ordentlich gelesen wird.", "source": "vision_ocr", "region_label": "text"}]})
|
|
mag.save_articles(self.dir, [{"id": "a1", "blocks": ["b1"], "images": []}])
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
def call(self, fn, *args, body=None):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
request = AsyncMock(); request.json.return_value = body if body is not None else {}
|
|
with patch('routes.magazine._require_magazine', return_value=(self.dir, {})), \
|
|
patch('routes.magazine._load_settings', return_value={}), \
|
|
patch('routes.magazine.asyncio.to_thread', new=_run_sync), \
|
|
patch('routes.magazine.asyncio.create_task', side_effect=lambda coro: asyncio.ensure_future(coro)):
|
|
return asyncio.run(fn('a' * 32, *args, *([request] if body is not None else [])))
|
|
|
|
def test_the_live_audit_reports_the_clipped_edge(self):
|
|
from routes.magazine import magazine_page_audit
|
|
audit = self.call(magazine_page_audit, 1)
|
|
self.assertEqual(audit["clipped"], [{"id": "b1", "side": "right"}])
|
|
self.assertEqual(audit["spelling"], 0)
|
|
self.assertFalse(audit["advertisement"])
|
|
|
|
def test_the_audit_of_an_unknown_page_is_a_404(self):
|
|
from routes.magazine import magazine_page_audit
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
self.call(magazine_page_audit, 9)
|
|
self.assertEqual(ctx.exception.status_code, 404)
|
|
|
|
def test_the_box_fix_removes_the_clipped_edge_and_can_be_undone(self):
|
|
from unittest.mock import patch
|
|
from core import magazine as mag
|
|
from routes.magazine import _autofix_page_sync, magazine_page_audit, magazine_undo
|
|
with patch('routes.magazine._LIB', self.dir.parent), patch('routes.magazine.background_job_update') as job:
|
|
_autofix_page_sync(self.dir.name, "j", 1, "boxes", {})
|
|
result = job.call_args.kwargs
|
|
self.assertTrue(result["done"], result)
|
|
self.assertGreaterEqual(result["result"]["boxes_fitted"], 1)
|
|
self.assertEqual(self.call(magazine_page_audit, 1)["clipped"], [])
|
|
self.call(magazine_undo, body={"token": result["result"]["undo"]})
|
|
self.assertEqual(mag.load_layout(self.dir)["blocks"][0]["bbox"], [98, 98, 182, 108])
|
|
|
|
def test_a_failure_is_reported_on_the_job_not_raised(self):
|
|
from unittest.mock import patch
|
|
from routes.magazine import _autofix_page_sync
|
|
(self.dir / "layout.json").unlink()
|
|
with patch('routes.magazine._LIB', self.dir.parent), patch('routes.magazine.background_job_update') as job:
|
|
_autofix_page_sync(self.dir.name, "j", 1, "all", {})
|
|
self.assertTrue(job.call_args.kwargs["done"] and job.call_args.kwargs["error"])
|
|
|
|
|
|
class ArticleLinkingRouteTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from core import magazine as mag
|
|
self.dir = Path(tempfile.mkdtemp(prefix="ttvc-link-"))
|
|
blk = lambda i, pg, ty, t, n: {"id": i, "page": pg, "type": ty, "text": t, "reading_order": n, "bbox": [0, 0, 1, 1]}
|
|
mag.save_layout(self.dir, {"pages": [{"page": 1}, {"page": 2}, {"page": 3}], "images": [], "blocks": [
|
|
blk("t", 1, "title", "Der Butler", 0), blk("a", 1, "body", "Der Text beginnt und", 1),
|
|
blk("b", 2, "body", "geht weiter und endet. (hag)", 0), blk("z", 3, "body", "Etwas ganz anderes.", 0)]})
|
|
art = lambda i, ti, pg, bl: {"id": i, "title": ti, "author": "", "pages": pg, "blocks": bl, "images": [], "status": "approved",
|
|
"confidence": 0.5, "section": "", "continued_to": [], "audio_status": "none"}
|
|
mag.save_articles(self.dir, [art("article-001", "Der Butler", [1], ["t", "a"]), art("article-002", "Page 2", [2], ["b"])])
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
def call(self, fn, *args, body=None):
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
request = AsyncMock(); request.json.return_value = body if body is not None else {}
|
|
with patch('routes.magazine._require_magazine', return_value=(self.dir, {})), \
|
|
patch('routes.magazine.asyncio.to_thread', new=_run_sync):
|
|
return asyncio.run(fn('a' * 32, *args, *([request] if body is not None else [])))
|
|
|
|
def test_auto_link_merges_the_stand_in_page_and_can_be_undone(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_auto_link, magazine_undo
|
|
out = self.call(magazine_auto_link)
|
|
self.assertEqual((out["before"], out["after"], out["text_blocks_linked"]), (2, 1, 1))
|
|
self.assertEqual(mag.load_articles(self.dir)[0]["pages"], [1, 2])
|
|
self.call(magazine_undo, body={"token": out["undo"]})
|
|
self.assertEqual(len(mag.load_articles(self.dir)), 2)
|
|
|
|
def test_linking_a_page_by_hand_adds_the_right_text_ticked_and_is_undoable(self):
|
|
from core import magazine as mag
|
|
from routes.magazine import magazine_link_page
|
|
out = self.call(magazine_link_page, "article-001", body={"page": 2, "where": "after"})
|
|
self.assertEqual(out["added"], ["b"])
|
|
article = mag.load_articles(self.dir)[0]
|
|
self.assertEqual((article["pages"], article["blocks"]), ([1, 2], ["t", "a", "b"]))
|
|
|
|
def test_linking_needs_a_real_page_with_something_to_link(self):
|
|
from routes.magazine import magazine_link_page
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
self.call(magazine_link_page, "article-001", body={"page": 99})
|
|
self.assertEqual(ctx.exception.status_code, 404)
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
self.call(magazine_link_page, "article-001", body={})
|
|
self.assertEqual(ctx.exception.status_code, 400)
|
|
|
|
|
|
class BoxFixesReorderThePageTests(unittest.TestCase):
|
|
def test_the_per_page_fix_applies_the_column_order_the_engine_order_lacked(self):
|
|
"""Regression: after import or a re-OCR the order was whatever the vision/LLM step
|
|
returned, because the column-aware order only ran from a manual button."""
|
|
from routes.magazine import _fix_page_geometry
|
|
d = Path(tempfile.mkdtemp(prefix="ttvc-order-"))
|
|
self.addCleanup(shutil.rmtree, d, True)
|
|
blk = lambda i, x, y, n: {"id": i, "page": 1, "type": "body", "text": "Text " + i, "reading_order": n, "bbox": [x, y, 90, 60]}
|
|
layout = {"pages": [{"page": 1, "width": 210, "height": 300}], "images": [],
|
|
"blocks": [blk("r1", 110, 10, 0), blk("l1", 10, 10, 1), blk("r2", 110, 80, 2), blk("l2", 10, 80, 3)]}
|
|
fixed, _, _ = _fix_page_geometry(d, layout, [], 1, {})
|
|
self.assertEqual([b["id"] for b in sorted(fixed["blocks"], key=lambda b: b["reading_order"])], ["l1", "l2", "r1", "r2"])
|
|
|
|
|
|
class FastLlmCallTests(unittest.TestCase):
|
|
def test_thinking_is_switched_off_and_a_rejecting_backend_is_asked_again_without_the_switch(self):
|
|
from unittest.mock import MagicMock, patch
|
|
from routes.magazine import _llm_chat_fast
|
|
bad, good = MagicMock(status_code=422), MagicMock(status_code=200)
|
|
with patch('routes.conversation._post_llm_chat_completion', side_effect=[bad, good]) as post:
|
|
out = _llm_chat_fast({"llm_model": "m", "llm_url": "http://x"}, [{"role": "user", "content": "hi"}])
|
|
self.assertIs(out, good)
|
|
first, second = post.call_args_list[0].args[1], post.call_args_list[1].args[1]
|
|
self.assertEqual(first["chat_template_kwargs"], {"enable_thinking": False})
|
|
self.assertEqual((first["temperature"], first["max_tokens"]), (0.0, 800))
|
|
self.assertNotIn("chat_template_kwargs", second)
|
|
|
|
def test_a_backend_that_accepts_it_is_called_once(self):
|
|
from unittest.mock import MagicMock, patch
|
|
from routes.magazine import _llm_chat_fast
|
|
with patch('routes.conversation._post_llm_chat_completion', return_value=MagicMock(status_code=200)) as post:
|
|
_llm_chat_fast({"llm_model": "m"}, [])
|
|
self.assertEqual(post.call_count, 1)
|