diff --git a/copienator/commands/clean.py b/copienator/commands/clean.py index 6d5dc27..c64a62a 100644 --- a/copienator/commands/clean.py +++ b/copienator/commands/clean.py @@ -153,7 +153,8 @@ def build_cleanup_plan(workspace: EvaluationWorkspace) -> CleanupPlan: files, directories = _workspace_entries(workspace.root) - kept_files = set(processed_copies) + kept_files = {path for path in files if path.is_relative_to(workspace.root / "BackUp")} + kept_files.update(processed_copies) kept_files.add(workspace.correction_file) lock_file = workspace.root / ".copienator-finalisation.lock" if lock_file in files: diff --git a/copienator/copy_removal.py b/copienator/copy_removal.py new file mode 100644 index 0000000..6572fc7 --- /dev/null +++ b/copienator/copy_removal.py @@ -0,0 +1,195 @@ +"""Reversible removal of one stable copy ID and its working artifacts.""" +from __future__ import annotations + +import io +import json +import os +import re +import shutil +import uuid +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + +from PIL import Image, ImageDraw + +from copienator import atomic_write_bytes, read_json +from copienator.evaluation_lock import EvaluationLock + + +@dataclass(frozen=True) +class RemovalPlan: + root: Path + copy: str + removed: tuple[Path, ...] + updated: dict[Path, bytes] + + +def available_copies(root: Path) -> list[str]: + names = set() + for folder in (root, root / "Copies", root / "Copies Originales"): + names.update(p.stem for p in folder.glob("Copie*.pdf") + if re.fullmatch(r"Copie[0-9]+", p.stem)) + return sorted(names, key=lambda name: int(name[5:])) + + +def _json_bytes(data) -> bytes: + return (json.dumps(data, ensure_ascii=False, indent=2) + "\n").encode() + + +def removal_plan(root: Path, copy: str) -> RemovalPlan: + root = root.resolve() + if not re.fullmatch(r"Copie[0-9]+", copy) or copy not in available_copies(root): + raise ValueError("Choisissez une copie existante.") + ident = copy[5:] + def same_id(value): + return isinstance(value, (str, int)) and str(value).removeprefix("Copie").isdecimal() and int(str(value).removeprefix("Copie")) == int(ident) + def owned(name): + match = re.match(r"^Copie([0-9]+)(?:$|[_. -])", name) + return bool(match and same_id(match[1])) + + removed = set() + files = [] + # Never traverse symlinks or the backups themselves. + for current, dirs, names in os.walk(root, followlinks=False): + directory = Path(current) + dirs[:] = [name for name in dirs if name not in {"BackUp", ".git"} + and not (directory / name).is_symlink()] + for name in list(dirs): + returned = re.search(r" \(([0-9]+)\)$", name) if directory.name == "A Rendre" else None + if owned(name) or (returned and same_id(returned[1])): + removed.add(directory / name) + dirs.remove(name) + for name in names: + path = directory / name + if path.is_symlink(): + continue + if owned(name): + removed.add(path) + else: + files.append(path) + + updated = {} + dropped = object() + def prune(value, *, rows=False, numeric_keys=False): + if isinstance(value, dict): + if any(same_id(value.get(key)) for key in ("id", "copy_id", "copy", "copie")): + return dropped + result = {} + for key, item in value.items(): + if owned(key) or (numeric_keys and same_id(key)): + continue + new = prune(item, rows=rows, numeric_keys=numeric_keys) + if new is not dropped: + result[key] = new + return result + if isinstance(value, list): + if value and ((rows and same_id(value[0])) or + (isinstance(value[0], str) and value[0] == copy)): + return dropped + return [new for item in value + if (new := prune(item, rows=rows, numeric_keys=numeric_keys)) is not dropped] + return value + + for path in files: + relative = path.relative_to(root) + # Historical logs and GUI history are audit records, not active inputs. + if "logs" in relative.parts or path.name == ".copienator-gui.json": + continue + if path.name in {"manual_resolutions.txt", "annotation_modifications.txt"}: + text = path.read_text() + pattern = re.compile(r"\bCopie0*" + str(int(ident)) + r"\b") + new_text = "".join(line for line in text.splitlines(keepends=True) if not pattern.search(line)) + if new_text != text: + updated[path] = new_text.encode() + if path.suffix == ".json": + data = read_json(path) # Fail before moving anything if an input is malformed. + is_group = path.name.startswith("Group_") and "Par label" in relative.parts + if is_group and isinstance(data, list): + targets = [entry for entry in data if isinstance(entry, list) and entry and same_id(entry[0])] + if targets: + remaining = [entry for entry in data if entry not in targets] + image_path = path.with_suffix(".jpg") + if not remaining: + removed.add(path) + if image_path.exists(): + removed.add(image_path) + continue + # Blank the removed answer without shifting other students' coordinates. + from copienator.commands.grouping import LABEL_HEIGHT, SEPARATOR_HEIGHT + with Image.open(image_path) as source: + image = source.convert("RGB") + draw = ImageDraw.Draw(image) + for entry in targets: + top = int(entry[1]) - LABEL_HEIGHT + draw.rectangle((0, max(0, top - SEPARATOR_HEIGHT), image.width, int(entry[2]) - 1), fill="white") + buffer = io.BytesIO() + image.save(buffer, format="JPEG", quality=95) + updated[image_path] = buffer.getvalue() + numeric_keys = path.name in {"custom_remarks.json", "general_remarks.json"} + new = prune(data, rows=is_group or path.name == "refaire.json", numeric_keys=numeric_keys) + if new is dropped: + removed.add(path) + elif new != data: + updated[path] = _json_bytes(new) + + # Cached contact sheets and batch payloads contain several students. Discard + # them as a unit so stale recognition/results cannot reintroduce the copy. + for directory in (root / "Cutleft/Names/Grids",): + if directory.is_dir() and not directory.is_symlink(): + removed.add(directory) + for path in files: + if (path.name.startswith(("batch_", "batched_")) or + path.name in {"correction_progress.json", "correction_pending_responses.json", + "histogramme.pdf", "merged_histogramme.pdf"}): + removed.add(path) + # Collapse children when a whole directory is being removed. + removed = {p for p in removed if not any(parent in removed for parent in p.parents)} + updated = {p: data for p, data in updated.items() + if p not in removed and not any(parent in removed for parent in p.parents)} + return RemovalPlan(root, copy, tuple(sorted(removed)), updated) + + +def remove_copy(root: Path, copy: str) -> Path: + """Back up every changed file first; roll back all edits on failure.""" + root = root.resolve() + if (root / "BackUp").is_symlink(): + raise ValueError("BackUp doit être un dossier dans l’évaluation, pas un lien symbolique.") + with EvaluationLock(root): + plan = removal_plan(root, copy) + backup = plan.root / "BackUp" / (datetime.now().strftime("%Y%m%d-%H%M%S") + "-" + copy + "-" + uuid.uuid4().hex[:8]) + originals = backup / "files" + originals.mkdir(parents=True) + paths = [*plan.removed, *plan.updated] + for path in paths: + target = originals / path.relative_to(plan.root) + target.parent.mkdir(parents=True, exist_ok=True) + if path.is_dir(): + shutil.copytree(path, target, symlinks=True) + else: + shutil.copy2(path, target) + manifest = {"copy": copy, "removed": [str(p.relative_to(plan.root)) for p in plan.removed], + "updated": [str(p.relative_to(plan.root)) for p in plan.updated], "status": "prepared"} + atomic_write_bytes(backup / "manifest.json", _json_bytes(manifest)) + changed = [] + try: + for path in plan.removed: + changed.append(path) + if path.is_dir(): + shutil.rmtree(path) + else: + path.unlink() + for path, data in plan.updated.items(): + changed.append(path) + atomic_write_bytes(path, data) + manifest["status"] = "complete" + atomic_write_bytes(backup / "manifest.json", _json_bytes(manifest)) + except Exception: + for path in reversed(changed): + source = originals / path.relative_to(plan.root) + if source.is_dir(): + shutil.copytree(source, path, dirs_exist_ok=True, symlinks=True) + else: + shutil.copy2(source, path) + raise + return backup diff --git a/copienator_gui/app.py b/copienator_gui/app.py index 9b3cf0a..69ecd16 100644 --- a/copienator_gui/app.py +++ b/copienator_gui/app.py @@ -921,7 +921,7 @@ class CopienatorApp(tk.Tk): if not step: return self._rendering = True - if step.id in {"personal_read", "edit_scores", "general_remarks"}: + if step.id in {"personal_read", "edit_scores", "general_remarks", "remove_copy"}: self.command_box.grid_remove() else: self.command_box.grid() @@ -997,7 +997,7 @@ class CopienatorApp(tk.Tk): continue value = values.get(spec.name, value_for_default(spec.default, evaluation_arg)) if step.id == 'statement_persp' and spec.name == 'delete_persp': - value = False # Deletion is a one-run choice, never a saved default. + value = spec.default # Reset on entry; retries in this form keep completed results. if step.id == 'statement' and spec.name == 'restart': value = False if step.id == "final_score" and spec.name == "copy_merged_histogramme" and spec.name not in values: @@ -1045,6 +1045,9 @@ class CopienatorApp(tk.Tk): ttk.Button(self.form, text='Actualiser les barèmes présents', command=self._refresh_persp_inventory).grid(row=row, column=1, sticky='w') row += 1 + ttk.Button(self.form, text='Ouvrir le dossier Persp', + command=self._open_persp).grid(row=row, column=1, sticky='w', pady=(8, 0)) + row += 1 self.arg_vars['target'].trace_add('write', lambda *_args: self._refresh_persp_inventory()) self._refresh_persp_inventory() @@ -1091,6 +1094,8 @@ class CopienatorApp(tk.Tk): row += 1 self.run_button.configure(text="Make modifications" if step.id == "modify_annotations" else "Enregistrer la sélection" if step.id == "refaire_selection" else ("Marquer terminée" if step.is_manual else "Exécuter")) + if step.id == "remove_copy": + self.run_button.configure(text="Supprimer la copie sélectionnée") if step.id == "personal_create": self.run_button.configure(text="Marquer Réussie") if step.id == "personal_read": @@ -1118,6 +1123,31 @@ class CopienatorApp(tk.Tk): self.persp_inventory_label.configure(text=text) def _render_context_controls(self, step: StepDefinition, row: int) -> int: + if step.id == "remove_copy": + from copienator.copy_removal import available_copies + self.removal_copy_var = tk.StringVar() + ttk.Label(self.form, text="Copie à supprimer").grid(row=row, column=0, sticky="w") + choices = available_copies(self.evaluation) if self.evaluation else [] + combo = ttk.Combobox(self.form, textvariable=self.removal_copy_var, + values=choices, state="readonly") + combo.grid(row=row, column=1, sticky="ew") + self.removal_preview = tk.Text(self.form, height=7, wrap="word", state="disabled") + self.removal_preview.grid(row=row + 1, column=0, columnspan=2, sticky="ew", pady=8) + def preview(*_args): + from copienator.copy_removal import removal_plan + try: + plan = removal_plan(self.evaluation, self.removal_copy_var.get()) + text = "Retirer :\n" + "\n".join(str(p.relative_to(plan.root)) for p in plan.removed) + if plan.updated: + text += "\nMettre à jour :\n" + "\n".join(str(p.relative_to(plan.root)) for p in plan.updated) + except (OSError, ValueError, TypeError) as error: + text = str(error) + self.removal_preview.configure(state="normal") + self.removal_preview.delete("1.0", "end") + self.removal_preview.insert("1.0", text) + self.removal_preview.configure(state="disabled") + combo.bind("<>", preview) + return row + 2 if step.id == "statement_groups" or ( step.id == "annotation" and self.variant_var.get() == "grouped" ): @@ -1252,7 +1282,7 @@ class CopienatorApp(tk.Tk): summary = str(exc) ttk.Label(self.form, text=summary, wraplength=600, justify="left").grid(row=row, column=0, columnspan=2, sticky="w") return row + 1 - if step.id in {"review_persp", "correction"}: + if step.id == "correction": row = self._correction_folder_buttons(row) copy_selection_steps = { "page_splitter", @@ -1435,6 +1465,11 @@ class CopienatorApp(tk.Tk): def _open_persp(self) -> None: evaluation = self.evaluation + if self.current_step and self.current_step.id == "statement_persp": + target = self.arg_vars.get("target") + value = str(target.get()).strip() if target else "" + if value: + evaluation = (self.repository / value).resolve() self._open_desktop_path(evaluation / "Persp" if evaluation else None, "dossier Persp") def _open_label_groups(self) -> None: @@ -1738,6 +1773,9 @@ class CopienatorApp(tk.Tk): if self.active_step_id or self.runner.running: messagebox.showwarning("Traitement en cours", "Interrompez le traitement actuel avant d’en lancer un autre.") return + if step.id == "remove_copy": + self._remove_selected_copy() + return if step.id == "refaire_selection": self._save_refaire_selection() return @@ -1893,7 +1931,8 @@ class CopienatorApp(tk.Tk): return self._save_current_form() # Score edits already invalidate downstream steps when Set saves a change. - if self.current_step.id not in {"edit_scores", "general_remarks"}: + if self.current_step.id not in {"edit_scores", "general_remarks"} and not ( + status == "skipped" and self.current_step.skip_on_forward): self._invalidate_after_step(self.current_step, self._current_variant().id) self.state_store.update_step(self.current_step.id, status=status) history: dict[str, object] = { @@ -1910,6 +1949,32 @@ class CopienatorApp(tk.Tk): f"{self.current_step.title} : {STATUS_LABELS.get(status, status).lower()}{detail}." ) + def _remove_selected_copy(self) -> None: + from copienator.copy_removal import removal_plan, remove_copy + copy = self.removal_copy_var.get() + try: + plan = removal_plan(self.evaluation, copy) + except (OSError, ValueError, TypeError) as error: + messagebox.showerror("Suppression impossible", str(error)) + return + if not messagebox.askyesno( + "Supprimer une copie", + f"Supprimer {copy} et ses artefacts ?\n\n" + f"{len(plan.removed)} fichiers ou dossiers retirés ; {len(plan.updated)} fichiers partagés mis à jour.\n" + "Les originaux seront sauvegardés dans BackUp avant toute modification.\n" + "Les autres copies ne seront pas renumérotées. Les lots de correction existants " + "devront être recréés ; les étapes suivantes seront à revalider.", + ): + return + try: + backup = remove_copy(self.evaluation, copy) + except Exception as error: + messagebox.showerror("Suppression impossible", str(error)) + return + self._mark_step("success", reason=f"{copy} supprimée ; sauvegarde : {backup}") + self._render_step() + self.info_var.set(f"{copy} supprimée — sauvegarde : {backup}") + def _validate_rename_step(self) -> None: if (not self.current_step or self.current_step.id != "rename" or not self.state_store.evaluation or self.active_step_id or self.runner.running): @@ -2254,6 +2319,7 @@ class CopienatorApp(tk.Tk): self._render_step() if step_id == 'statement_persp': self._refresh_persp_inventory() + return if step_id == "statement" and self._latex_warning_paths: self.after_idle(self._show_current_latex_warnings) if step_id == "personal_read": @@ -2394,6 +2460,14 @@ class CopienatorApp(tk.Tk): target = max(0, min(len(ids) - 1, index + delta)) if target == index: return + while delta > 0 and target < len(ids) - 1 and self.step_by_id[ids[target]].skip_on_forward: + skipped = ids[target] + entry = self.state_store.step(skipped) + if not entry.get("status"): + self.state_store.update_step(skipped, status="skipped", visited=True) + if self.tree.exists(skipped): + self.tree.item(skipped, values=(STATUS_LABELS["skipped"],)) + target += 1 self.tree.selection_set(ids[target]) self.tree.see(ids[target]) diff --git a/copienator_gui/workflow.py b/copienator_gui/workflow.py index 754a680..209d88d 100644 --- a/copienator_gui/workflow.py +++ b/copienator_gui/workflow.py @@ -53,6 +53,7 @@ class StepDefinition: personal: bool = False requires: tuple[str, ...] = () artifacts: tuple[str, ...] = () + skip_on_forward: bool = False auto_start_first_visit: bool = False skip_for_live_correction: bool = False skip_without_manual_conflicts: bool = False @@ -194,27 +195,19 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: optional=True, personal=True, requires=("labels", "Text2", "Sol2"), ), StepDefinition( - "statement_persp", "Prétraitement de l’énoncé", "Générer les barèmes manquants avec Gemini", - "Génère uniquement les barèmes manquants dans Persp et conserve ceux déjà présents. " + "statement_persp", "Prétraitement de l’énoncé", "Générer les barèmes avec Gemini", + "Par défaut, supprime les barèmes actuels dans Persp et tous les régénère. " + "Décochez Delete current Persp pour générer uniquement les barèmes manquants et conserver ceux déjà présents. " "Les fichiers vides ou contenant seulement le label sont considérés comme manquants. " "Après un échec, relancez cette étape pour compléter les barèmes. " - "Cochez Delete current Persp pour supprimer les barèmes actuels au lancement et tous les régénérer.", + "Après la génération, ouvrez Persp pour vérifier les résultats, puis passez à l’étape suivante.", (python("default", "Barèmes Gemini", "statement", fixed_args=("--persp-only",)),), arguments=(arg_target("le dossier de l’évaluation"), ArgumentSpec('delete_persp', 'Delete current Persp', kind='bool', - flag='--delete-persp', default=False, + flag='--delete-persp', default=True, help='Supprime tous les fichiers actuels de Persp au lancement, y compris les barèmes modifiés manuellement. Puis régénère tous les barèmes.')), optional=True, personal=True, requires=("labels", "label_groups", "Text2", "Sol2"), ), - StepDefinition( - "review_persp", - "Prétraitement de l’énoncé", - "Relire les barèmes", - "Étape manuelle facultative : vérifier et modifier les instructions de correction.", - (manual("review"),), - optional=True, - requires=("Persp",), - ), StepDefinition( "rotate", "Prétraitement des copies", @@ -252,6 +245,17 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]: ), arguments=(arg_target("le dossier de l’évaluation"),), ), + StepDefinition( + "remove_copy", + "Prétraitement des copies", + "Supprimer une copie", + "Étape facultative, ignorée au premier passage. Sélectionnez une copie pour retirer ses fichiers " + "et artefacts. Chaque fichier supprimé ou modifié est sauvegardé dans BackUp, dans l’évaluation. " + "Les autres copies gardent leur numéro. Les lots de correction partagés devront être recréés.", + (CommandVariant("default", "Suppression avec sauvegarde", None, "manual"),), + optional=True, + skip_on_forward=True, + ), StepDefinition( "page_splitter", "Prétraitement des copies", diff --git a/docs/guide_utilisateur_gui.md b/docs/guide_utilisateur_gui.md index 36762ac..2bd0ffb 100644 --- a/docs/guide_utilisateur_gui.md +++ b/docs/guide_utilisateur_gui.md @@ -51,6 +51,14 @@ Exécutez seulement si toutes les copies sont à l’envers. Vérifiez qu’un PDF correspond à une copie et que le dossier ne contient pas de documents parasites. Exécutez puis contrôlez le nombre de copies. +### Supprimer une copie — facultatif + +Cette étape suit le renommage et est ignorée lorsque vous avancez dans le parcours. Cliquez dessus dans la liste, ou revenez en arrière, pour l’utiliser. + +Choisissez une copie, examinez les fichiers affichés, puis cliquez sur **Supprimer la copie sélectionnée** et confirmez. Les scans, labels, découpages, annotations et livrables de cette copie sont retirés sans renuméroter les autres. Les données de correction partagées sont mises à jour ; les anciennes requêtes par lots et les grilles de noms sont retirées pour éviter de réintroduire la copie. + +Chaque fichier supprimé ou modifié est sauvegardé dans `BackUp/-CopieXX-/files`, avec son chemin d’origine et un `manifest.json`. Le nettoyage final conserve ces sauvegardes. Les étapes suivantes passent à **À revalider** ; les lots de correction retirés doivent être recréés. Les fichiers déjà exportés hors de l’évaluation restent à gérer séparément. + ### Séparer et réordonner les pages **Effet :** Découpe les scans A3 en pages A4 correctement ordonnées. diff --git a/tests/test_copy_removal.py b/tests/test_copy_removal.py new file mode 100644 index 0000000..b50988d --- /dev/null +++ b/tests/test_copy_removal.py @@ -0,0 +1,140 @@ +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from PIL import Image + +from copienator.copy_removal import available_copies, removal_plan, remove_copy + + +class CopyRemovalTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + for name in ('Copies/Copie24.pdf', 'Copies/Copie25.pdf', 'Copies Originales/Copie24.pdf', + 'Cutleft/Copie24_01.jpg', 'Copies/Copie24/Ex 1.pdf', + 'BGnot/Copie24/Concat.pdf', 'A Rendre/Antoine (24)/score.json', + '.copienator/runs/old/Copies/Copie24.pdf', + 'BackUp/previous/Copie24.pdf'): + self.write(name, b'original bytes') + + def write(self, name, value): + p = self.root / name + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes(value) + return p + + def test_removes_all_copy_artifacts_and_preserves_neighbors_and_backups(self): + self.write('Copies/Copie240.pdf', b'other copy') + self.write('Copies/Copie24_01.json', b'{"name": "Antoine"}') + self.write('.copienator/copy_errors.json', b'{"Copie24.pdf":{"plotting":"bad"}, "Copie25.pdf":{"plotting":"keep"}}') + self.write('correction.json', b'{"Ex 1":[[{"id":"24","result":{"score":3}}, {"id":"25","result":{"score":4}}]]}') + self.write('refaire.json', b'[["Copie24", []], ["Copie25", []]]') + self.write('general_remarks.json', b'{"24":"remove", "25":"keep"}') + self.write('batch_requests_flash.jsonl', b'old payload') + self.write('correction_pending_responses.json', b'{"group": "cached raw response"}') + self.write('manual_resolutions.txt', b'Copie24 Ex 1 -> Ex 2\nCopie25 Ex 1 -> Ex 2\n') + before = {p.relative_to(self.root): p.read_bytes() for p in self.root.rglob('*') if p.is_file()} + plan = removal_plan(self.root, 'Copie24') + backup = remove_copy(self.root, 'Copie24') + self.assertEqual(available_copies(self.root), ['Copie25', 'Copie240']) + self.assertFalse((self.root/'correction_pending_responses.json').exists()) + self.assertEqual((self.root/'manual_resolutions.txt').read_text(), 'Copie25 Ex 1 -> Ex 2\n') + for p in plan.removed: + self.assertFalse(p.exists()) + for relative in [p.relative_to(self.root) for p in (*plan.removed, *plan.updated) if p.relative_to(self.root) in before]: + self.assertEqual((backup / 'files' / relative).read_bytes(), before[relative]) + self.assertEqual((self.root / 'Copies/Copie25.pdf').read_bytes(), b'original bytes') + self.assertEqual((self.root / 'BackUp/previous/Copie24.pdf').read_bytes(), b'original bytes') + self.assertEqual(json.loads((self.root / 'correction.json').read_text())['Ex 1'][0], [{'id':'25','result':{'score':4}}]) + self.assertEqual(json.loads((self.root / 'refaire.json').read_text()), [['Copie25', []]]) + self.assertEqual(json.loads((self.root / 'general_remarks.json').read_text()), {'25':'keep'}) + self.assertEqual(json.loads((backup / 'manifest.json').read_text())['status'], 'complete') + + def test_group_removal_preserves_other_students_coordinates(self): + path = self.write('Par label/Ex 1/Group_1.json', b'[["24", 50, 100, 1, "Ex 1"], ["25", 160, 210, 1, "Ex 1"]]') + Image.new('RGB', (100, 220), 'blue').save(path.with_suffix('.jpg')) + remove_copy(self.root, 'Copie24') + self.assertEqual(json.loads(path.read_text()), [['25',160,210,1,'Ex 1']]) + with Image.open(path.with_suffix('.jpg')) as image: + self.assertGreater(image.getpixel((20,60))[0], 240) + self.assertLess(image.getpixel((20,180))[0], 30) + + def test_failure_restores_already_removed_files(self): + from copienator.copy_removal import atomic_write_bytes + self.write('refaire.json', b'[["Copie24", []]]') + original = (self.root/'Copies/Copie24.pdf').read_bytes() + def failing_write(path, data): + if path == self.root/'refaire.json': + raise OSError('simulated full disk') + return atomic_write_bytes(path, data) + with patch('copienator.copy_removal.atomic_write_bytes', side_effect=failing_write): + with self.assertRaises(OSError): + remove_copy(self.root, 'Copie24') + self.assertEqual((self.root/'Copies/Copie24.pdf').read_bytes(), original) + self.assertEqual((self.root/'refaire.json').read_bytes(), b'[["Copie24", []]]') + + def test_cleanup_keeps_removal_backups(self): + from copienator import EvaluationWorkspace + from copienator.commands.clean import build_cleanup_plan + self.write('correction.json', b'{}') + self.write('A Rendre/Antoine (24)/Antoine.jpg', b'returned image') + plan = build_cleanup_plan(EvaluationWorkspace(self.root)) + saved = self.root/'BackUp/previous/Copie24.pdf' + self.assertIn(saved, plan.kept_files) + self.assertNotIn(saved, plan.deleted_files) + self.assertNotIn(saved.parent, plan.deleted_directories) + + def test_invalid_selection_and_external_symlink(self): + for invalid in ('../Copie24', 'Copie99', ''): + with self.assertRaises(ValueError): + removal_plan(self.root, invalid) + with tempfile.TemporaryDirectory() as external: + p = Path(external)/'Copie24.pdf' + p.write_bytes(b'outside') + (self.root/'external').symlink_to(external, target_is_directory=True) + remove_copy(self.root, 'Copie24') + self.assertEqual(p.read_bytes(), b'outside') + + +@unittest.skipUnless(os.environ.get('DISPLAY'), 'Tk requires a display') +class CopyRemovalGuiTests(unittest.TestCase): + def test_first_visit_skips_without_invalidating_then_revisit_can_remove(self): + from copienator_gui.app import CopienatorApp + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + evaluation = root/'Eval' + evaluation.mkdir() + (evaluation/'Copie24.pdf').write_bytes(b'copy') + app = CopienatorApp(root, False, evaluation) + app.update() + try: + steps = [s.id for s in app.steps] + self.assertEqual(steps[steps.index('rename')+1], 'remove_copy') + app.state_store.update_step('page_splitter', status='success') + step = app.step_by_id['remove_copy'] + app._move_selection_from('rename', 1) + app.update() + self.assertEqual(app.current_step.id, 'page_splitter') + self.assertEqual(app.state_store.step('remove_copy')['status'], 'skipped') + self.assertEqual(app.state_store.step('page_splitter')['status'], 'success') + app.current_step = step + app._render_step() + app._handle_first_visit(step) + app.removal_copy_var.set('Copie24') + with patch('copienator_gui.app.messagebox.askyesno', return_value=False): + app._run_current_step() + self.assertTrue((evaluation/'Copie24.pdf').exists()) + with patch('copienator_gui.app.messagebox.askyesno', return_value=True): + app._run_current_step() + self.assertFalse((evaluation/'Copie24.pdf').exists()) + self.assertEqual(app.state_store.step('page_splitter')['status'], 'stale') + self.assertEqual(len(list((evaluation/'BackUp').glob('*/manifest.json'))), 1) + finally: + for callback in app.tk.splitlist(app.tk.call('after', 'info')): + app.after_cancel(callback) + app.destroy() diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py index 039b0e5..2da7ff5 100644 --- a/tests/test_gui_core.py +++ b/tests/test_gui_core.py @@ -2530,8 +2530,8 @@ class WorkflowTests(unittest.TestCase): self.assertFalse(default_config.ALWAYS_CROP) self.assertTrue(config.ALWAYS_CROP) - def test_review_persp_has_shorter_title(self) -> None: - self.assertEqual(self.steps["review_persp"].title, "Relire les barèmes") + def test_rubric_review_is_part_of_generation(self) -> None: + self.assertNotIn("review_persp", self.steps) def test_cutleft_title_and_shortcuts_are_documented(self) -> None: step = self.steps["cutleft"] diff --git a/tests/test_statement_personal_flow.py b/tests/test_statement_personal_flow.py index b8a004a..9aeb054 100644 --- a/tests/test_statement_personal_flow.py +++ b/tests/test_statement_personal_flow.py @@ -82,9 +82,12 @@ class PersonalStatementTests(unittest.TestCase): def test_personal_choice_and_optional_steps_only_in_personal_profile(self): standard = {step.id: step for step in build_workflow(False)} enabled = {step.id: step for step in build_workflow(True)} - self.assertEqual([variant.id for variant in standard["statement"].variants], ["gemini"]) + self.assertEqual([variant.id for variant in standard["statement"].variants], + ["gemini", "compile"]) + self.assertEqual([variant.id for variant in enabled["statement"].variants], + ["personal", "gemini", "compile"]) self.assertEqual([variant.program for variant in enabled["statement"].variants], - ["statement-personal", "statement"]) + ["statement-personal", "statement", "statement"]) for ident in ("statement_groups", "statement_persp"): self.assertNotIn(ident, standard) self.assertTrue(enabled[ident].optional) @@ -392,13 +395,26 @@ class PersonalStatementGuiTests(unittest.TestCase): app._refresh_persp_inventory() self.assertIn('Barèmes présents : 1/2', app.persp_inventory_label.cget('text')) self.assertIn('Manquants : Ex 2', app.persp_inventory_label.cget('text')) - self.assertFalse(app.arg_vars['delete_persp'].get()) - app.arg_vars['delete_persp'].set(True) + self.assertTrue(app.arg_vars['delete_persp'].get()) self.assertIn('--delete-persp', app.command_var.get()) + app.arg_vars['delete_persp'].set(False) app._save_current_form() app._render_step() - self.assertFalse(app.arg_vars['delete_persp'].get()) - self.assertNotIn('--delete-persp', app.command_var.get()) + self.assertTrue(app.arg_vars['delete_persp'].get()) + self.assertIn('--delete-persp', app.command_var.get()) + folder_button = next(child for child in app.form.winfo_children() + if isinstance(child, ttk.Button) + and child.cget('text') == 'Ouvrir le dossier Persp') + with patch('copienator_gui.app.open_path') as opened: + folder_button.invoke() + opened.assert_called_once_with(evaluation / 'Persp') + with patch.object(app, '_move_selection_from') as advance: + app.active_step_id = 'statement_persp' + app._finish_process(0, False) + app.update() + advance.assert_not_called() + self.assertEqual(app.current_step.id, 'statement_persp') + self.assertEqual(app.state_store.step('statement_persp')['status'], 'success') app.tree.selection_set('statement') app.update() app._select_variant(1)