452 lines
24 KiB
Python
452 lines
24 KiB
Python
"""Turning a page that was scanned sideways: geometry, persistence, failure handling, validation."""
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-test-config-"))
|
|
|
|
from PIL import Image
|
|
|
|
from core import magazine as mag
|
|
|
|
|
|
class GeometryTests(unittest.TestCase):
|
|
def test_a_box_turns_with_the_page(self):
|
|
# 100 wide, 200 tall page; a box near the top-left corner.
|
|
self.assertEqual(mag.rotate_bbox([10, 20, 30, 40], 100, 200, 90), [140, 10, 40, 30]) # lands top-right
|
|
self.assertEqual(mag.rotate_bbox([10, 20, 30, 40], 100, 200, 180), [60, 140, 30, 40]) # bottom-right
|
|
self.assertEqual(mag.rotate_bbox([10, 20, 30, 40], 100, 200, 270), [20, 60, 40, 30]) # bottom-left
|
|
|
|
def test_four_quarter_turns_are_no_turn(self):
|
|
box, w, h = [10.0, 20.0, 30.0, 40.0], 100.0, 200.0
|
|
for _ in range(4):
|
|
box = mag.rotate_bbox(box, w, h, 90)
|
|
w, h = h, w
|
|
self.assertEqual(box, [10.0, 20.0, 30.0, 40.0])
|
|
|
|
def test_a_turn_and_its_opposite_cancel(self):
|
|
turned = mag.rotate_bbox([10, 20, 30, 40], 100, 200, 90)
|
|
self.assertEqual(mag.rotate_bbox(turned, 200, 100, 270), [10, 20, 30, 40])
|
|
|
|
def test_only_quarter_turns_are_accepted(self):
|
|
self.assertEqual([mag.normalise_rotation(v) for v in (90, -90, 270, 180, "90", 450, -270)], [90, 270, 270, 180, 90, 90, 90])
|
|
for bad in (0, 45, 360, "x", None, 1):
|
|
with self.assertRaises(ValueError):
|
|
mag.normalise_rotation(bad)
|
|
|
|
|
|
class ImageTests(unittest.TestCase):
|
|
def png(self, w=100, h=200):
|
|
p = Path(tempfile.mkdtemp(prefix="ttvc-rot-")) / "0001.png"
|
|
im = Image.new("RGB", (w, h), (255, 255, 255))
|
|
im.putpixel((10, 20), (255, 0, 0))
|
|
im.save(p)
|
|
return p
|
|
|
|
def test_a_quarter_turn_is_clockwise_and_swaps_the_size(self):
|
|
p = self.png()
|
|
mag.rotate_page_image(p, 90)
|
|
with Image.open(p) as im:
|
|
self.assertEqual(im.size, (200, 100))
|
|
self.assertEqual(im.getpixel((200 - 1 - 20, 10)), (255, 0, 0)) # (10,20) -> (H-1-y, x)
|
|
|
|
def test_a_half_turn_keeps_the_size(self):
|
|
p = self.png()
|
|
mag.rotate_page_image(p, 180)
|
|
with Image.open(p) as im:
|
|
self.assertEqual(im.size, (100, 200))
|
|
self.assertEqual(im.getpixel((100 - 1 - 10, 200 - 1 - 20)), (255, 0, 0))
|
|
|
|
def test_the_pixel_agrees_with_the_box_maths(self):
|
|
p = self.png()
|
|
x, y, _, _ = mag.rotate_bbox([10, 20, 1, 1], 100, 200, 90)
|
|
mag.rotate_page_image(p, 90)
|
|
with Image.open(p) as im:
|
|
self.assertEqual(im.getpixel((int(x), int(y))), (255, 0, 0))
|
|
|
|
def test_no_temporary_file_is_left_behind(self):
|
|
p = self.png()
|
|
mag.rotate_page_image(p, 270)
|
|
self.assertEqual(sorted(f.name for f in p.parent.iterdir()), ["0001.png"])
|
|
|
|
|
|
class LayoutTests(unittest.TestCase):
|
|
def layout(self):
|
|
return {"pages": [{"page": 1, "width": 100.0, "height": 200.0}, {"page": 2, "width": 100.0, "height": 200.0}],
|
|
"blocks": [{"id": "a", "page": 1, "bbox": [10, 20, 30, 40]}, {"id": "b", "page": 2, "bbox": [10, 20, 30, 40]}],
|
|
"images": [{"id": "i", "page": 1, "bbox": [0, 0, 50, 50]}]}
|
|
|
|
def test_only_the_turned_page_changes(self):
|
|
out = mag.layout_with_rotated_page(self.layout(), 1, 90)
|
|
self.assertEqual((out["pages"][0]["width"], out["pages"][0]["height"], out["pages"][0]["rotation"]), (200.0, 100.0, 90))
|
|
self.assertEqual(out["pages"][1], {"page": 2, "width": 100.0, "height": 200.0})
|
|
self.assertEqual(out["blocks"][0]["bbox"], [140, 10, 40, 30])
|
|
self.assertEqual(out["blocks"][1]["bbox"], [10, 20, 30, 40])
|
|
self.assertEqual(out["images"][0]["bbox"], [150, 0, 50, 50])
|
|
|
|
def test_the_input_is_not_modified_and_rotation_accumulates(self):
|
|
original = self.layout()
|
|
snapshot = json.dumps(original)
|
|
once = mag.layout_with_rotated_page(original, 1, 90)
|
|
self.assertEqual(json.dumps(original), snapshot)
|
|
twice = mag.layout_with_rotated_page(once, 1, 270)
|
|
self.assertEqual(twice["pages"][0]["rotation"], 0)
|
|
self.assertEqual((twice["pages"][0]["width"], twice["pages"][0]["height"]), (100.0, 200.0))
|
|
self.assertEqual(twice["blocks"][0]["bbox"], [10, 20, 30, 40])
|
|
|
|
def test_an_unknown_page_is_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
mag.layout_with_rotated_page(self.layout(), 9, 90)
|
|
|
|
|
|
class LadderTests(unittest.TestCase):
|
|
"""read_turned_page_regions: which reading wins, and how boxes from the original orientation are carried."""
|
|
GOOD = "Dies ist ein vollstaendiger deutscher Satz mit genug Woertern fuer eine brauchbare Lesung der Seite."
|
|
|
|
def setUp(self):
|
|
d = Path(tempfile.mkdtemp(prefix="ttvc-ladder-"))
|
|
self.turned = d / "turned.png"
|
|
Image.new("RGB", (200, 100), "white").save(self.turned) # the page after a 90 degree turn (was 100 wide, 200 tall)
|
|
buf = __import__("io").BytesIO()
|
|
Image.new("RGB", (100, 200), "white").save(buf, "PNG")
|
|
self.original = buf.getvalue()
|
|
self.calls = []
|
|
|
|
def run_ladder(self, vision, tesseract=None, which="/usr/bin/tesseract"):
|
|
def chain(png, engines, timeout=0):
|
|
self.calls.append(Path(png).name)
|
|
return vision[len(self.calls) - 1] if len(self.calls) <= len(vision) else None
|
|
with mock.patch.object(mag, "vision_ocr_page_chain", side_effect=chain), \
|
|
mock.patch.object(mag, "ocr_page", return_value=tesseract or []) as ocr, \
|
|
mock.patch("shutil.which", return_value=which):
|
|
out = mag.read_turned_page_regions(self.turned, self.original, 90, [("http://x", "m")])
|
|
self.ocr = ocr
|
|
return out
|
|
|
|
def region(self, text, bbox=(10, 20, 30, 40)):
|
|
return {"label": "text", "bbox": list(bbox), "text": text}
|
|
|
|
def test_a_good_reading_of_the_turned_page_wins_at_once(self):
|
|
regions, how = self.run_ladder([[self.region(self.GOOD)]])
|
|
self.assertEqual(how, "vision")
|
|
self.assertEqual(self.calls, ["turned.png"]) # nothing else was tried
|
|
self.ocr.assert_not_called()
|
|
|
|
def test_when_the_engine_returns_nothing_the_original_orientation_is_read_and_its_boxes_carried_across(self):
|
|
regions, how = self.run_ladder([None, [self.region(self.GOOD, (10, 20, 30, 40))]])
|
|
self.assertEqual(how, "vision-original")
|
|
self.assertEqual(regions[0]["bbox"], [140, 10, 40, 30]) # rotate_bbox of the box on the 100x200 original, 90 degrees
|
|
self.assertEqual(regions[0]["text"], self.GOOD)
|
|
self.assertEqual(self.calls, ["turned.png", "original.png"])
|
|
|
|
def test_tesseract_is_the_last_rung(self):
|
|
para = {"bbox": [5, 5, 50, 20], "text": self.GOOD, "confidence": 90, "line_height": 8}
|
|
regions, how = self.run_ladder([None, None], tesseract=[para])
|
|
self.assertEqual(how, "tesseract")
|
|
self.assertEqual(regions, [{"label": "text", "bbox": [5, 5, 50, 20], "text": self.GOOD}])
|
|
|
|
def test_a_short_or_implausible_reading_does_not_win_but_the_best_of_them_is_used_if_nothing_else_does(self):
|
|
regions, how = self.run_ladder([[self.region("Seite 7")], [self.region("zzz qqq xxx")]], tesseract=[])
|
|
self.assertIn(how, ("vision", "vision-original"))
|
|
self.assertTrue(regions)
|
|
self.assertEqual(regions[0]["text"], "zzz qqq xxx" if how == "vision-original" else "Seite 7")
|
|
|
|
def test_a_picture_page_is_an_empty_reading_not_a_failure_when_tesseract_ran(self):
|
|
self.assertEqual(self.run_ladder([None, None], tesseract=[]), ([], "empty"))
|
|
|
|
def test_without_tesseract_and_without_any_reading_it_is_a_failure(self):
|
|
self.assertEqual(self.run_ladder([None, None], tesseract=[], which=None), (None, "failed"))
|
|
|
|
def test_the_original_bytes_are_used_not_the_turned_file(self):
|
|
self.run_ladder([None, [self.region(self.GOOD)]])
|
|
self.assertEqual(self.calls[1], "original.png")
|
|
|
|
def test_bad_degrees_are_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
mag.read_turned_page_regions(self.turned, self.original, 45, [])
|
|
|
|
|
|
class JobTests(unittest.TestCase):
|
|
"""_reocr_page_sync with a turn: success persists it, failure puts the page back."""
|
|
|
|
def setUp(self):
|
|
import routes.magazine as routes
|
|
self.routes = routes
|
|
self.lib = Path(tempfile.mkdtemp(prefix="ttvc-lib-"))
|
|
self.d = self.lib / ("a" * 32)
|
|
(self.d / "pages").mkdir(parents=True)
|
|
self.png = self.d / "pages" / "0003.png"
|
|
Image.new("RGB", (100, 200), (255, 255, 255)).save(self.png)
|
|
self.original = self.png.read_bytes()
|
|
layout = {"pages": [{"page": 3, "width": 100.0, "height": 200.0}], "images": [],
|
|
"blocks": [{"id": "b1", "page": 3, "bbox": [10, 20, 30, 40], "type": "body", "text": "x", "reading_order": 0}]}
|
|
(self.d / "layout.json").write_text(json.dumps(layout))
|
|
(self.d / "articles.json").write_text(json.dumps({"articles": []}))
|
|
(self.d / "meta.json").write_text(json.dumps({"kind": "magazine", "title": "T"}))
|
|
self.jobs = []
|
|
patches = [mock.patch.object(routes, "_LIB", self.lib),
|
|
mock.patch.object(routes, "background_job_update", side_effect=lambda jid, **kw: self.jobs.append(kw)),
|
|
mock.patch.object(routes, "_recognise_advertisements"), mock.patch.object(routes, "_recognise_semantic_layout"),
|
|
mock.patch.object(routes, "_refresh_quality_rows"), mock.patch.object(routes, "_load_settings", return_value={}),
|
|
mock.patch.object(routes, "_snapshot_for_undo", return_value="tok"), mock.patch.object(routes, "_seal_undo"),
|
|
mock.patch.object(routes.mag, "refresh_visual_regions"),
|
|
mock.patch.object(routes.mag, "clean_up_regions", side_effect=lambda lay, pages=None: (lay, {})),
|
|
mock.patch.object(routes.mag, "reorder_page", side_effect=lambda lay, p: lay),
|
|
mock.patch.object(routes.mag, "detect_inset_pullquotes", side_effect=lambda d, lay, pages, *a: (lay, 0)),
|
|
mock.patch.object(routes.mag, "split_around_pullquotes", side_effect=lambda d, lay, pages: (lay, 0)),
|
|
mock.patch.object(routes.mag, "snap_text_boxes", side_effect=lambda d, lay, pages=None: (lay, [])),
|
|
mock.patch.object(routes.mag, "remap_reviewed_page")]
|
|
for p in patches:
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
|
|
def run_job(self, degrees, reread, ladder=None):
|
|
ladder = ladder or (lambda *a, **k: ([{"label": "text", "bbox": [1, 1, 5, 5], "text": "x" * 50}], "vision"))
|
|
with mock.patch.object(self.routes.mag, "reocr_page", side_effect=reread) as m, \
|
|
mock.patch.object(self.routes.mag, "read_turned_page_regions", side_effect=ladder) as lad:
|
|
self.routes._reocr_page_sync("a" * 32, "job", 3, "http://ocr", "m", None, None, degrees)
|
|
self.ladder = lad
|
|
return m
|
|
|
|
def test_success_turns_the_image_remembers_it_and_re_measures_the_page(self):
|
|
seen = {}
|
|
|
|
def reread(d, layout, page, *a, **k):
|
|
seen["meta"] = next(p for p in layout["pages"] if p["page"] == page)
|
|
seen["bbox"] = layout["blocks"][0]["bbox"]
|
|
return layout
|
|
self.run_job(90, reread)
|
|
self.assertEqual((seen["meta"]["width"], seen["meta"]["height"]), (200.0, 100.0)) # the read sees the turned page
|
|
self.assertEqual(seen["bbox"], [140, 10, 40, 30]) # ...with the old boxes carried along
|
|
with Image.open(self.png) as im:
|
|
self.assertEqual(im.size, (200, 100))
|
|
self.assertEqual(json.loads((self.d / "meta.json").read_text())["page_rotations"], {"3": 90})
|
|
self.routes._refresh_quality_rows.assert_called_once()
|
|
self.assertEqual(self.jobs[-1]["stage"], "done")
|
|
self.assertIsNone(self.jobs[-1]["result"]["undo"])
|
|
self.routes._snapshot_for_undo.assert_not_called() # a turn is not undoable
|
|
|
|
def test_turns_add_up_and_a_full_circle_forgets_itself(self):
|
|
for _ in range(4):
|
|
self.run_job(90, lambda d, layout, page, *a, **k: layout)
|
|
self.assertNotIn("3", json.loads((self.d / "meta.json").read_text()).get("page_rotations", {}))
|
|
|
|
def test_a_failed_read_puts_the_page_back_and_saves_nothing(self):
|
|
before_layout = (self.d / "layout.json").read_text()
|
|
self.run_job(90, lambda *a, **k: None) # the engine returned nothing usable
|
|
self.assertEqual(self.png.read_bytes(), self.original)
|
|
self.assertEqual((self.d / "layout.json").read_text(), before_layout)
|
|
self.assertNotIn("page_rotations", json.loads((self.d / "meta.json").read_text()))
|
|
self.assertEqual(self.jobs[-1]["stage"], "failed")
|
|
self.routes._refresh_quality_rows.assert_not_called()
|
|
|
|
def test_an_exception_during_the_read_also_restores_the_page(self):
|
|
def boom(*a, **k):
|
|
raise RuntimeError("engine down")
|
|
self.run_job(180, boom)
|
|
self.assertEqual(self.png.read_bytes(), self.original)
|
|
self.assertIn("engine down", self.jobs[-1]["error"])
|
|
|
|
def test_a_turn_reads_through_the_ladder_and_hands_its_regions_to_the_re_read(self):
|
|
regions = [{"label": "text", "bbox": [1, 2, 3, 4], "text": "gelesen"}]
|
|
m = self.run_job(90, lambda d, layout, page, *a, **k: layout, ladder=lambda *a, **k: (regions, "vision-original"))
|
|
self.assertEqual(m.call_args.kwargs["regions"], regions)
|
|
self.assertEqual(self.jobs[-1]["result"]["read_with"], "vision-original")
|
|
# the ladder gets the turned page, the bytes of the page as it was, and the turn
|
|
args = self.ladder.call_args.args
|
|
self.assertEqual((args[0], args[1], args[2]), (self.png, self.original, 90))
|
|
|
|
def test_when_nothing_can_read_the_page_it_is_put_back(self):
|
|
self.run_job(90, lambda d, layout, page, *a, **k: layout, ladder=lambda *a, **k: (None, "failed"))
|
|
self.assertEqual(self.png.read_bytes(), self.original)
|
|
self.assertIn("Neither the OCR engines nor tesseract", self.jobs[-1]["error"])
|
|
self.assertNotIn("page_rotations", json.loads((self.d / "meta.json").read_text()))
|
|
|
|
def test_a_picture_page_with_no_text_can_still_be_turned(self):
|
|
self.run_job(90, lambda d, layout, page, *a, **k: layout, ladder=lambda *a, **k: ([], "empty"))
|
|
self.assertEqual(self.jobs[-1]["stage"], "done")
|
|
self.assertEqual(json.loads((self.d / "meta.json").read_text())["page_rotations"], {"3": 90})
|
|
|
|
def test_without_a_turn_the_job_behaves_as_before(self):
|
|
self.run_job(0, lambda d, layout, page, *a, **k: layout)
|
|
self.ladder.assert_not_called()
|
|
self.routes._snapshot_for_undo.assert_called_once()
|
|
self.assertEqual(self.png.read_bytes(), self.original)
|
|
self.assertNotIn("page_rotations", json.loads((self.d / "meta.json").read_text()))
|
|
|
|
|
|
class RenderTests(unittest.TestCase):
|
|
"""A stored turn survives re-rendering: extract_pages applies it, so a re-import does not flip the page back."""
|
|
|
|
def pdf(self):
|
|
import pymupdf
|
|
d = Path(tempfile.mkdtemp(prefix="ttvc-pdf-"))
|
|
doc = pymupdf.open()
|
|
page = doc.new_page(width=300, height=200)
|
|
page.insert_text((20, 50), "A sentence with enough words to count as native text on this page, repeated. " * 3, fontsize=8)
|
|
doc.save(str(d / "source.pdf"))
|
|
doc.close()
|
|
return d
|
|
|
|
def test_extract_pages_renders_a_stored_turn_and_ignores_the_unturned_text_layer(self):
|
|
d = self.pdf()
|
|
plain = mag.extract_pages(d / "source.pdf", d / "plain", render_dpi=50)
|
|
self.assertEqual((plain["pages"][0]["width"], plain["pages"][0]["height"]), (300.0, 200.0))
|
|
self.assertTrue(any(b["text"].startswith("A sentence") for b in plain["blocks"]))
|
|
turned = mag.extract_pages(d / "source.pdf", d / "turned", render_dpi=50, page_rotations={1: 90})
|
|
self.assertEqual((turned["pages"][0]["width"], turned["pages"][0]["height"]), (200.0, 300.0))
|
|
self.assertEqual(turned["pages"][0].get("rotation"), 90) # the stored turn is visible in the layout again
|
|
self.assertNotIn("rotation", plain["pages"][0])
|
|
with Image.open(d / "turned" / "pages" / "0001.png") as im, Image.open(d / "plain" / "pages" / "0001.png") as ref:
|
|
self.assertEqual(im.size, (ref.size[1], ref.size[0]))
|
|
# The PDF's own text boxes belong to the unturned page; on a turned page only OCR of the render counts.
|
|
self.assertFalse(any(b["source"] == "native" or b["text"].startswith("A sentence") for b in turned["blocks"]))
|
|
|
|
def test_no_stored_turn_changes_nothing(self):
|
|
d = self.pdf()
|
|
a = mag.extract_pages(d / "source.pdf", d / "a", render_dpi=50)
|
|
b = mag.extract_pages(d / "source.pdf", d / "b", render_dpi=50, page_rotations={})
|
|
self.assertEqual(a["pages"], b["pages"])
|
|
self.assertEqual([x["text"] for x in a["blocks"]], [x["text"] for x in b["blocks"]])
|
|
|
|
|
|
class ImageVersionAndJobGuardTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
import routes.magazine as routes
|
|
self.routes = routes
|
|
self.lib = Path(tempfile.mkdtemp(prefix="ttvc-lib-"))
|
|
p = mock.patch.object(routes, "_LIB", self.lib)
|
|
p.start(); self.addCleanup(p.stop)
|
|
app = FastAPI(); app.include_router(routes.router)
|
|
self.client = TestClient(app)
|
|
self.doc = self.client.post("/api/magazine/docs", json={"title": "T"}).json()["id"]
|
|
self.d = self.lib / self.doc
|
|
(self.d / "pages").mkdir()
|
|
self.png = self.d / "pages" / "0002.png"
|
|
Image.new("RGB", (50, 80), "white").save(self.png)
|
|
(self.d / "layout.json").write_text(json.dumps({"pages": [{"page": 2, "width": 50.0, "height": 80.0}, {"page": 3, "width": 50.0, "height": 80.0}],
|
|
"blocks": [], "images": []}))
|
|
|
|
def test_each_page_is_stamped_with_its_picture_version_and_it_changes_with_the_picture(self):
|
|
first = self.client.get(f"/api/magazine/{self.doc}/layout").json()["pages"]
|
|
self.assertEqual(first[0]["image_v"], int(self.png.stat().st_mtime))
|
|
self.assertEqual(first[1]["image_v"], 0) # page 3 has no picture: 0, not an error
|
|
os.utime(self.png, (self.png.stat().st_atime, self.png.stat().st_mtime + 100))
|
|
again = self.client.get(f"/api/magazine/{self.doc}/layout").json()["pages"]
|
|
self.assertEqual(again[0]["image_v"], first[0]["image_v"] + 100)
|
|
|
|
def test_the_stamp_is_also_there_when_one_page_is_asked_for(self):
|
|
pages = self.client.get(f"/api/magazine/{self.doc}/layout?page=2").json()["pages"]
|
|
self.assertEqual([p["page"] for p in pages], [2])
|
|
self.assertGreater(pages[0]["image_v"], 0)
|
|
|
|
def test_the_stamp_is_not_written_into_the_saved_layout(self):
|
|
self.client.get(f"/api/magazine/{self.doc}/layout")
|
|
self.assertNotIn("image_v", json.loads((self.d / "layout.json").read_text())["pages"][0])
|
|
|
|
def job(self, **kw):
|
|
base = {"done": False, "fingerprint": f"{self.doc}:2", "updated": __import__("time").time()}
|
|
base.update(kw)
|
|
return base
|
|
|
|
def running(self, jobs, page=2):
|
|
with mock.patch.object(self.routes, "background_job_list", side_effect=lambda kind, limit=50: jobs if kind == "magazine-rotate-page" else []):
|
|
return self.routes._page_job_running(self.doc, page)
|
|
|
|
def test_a_job_in_flight_for_the_same_page_counts(self):
|
|
self.assertTrue(self.running([self.job()]))
|
|
|
|
def test_finished_stale_and_other_page_jobs_do_not(self):
|
|
self.assertFalse(self.running([self.job(done=True)]))
|
|
self.assertFalse(self.running([self.job(updated=__import__("time").time() - 3600)])) # died with the server
|
|
self.assertFalse(self.running([self.job(fingerprint=f"{self.doc}:3")]))
|
|
self.assertFalse(self.running([self.job(fingerprint="other:2")]))
|
|
self.assertFalse(self.running([]))
|
|
|
|
def test_a_second_turn_or_re_read_while_one_runs_is_refused_and_changes_nothing(self):
|
|
before = self.png.read_bytes()
|
|
settings = {"vision_ocr_url": "http://ocr", "vision_ocr_model": "m"}
|
|
with mock.patch.object(self.routes, "_load_settings", return_value=settings), \
|
|
mock.patch.object(self.routes, "_page_job_running", return_value=True), \
|
|
mock.patch.object(self.routes.asyncio, "create_task") as task:
|
|
turn = self.client.post(f"/api/magazine/{self.doc}/pages/2/rotate", json={"degrees": 90})
|
|
reread = self.client.post(f"/api/magazine/{self.doc}/pages/2/reocr", json={})
|
|
self.assertEqual((turn.status_code, reread.status_code), (409, 409))
|
|
self.assertIn("wait for it to finish", turn.json()["detail"])
|
|
task.assert_not_called()
|
|
self.assertEqual(self.png.read_bytes(), before)
|
|
|
|
|
|
class EndpointTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
import routes.magazine as routes
|
|
self.routes = routes
|
|
self.lib = Path(tempfile.mkdtemp(prefix="ttvc-lib-"))
|
|
self.patches = [mock.patch.object(routes, "_LIB", self.lib)]
|
|
for p in self.patches:
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
app = FastAPI(); app.include_router(routes.router)
|
|
self.client = TestClient(app)
|
|
self.doc = self.client.post("/api/magazine/docs", json={"title": "T"}).json()["id"]
|
|
self.d = self.lib / self.doc
|
|
(self.d / "pages").mkdir()
|
|
Image.new("RGB", (50, 80), (255, 255, 255)).save(self.d / "pages" / "0002.png")
|
|
(self.d / "layout.json").write_text(json.dumps({"pages": [{"page": 2, "width": 50.0, "height": 80.0}], "blocks": [], "images": []}))
|
|
|
|
def post(self, degrees, page=2, vision=True):
|
|
settings = {"vision_ocr_url": "http://ocr", "vision_ocr_model": "m"} if vision else {}
|
|
created = []
|
|
|
|
async def fake_task(coro):
|
|
created.append(coro)
|
|
with mock.patch.object(self.routes, "_load_settings", return_value=settings), \
|
|
mock.patch.object(self.routes.asyncio, "create_task", side_effect=lambda c: (c.close(), created.append(1), mock.MagicMock())[2]):
|
|
return self.client.post(f"/api/magazine/{self.doc}/pages/{page}/rotate", json={"degrees": degrees}), created
|
|
|
|
def test_a_valid_request_starts_a_job(self):
|
|
r, created = self.post(90)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
self.assertEqual(r.json()["degrees"], 90)
|
|
self.assertEqual(len(created), 1)
|
|
|
|
def test_minus_ninety_means_two_seventy(self):
|
|
self.assertEqual(self.post(-90)[0].json()["degrees"], 270)
|
|
|
|
def test_bad_degrees_are_refused_and_nothing_starts(self):
|
|
for bad in (0, 45, "sideways", None, 360):
|
|
r, created = self.post(bad)
|
|
self.assertEqual(r.status_code, 400, bad)
|
|
self.assertEqual(created, [])
|
|
|
|
def test_a_page_that_is_not_there_is_404(self):
|
|
self.assertEqual(self.post(90, page=7)[0].status_code, 404)
|
|
|
|
def test_no_vision_engine_is_a_clear_400_and_the_page_is_untouched(self):
|
|
before = (self.d / "pages" / "0002.png").read_bytes()
|
|
r, created = self.post(90, vision=False)
|
|
self.assertEqual(r.status_code, 400)
|
|
self.assertIn("vision OCR engine", r.json()["detail"])
|
|
self.assertEqual((self.d / "pages" / "0002.png").read_bytes(), before)
|
|
self.assertEqual(created, [])
|
|
|
|
def test_page_images_are_revalidated_so_a_turned_page_shows_at_once(self):
|
|
r = self.client.get(f"/api/magazine/{self.doc}/pages/2")
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(r.headers["cache-control"], "no-cache")
|
|
|
|
def test_stored_turns_reach_ingestion(self):
|
|
meta = {"page_rotations": {"3": 90, "4": "180", "5": 45, "x": 90, "0": 90}}
|
|
self.assertEqual(self.routes._page_rotations(meta), {3: 90, 4: 180})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|