from __future__ import annotations import shutil import tempfile import unittest from pathlib import Path from unittest.mock import patch from PIL import Image, ImageDraw from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, read_json from copienator.annotation_data import AnnotationLoadResult from copienator.commands import annotating_by_label as grouped from copienator.commands import annotating_with_checks as checks from copienator.commands import export, import_annotations from copienator.commands import reading_grouped_annotations as reader class GroupedRedoTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) / "Exam" for directory in ("Copies", "Par label", "BGnot", "BRnot"): (self.root / directory).mkdir(parents=True) (self.root / "labels").write_text("Ex 1\nEx 2\n") atomic_write_json(self.root / "correction.json", {}) atomic_write_json( self.root / "refaire.json", [["Copie01", ["Ex 1"]], ["Copie02", ["Ex 1"]]] ) (self.root / "BRnot/previous.txt").write_text("previous redo") (self.root / "BGnot/main.txt").write_text("main run") self.workspace = EvaluationWorkspace(self.root) self.data = {"01": {"Ex 1": {}}, "02": {"Ex 1": {}}} def tearDown(self): self.temp.cleanup() @staticmethod def render(item): student_id, label, _content = item image = Image.new("RGB", (100, 100), "white") ImageDraw.Draw(image).rectangle((10, 10, 50, 50), outline="black", width=2) return ( student_id, label, image, 0, [ { "type": "score", "label": label, "value": 3, "final_box": [10, 10, 50, 50], } ], ) def generate(self): with ( patch.object( grouped, "load_annotation_data", return_value=AnnotationLoadResult(self.data, []), ) as load, patch.object(grouped, "render_item", side_effect=self.render), patch.object(grouped, "_load_label_groups") as label_groups, ): self.assertEqual( grouped.run(self.workspace, refaire=True, overwrite=True), ExitCode.SUCCESS, ) self.assertEqual( load.call_args.kwargs["refaire_list"], read_json(self.root / "refaire.json"), ) label_groups.assert_not_called() directories = list((self.root / "BRnot").iterdir()) self.assertEqual(len(directories), 1) self.assertTrue(directories[0].is_dir()) self.assertEqual((self.root / "BGnot/main.txt").read_text(), "main run") return directories[0] def test_grouped_redo_export_import_and_actual_annotation_detection(self): directory = self.generate() metadata = read_json(directory / "bnote.json")["images"] self.assertEqual( [(item["id"], item["label"]) for item in metadata], [("01", "Ex 1"), ("02", "Ex 1")], ) export_root = Path(self.temp.name) / "Export" with patch.object(export, "EXPORT_DIR", export_root): self.assertEqual(export.run(self.workspace, refaire=True), ExitCode.SUCCESS) self.assertEqual(len(list((export_root / "Exam").glob("*.pdf"))), 1) imported_root = Path(self.temp.name) / "Import" imported_root.mkdir() with Image.open(directory / "Reference.jpg") as reference: annotated = reference.convert("RGB") draw = ImageDraw.Draw(annotated) draw.rectangle((17, 17, 43, 43), fill="black") draw.rectangle((70, 170, 90, 190), fill="black") annotated.save(imported_root / f"{directory.name}.pdf", "PDF", resolution=72) with patch.object(import_annotations, "IMPORT_DIR", imported_root): self.assertEqual( import_annotations.run(self.workspace, refaire=True), ExitCode.SUCCESS ) actions, notes, incomplete = reader._scan_redo_annotations( self.root / "BRnot", {"01": {"Ex 1"}, "02": {"Ex 1"}} ) self.assertFalse(incomplete) self.assertEqual(actions["01"][0]["value"], 3) self.assertFalse(actions.get("02")) self.assertIn("Ex 1", notes["02"]) full_data = {student: {"Ex 1": {}, "Ex 2": {}} for student in ("01", "02")} for mode in ("BGnot", "Bnot", "Anot"): (self.root / mode).mkdir(exist_ok=True) with ( self.subTest(mode=mode), patch.object( reader, "load_annotation_data", return_value=AnnotationLoadResult(full_data, []), ), patch.object( reader, "apply_actions_and_regenerate_grouped", return_value=(ExitCode.SUCCESS, ""), ) as regenerate, ): self.assertEqual( reader.run(self.workspace, refaire=True, annotation_dir=mode), ExitCode.SUCCESS, ) calls = {call.args[2]: call for call in regenerate.call_args_list} self.assertEqual(set(calls), {"01", "02"}) self.assertEqual(set(calls["01"].args[1]["01"]), {"Ex 1", "Ex 2"}) self.assertEqual(calls["01"].args[3][0]["value"], 3) self.assertIn("Ex 1", calls["02"].args[4]) def test_missing_group_leaves_affected_copy_incomplete(self): directory = self.generate() shutil.copy2(directory / "Concat.pdf", directory / "Concat_annotated.pdf") extra = self.root / "BRnot/Ex 2 G1" extra.mkdir() atomic_write_json( extra / "bnote.json", {"images": [{"id": "01", "label": "Ex 2"}]} ) _actions, _notes, incomplete = reader._scan_redo_annotations( self.root / "BRnot", {"01": {"Ex 1", "Ex 2"}, "02": {"Ex 1"}} ) self.assertEqual(incomplete, {"01"}) def test_failed_generation_preserves_previous_redo_for_both_layouts(self): for module, worker in ((grouped, "render_item"), (checks, "_render_student")): with ( self.subTest(module=module), patch.object( module, "load_annotation_data", return_value=AnnotationLoadResult(self.data, []), ), patch.object( module, worker, return_value=None if module is grouped else "partial", ), ): if module is grouped: status = module.run(self.workspace, refaire=True, overwrite=True) else: status = module.run( self.workspace, self.root, refaire=True, overwrite=True ) self.assertEqual(status, ExitCode.PARTIAL) self.assertEqual( (self.root / "BRnot/previous.txt").read_text(), "previous redo" ) def test_switching_to_per_copy_removes_previous_group_layout(self): self.generate() def render(_workspace, student_id, _labels, **kwargs): output = kwargs["output_root"] / f"Copie{student_id}" output.mkdir() (output / "Concat.pdf").touch() return "success" with ( patch.object( checks, "load_annotation_data", return_value=AnnotationLoadResult(self.data, []), ), patch.object(checks, "_render_student", side_effect=render), ): self.assertEqual( checks.run(self.workspace, self.root, refaire=True, overwrite=True), ExitCode.SUCCESS, ) self.assertEqual( {path.name for path in (self.root / "BRnot").iterdir()}, {"Copie01", "Copie02"}, ) if __name__ == "__main__": unittest.main()