import copy import tempfile import unittest from pathlib import Path from unittest.mock import Mock, patch import pymupdf from copienator import CliError, EvaluationWorkspace, atomic_write_json, read_json from copienator.commands import correction, resolve_manual from copienator.answer_origins import origins_for, save_origins from copienator.annotation_layout import save_parts, answer_parts, layout_for, page_parts from copienator.annotation_data import load_annotation_data from copienator_gui.app import has_manual_conflicts class ManualResolutionStateTests(unittest.TestCase): def test_merge_preserves_low_margin_and_page_order(self): for pipe_first in (False, True): with self.subTest(pipe_first=pipe_first), tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), '->') workspace.manual_resolutions_file.write_text( 'Copie01 A -> |B\n' if pipe_first else 'Copie01 A -> B|\n') copies = workspace.copies_dir / 'Copie01' for label, status in (('A', 'low'), ('B', 'normal')): pdf = copies / f'{label}.pdf' save_parts(pdf, [{'margin_status': status} for _ in page_parts(pdf)]) resolve_manual.resolve_manual(workspace) target = resolve_manual.get_actual_pdf(workspace.copies_dir, '01', 'B') self.assertEqual(layout_for(target)['margin_left'], 400) statuses = [part['margin_status'] for part in answer_parts(target)] self.assertEqual(statuses[0], 'low' if pipe_first else 'normal') self.assertEqual(statuses[-1], 'normal' if pipe_first else 'low') def test_automatic_additional_content_records_origin_but_wrong_label_does_not(self): for kind in ("additional-answer", "wrong-label"): with self.subTest(kind=kind), tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), "->") copies = workspace.copies_dir / "Copie01" (copies / "B.pdf").unlink() (workspace.root / "labels").write_text("A\nB\n") with ( patch.object(correction, "INPUT_DIR", workspace.root, create=True), patch.object(correction, "COPIES_DIR", workspace.copies_dir, create=True), patch.object(correction, "GROUPS_DIR", workspace.groups_dir, create=True), patch.object(correction, "enonce_total", return_value=""), patch.object(correction.prompting, "request_for_additional_answer", return_value=([], {})), patch.object(correction.prompting, "request_for_wrong_label", return_value=([], {})), patch.object(correction, "call_gemini_with_retries", return_value='["B"]' if kind == "additional-answer" else '"B"'), patch.object(correction, "reserve_next_group_idx", return_value=0), patch.object(correction.grouping, "get_pdf_height", return_value=200), patch.object(correction.grouping, "create_jpg"), ): correction.handle_label_errors("01", "A", {"error": kind}, copies / "A.pdf") self.assertEqual( origins_for(workspace.root, "01", "B", copies / "B_new.pdf"), ["A"] if kind == "additional-answer" else [], ) def test_target_provenance_only_for_uncut_retained_sources(self): for operator in (*resolve_manual.OPERATORS, "c{50}1>", "c{50}2x"): with self.subTest(operator=operator), tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), operator) resolve_manual.resolve_manual(workspace) target = resolve_manual.get_actual_pdf(workspace.copies_dir, "01", "B") self.assertEqual( origins_for(workspace.root, "01", "B", target), ["A"] if operator in ("->", "-x") else [], ) def test_provenance_survives_new_correction_but_not_replaced_pdf(self): with tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), "->") target = workspace.copies_dir / "Copie01" / "B.pdf" save_origins(workspace.root, "01", "B", target, ["A", "A"]) atomic_write_json(workspace.correction_file, { "B": [[{"id": "01", "result": {"score": 4, "error": "", "feedback": []}}]] }) self.assertEqual(origins_for(workspace.root, "01", "B", target), ["A"]) loaded = load_annotation_data(workspace) self.assertEqual(loaded.data["01"]["B"]["result"]["content_origins"], ["A"]) target.write_bytes(b"replacement") self.assertEqual(origins_for(workspace.root, "01", "B", target), []) def test_merge_into_missing_target_uses_source_and_schedules_redo(self): for operator in ("->", "x>"): for target in ("B|", "|B"): with self.subTest(operator=operator, target=target), tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), operator) copies = workspace.copies_dir / "Copie01" (copies / "B.pdf").unlink() workspace.manual_resolutions_file.write_text(f"Copie01 A {operator} {target}\n") resolve_manual.resolve_manual(workspace) with pymupdf.open(copies / "B_new.pdf") as document: self.assertEqual(len(document), 1) self.assertIn("A", document[0].get_text()) self.assertEqual((copies / "A_old.pdf").exists(), operator == "x>") self.assertEqual((copies / "A.pdf").exists(), operator == "->") self.assertFalse((copies / "B_old.pdf").exists()) self.assertEqual(read_json(workspace.refaire_file), [["Copie01", ["B"]]]) self.assertNotIn("delayed", read_json(workspace.correction_file)["A"][0][0]["result"]) def test_merge_still_requires_source_without_changing_state(self): with tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), "x>") (workspace.copies_dir / "Copie01" / "A.pdf").unlink() before = workspace.correction_file.read_bytes() with self.assertRaises(CliError): resolve_manual.resolve_manual(workspace) self.assertEqual(workspace.correction_file.read_bytes(), before) self.assertTrue(workspace.manual_resolutions_file.exists()) def test_acknowledgement_clears_only_this_target_and_copy(self): for error in ("wrg-lbl:B?", "wrg-lbl:B?delayed", "wrg-lbl:B?exists", "al:(->)B?(->)C?", "al:(delayed)B"): with self.subTest(error=error): source = {"id": "01", "result": {"error": error, "delayed": [ ["wrong-label", "B"], ["add-label", "B"], ["add-label", "C"]]}} other = {"id": "02", "result": {"error": error, "delayed": [["wrong-label", "B"]]}} before_other = copy.deepcopy(other) results = {"A": [[source, other]]} resolve_manual.set_suffix_and_clean_error(results, "01", "A", None, "B") self.assertEqual(source["result"]["delayed"], [["add-label", "C"]]) self.assertNotIn("B?", source["result"]["error"]) self.assertEqual(other, before_other) resolve_manual.set_suffix_and_clean_error(results, "01", "A", None, "C") self.assertNotIn("delayed", source["result"]) def fixture(self, root, operator): copies = root / "Copies" / "Copie01" copies.mkdir(parents=True) for label in ("A", "B"): with pymupdf.open() as doc: page = doc.new_page(width=200, height=200) page.insert_text((20, 40), label) doc.save(copies / f"{label}.pdf") atomic_write_json(root / "correction.json", { "A": [[{"id": "01", "result": {"error": "wrg-lbl:B?", "delayed": [["wrong-label", "B"]]}}]], "B": [[{"id": "01", "result": {"error": ""}}]], }) (root / "manual_resolutions.txt").write_text(f"Copie01 A {operator} B|\n") return EvaluationWorkspace(root) def test_every_resolution_acknowledges_pending_conflict(self): for operator in (*resolve_manual.OPERATORS, "c{50}1>", "c{50}2x"): with self.subTest(operator=operator), tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), operator) resolve_manual.resolve_manual(workspace) result = read_json(workspace.correction_file)["A"][0][0]["result"] self.assertNotIn("delayed", result) self.assertFalse(workspace.manual_resolutions_file.exists()) def test_refaire_does_not_recreate_a_resolved_source_conflict(self): with tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), "x>") resolve_manual.resolve_manual(workspace) (workspace.groups_dir / "B").mkdir(parents=True) args = correction.build_parser().parse_args([str(workspace.root), "--refaire"]) correction.configure_runtime(workspace, [], args, api_client=Mock()) with patch.object(correction.grouping, "get_pdf_height", return_value=400), patch.object( correction.grouping, "create_jpg" ), patch.object(correction, "process_single_task", return_value=[]): self.assertEqual(correction.run_configured(args), 0) self.assertFalse(workspace.manual_resolutions_file.exists()) self.assertNotIn("delayed", correction.results["A"][0][0]["result"]) def test_deleted_conflict_stays_resolved_when_other_resolution_generates_refaire(self): for kind in ("wrong-label", "add-label"): with self.subTest(kind=kind), tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), "x>") other = workspace.copies_dir / "Copie02" other.mkdir() for label in ("A", "B"): (other / f"{label}.pdf").write_bytes( (workspace.copies_dir / "Copie01" / f"{label}.pdf").read_bytes()) before = {path.name: path.read_bytes() for path in other.iterdir()} results = read_json(workspace.correction_file) results["A"][0].append({"id": "02", "result": { "error": "wrg-lbl:B?" if kind == "wrong-label" else "al:(->)B?", "delayed": [[kind, "B"]], "score": 3, }}) atomic_write_json(workspace.correction_file, results) # Only Copie01 remains in the edited file; Copie02 was deleted. resolve_manual.resolve_manual(workspace) self.assertEqual(read_json(workspace.refaire_file), [["Copie01", ["B"]]]) self.assertEqual(before, {path.name: path.read_bytes() for path in other.iterdir()}) saved = read_json(workspace.correction_file)["A"][0][1]["result"] self.assertNotIn("delayed", saved) self.assertEqual(saved["score"], 3) self.assertNotIn("B?", saved["error"]) (workspace.groups_dir / "B").mkdir(parents=True) args = correction.build_parser().parse_args([str(workspace.root), "--refaire"]) correction.configure_runtime(workspace, [], args, api_client=Mock()) with patch.object(correction.grouping, "get_pdf_height", return_value=400), \ patch.object(correction.grouping, "create_jpg"), \ patch.object(correction, "process_single_task", return_value=[]): self.assertEqual(correction.run_configured(args), 0) self.assertFalse(workspace.manual_resolutions_file.exists()) self.assertNotIn("delayed", correction.results["A"][0][1]["result"]) def test_deleting_all_lines_can_be_applied_and_survives_resume(self): for text in ("", "### Instructions\n", "### Copie01 A x> B|\n"): with self.subTest(text=text), tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), "x>") source = workspace.copies_dir / "Copie01" before = {path.name: path.read_bytes() for path in source.iterdir()} workspace.manual_resolutions_file.write_text(text) self.assertTrue(has_manual_conflicts(workspace.manual_resolutions_file)) resolve_manual.resolve_manual(workspace) self.assertEqual(before, {path.name: path.read_bytes() for path in source.iterdir()}) self.assertFalse(has_manual_conflicts(workspace.manual_resolutions_file)) atomic_write_json(workspace.refaire_file, []) args = correction.build_parser().parse_args([str(workspace.root), "--refaire"]) correction.configure_runtime(workspace, [], args, api_client=Mock()) self.assertEqual(correction.run_configured(args), 0) self.assertFalse(workspace.manual_resolutions_file.exists()) def test_invalid_remaining_instruction_does_not_commit_deletions(self): with tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), "x>") before = workspace.correction_file.read_bytes() workspace.manual_resolutions_file.write_text("invalid instruction\n") with self.assertRaises(CliError): resolve_manual.resolve_manual(workspace) self.assertEqual(workspace.correction_file.read_bytes(), before) def test_generated_first_comment_documents_deleting_lines(self): with tempfile.TemporaryDirectory() as temporary: workspace = self.fixture(Path(temporary), "x>") atomic_write_json(workspace.refaire_file, []) args = correction.build_parser().parse_args([str(workspace.root), "--refaire"]) correction.configure_runtime(workspace, [], args, api_client=Mock()) self.assertEqual(correction.run_configured(args), 0) first_line = workspace.manual_resolutions_file.read_text().splitlines()[0] self.assertTrue(first_line.startswith("###")) self.assertIn("delete a line", first_line) self.assertIn("resolve-manual", first_line)