mARTin-B78
1edc941dbf
Some checks are pending
Test and SonarQube / quality (push) Waiting to run
The preview carries each page's regions (percent of the page, type, reading order) for the reading pass and the labelling pass, drawn over the page image with a legend. A server restart that killed the import left it "running" for ever; the status route now reports it as failed. v1.42.0. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1247 lines
68 KiB
Python
1247 lines
68 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, _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 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_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 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)
|
|
|
|
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 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 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.assertFalse((self.dir / "undo.json").exists())
|
|
|
|
|
|
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_old_snapshot_is_refused(self):
|
|
import json
|
|
from routes.magazine import magazine_undo
|
|
from core import magazine as mag
|
|
(self.dir / "undo.json").write_text(json.dumps({"token": "t", "label": "x", "layout": mag.load_layout(self.dir),
|
|
"articles": mag.load_articles(self.dir)}), "utf-8")
|
|
with self.assertRaises(HTTPException):
|
|
self.run_route(magazine_undo, body={"token": "t"})
|
|
|
|
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_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_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 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()
|