diff --git a/Script.org b/Script.org index 91965a8..3a1f552 100644 --- a/Script.org +++ b/Script.org @@ -377,6 +377,39 @@ Ce parcours a sa propre progression : les boutons de navigation du passage principal ne l'ouvrent pas automatiquement. Après la fusion, les étapes de restitution déjà terminées sont marquées à revalider. +Pour enchaîner les reprises, deux boutons sont disponibles dans ce parcours : +- =Nouvelle reprise= vide la sélection et remet les étapes de reprise à zéro. +- =Refaire la même sélection= conserve les copies et les questions du dernier + enregistrement, mais remet également les étapes de reprise à zéro. +Les choix du passage principal et de présentation des PDF sont conservés. +Enregistrer ensuite la sélection avant de poursuivre. + +Attention : appeler =Nouvelle reprise= seulement après avoir importé les +résultats et exécuté =Mettre à jour les copies finales=. La même précaution +s'applique à =Refaire la même sélection=. Un avertissement est affiché avant +les deux actions, avec une mention supplémentaire si la fusion n'est pas +marquée réussie. Annuler conserve la reprise actuelle. Après une fusion +réussie, utiliser ces boutons pour recommencer, plutôt que modifier la +sélection du passage terminé. + +Chaque nouveau passage utilise =Reprises/reprise-DATE-HEURE-ID/BRnot=. +Le passage précédent garde ses PDF, ses retours manuscrits, sa sélection +et une copie de la progression du GUI. Au premier changement de passage, +l'ancien =BRnot= à la racine est également copié dans =Reprises=. +Ces archives concernent les fichiers de vérification, pas un mécanisme +permettant d'annuler les modifications des copies finales. +Le fichier =refaire-session.json= désigne le passage actif ; les commandes +habituelles =--refaire= le suivent automatiquement. Sans ce fichier, le +fonctionnement historique dans =BRnot= à la racine reste disponible. +Dans la suite, =BRnot= désigne le dossier de la reprise active. + +L'export d'un passage identifié utilise son propre sous-dossier dans +=EXPORT_DIR/Évaluation= et préfixe les noms des PDF par son identifiant. +Conserver les noms complets au retour et placer les PDF directement dans +=IMPORT_DIR=. L'import ignore les retours des autres passages ; aucun retour +du passage actif donne un résultat partiel. Ainsi, deux reprises de la même +question ne partagent pas les mêmes noms de fichiers exportés. + Ce flux fonctionne après =annotate-grouped= (=BGnot=), =annotate-checks= (=Bnot=) ou =annotate-simple= (=Anot=). Terminer d'abord la lecture des annotations du passage principal @@ -405,8 +438,8 @@ Conserver les dossiers d'annotation et leurs fichiers de référence. principal. Le mode groupé garde les identifiants des élèves et regroupe les réponses par label sans demander de modifier =label_groups=. Cela ne nécessite pas d'avoir généré =Bnot= ou =BGnot= auparavant. - Attention : =--overwrite= remplace tout le précédent passage dans - =BRnot=, y compris ses annotations manuscrites. Une génération incomplète + Attention : =--overwrite= remplace le contenu du =BRnot= actif, + y compris ses annotations manuscrites, mais pas les autres reprises. Une génération incomplète conserve l'ancien =BRnot=. Sans =--overwrite=, un =BRnot= existant est refusé. 5. Vider les dossiers personnels d'export/import des anciens fichiers, puis =python -m copienator export Interro --refaire=. diff --git a/copienator/commands/export.py b/copienator/commands/export.py index f401122..e4f100e 100644 --- a/copienator/commands/export.py +++ b/copienator/commands/export.py @@ -3,7 +3,6 @@ import sys from collections.abc import Sequence from pathlib import Path -from copienator.configuration import EXPORT_DIR from copienator import ( EvaluationWorkspace, ExitCode, @@ -11,6 +10,7 @@ from copienator import ( execute, workspace_from_args, ) +from copienator.configuration import EXPORT_DIR from copienator.platform import replace_with_link_or_copy ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot") @@ -21,8 +21,12 @@ def export_directory( source_dir_name: str, ) -> ExitCode: workspace.require_directories(source_dir_name) - source_dir = workspace.root / source_dir_name + source_dir = workspace.annotation_dir("refaire") if source_dir_name == "BRnot" else workspace.root / source_dir_name + session_id = workspace.refaire_session_id if source_dir_name == "BRnot" else None + prefix = f"{session_id}__" if session_id else "" sync_dir = Path(EXPORT_DIR).expanduser() / workspace.name + if session_id: + sync_dir /= session_id sync_dir.mkdir(parents=True, exist_ok=True) subdirs = [directory for directory in source_dir.iterdir() if directory.is_dir()] @@ -48,7 +52,7 @@ def export_directory( ) missing_outputs += 1 continue - destination = sync_dir / f"{subdir.name}{concat_file.suffix.lower()}" + destination = sync_dir / f"{prefix}{subdir.name}{concat_file.suffix.lower()}" method = replace_with_link_or_copy(concat_file, destination, prefer="hardlink") print(f"Exported: {destination} ({method})") return ExitCode.PARTIAL if missing_outputs else ExitCode.SUCCESS diff --git a/copienator/commands/import_annotations.py b/copienator/commands/import_annotations.py index 55c8c9d..9eb1292 100644 --- a/copienator/commands/import_annotations.py +++ b/copienator/commands/import_annotations.py @@ -4,7 +4,6 @@ import sys from collections.abc import Sequence from pathlib import Path -from copienator.configuration import IMPORT_DIR from copienator import ( EvaluationWorkspace, ExitCode, @@ -12,6 +11,7 @@ from copienator import ( execute, workspace_from_args, ) +from copienator.configuration import IMPORT_DIR ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot") @@ -23,7 +23,10 @@ def sync_annotated( import_dir: Path, ) -> ExitCode: workspace.require_directories(annotation_dir_name) - annotation_dir = workspace.root / annotation_dir_name + annotation_dir = workspace.annotation_dir("refaire") if annotation_dir_name == "BRnot" else workspace.root / annotation_dir_name + session_id = workspace.refaire_session_id if annotation_dir_name == "BRnot" else None + prefix = f"{session_id}__" if session_id else "" + accepted = 0 annotated_dir = Path(import_dir).expanduser() if not annotated_dir.is_dir(): print(f"Error: directory does not exist: {annotated_dir}", file=sys.stderr) @@ -39,7 +42,10 @@ def sync_annotated( key=lambda path: path.name.casefold(), ) for annotated_file in annotated_files: - target_subdir = annotation_dir / annotated_file.stem + if prefix and not annotated_file.stem.startswith(prefix): + print(f"Ignoring return from another pass: {annotated_file.name}") + continue + target_subdir = annotation_dir / annotated_file.stem.removeprefix(prefix) if not target_subdir.is_dir(): print(f"Warning: directory not found: {target_subdir}", file=sys.stderr) @@ -49,6 +55,10 @@ def sync_annotated( dest_file = target_subdir / f"Concat_annotated{suffix}" print(f"Copying {annotated_file} to {dest_file}") shutil.copy2(annotated_file, dest_file) + accepted += 1 + if prefix and not accepted: + print(f"No returns for the active pass {session_id} were imported.") + return ExitCode.PARTIAL return ExitCode.PARTIAL if missing_targets else ExitCode.SUCCESS diff --git a/copienator/workspace.py b/copienator/workspace.py index f9199a0..c6245a9 100644 --- a/copienator/workspace.py +++ b/copienator/workspace.py @@ -165,7 +165,26 @@ class EvaluationWorkspace: def return_dir(self) -> Path: return self.root / "A Rendre" + @property + def refaire_session_id(self) -> str | None: + from .json_io import read_json + + session = read_json(self.root / "refaire-session.json", default=None) + if session is None: + return None + ident = session.get("id") if isinstance(session, dict) else None + if not isinstance(ident, str) or not re.fullmatch(r"reprise-[0-9]{8}-[0-9]{6}-[a-f0-9]{8}", ident): + raise ValueError("Invalid refaire-session.json") + return ident + + @property + def refaire_session_dir(self) -> Path | None: + ident = self.refaire_session_id + return self.root / "Reprises" / ident if ident else None + def annotation_dir(self, mode: str) -> Path: + if mode == "refaire" and self.refaire_session_dir is not None: + return self.refaire_session_dir / "BRnot" directories = { "simple": "Anot", "checks": "Bnot", @@ -215,7 +234,7 @@ class EvaluationWorkspace: missing = [ relative_path for relative_path in relative_paths - if not (self.root / relative_path).is_dir() + if not (self.annotation_dir("refaire") if relative_path == "BRnot" else self.root / relative_path).is_dir() ] if missing: raise WorkspaceValidationError(self.root, missing) diff --git a/copienator_gui/app.py b/copienator_gui/app.py index 6351473..7b60ded 100644 --- a/copienator_gui/app.py +++ b/copienator_gui/app.py @@ -25,6 +25,7 @@ from .refaire import ( resolve_layout, validate_selection, ) +from .refaire_sessions import RESTART_WARNING, begin_pass from .runner import ProcessRunner from .state import StateStore from .workflow import ( @@ -470,7 +471,8 @@ class CopienatorApp(tk.Tk): if any(char in pattern for char in "*?["): exists = next(evaluation.glob(pattern), None) is not None else: - exists = (evaluation / pattern).exists() + path = EvaluationWorkspace(evaluation).annotation_dir("refaire") if pattern == "BRnot" else evaluation / pattern + exists = path.exists() if not exists: missing.append(pattern) return missing @@ -635,6 +637,7 @@ class CopienatorApp(tk.Tk): evaluation = self.evaluation if step.id in {"refaire_selection", "refaire_correct"}: row = self._correction_folder_buttons(row) + row = self._refaire_restart_controls(row) if step.id == "refaire_selection" and evaluation: choices, preferred = self._annotation_directory_choices("export") draft = self.state_store.step(step.id).get("values", {}) @@ -699,6 +702,50 @@ class CopienatorApp(tk.Tk): row += 1 return row + def _refaire_restart_controls(self, row: int) -> int: + controls = ttk.Frame(self.form) + controls.grid(row=row, column=0, columnspan=2, sticky="w", pady=(0, 5)) + running = bool(self.active_step_id) or self.runner.running + ttk.Button(controls, text="Nouvelle reprise", command=lambda: self._start_refaire_pass(False), + state="disabled" if running else "normal").pack(side="left", padx=(0, 6)) + ttk.Button(controls, text="Refaire la même sélection", command=lambda: self._start_refaire_pass(True), + state="disabled" if running else "normal").pack(side="left") + ttk.Label(self.form, text="« Nouvelle reprise » : à utiliser après avoir importé les résultats et mis à jour les copies finales.", + wraplength=580, justify="left").grid(row=row + 1, column=0, columnspan=2, sticky="w", pady=(0, 6)) + return row + 2 + + def _start_refaire_pass(self, keep_selection: bool) -> None: + workspace = self.state_store.workspace + if workspace is None: + messagebox.showerror("Évaluation absente", "Chargez d’abord une évaluation.") + return + if self.active_step_id or self.runner.running: + messagebox.showwarning("Traitement en cours", "Attendez la fin du traitement avant de commencer une reprise.") + return + try: + if keep_selection: + load_selection(workspace.root) + title = "Refaire la même sélection" if keep_selection else "Nouvelle reprise" + message = RESTART_WARNING + if self.state_store.step("refaire_merge").get("status") != "success": + message += "\n\nLa reprise actuelle n’est pas marquée comme fusionnée. Les résultats non fusionnés ne seront pas intégrés aux copies finales." + message += "\n\nLes fichiers de la reprise précédente seront conservés. Continuer ?" + if not messagebox.askyesno(title, message, icon="warning", default="no"): + return + self._save_current_form() + updated, ident = begin_pass(workspace, self.state_store.data, keep_selection=keep_selection) + self.state_store.data = updated + # Do not save the old form over the fresh pass during navigation. + self.current_step = None + self.refaire_panel = None + self.arg_vars.clear() + self._populate_tree() + self.tree.selection_set("refaire_selection") + self.tree.see("refaire_selection") + self.info_var.set(f"Nouvelle reprise : {ident}. Enregistrez la sélection pour continuer.") + except (OSError, ValueError, TypeError) as exc: + messagebox.showerror("Nouvelle reprise impossible", str(exc)) + def _correction_folder_buttons(self, row: int) -> int: buttons = ttk.Frame(self.form) buttons.grid(row=row, column=0, columnspan=2, sticky="w", pady=(0, 6)) @@ -804,11 +851,23 @@ class CopienatorApp(tk.Tk): if not self.refaire_panel or not self.evaluation: return try: + if self.state_store.step("refaire_merge").get("status") == "success": + messagebox.showinfo("Reprise terminée", "Utilisez « Nouvelle reprise » ou « Refaire la même sélection » pour commencer un autre passage.") + return values = self.refaire_panel.values() entries = validate_selection(values["selection"], available_copies(self.evaluation), EvaluationWorkspace(self.evaluation).read_labels()) source = values["annotation_dir"] if source not in detected_annotation_directories(self.evaluation): raise ValueError("Choisissez un dossier existant pour le passage principal.") + workspace = EvaluationWorkspace(self.evaluation) + output = workspace.annotation_dir("refaire") + if output.is_dir() and any(output.iterdir()) and workspace.refaire_file.is_file(): + from copienator import read_json + if read_json(workspace.refaire_file) != entries: + raise ValueError("Des PDF de vérification existent déjà. Importez et fusionnez leurs résultats, puis utilisez « Nouvelle reprise » pour changer de sélection.") + if workspace.refaire_session_dir is not None: + atomic_write_json(workspace.refaire_session_dir / "refaire.json", entries) + atomic_write_json(workspace.refaire_session_dir / "session.json", {"id": workspace.refaire_session_id, "values": values}) atomic_write_json(self.evaluation / "refaire.json", entries) self._mark_step("success") self._move_selection_from("refaire_selection", 1) diff --git a/copienator_gui/refaire_sessions.py b/copienator_gui/refaire_sessions.py new file mode 100644 index 0000000..ad00e7c --- /dev/null +++ b/copienator_gui/refaire_sessions.py @@ -0,0 +1,86 @@ +"""Preserve completed redo passes and activate a fresh working directory.""" + +from __future__ import annotations + +import copy +import shutil +import uuid +from datetime import datetime +from typing import Any + +from copienator import EvaluationWorkspace, atomic_write_json, read_json +from copienator.filesystem import staged_files + +RESTART_WARNING = ( + "Appelez « Nouvelle reprise » seulement après avoir importé les résultats " + "de la reprise en cours et exécuté « Mettre à jour les copies finales ». " + "Cette consigne s’applique aussi à « Refaire la même sélection »." +) + + +def _identifier() -> str: + return f"reprise-{datetime.now().astimezone():%Y%m%d-%H%M%S}-{uuid.uuid4().hex[:8]}" + + +def begin_pass( + workspace: EvaluationWorkspace, + state: dict[str, Any], + *, + keep_selection: bool, +) -> tuple[dict[str, Any], str]: + """Activate a pass atomically with its selection and reset GUI state. + + Existing review files remain in place. Legacy BRnot is copied once into + the archive; failures before activation leave the old pass active. + """ + selection = read_json(workspace.refaire_file, default=[]) + previous = workspace.refaire_session_dir + if previous is None and ( + workspace.refaire_file.exists() or workspace.annotation_dir("refaire").exists() + ): + previous = workspace.root / "Reprises" / _identifier() + previous.mkdir(parents=True) + legacy = workspace.annotation_dir("refaire") + if legacy.is_dir(): + shutil.copytree(legacy, previous / "BRnot") + if previous is not None: + atomic_write_json(previous / "refaire.json", selection) + atomic_write_json( + previous / "progression.json", + { + name: entry + for name, entry in state.get("steps", {}).items() + if name.startswith("refaire_") + }, + ) + + ident = _identifier() + directory = workspace.root / "Reprises" / ident + directory.mkdir(parents=True) + new_selection = selection if keep_selection else [] + updated = copy.deepcopy(state) + values = dict( + updated.get("steps", {}).get("refaire_selection", {}).get("values", {}) + ) + values["selection"] = new_selection + updated["steps"] = { + name: entry + for name, entry in updated.get("steps", {}).items() + if not name.startswith("refaire_") + } + updated["steps"]["refaire_selection"] = {"values": values} + updated.setdefault("history", []).append( + { + "step": "refaire_selection", + "action": "repeat" if keep_selection else "new", + "session": ident, + "timestamp": datetime.now().astimezone().isoformat(), + } + ) + atomic_write_json(directory / "refaire.json", new_selection) + atomic_write_json(directory / "session.json", {"id": ident, "values": values}) + with staged_files(workspace.root) as staging: + atomic_write_json(staging / workspace.refaire_file.name, new_selection) + atomic_write_json(staging / workspace.gui_state_file.name, updated) + atomic_write_json(staging / "refaire-session.json", {"id": ident}) + return updated, ident diff --git a/tests/test_gui_refaire.py b/tests/test_gui_refaire.py index 6a2b59c..d1785ba 100644 --- a/tests/test_gui_refaire.py +++ b/tests/test_gui_refaire.py @@ -194,6 +194,49 @@ class RefaireGuiTests(unittest.TestCase): [self.root / "Sol", self.root / "Persp"], ) + def test_restart_actions_warn_reset_and_preserve_or_clear_selection(self): + self.save_selection() + self.app.state_store.update_step("refaire_merge", status="success") + self.app.state_store.update_step("annotation", status="success") + selection = read_json(self.root / "refaire.json") + with patch( + "copienator_gui.app.messagebox.askyesno", return_value=False + ) as confirm: + self.app._start_refaire_pass(False) + self.assertIn("importé", confirm.call_args.args[1]) + self.assertIn("Mettre à jour", confirm.call_args.args[1]) + self.assertEqual(read_json(self.root / "refaire.json"), selection) + with patch("copienator_gui.app.messagebox.askyesno", return_value=True): + self.app._start_refaire_pass(True) + self.app.update() + self.assertEqual(self.app.current_step.id, "refaire_selection") + self.assertEqual(self.app.refaire_panel.values()["selection"], selection) + self.assertIsNone(self.app.state_store.step("refaire_merge").get("status")) + self.assertEqual(self.app.state_store.step("annotation")["status"], "success") + first = self.app.state_store.workspace.refaire_session_id + with patch( + "copienator_gui.app.messagebox.askyesno", return_value=True + ) as confirm: + self.app._start_refaire_pass(False) + self.app.update() + self.assertIn("n’est pas marquée comme fusionnée", confirm.call_args.args[1]) + self.assertEqual(self.app.refaire_panel.entries, {}) + self.assertEqual(read_json(self.root / "refaire.json"), []) + self.assertNotEqual(self.app.state_store.workspace.refaire_session_id, first) + self.assertEqual(self.app.refaire_panel.source_var.get(), "Anot") + + def test_restart_is_blocked_while_a_command_is_active(self): + self.save_selection() + self.app.active_step_id = "refaire_correct" + with ( + patch("copienator_gui.app.messagebox.showwarning") as warning, + patch("copienator_gui.app.messagebox.askyesno") as confirm, + ): + self.app._start_refaire_pass(False) + warning.assert_called_once() + confirm.assert_not_called() + self.assertIsNone(self.app.state_store.workspace.refaire_session_id) + def test_small_window_keeps_selection_accessible_by_scrolling(self): self.select("refaire_selection") self.app.geometry("900x640") diff --git a/tests/test_refaire_sessions.py b/tests/test_refaire_sessions.py new file mode 100644 index 0000000..7347099 --- /dev/null +++ b/tests/test_refaire_sessions.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +import copy +import shutil +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 +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 +from copienator_gui.refaire_sessions import begin_pass + + +class RefaireSessionTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name) / "Exam" + self.root.mkdir() + self.workspace = EvaluationWorkspace(self.root) + atomic_write_json(self.workspace.refaire_file, [["Copie01", ["Ex 1"]]]) + (self.root / "BRnot/Copie01").mkdir(parents=True) + (self.root / "BRnot/Copie01/Concat_annotated.pdf").write_bytes(b"legacy return") + self.state = { + "steps": { + "annotation": {"status": "success"}, + "refaire_selection": { + "status": "success", + "values": { + "selection": [["Copie01", ["Ex 1"]]], + "annotation_dir": "Anot", + "layout": "grouped", + }, + }, + "refaire_merge": {"status": "success"}, + }, + "history": [], + } + atomic_write_json(self.workspace.gui_state_file, self.state) + + def tearDown(self): + self.temp.cleanup() + + def test_new_then_repeat_preserve_each_pass_and_reset_only_redo_progress(self): + state, first = begin_pass(self.workspace, self.state, keep_selection=False) + self.assertEqual( + state["steps"]["annotation"], self.state["steps"]["annotation"] + ) + self.assertNotIn("refaire_merge", state["steps"]) + self.assertEqual(state["steps"]["refaire_selection"]["values"]["selection"], []) + self.assertEqual( + state["steps"]["refaire_selection"]["values"]["layout"], "grouped" + ) + self.assertEqual(read_json(self.workspace.refaire_file), []) + archives = [ + path for path in (self.root / "Reprises").iterdir() if path.name != first + ] + self.assertEqual(len(archives), 1) + self.assertEqual( + (archives[0] / "BRnot/Copie01/Concat_annotated.pdf").read_bytes(), + b"legacy return", + ) + selection = [["Copie02", ["Ex 2"]]] + atomic_write_json(self.workspace.refaire_file, selection) + first_output = self.workspace.annotation_dir("refaire") + first_output.mkdir(parents=True) + (first_output / "review.pdf").write_bytes(b"first review") + state["steps"]["refaire_merge"] = {"status": "success"} + updated, second = begin_pass(self.workspace, state, keep_selection=True) + self.assertNotEqual(first, second) + self.assertEqual(read_json(self.workspace.refaire_file), selection) + self.assertEqual( + updated["steps"]["refaire_selection"]["values"]["selection"], selection + ) + self.assertEqual((first_output / "review.pdf").read_bytes(), b"first review") + self.assertEqual(read_json(first_output.parent / "refaire.json"), selection) + self.assertNotIn("status", updated["steps"]["refaire_selection"]) + self.assertEqual( + self.workspace.annotation_dir("refaire"), + self.root / "Reprises" / second / "BRnot", + ) + reloaded = EvaluationWorkspace(self.root) + self.assertEqual(reloaded.refaire_session_id, second) + + def test_failed_activation_leaves_current_pass_and_state_unchanged(self): + before = copy.deepcopy(self.state) + with ( + patch( + "copienator_gui.refaire_sessions.staged_files", + side_effect=OSError("disk unavailable"), + ), + self.assertRaises(OSError), + ): + begin_pass(self.workspace, self.state, keep_selection=False) + self.assertIsNone(self.workspace.refaire_session_id) + self.assertEqual(read_json(self.workspace.gui_state_file), before) + self.assertEqual( + read_json(self.workspace.refaire_file), [["Copie01", ["Ex 1"]]] + ) + self.assertEqual(self.state, before) + self.assertEqual( + (self.root / "BRnot/Copie01/Concat_annotated.pdf").read_bytes(), + b"legacy return", + ) + + def test_exports_are_unique_and_old_returns_cannot_enter_new_pass(self): + export_root = Path(self.temp.name) / "Export" + import_root = Path(self.temp.name) / "Import" + import_root.mkdir() + state, first = begin_pass(self.workspace, self.state, keep_selection=True) + for expected_id in (first, None): + if expected_id is None: + state, second = begin_pass(self.workspace, state, keep_selection=True) + expected_id = second + output = self.workspace.annotation_dir("refaire") / "Ex 1 G1" + output.mkdir(parents=True) + (output / "Concat.pdf").write_bytes(expected_id.encode()) + with patch.object(export, "EXPORT_DIR", export_root): + self.assertEqual( + export.run(self.workspace, refaire=True), ExitCode.SUCCESS + ) + exported = ( + export_root / "Exam" / expected_id / f"{expected_id}__Ex 1 G1.pdf" + ) + self.assertEqual(exported.read_bytes(), expected_id.encode()) + if expected_id == first: + (import_root / exported.name).write_bytes(b"old return") + current = self.workspace.annotation_dir("refaire") / "Ex 1 G1" + with patch.object(import_annotations, "IMPORT_DIR", import_root): + self.assertEqual( + import_annotations.run(self.workspace, refaire=True), ExitCode.PARTIAL + ) + self.assertFalse((current / "Concat_annotated.pdf").exists()) + (import_root / f"{second}__Ex 1 G1.pdf").write_bytes(b"new return") + self.assertEqual( + import_annotations.run(self.workspace, refaire=True), ExitCode.SUCCESS + ) + self.assertEqual((current / "Concat_annotated.pdf").read_bytes(), b"new return") + self.workspace.require_directories("BRnot") + + def test_generation_and_merge_follow_the_active_pass(self): + for directory in ("Copies", "Par label", "Anot"): + (self.root / directory).mkdir() + (self.root / "labels").write_text("Ex 1\n") + atomic_write_json(self.root / "correction.json", {}) + answer = self.root / "answer.pdf" + answer.touch() + loaded = AnnotationLoadResult( + { + "01": { + "Ex 1": { + "pdf_path": answer, + "result": {"score": 2, "feedback": []}, + "coordinates": (0, 0), + } + } + }, + [], + ) + begin_pass(self.workspace, self.state, keep_selection=True) + with ( + patch.object(checks, "load_annotation_data", return_value=loaded), + patch.object(reader, "load_annotation_data", return_value=loaded), + patch.object( + checks.annotating, + "make_base_image", + return_value=(Image.new("RGB", (100, 100), "white"), 0, 0), + ), + patch.object( + checks.annotating, + "compose_label_image", + return_value=(Image.new("RGB", (100, 100), "white"), 0), + ), + ): + self.assertEqual( + checks.run(self.workspace, self.root, refaire=True, overwrite=True), + ExitCode.SUCCESS, + ) + output = self.workspace.annotation_dir("refaire") / "Copie01" + shutil.copy2(output / "Concat.pdf", output / "Concat_annotated.pdf") + self.assertEqual( + reader.run(self.workspace, refaire=True, annotation_dir="Anot"), + ExitCode.SUCCESS, + ) + self.assertEqual( + read_json(self.root / "Anot/Copie01/score.json"), {"Ex 1": "2"} + ) + self.assertEqual( + (self.root / "BRnot/Copie01/Concat_annotated.pdf").read_bytes(), + b"legacy return", + ) + + def test_invalid_session_id_cannot_escape_workspace(self): + atomic_write_json(self.root / "refaire-session.json", {"id": "../elsewhere"}) + with self.assertRaises(ValueError): + self.workspace.annotation_dir("refaire") + + +if __name__ == "__main__": + unittest.main()