from __future__ import annotations import importlib.util import io import os import queue import sys import tempfile import time import unittest from concurrent.futures import ThreadPoolExecutor from contextlib import redirect_stderr from pathlib import Path from unittest.mock import patch from pypdf import PdfReader, PdfWriter from copienator import ( EvaluationWorkspace, WorkspaceNotFoundError, WorkspaceValidationError, atomic_update_json, atomic_write_json, read_json, ) from copienator_gui.app import process_status from copienator_gui.diagnostics import collect_diagnostics from copienator_gui.runner import ProcessRunner from copienator_gui.state import StateStore from copienator_gui.workflow import build_command, build_workflow, evaluation_argument from copies_tools import rename_all, rotate_all from platform_utils import ( WindowsLabelError, replace_with_link_or_copy, safe_filename, validate_windows_labels, windows_filename_problems, ) REPOSITORY = Path(__file__).resolve().parents[1] def load_script_module(filename: str, module_name: str): spec = importlib.util.spec_from_file_location(module_name, REPOSITORY / filename) if spec is None or spec.loader is None: raise RuntimeError(f"Could not load {filename}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module class WorkspaceTests(unittest.TestCase): def test_canonical_paths_and_no_constructor_side_effects(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) / "Exam" root.mkdir() workspace = EvaluationWorkspace(root, Path(directory)) self.assertFalse(workspace.metadata_dir.exists()) self.assertEqual(workspace.labels_file, root / "labels") self.assertEqual(workspace.copies_dir, root / "Copies") self.assertEqual(workspace.annotation_dir("grouped"), root / "BGnot") self.assertEqual(workspace.state_database, root / ".copienator" / "state.sqlite3") self.assertEqual(workspace.command_argument(), "Exam") def test_discover_from_nested_artifact(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) / "Exam" nested = root / "Copies" / "Copie01" nested.mkdir(parents=True) (root / "labels").write_text("Ex 1\n", encoding="utf-8") artifact = nested / "Ex 1.pdf" artifact.touch() workspace = EvaluationWorkspace.discover(artifact) self.assertEqual(workspace.root, root) def test_discover_and_require_report_clear_errors(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) with self.assertRaises(WorkspaceNotFoundError): EvaluationWorkspace.discover(root) workspace = EvaluationWorkspace(root) with self.assertRaises(WorkspaceValidationError) as context: workspace.require("enonce.pdf", "labels") self.assertEqual(context.exception.missing, ["enonce.pdf", "labels"]) (root / "Copies").write_text("not a directory", encoding="utf-8") with self.assertRaises(WorkspaceValidationError): workspace.require_directories("Copies") def test_names_file_prefers_evaluation_then_repository(self) -> None: with tempfile.TemporaryDirectory() as directory: repository = Path(directory) root = repository / "Exam" root.mkdir() (repository / "names").write_text("Global\n", encoding="utf-8") workspace = EvaluationWorkspace(root, repository) self.assertEqual(workspace.names_file(), repository / "names") (root / "names").write_text("Local\n", encoding="utf-8") self.assertEqual(workspace.names_file(), root / "names") class AtomicJsonTests(unittest.TestCase): def test_atomic_round_trip_and_unicode(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "nested" / "state.json" value = {"name": "Élève", "steps": [1, 2]} atomic_write_json(path, value) self.assertEqual(read_json(path), value) self.assertEqual(list(path.parent.glob(f".{path.name}.*.tmp")), []) def test_serialization_failure_preserves_previous_file(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "state.json" atomic_write_json(path, {"version": 1}) before = path.read_bytes() with self.assertRaises(TypeError): atomic_write_json(path, {"invalid": object()}) # type: ignore[dict-item] self.assertEqual(path.read_bytes(), before) def test_locked_updates_do_not_lose_concurrent_changes(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "state.json" def increment() -> None: def update(value): value["count"] += 1 atomic_update_json( path, update, default_factory=lambda: {"count": 0}, ) with ThreadPoolExecutor(max_workers=8) as executor: list(executor.map(lambda _index: increment(), range(40))) self.assertEqual(read_json(path), {"count": 40}) def test_gui_state_uses_workspace_and_atomic_writer(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) store = StateStore() store.load(root) store.update_step("labels", status="success") self.assertEqual(store.workspace, EvaluationWorkspace(root)) persisted = read_json(root / ".copienator-gui.json") self.assertEqual(persisted["steps"]["labels"]["status"], "success") class StandardCliTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.modules = { "export": load_script_module("export.py", "copienator_export_test"), "import": load_script_module("import.py", "copienator_import_test"), "giving_names": load_script_module( "giving_names.py", "copienator_giving_names_test" ), "grouping": load_script_module("grouping.py", "copienator_grouping_test"), } def test_missing_evaluation_has_standard_exit_code(self) -> None: with tempfile.TemporaryDirectory() as directory: missing = str(Path(directory) / "missing") invocations = { "export": [missing], "import": [missing], "giving_names": [missing, "BGnot"], "grouping": [missing], } for name, arguments in invocations.items(): with self.subTest(script=name), redirect_stderr(io.StringIO()): self.assertEqual(self.modules[name].main(arguments), 3) def test_invalid_arguments_use_argparse_exit_code(self) -> None: with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as context: self.modules["grouping"].main([]) self.assertEqual(context.exception.code, 2) def test_unexpected_processing_error_has_failure_exit_code(self) -> None: module = self.modules["grouping"] with tempfile.TemporaryDirectory() as directory: with patch.object( module, "run", side_effect=RuntimeError("broken") ), redirect_stderr(io.StringIO()) as errors: self.assertEqual(module.main([directory]), 1) self.assertIn("broken", errors.getvalue()) def test_prerequisites_are_checked_before_creating_outputs(self) -> None: module = self.modules["export"] with tempfile.TemporaryDirectory() as directory: base = Path(directory) evaluation = base / "Exam" evaluation.mkdir() export_dir = base / "Export" with patch.object(module, "EXPORT_DIR", export_dir), redirect_stderr( io.StringIO() ): self.assertEqual(module.main([str(evaluation)]), 3) self.assertFalse(export_dir.exists()) def test_gui_understands_standard_process_exit_codes(self) -> None: self.assertEqual(process_status(0), "success") self.assertEqual(process_status(1), "failed") self.assertEqual(process_status(3), "failed") self.assertEqual(process_status(4), "partial") self.assertEqual(process_status(130), "interrupted") self.assertEqual(process_status(0, interrupted=True), "interrupted") def test_gui_commands_are_accepted_by_script_parsers(self) -> None: steps = {step.id: step for step in build_workflow(True)} evaluation = "Evaluation with spaces" cases = { "export": ("export", "default", {"target": evaluation, "refaire": True}), "import": ("import", "default", {"target": evaluation, "refaire": True}), "giving_names": ( "giving_names", "default", {"target": evaluation, "annotation_dir": "BGnot"}, ), "grouping": ("grouping", "default", {"target": evaluation}), } for module_name, (step_id, variant_id, values) in cases.items(): step = steps[step_id] variant = next(item for item in step.variants if item.id == variant_id) command = build_command(REPOSITORY, step, variant, values, evaluation) with self.subTest(script=module_name): parsed = self.modules[module_name].build_parser().parse_args(command[3:]) self.assertEqual(str(parsed.evaluation), evaluation) def test_export_main_copies_outputs(self) -> None: module = self.modules["export"] with tempfile.TemporaryDirectory() as directory: base = Path(directory) evaluation = base / "Exam" source = evaluation / "BGnot" / "Ex 1" source.mkdir(parents=True) (source / "Concat.pdf").write_bytes(b"annotated") with patch.object(module, "EXPORT_DIR", base / "Export"): self.assertEqual(module.main([str(evaluation)]), 0) exported = base / "Export" / "Exam" / "Ex 1.pdf" self.assertEqual(exported.read_bytes(), b"annotated") def test_import_main_copies_handwritten_annotations(self) -> None: module = self.modules["import"] with tempfile.TemporaryDirectory() as directory: base = Path(directory) evaluation = base / "Exam" target = evaluation / "BGnot" / "Ex 1" target.mkdir(parents=True) import_dir = base / "Import" import_dir.mkdir() (import_dir / "Ex 1.pdf").write_bytes(b"handwritten") with patch.object(module, "IMPORT_DIR", import_dir): self.assertEqual(module.main([str(evaluation)]), 0) self.assertEqual( (target / "Concat_annotated.pdf").read_bytes(), b"handwritten" ) def test_giving_names_main_builds_return_directory(self) -> None: module = self.modules["giving_names"] with tempfile.TemporaryDirectory() as directory: evaluation = Path(directory) / "Exam" copies = evaluation / "Copies" annotations = evaluation / "BGnot" / "Copie01" copies.mkdir(parents=True) annotations.mkdir(parents=True) atomic_write_json(copies / "Copie01.json", {"name": "Élève Test"}) atomic_write_json(annotations / "score.json", {"total": 10}) (annotations / "Concat.jpg").write_bytes(b"image") (evaluation / "names").write_text("Élève Test\n", encoding="utf-8") self.assertEqual(module.main([str(evaluation), "BGnot"]), 0) destination = evaluation / "A Rendre" / "Élève Test (01)" self.assertEqual((destination / "Élève Test.jpg").read_bytes(), b"image") self.assertEqual(read_json(destination / "score.json"), {"total": 10}) def test_grouping_main_accepts_empty_copies_directory(self) -> None: module = self.modules["grouping"] with tempfile.TemporaryDirectory() as directory: evaluation = Path(directory) / "Exam" (evaluation / "Copies").mkdir(parents=True) self.assertEqual(module.main([str(evaluation)]), 0) self.assertTrue((evaluation / "Par label").is_dir()) class WorkflowTests(unittest.TestCase): def setUp(self) -> None: self.steps = {step.id: step for step in build_workflow(True)} self.evaluation = evaluation_argument(REPOSITORY, REPOSITORY / "Example") def command(self, step_id: str, variant_id: str, values: dict[str, object]) -> list[str]: step = self.steps[step_id] variant = next(item for item in step.variants if item.id == variant_id) return build_command(REPOSITORY, step, variant, values, self.evaluation) def test_personal_steps_are_configurable(self) -> None: standard_ids = {step.id for step in build_workflow(False)} personal_ids = {step.id for step in build_workflow(True)} self.assertNotIn("update_ods", standard_ids) self.assertIn("update_ods", personal_ids) def test_live_correction_arguments(self) -> None: command = self.command( "correction", "live", {"target": self.evaluation, "overwrite": True, "limit": "0"}, ) self.assertEqual(command[3:], [self.evaluation, "--overwrite", "--limit", "0"]) self.assertIn("correction.py", command[2]) def test_hybrid_correction_arguments(self) -> None: command = self.command( "correction", "hybrid", {"target": self.evaluation, "batch_from": "Ex 4"} ) self.assertEqual(command[3:], [self.evaluation, "--batch-from", "Ex 4"]) def test_annotation_variants_are_exclusive_commands(self) -> None: command = self.command( "annotation", "grouped", {"target": self.evaluation, "overwrite": True} ) self.assertIn("annotating_by_label.py", command[2]) self.assertNotIn("annotating.py", command[2]) def test_copy_preparation_commands_are_python(self) -> None: rotate = self.command("rotate", "rotate", {"target": self.evaluation}) rename = self.command("rename", "rename", {"target": self.evaluation}) self.assertEqual(rotate[2:], [str(REPOSITORY / "copies_tools.py"), "rotate", self.evaluation]) self.assertEqual(rename[2:], [str(REPOSITORY / "copies_tools.py"), "rename", self.evaluation]) class CrossPlatformFileTests(unittest.TestCase): def test_rotate_all_skips_statement(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) for filename in ("Copie.pdf", "enonce.pdf"): writer = PdfWriter() writer.add_blank_page(width=100, height=200) with (root / filename).open("wb") as output: writer.write(output) self.assertEqual(rotate_all(root), 1) self.assertEqual(PdfReader(root / "Copie.pdf").pages[0].rotation, 180) self.assertEqual(PdfReader(root / "enonce.pdf").pages[0].rotation, 0) def test_rename_all_is_collision_safe(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / "z.pdf").write_bytes(b"z") (root / "a.pdf").write_bytes(b"a") (root / "Copie01.pdf").write_bytes(b"existing") (root / "enonce.pdf").write_bytes(b"statement") rename_all(root) self.assertEqual((root / "Copie01.pdf").read_bytes(), b"a") self.assertEqual((root / "Copie02.pdf").read_bytes(), b"existing") self.assertEqual((root / "Copie03.pdf").read_bytes(), b"z") self.assertEqual((root / "enonce.pdf").read_bytes(), b"statement") def test_link_falls_back_to_copy(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) source = root / "source.txt" destination = root / "destination.txt" source.write_text("content", encoding="utf-8") with patch("platform_utils.os.link", side_effect=OSError), patch( "platform_utils.os.symlink", side_effect=OSError ): method = replace_with_link_or_copy(source, destination) self.assertEqual(method, "copy") self.assertEqual(destination.read_text(encoding="utf-8"), "content") def test_windows_safe_student_filename(self) -> None: self.assertEqual(safe_filename('Jean: Dupont? '), "Jean Dupont") self.assertEqual(safe_filename("CON"), "_CON") self.assertEqual(safe_filename("name..."), "name") def test_windows_labels_are_rejected_without_modification(self) -> None: labels = ["Ex 1", "Ex 2 : a)", "AUX", "Ex 3."] with self.assertRaises(WindowsLabelError) as context: validate_windows_labels(labels, platform_name="nt") message = str(context.exception) self.assertIn("Ex 2 : a)", message) self.assertIn("AUX", message) self.assertIn("Ex 3.", message) self.assertEqual(labels[1], "Ex 2 : a)") def test_windows_safe_labels_are_accepted(self) -> None: validate_windows_labels(["Ex 1 - a)", "Question 2"], platform_name="nt") validate_windows_labels(["Ex 1 : a)"], platform_name="posix") self.assertTrue(windows_filename_problems("Ex 1 : a)")) def test_diagnostics_include_supported_platform(self) -> None: checks = collect_diagnostics(False, "test-key") system = next(check for check in checks if check.name == "Système") key = next(check for check in checks if check.name == "Clé Gemini") self.assertTrue(system.ok) self.assertTrue(key.ok) def test_diagnostics_detect_windows_unsafe_labels(self) -> None: with tempfile.TemporaryDirectory() as directory: evaluation = Path(directory) (evaluation / "labels").write_text("Ex 1 : a)\nEx 2\n", encoding="utf-8") checks = collect_diagnostics(False, "test-key", evaluation) labels = next(check for check in checks if check.name == "Labels compatibles Windows") self.assertFalse(labels.ok) self.assertIn("Ex 1 : a)", labels.detail) class RunnerTests(unittest.TestCase): def wait_for_finish(self, runner: ProcessRunner, timeout: float = 5) -> tuple[int, bool]: deadline = time.time() + timeout while time.time() < deadline: try: event, payload = runner.events.get(timeout=0.2) except queue.Empty: continue if event == "finished": return payload self.fail("Le sous-processus ne s’est pas terminé") def test_interactive_input_and_output(self) -> None: with tempfile.TemporaryDirectory() as directory: runner = ProcessRunner() runner.start( [ sys.executable, "-u", "-c", "value=input('réponse ? '); print('reçu:', value)", ], REPOSITORY, os.environ.copy(), Path(directory) / "runner.log", ) time.sleep(0.1) runner.send_input("oui") self.assertEqual(self.wait_for_finish(runner), (0, False)) self.assertIn("reçu: oui", (Path(directory) / "runner.log").read_text(encoding="utf-8")) if __name__ == "__main__": unittest.main()