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()) 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_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()