167 lines
10 KiB
Python
167 lines
10 KiB
Python
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_gui.app import has_manual_conflicts
|
|
|
|
|
|
class ManualResolutionStateTests(unittest.TestCase):
|
|
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)
|