mARTin-B78
2dd5cea7b7
Some checks failed
Test and SonarQube / quality (push) Has been cancelled
Adds "Edit boxes" mode to article review: drag an already-detected
region to move it, its corner handle to resize it, or press its x to
delete it -- for correcting a box the layout engine got wrong, not one
it missed entirely ("Draw it" still covers that case). Moving or
resizing opens a confirm panel; "Save" keeps the region's existing
text at its new position, and for text regions "Save & re-OCR" reads
the corrected area again and replaces the text.
New core/magazine.py update_region()/delete_region() and PATCH/DELETE
/api/magazine/{doc}/pages/{page}/regions/{block_id}. An image region
is re-cropped from the page render whenever its bbox changes, since
its saved file is exactly that crop.
Live-verified against the real ingested c't KI Wissen 2026 magazine:
add -> move -> move+re-OCR (correctly kept old text when the corrected
area had nothing readable) -> delete -> confirmed gone, repeat delete
404s.
275 tests passing.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
580 lines
30 KiB
Python
580 lines
30 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
|
|
|
|
_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:
|
|
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'):
|
|
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)
|