219 lines
10 KiB
Python
219 lines
10 KiB
Python
import contextlib
|
|
import io
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pymupdf
|
|
|
|
from copienator.cli import CliError
|
|
from copienator.commands import crop_margins
|
|
from copienator.commands.clean import build_cleanup_plan, apply_cleanup
|
|
from copienator.dispatcher import main
|
|
from copienator.workspace import EvaluationWorkspace
|
|
from copienator_gui.workflow import build_workflow
|
|
|
|
|
|
class CropMarginsCommandTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.workspace = EvaluationWorkspace(Path(self.temp.name)/"Évaluation")
|
|
self.workspace.copies_dir.mkdir(parents=True)
|
|
self.source = self.workspace.copies_dir/"Copie01.pdf"
|
|
with pymupdf.open() as doc:
|
|
page = doc.new_page(width=300, height=420)
|
|
page.insert_text((50, 180), "answer = 42", fontsize=15)
|
|
doc.new_page(width=300, height=420)
|
|
doc.save(self.source)
|
|
self.original = self.source.read_bytes()
|
|
|
|
def test_dispatcher_replaces_copies_and_keeps_recoverable_original(self):
|
|
with contextlib.redirect_stdout(io.StringIO()) as log:
|
|
self.assertEqual(main(["crop-margins", str(self.workspace.root)]), 0)
|
|
self.assertIn("Page 2/2", log.getvalue())
|
|
self.assertIn("1/2 pages rognées", log.getvalue())
|
|
self.assertIn("Rognage moyen des pages modifiées", log.getvalue())
|
|
self.assertIn("page(s) rognée(s) de plus de 30 %", log.getvalue())
|
|
with pymupdf.open(self.source) as result:
|
|
self.assertEqual(len(result), 2)
|
|
self.assertLess(result[0].rect.height, 150)
|
|
self.assertIn("answer = 42", result[0].get_text())
|
|
self.assertEqual(result[1].rect.height, 420)
|
|
for page in result:
|
|
self.assertEqual(page.rect.width, 300)
|
|
page.get_pixmap()
|
|
backups = list(self.workspace.runs_dir.glob("crop-margins-*/Copies/Copie01.pdf"))
|
|
self.assertEqual(len(backups), 1)
|
|
self.assertEqual(backups[0].read_bytes(), self.original)
|
|
self.assertTrue((backups[0].parent.parent/"report.json").is_file())
|
|
|
|
def test_crop_statistics_use_only_modified_pages(self):
|
|
records = [
|
|
{
|
|
"top_removed_mm": 10.0,
|
|
"bottom_removed_mm": 20.0,
|
|
"original_cropbox": [0, 0, 200, 300],
|
|
"rotation": 0,
|
|
},
|
|
{
|
|
"top_removed_mm": 40.0,
|
|
"bottom_removed_mm": 0.0,
|
|
"original_cropbox": [0, 0, 200, 300],
|
|
"rotation": 90,
|
|
},
|
|
{
|
|
"top_removed_mm": 0.0,
|
|
"bottom_removed_mm": 0.0,
|
|
"original_cropbox": [0, 0, 200, 300],
|
|
"rotation": 0,
|
|
},
|
|
]
|
|
count, average, over_thirty = crop_margins.crop_statistics(records)
|
|
self.assertEqual(count, 2)
|
|
self.assertAlmostEqual(average, (28.3465 + 56.6929) / 2, places=3)
|
|
self.assertEqual(over_thirty, 1)
|
|
|
|
def test_failure_or_interruption_never_publishes_partial_batch(self):
|
|
second = self.workspace.copies_dir/"Copie02.pdf"
|
|
second.write_bytes(self.original)
|
|
real = crop_margins.process_pdf
|
|
for error in (OSError("bad scan"), KeyboardInterrupt()):
|
|
with self.subTest(error=type(error).__name__):
|
|
def process(source, *args, **kwargs):
|
|
if source == second:
|
|
raise error
|
|
return real(source, *args, **kwargs)
|
|
with patch.object(crop_margins, "process_pdf", side_effect=process):
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
with self.assertRaises(type(error)):
|
|
crop_margins.run(self.workspace, self.workspace.root, workers=1)
|
|
self.assertEqual(self.source.read_bytes(), self.original)
|
|
self.assertEqual(second.read_bytes(), self.original)
|
|
|
|
def test_existing_coordinates_are_not_silently_invalidated(self):
|
|
self.source.with_suffix(".json").write_text('{"list": []}')
|
|
with self.assertRaisesRegex(CliError, "coordonnées"):
|
|
crop_margins.run(self.workspace, self.workspace.root)
|
|
self.assertEqual(self.source.read_bytes(), self.original)
|
|
|
|
def test_archiving_removes_crop_backups_but_keeps_processed_pdf_and_logs(self):
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
crop_margins.run(self.workspace, self.workspace.root)
|
|
processed = self.source.read_bytes()
|
|
self.workspace.correction_file.write_text('{}')
|
|
student = self.workspace.return_dir/"Student"
|
|
student.mkdir(parents=True)
|
|
(student/"answer.jpg").write_bytes(b"return image")
|
|
(student/"score.json").write_text('{}')
|
|
self.workspace.logs_dir.mkdir(parents=True)
|
|
log = self.workspace.logs_dir/"crop_blank_margins.log"
|
|
log.write_text("Completed crop")
|
|
backups = list(self.workspace.runs_dir.glob("crop-margins-*/Copies/*.pdf"))
|
|
reports = list(self.workspace.runs_dir.glob("crop-margins-*/report.json"))
|
|
self.assertTrue(backups)
|
|
self.assertTrue(reports)
|
|
plan = build_cleanup_plan(self.workspace)
|
|
for path in backups+reports:
|
|
self.assertIn(path, plan.deleted_files)
|
|
self.assertIn(self.source, plan.kept_files)
|
|
self.assertIn(log, plan.kept_files)
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
apply_cleanup(self.workspace, plan)
|
|
self.assertFalse(self.workspace.runs_dir.exists())
|
|
self.assertEqual(self.source.read_bytes(), processed)
|
|
self.assertEqual(log.read_text(), "Completed crop")
|
|
self.assertTrue((student/"answer.jpg").exists())
|
|
self.assertTrue((student/"score.json").exists())
|
|
|
|
def test_single_copy_selection_cannot_target_original_scans(self):
|
|
self.assertEqual(crop_margins.selected_files(self.workspace, self.source), [self.source])
|
|
original = self.workspace.root/"Original.pdf"
|
|
original.write_bytes(self.original)
|
|
with self.assertRaises(CliError):
|
|
crop_margins.selected_files(self.workspace, original)
|
|
|
|
def test_parallel_workers_match_serial_results_in_copy_and_page_order(self):
|
|
second = self.workspace.copies_dir/"Copie02.pdf"
|
|
second.write_bytes(self.original)
|
|
serial, parallel = self.workspace.root/"serial", self.workspace.root/"parallel"
|
|
serial.mkdir()
|
|
parallel.mkdir()
|
|
files = [self.source, second]
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
expected = crop_margins.process_copies(files, serial, 1)
|
|
actual = crop_margins.process_copies(files, parallel, 2)
|
|
self.assertEqual(actual, expected)
|
|
self.assertEqual([(r['file'],r['page']) for r in actual],
|
|
[(p.name,i) for p in files for i in (1,2)])
|
|
for source in files:
|
|
with pymupdf.open(serial/source.name) as a, pymupdf.open(parallel/source.name) as b:
|
|
self.assertEqual([list(p.cropbox) for p in a], [list(p.cropbox) for p in b])
|
|
|
|
def test_worker_failure_does_not_replace_any_copy(self):
|
|
broken = self.workspace.copies_dir/"Copie02.pdf"
|
|
broken.write_bytes(b"not a PDF")
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
with self.assertRaises(Exception):
|
|
crop_margins.run(self.workspace, self.workspace.root, workers=2)
|
|
self.assertEqual(self.source.read_bytes(), self.original)
|
|
self.assertEqual(broken.read_bytes(), b"not a PDF")
|
|
self.assertFalse(list(self.workspace.root.glob(".Copies.*.files.tmp")))
|
|
|
|
def test_invalid_worker_count_is_rejected_before_processing(self):
|
|
with self.assertRaises(CliError):
|
|
crop_margins.run(self.workspace, self.workspace.root, workers=0)
|
|
self.assertEqual(self.source.read_bytes(), self.original)
|
|
|
|
@unittest.skipIf(os.name == "nt", "SIGINT subprocess check uses Unix signals")
|
|
def test_interrupt_stops_parallel_workers_before_removing_staging(self):
|
|
import select
|
|
with pymupdf.open() as doc:
|
|
for _ in range(30):
|
|
page = doc.new_page(width=595, height=842)
|
|
page.insert_text((100,400), "answer = 42", fontsize=20)
|
|
data = doc.tobytes()
|
|
self.source.write_bytes(data)
|
|
second = self.workspace.copies_dir/"Copie02.pdf"
|
|
second.write_bytes(data)
|
|
proc = subprocess.Popen(
|
|
[sys.executable,"-u","-m","copienator","crop-margins",
|
|
str(self.workspace.root),"--workers","2"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
|
|
cwd=Path(__file__).resolve().parents[1])
|
|
try:
|
|
while True:
|
|
ready, _, _ = select.select([proc.stdout], [], [], 20)
|
|
self.assertTrue(ready, "No progress from parallel crop command")
|
|
line = proc.stdout.readline()
|
|
self.assertTrue(line, "Crop command exited before processing a page")
|
|
if "Page " in line:
|
|
break
|
|
proc.send_signal(signal.SIGINT)
|
|
output, _ = proc.communicate(timeout=20)
|
|
self.assertEqual(proc.returncode, 130, output)
|
|
finally:
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
proc.communicate()
|
|
self.assertEqual(self.source.read_bytes(), data)
|
|
self.assertEqual(second.read_bytes(), data)
|
|
self.assertFalse(list(self.workspace.root.glob(".Copies.*.files.tmp")))
|
|
|
|
def test_optional_step_sits_between_page_split_and_label_crop(self):
|
|
steps = build_workflow(False)
|
|
index = next(i for i, step in enumerate(steps) if step.id == "crop_blank_margins")
|
|
self.assertEqual(steps[index-1].id, "page_splitter")
|
|
self.assertEqual(steps[index+1].id, "cutleft")
|
|
self.assertTrue(steps[index].optional)
|
|
self.assertTrue(steps[index].auto_start_first_visit)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|