from __future__ import annotations import tempfile import unittest from pathlib import Path from unittest.mock import patch from PIL import Image from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, read_json from copienator.annotation_data import AnnotationLoadResult, _coordinate_index from copienator.commands import reading_grouped_annotations as reader class RefaireTests(unittest.TestCase): def workspace(self, root, mode): for name in ("Copies", "Par label", mode, "BRnot/Copie01"): (root / name).mkdir(parents=True) (root / "labels").write_text("Ex 1\nEx 2\n") atomic_write_json(root / "correction.json", {}) atomic_write_json(root / "refaire.json", [["Copie01", ["Ex 2"]]]) for name in ( "bnote.json", "checkboxes.json", "Reference.jpg", "Concat_annotated.pdf", ): (root / "BRnot/Copie01" / name).touch() atomic_write_json( root / "BRnot/Copie01/bnote.json", {"images": [{"label": "Ex 2"}]} ) return EvaluationWorkspace(root) def test_reader_merges_full_copy_for_every_original_mode(self): for mode in ("BGnot", "Bnot", "Anot"): with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp: workspace = self.workspace(Path(tmp), mode) data = {"01": {"Ex 1": {}, "Ex 2": {}}, "02": {"Ex 1": {}}} def load(_workspace, data=data, **kwargs): return AnnotationLoadResult( {"01": {"Ex 2": {}}} if kwargs else data, [] ) with ( patch.object(reader, "load_annotation_data", side_effect=load), patch.object( reader, "_scan_annotation_directory", return_value=( {"01": [{"label": "Ex 2", "type": "score", "value": 3}]}, {}, ), ), patch.object( reader, "apply_actions_and_regenerate_grouped", return_value=(ExitCode.SUCCESS, ""), ) as render, ): self.assertEqual( reader.run(workspace, refaire=True, annotation_dir=mode), ExitCode.SUCCESS, ) render.assert_called_once() self.assertEqual(set(render.call_args.args[1]["01"]), {"Ex 1", "Ex 2"}) self.assertEqual(render.call_args.args[2], "01") self.assertEqual(render.call_args.kwargs["selected_labels"], {"Ex 2"}) self.assertEqual(render.call_args.kwargs["annotation_dir"], mode) def test_whole_copy_selection_replaces_all_old_actions(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) workspace = self.workspace(root, "BGnot") (root / "BGnot/Ex 1").mkdir() atomic_write_json(root / "refaire.json", [["Copie01", []]]) atomic_write_json( root / "BRnot/Copie01/bnote.json", {"images": [{"label": label} for label in ("Ex 1", "Ex 2")]}, ) loaded = AnnotationLoadResult({"01": {"Ex 1": {}, "Ex 2": {}}}, []) def scan(directory, *args, **kwargs): return ( ({"01": [{"label": "Ex 1", "type": "score", "value": 1}]}, {}) if directory.parent.name == "BGnot" else ({"01": [{"label": "Ex 2", "type": "score", "value": 4}]}, {}) ) with ( patch.object(reader, "load_annotation_data", return_value=loaded), patch.object(reader, "_scan_annotation_directory", side_effect=scan), patch.object( reader, "apply_actions_and_regenerate_grouped", return_value=(ExitCode.SUCCESS, ""), ) as render, ): self.assertEqual(reader.run(workspace, refaire=True), ExitCode.SUCCESS) self.assertEqual( render.call_args.kwargs["selected_labels"], {"Ex 1", "Ex 2"} ) self.assertEqual( render.call_args.args[3], [{"label": "Ex 2", "type": "score", "value": 4}], ) def test_stale_redo_selection_is_rejected(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) workspace = self.workspace(root, "Anot") atomic_write_json( root / "BRnot/Copie01/bnote.json", {"images": [{"label": "Ex 1"}]} ) with ( patch.object( reader, "load_annotation_data", return_value=AnnotationLoadResult({"01": {"Ex 2": {}}}, []), ), patch.object(reader, "apply_actions_and_regenerate_grouped") as render, ): self.assertEqual( reader.run(workspace, refaire=True, annotation_dir="Anot"), ExitCode.PARTIAL, ) render.assert_not_called() def test_missing_redo_return_leaves_copy_untouched(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) workspace = self.workspace(root, "BGnot") (root / "BRnot/Copie01/Concat_annotated.pdf").unlink() with ( patch.object( reader, "load_annotation_data", return_value=AnnotationLoadResult({"01": {"Ex 2": {}}}, []), ), patch.object(reader, "apply_actions_and_regenerate_grouped") as render, ): self.assertEqual(reader.run(workspace, refaire=True), ExitCode.PARTIAL) render.assert_not_called() def test_regeneration_preserves_untouched_image_and_score_and_saves_redo(self): for mode in ("BGnot", "Bnot", "Anot"): with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp: root = Path(tmp) workspace = self.workspace(root, mode) output = root / mode / "Copie01" output.mkdir() Image.new("RGB", (40, 30), "red").save(output / "Ex 1.jpg") before = (output / "Ex 1.jpg").read_bytes() atomic_write_json(output / "score.json", {"Ex 1": "3.5", "Ex 2": "0"}) answer = root / "answer.pdf" answer.touch() data = { "01": { label: { "result": {"score": 2, "feedback": []}, "pdf_path": answer, "coordinates": (0, 0), } for label in ("Ex 1", "Ex 2") } } with ( patch.object( reader.annotating, "make_base_image", return_value=(Image.new("RGB", (40, 20)), 0, 0), ), patch.object( reader.annotating, "compose_label_image", return_value=(Image.new("RGB", (40, 20), "blue"), 0), ), patch.object(reader, "get_extra_pdfs_as_images", return_value=[]), ): status, _ = reader.apply_actions_and_regenerate_grouped( workspace, data, "01", [], {}, ["Ex 1", "Ex 2"], annotation_dir=mode, selected_labels={"Ex 2"}, ) self.assertEqual(status, ExitCode.SUCCESS) self.assertEqual( read_json(output / "score.json"), {"Ex 1": "3.5", "Ex 2": "2"} ) self.assertEqual((output / "Ex 1.jpg").read_bytes(), before) self.assertTrue((output / "Ex 2.jpg").is_file()) with Image.open(output / "Concat.jpg") as concat: self.assertEqual(concat.size, (40, 50)) def test_simple_import_survives_successive_redos(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) workspace = self.workspace(root, "Anot") output = root / "Anot/Copie01" output.mkdir() for label in ("Ex 1", "Ex 2"): Image.new("RGB", (40, 30), "white").save(output / f"{label}.jpg") Image.new("RGB", (40, 60), "red").save(output / "Concat_annotated.jpg") atomic_write_json(output / "score.json", {"Ex 1": "1", "Ex 2": "1"}) answer = root / "answer.pdf" answer.touch() data = { "01": { label: { "result": {"score": 2, "feedback": []}, "pdf_path": answer, "coordinates": (0, 0), } for label in ("Ex 1", "Ex 2") } } with ( patch.object( reader.annotating, "make_base_image", return_value=(Image.new("RGB", (40, 20)), 0, 0), ), patch.object( reader.annotating, "compose_label_image", return_value=(Image.new("RGB", (40, 20), "blue"), 0), ), ): for selected in ({"Ex 2"}, {"Ex 1"}): status, _ = reader.apply_actions_and_regenerate_grouped( workspace, data, "01", [], {}, ["Ex 1", "Ex 2"], annotation_dir="Anot", selected_labels=selected, ) self.assertEqual(status, ExitCode.SUCCESS) if selected == {"Ex 2"}: with Image.open(output / "Ex 1.jpg") as untouched: self.assertGreater(untouched.getpixel((10, 10))[0], 240) with Image.open(output / "Concat.jpg") as concat: self.assertEqual(concat.size, (40, 40)) self.assertGreater(concat.getpixel((10, 30))[2], 240) def test_latest_group_coordinates_win_numerically(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) group = root / "Par label" / "Ex 1" group.mkdir(parents=True) for number, height in ((2, 100), (10, 40)): atomic_write_json( group / f"Group_{number}.json", [["01", 0, height, "", "Ex 1"]] ) Image.new("RGB", (20, height)).save(group / f"Group_{number}.jpg") index, warnings = _coordinate_index(EvaluationWorkspace(root)) self.assertFalse(warnings) self.assertEqual(index[("Ex 1", "01")].height, 40) def test_unreadable_redo_is_rejected(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) atomic_write_json(root / "bnote.json", {"images": []}) with ( patch.object( reader, "detect_checks_and_notes", return_value=([], None) ), self.assertRaises(ValueError), ): reader._scan_annotation_directory(root, required=True) if __name__ == "__main__": unittest.main()