From 2f1cd00e32605594c74217925dee11edf9d476c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Thu, 20 Aug 2026 12:44:58 +0200 Subject: [PATCH] EvaluationWorkspace --- Readme.org | 28 ++++ copienator/__init__.py | 25 ++++ copienator/json_io.py | 168 +++++++++++++++++++++++ copienator/workspace.py | 204 ++++++++++++++++++++++++++++ copienator_gui/app.py | 8 +- copienator_gui/state.py | 19 ++- correction.py | 17 +-- post-correction.py | 5 +- resolve_manual.py | 8 +- tests/test_gui_core.py | 294 ++++++++++++++++++++++++++++++++++++++++ utils.py | 3 +- 11 files changed, 746 insertions(+), 33 deletions(-) create mode 100644 copienator/__init__.py create mode 100644 copienator/json_io.py create mode 100644 copienator/workspace.py create mode 100644 tests/test_gui_core.py diff --git a/Readme.org b/Readme.org index fd161dd..51ed7cd 100644 --- a/Readme.org +++ b/Readme.org @@ -119,6 +119,34 @@ Les chemins des étapes personnelles peuvent être adaptés avec =CURRENT_SCORE_ODS_PATH=, =FINAL_SCORE_ODS_PATH=, =FINAL_SCORE_OUTPUT_DIR= et =FINAL_SCORE_FONT_PATH= dans =config.py=. +*** API commune pour les scripts + +Le paquet =copienator= centralise les chemins d'une évaluation et les +écritures JSON sûres. Un script ne devrait donc plus reconstruire les +chemins partagés à la main : + +#+BEGIN_SRC python +from copienator import EvaluationWorkspace, atomic_write_json + +workspace = EvaluationWorkspace("Interro") +atomic_write_json(workspace.correction_file, corrections) +#+END_SRC + +=EvaluationWorkspace.discover(path)= retrouve également la racine d'une +évaluation à partir d'un fichier ou d'un sous-dossier. Sa construction +ne crée aucun fichier. La création explicite de +=.copienator/logs/= et =.copienator/runs/= se fait avec +=workspace.ensure_control_directories()=. Le chemin +=.copienator/state.sqlite3= est réservé à une future couche d'état +transactionnelle. + +=atomic_write_json= écrit d'abord dans un fichier temporaire situé dans +le même dossier, synchronise son contenu, puis remplace la destination. +Une interruption ne laisse donc pas un JSON partiellement écrit. Pour +une modification concurrente de type lire-modifier-écrire, utiliser +=atomic_update_json= : cet utilitaire protège l'opération complète avec +un verrou inter-processus Linux/Windows. + ** Correction d'un paquet de copies 1. Créer un fichier =names= dans le dossier courant, avec les diff --git a/copienator/__init__.py b/copienator/__init__.py new file mode 100644 index 0000000..ffc8901 --- /dev/null +++ b/copienator/__init__.py @@ -0,0 +1,25 @@ +"""Core building blocks shared by Copienator scripts and interfaces.""" + +from .json_io import ( + JsonLockTimeout, + atomic_update_json, + atomic_write_json, + atomic_write_text, + read_json, +) +from .workspace import ( + EvaluationWorkspace, + WorkspaceNotFoundError, + WorkspaceValidationError, +) + +__all__ = [ + "EvaluationWorkspace", + "JsonLockTimeout", + "WorkspaceNotFoundError", + "WorkspaceValidationError", + "atomic_update_json", + "atomic_write_json", + "atomic_write_text", + "read_json", +] diff --git a/copienator/json_io.py b/copienator/json_io.py new file mode 100644 index 0000000..3585717 --- /dev/null +++ b/copienator/json_io.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import json +import os +import stat +import tempfile +import time +from collections.abc import Callable +from contextlib import contextmanager +from pathlib import Path +from typing import Any, BinaryIO, TypeVar, cast + +JsonValue = dict[str, Any] | list[Any] | str | int | float | bool | None +T = TypeVar("T", bound=JsonValue) +_MISSING = object() + + +class JsonLockTimeout(TimeoutError): + pass + + +def read_json(path: str | Path, *, default: T | object = _MISSING) -> T: + target = Path(path) + try: + with target.open("r", encoding="utf-8") as stream: + return cast(T, json.load(stream)) + except FileNotFoundError: + if default is _MISSING: + raise + return cast(T, default) + + +def _sync_directory(directory: Path) -> None: + if os.name == "nt" or not hasattr(os, "O_DIRECTORY"): + return + descriptor = os.open(directory, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _atomic_write(path: Path, payload: bytes) -> None: + path = path.expanduser() + path.parent.mkdir(parents=True, exist_ok=True) + previous_mode = None + try: + previous_mode = stat.S_IMODE(path.stat().st_mode) + except FileNotFoundError: + pass + + descriptor, temporary_name = tempfile.mkstemp( + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + ) + temporary = Path(temporary_name) + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + if previous_mode is not None: + os.chmod(temporary, previous_mode) + os.replace(temporary, path) + _sync_directory(path.parent) + finally: + if temporary.exists(): + temporary.unlink() + + +def atomic_write_text( + path: str | Path, + text: str, + *, + encoding: str = "utf-8", +) -> None: + _atomic_write(Path(path), text.encode(encoding)) + + +def atomic_write_json( + path: str | Path, + value: JsonValue, + *, + indent: int | None = 2, + ensure_ascii: bool = False, + sort_keys: bool = False, +) -> None: + serialized = json.dumps( + value, + indent=indent, + ensure_ascii=ensure_ascii, + sort_keys=sort_keys, + ) + atomic_write_text(path, serialized + "\n") + + +def _prepare_lock_file(stream: BinaryIO) -> None: + stream.seek(0, os.SEEK_END) + if stream.tell() == 0: + stream.write(b"\0") + stream.flush() + + +def _try_lock(stream: BinaryIO) -> None: + stream.seek(0) + if os.name == "nt": + import msvcrt + + msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + + fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + + +def _unlock(stream: BinaryIO) -> None: + stream.seek(0) + if os.name == "nt": + import msvcrt + + msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(stream.fileno(), fcntl.LOCK_UN) + + +@contextmanager +def json_file_lock(path: str | Path, *, timeout: float = 10.0): + target = Path(path).expanduser() + target.parent.mkdir(parents=True, exist_ok=True) + lock_path = target.with_name(f".{target.name}.lock") + deadline = time.monotonic() + timeout + + with lock_path.open("a+b") as stream: + _prepare_lock_file(stream) + while True: + try: + _try_lock(stream) + break + except (BlockingIOError, PermissionError, OSError) as exc: + if time.monotonic() >= deadline: + raise JsonLockTimeout( + f"Could not acquire JSON lock within {timeout:.1f}s: {lock_path}" + ) from exc + time.sleep(0.05) + try: + yield + finally: + _unlock(stream) + + +def atomic_update_json( + path: str | Path, + update: Callable[[T], T | None], + *, + default_factory: Callable[[], T], + timeout: float = 10.0, + indent: int | None = 2, +) -> T: + target = Path(path) + with json_file_lock(target, timeout=timeout): + current = read_json(target, default=default_factory()) + updated = update(current) + result = current if updated is None else updated + atomic_write_json(target, result, indent=indent) + return result diff --git a/copienator/workspace.py b/copienator/workspace.py new file mode 100644 index 0000000..d86667a --- /dev/null +++ b/copienator/workspace.py @@ -0,0 +1,204 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + + +class WorkspaceNotFoundError(FileNotFoundError): + pass + + +class WorkspaceValidationError(ValueError): + def __init__(self, root: Path, missing: list[str]) -> None: + self.root = root + self.missing = missing + super().__init__( + f"Invalid evaluation workspace {root}: missing {', '.join(missing)}" + ) + + +@dataclass(frozen=True, slots=True) +class EvaluationWorkspace: + root: Path + repository: Path | None = None + + def __post_init__(self) -> None: + object.__setattr__(self, "root", Path(self.root).expanduser().resolve()) + if self.repository is not None: + object.__setattr__( + self, "repository", Path(self.repository).expanduser().resolve() + ) + + @classmethod + def discover( + cls, + target: str | Path, + *, + repository: str | Path | None = None, + ) -> EvaluationWorkspace: + candidate = Path(target).expanduser().resolve() + if candidate.is_file(): + candidate = candidate.parent + repository_path = Path(repository).expanduser().resolve() if repository else None + + for directory in (candidate, *candidate.parents): + if cls.looks_like_evaluation(directory): + return cls(directory, repository_path) + if repository_path is not None and directory == repository_path: + break + raise WorkspaceNotFoundError( + f"Could not find an evaluation workspace from: {target}" + ) + + @staticmethod + def looks_like_evaluation(path: Path) -> bool: + markers = ( + "enonce.pdf", + "enonce.tex", + "labels", + "Copies", + "correction.json", + ".copienator", + ".copienator-gui.json", + ) + return path.is_dir() and any((path / marker).exists() for marker in markers) + + @property + def name(self) -> str: + return self.root.name + + @property + def metadata_dir(self) -> Path: + return self.root / ".copienator" + + @property + def logs_dir(self) -> Path: + return self.metadata_dir / "logs" + + @property + def runs_dir(self) -> Path: + return self.metadata_dir / "runs" + + @property + def state_database(self) -> Path: + return self.metadata_dir / "state.sqlite3" + + @property + def gui_state_file(self) -> Path: + return self.root / ".copienator-gui.json" + + @property + def labels_file(self) -> Path: + return self.root / "labels" + + @property + def correction_file(self) -> Path: + return self.root / "correction.json" + + @property + def correction_progress_file(self) -> Path: + return self.root / "correction_progress.json" + + @property + def manual_resolutions_file(self) -> Path: + return self.root / "manual_resolutions.txt" + + @property + def refaire_file(self) -> Path: + return self.root / "refaire.json" + + @property + def copies_dir(self) -> Path: + return self.root / "Copies" + + @property + def original_copies_dir(self) -> Path: + return self.root / "Copies Originales" + + @property + def cutleft_dir(self) -> Path: + return self.root / "Cutleft" + + @property + def groups_dir(self) -> Path: + return self.root / "Par label" + + @property + def text_dir(self) -> Path: + return self.root / "Text" + + @property + def solution_dir(self) -> Path: + return self.root / "Sol" + + @property + def rendered_text_dir(self) -> Path: + return self.root / "Text2" + + @property + def rendered_solution_dir(self) -> Path: + return self.root / "Sol2" + + @property + def rubric_dir(self) -> Path: + return self.root / "Persp" + + @property + def return_dir(self) -> Path: + return self.root / "A Rendre" + + def annotation_dir(self, mode: str) -> Path: + directories = { + "simple": "Anot", + "checks": "Bnot", + "grouped": "BGnot", + "refaire": "BRnot", + } + try: + return self.root / directories[mode] + except KeyError as exc: + choices = ", ".join(directories) + raise ValueError( + f"Unknown annotation mode {mode!r}; expected one of: {choices}" + ) from exc + + def names_file(self) -> Path: + local = self.root / "names" + if local.exists() or self.repository is None: + return local + return self.repository / "names" + + def read_labels(self) -> list[str]: + return [ + line.strip() + for line in self.labels_file.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + def require(self, *relative_paths: str) -> None: + missing = [ + relative_path + for relative_path in relative_paths + if not (self.root / relative_path).exists() + ] + if missing: + raise WorkspaceValidationError(self.root, missing) + + def ensure_control_directories(self) -> None: + self.logs_dir.mkdir(parents=True, exist_ok=True) + self.runs_dir.mkdir(parents=True, exist_ok=True) + + def command_argument(self) -> str: + if self.repository is not None: + try: + return str(self.root.relative_to(self.repository)) + except ValueError: + pass + return str(self.root) + + def log_path(self, step_id: str, when: datetime | None = None) -> Path: + timestamp = (when or datetime.now().astimezone()).strftime("%Y%m%d-%H%M%S") + safe_step = re.sub(r"[^A-Za-z0-9_.-]+", "_", step_id) + return self.logs_dir / f"{timestamp}-{safe_step}.log" diff --git a/copienator_gui/app.py b/copienator_gui/app.py index b02a8a6..5171c75 100644 --- a/copienator_gui/app.py +++ b/copienator_gui/app.py @@ -2,9 +2,7 @@ from __future__ import annotations import os import queue -import re import tkinter as tk -from datetime import datetime from pathlib import Path from tkinter import filedialog, messagebox, ttk from typing import Any @@ -565,9 +563,9 @@ class CopienatorApp(tk.Tk): self.state_store.invalidate_after(ordered_ids, step.id) self.state_store.update_step(step.id, status="running", command=command_display(command)) self.active_step_id = step.id - timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S") - safe_step = re.sub(r"[^A-Za-z0-9_.-]+", "_", step.id) - log_path = evaluation / ".copienator" / "logs" / f"{timestamp}-{safe_step}.log" + workspace = self.state_store.workspace + assert workspace is not None + log_path = workspace.log_path(step.id) environment = os.environ.copy() environment["PYTHONUNBUFFERED"] = "1" diff --git a/copienator_gui/state.py b/copienator_gui/state.py index b2f1b04..bc14eec 100644 --- a/copienator_gui/state.py +++ b/copienator_gui/state.py @@ -5,12 +5,13 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any -STATE_FILENAME = ".copienator-gui.json" +from copienator import EvaluationWorkspace, atomic_write_json, read_json class StateStore: def __init__(self) -> None: self.evaluation: Path | None = None + self.workspace: EvaluationWorkspace | None = None self.data: dict[str, Any] = self._empty_data() @staticmethod @@ -18,10 +19,11 @@ class StateStore: return {"version": 1, "steps": {}, "history": []} def load(self, evaluation: Path) -> None: - self.evaluation = evaluation - path = evaluation / STATE_FILENAME + self.workspace = EvaluationWorkspace(evaluation) + self.evaluation = self.workspace.root + path = self.workspace.gui_state_file try: - loaded = json.loads(path.read_text(encoding="utf-8")) + loaded = read_json(path, default=self._empty_data()) self.data = loaded if isinstance(loaded, dict) else self._empty_data() except (OSError, json.JSONDecodeError): self.data = self._empty_data() @@ -30,14 +32,9 @@ class StateStore: self.data.setdefault("history", []) def save(self) -> None: - if not self.evaluation: + if not self.workspace: return - path = self.evaluation / STATE_FILENAME - temporary = path.with_suffix(path.suffix + ".tmp") - temporary.write_text( - json.dumps(self.data, indent=2, ensure_ascii=False), encoding="utf-8" - ) - temporary.replace(path) + atomic_write_json(self.workspace.gui_state_file, self.data) def step(self, step_id: str) -> dict[str, Any]: return self.data["steps"].setdefault(step_id, {}) diff --git a/correction.py b/correction.py index 841d77c..488d4a8 100644 --- a/correction.py +++ b/correction.py @@ -12,6 +12,8 @@ import json import threading import concurrent.futures +from copienator import atomic_write_json + if len(sys.argv) < 2: sys.exit("Usage: python script.py 'InterroTest/Ex 2/Group_1.jpg' OR OR 'file1' 'file2'") @@ -504,13 +506,11 @@ def process_single_task(task_tuple, precomputed_response=None): results[label] = [] results[label].append(json_data) - with open(output_path, "w", encoding="utf-8") as f: - json.dump(results, f, indent=2) + atomic_write_json(output_path, results) # To track progress completed_tasks.append((file_path, label)) - with open(progress_path, "w", encoding="utf-8") as f: - json.dump(completed_tasks, f, indent=2) + atomic_write_json(progress_path, completed_tasks) except json.JSONDecodeError: tprint(f"Error decoding JSON for {file_path}", file=sys.stderr) @@ -598,8 +598,7 @@ def resolve_delayed_moves(): del res["delayed"] if new_tasks: - with open(output_path, "w", encoding="utf-8") as f: - json.dump(results, f, indent=2) + atomic_write_json(output_path, results) return new_tasks @@ -675,10 +674,8 @@ if __name__ == "__main__": tasks_to_process.append((new_group_path, label, not is_new)) if dirty_results: - with open(output_path, "w", encoding="utf-8") as f: - json.dump(results, f, indent=2) - with open(overwritten_path, "w", encoding="utf-8") as f: - json.dump(overwritten_data, f, indent=2) + atomic_write_json(output_path, results) + atomic_write_json(overwritten_path, overwritten_data) else: print(f"Warning: --refaire flag used, but {refaire_path} not found.", file=sys.stderr) diff --git a/post-correction.py b/post-correction.py index 6bb5b7b..8c176c0 100644 --- a/post-correction.py +++ b/post-correction.py @@ -4,6 +4,8 @@ import time from pathlib import Path import argparse +from copienator import atomic_write_json + if len(sys.argv) < 2: sys.exit("Usage: python script.py ") @@ -148,7 +150,6 @@ with open(INPUT_FILE, "r", encoding="utf-8") as f: data = clean_obj(data) -with open(OUTPUT_FILE, "w", encoding="utf-8") as f: - json.dump(data, f, ensure_ascii=False, indent=2) +atomic_write_json(OUTPUT_FILE, data) print("Fixed JSON saved to", OUTPUT_FILE) diff --git a/resolve_manual.py b/resolve_manual.py index b4f73b5..b276842 100644 --- a/resolve_manual.py +++ b/resolve_manual.py @@ -6,6 +6,8 @@ import shutil from pathlib import Path from pypdf import PdfWriter +from copienator import atomic_write_json + if len(sys.argv) < 2: sys.exit("Usage: python resolve_manual.py ") @@ -180,12 +182,10 @@ for temp in temp_files: temp.unlink() # Finalize JSONs -with open(correction_file, "w", encoding="utf-8") as f: - json.dump(results, f, indent=2) +atomic_write_json(correction_file, results) if refaire_tasks: - with open(refaire_file, "w", encoding="utf-8") as f: - json.dump(refaire_tasks, f, indent=2) + atomic_write_json(refaire_file, refaire_tasks) manual_file.unlink(missing_ok=True) diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py new file mode 100644 index 0000000..72dd914 --- /dev/null +++ b/tests/test_gui_core.py @@ -0,0 +1,294 @@ +from __future__ import annotations + +import os +import queue +import sys +import tempfile +import time +import unittest +from concurrent.futures import ThreadPoolExecutor +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.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] + + +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"]) + + 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 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() diff --git a/utils.py b/utils.py index 5d05928..edf1ed9 100644 --- a/utils.py +++ b/utils.py @@ -1,13 +1,14 @@ import re from pathlib import Path +from copienator import EvaluationWorkspace from platform_utils import validate_windows_labels def natural_key(text): return [int(c) if c.isdigit() else c.lower() for c in re.split(r'(\d+)', str(text))] def read_all_labels(base_dir): - labels = list(filter(None, (Path(base_dir) / "labels").read_text().splitlines())) + labels = EvaluationWorkspace(Path(base_dir)).read_labels() validate_windows_labels(labels) return labels