diff --git a/Readme.org b/Readme.org index 51ed7cd..20d5afa 100644 --- a/Readme.org +++ b/Readme.org @@ -147,6 +147,29 @@ 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. +*** Convention des scripts standardisés + +Les scripts standardisés exposent =build_parser()=, =run(...)= et +=main(argv=None)=. Leur import ne lance aucun traitement. Ils acceptent +le dossier d'évaluation comme premier argument positionnel, utilisent +=EvaluationWorkspace= pour les chemins partagés et peuvent afficher la +trace complète d'une erreur avec =--verbose=. + +Les codes de sortie communs sont : + +| Code | Signification | +|------+---------------| +| 0 | réussite | +| 1 | erreur de traitement | +| 2 | arguments invalides | +| 3 | évaluation ou prérequis invalides | +| 4 | traitement partiel, avec avertissements | +| 130 | interruption par l'utilisateur | + +Le GUI distingue notamment un traitement partiel d'un échec. Les +premiers scripts migrés vers cette convention sont =export.py=, +=import.py=, =giving_names.py= et =grouping.py=. + ** 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 index ffc8901..bdc56e3 100644 --- a/copienator/__init__.py +++ b/copienator/__init__.py @@ -1,5 +1,13 @@ """Core building blocks shared by Copienator scripts and interfaces.""" +from .cli import ( + CliError, + ExitCode, + evaluation_parser, + evaluation_workspace, + execute, + workspace_from_args, +) from .json_io import ( JsonLockTimeout, atomic_update_json, @@ -14,12 +22,18 @@ from .workspace import ( ) __all__ = [ + "CliError", "EvaluationWorkspace", + "ExitCode", "JsonLockTimeout", "WorkspaceNotFoundError", "WorkspaceValidationError", "atomic_update_json", "atomic_write_json", "atomic_write_text", + "evaluation_parser", + "evaluation_workspace", + "execute", "read_json", + "workspace_from_args", ] diff --git a/copienator/cli.py b/copienator/cli.py new file mode 100644 index 0000000..509b6f8 --- /dev/null +++ b/copienator/cli.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import argparse +import sys +import traceback +from collections.abc import Callable, Sequence +from enum import IntEnum +from pathlib import Path +from typing import Any + +from .workspace import ( + EvaluationWorkspace, + WorkspaceNotFoundError, + WorkspaceValidationError, +) + + +class ExitCode(IntEnum): + SUCCESS = 0 + FAILURE = 1 + INVALID_ARGUMENTS = 2 + INVALID_WORKSPACE = 3 + PARTIAL = 4 + INTERRUPTED = 130 + + +class CliError(Exception): + def __init__(self, message: str, exit_code: ExitCode = ExitCode.FAILURE) -> None: + self.exit_code = exit_code + super().__init__(message) + + +def evaluation_parser(description: str) -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=description) + parser.add_argument( + "evaluation", + type=Path, + help="Evaluation directory", + ) + parser.add_argument( + "--verbose", + action="store_true", + help="Show a traceback when an unexpected error occurs", + ) + return parser + + +def evaluation_workspace( + path: str | Path, + *, + repository: str | Path | None = None, +) -> EvaluationWorkspace: + root = Path(path).expanduser().resolve() + if not root.exists(): + raise CliError( + f"Evaluation directory does not exist: {root}", + ExitCode.INVALID_WORKSPACE, + ) + if not root.is_dir(): + raise CliError( + f"Evaluation path is not a directory: {root}", + ExitCode.INVALID_WORKSPACE, + ) + return EvaluationWorkspace(root, Path(repository) if repository is not None else None) + + +def workspace_from_args( + args: argparse.Namespace, + *, + repository: str | Path | None = None, +) -> EvaluationWorkspace: + return evaluation_workspace(args.evaluation, repository=repository) + + +def execute( + parser: argparse.ArgumentParser, + argv: Sequence[str] | None, + handler: Callable[[argparse.Namespace], int | ExitCode | None], +) -> int: + args = parser.parse_args(argv) + try: + result = handler(args) + except CliError as exc: + print(f"Error: {exc}", file=sys.stderr) + return int(exc.exit_code) + except (WorkspaceNotFoundError, WorkspaceValidationError) as exc: + print(f"Error: {exc}", file=sys.stderr) + return int(ExitCode.INVALID_WORKSPACE) + except KeyboardInterrupt: + print("Interrupted by user.", file=sys.stderr) + return int(ExitCode.INTERRUPTED) + except Exception as exc: # noqa: BLE001 - executable boundary + if getattr(args, "verbose", False): + traceback.print_exc() + else: + print(f"Error: {exc}", file=sys.stderr) + return int(ExitCode.FAILURE) + return int(ExitCode.SUCCESS if result is None else result) + + +def stderr(message: Any) -> None: + print(message, file=sys.stderr) diff --git a/copienator/workspace.py b/copienator/workspace.py index d86667a..46e3dc7 100644 --- a/copienator/workspace.py +++ b/copienator/workspace.py @@ -186,6 +186,24 @@ class EvaluationWorkspace: if missing: raise WorkspaceValidationError(self.root, missing) + def require_files(self, *relative_paths: str) -> None: + missing = [ + relative_path + for relative_path in relative_paths + if not (self.root / relative_path).is_file() + ] + if missing: + raise WorkspaceValidationError(self.root, missing) + + def require_directories(self, *relative_paths: str) -> None: + missing = [ + relative_path + for relative_path in relative_paths + if not (self.root / relative_path).is_dir() + ] + 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) diff --git a/copienator_gui/app.py b/copienator_gui/app.py index 5171c75..fd566e7 100644 --- a/copienator_gui/app.py +++ b/copienator_gui/app.py @@ -7,6 +7,7 @@ from pathlib import Path from tkinter import filedialog, messagebox, ttk from typing import Any +from copienator import ExitCode from platform_utils import WindowsLabelError, validate_windows_labels from .diagnostics import collect_diagnostics @@ -26,6 +27,7 @@ STATUS_LABELS = { "ready": "Prête", "running": "En cours", "success": "Réussie", + "partial": "Partielle", "failed": "Échouée", "interrupted": "Interrompue", "skipped": "Ignorée", @@ -34,6 +36,16 @@ STATUS_LABELS = { } +def process_status(return_code: int, interrupted: bool = False) -> str: + if interrupted or return_code == ExitCode.INTERRUPTED: + return "interrupted" + if return_code == ExitCode.SUCCESS: + return "success" + if return_code == ExitCode.PARTIAL: + return "partial" + return "failed" + + class CopienatorApp(tk.Tk): def __init__( self, @@ -624,10 +636,7 @@ class CopienatorApp(tk.Tk): step_id = self.active_step_id if not step_id: return - if interrupted: - status = "interrupted" - else: - status = "success" if return_code == 0 else "failed" + status = process_status(return_code, interrupted) self.state_store.update_step(step_id, status=status, return_code=return_code) self.state_store.add_history( { diff --git a/export.py b/export.py index 00b0dde..25d5059 100644 --- a/export.py +++ b/export.py @@ -1,44 +1,65 @@ import argparse import sys +from collections.abc import Sequence from pathlib import Path from config import EXPORT_DIR +from copienator import ( + EvaluationWorkspace, + ExitCode, + evaluation_parser, + execute, + workspace_from_args, +) from platform_utils import replace_with_link_or_copy -def export_directory(base_dir, source_dir_name): - base_dir = Path(base_dir).expanduser().resolve() - source_dir = base_dir / source_dir_name - sync_dir = Path(EXPORT_DIR).expanduser() / base_dir.name +def export_directory( + workspace: EvaluationWorkspace, + source_dir_name: str, +) -> ExitCode: + workspace.require_directories(source_dir_name) + source_dir = workspace.root / source_dir_name + sync_dir = Path(EXPORT_DIR).expanduser() / workspace.name sync_dir.mkdir(parents=True, exist_ok=True) - if not source_dir.is_dir(): - print(f"Erreur : le sous-dossier {source_dir} n'existe pas.") - return - subdirs = [directory for directory in source_dir.iterdir() if directory.is_dir()] if source_dir_name == "BGnot" and subdirs: all_start_with_copie = all(directory.name.startswith("Copie") for directory in subdirs) if not all_start_with_copie: subdirs = [directory for directory in subdirs if not directory.name.startswith("Copie")] + missing_outputs = 0 for subdir in subdirs: concat_file = subdir / "Concat.pdf" if not concat_file.is_file(): - print(f"Attention : le fichier {concat_file} est introuvable.") + print(f"Warning: file not found: {concat_file}", file=sys.stderr) + missing_outputs += 1 continue destination = sync_dir / f"{subdir.name}.pdf" 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 + + +def build_parser() -> argparse.ArgumentParser: + parser = evaluation_parser("Export annotated PDFs to the tablet directory.") + parser.add_argument("--refaire", action="store_true", help="Process only copies/labels defined in refaire.json") + return parser + + +def run(workspace: EvaluationWorkspace, *, refaire: bool = False) -> ExitCode: + return export_directory(workspace, "BRnot" if refaire else "BGnot") + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + return execute( + parser, + argv, + lambda args: run(workspace_from_args(args), refaire=args.refaire), + ) if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Move to tablette folder.") - parser.add_argument("dir", help="The directory to process") - parser.add_argument("--refaire", action="store_true", help="Process only copies/labels defined in refaire.json") - - args = parser.parse_args() - if args.refaire: - export_directory(args.dir, "BRnot") - sys.exit(0) - export_directory(args.dir, "BGnot") + raise SystemExit(main()) diff --git a/giving_names.py b/giving_names.py index fd14c9c..5c68eb6 100644 --- a/giving_names.py +++ b/giving_names.py @@ -1,102 +1,165 @@ -import json -import os +from __future__ import annotations + +import argparse import re import sys from collections import defaultdict +from collections.abc import Sequence from pathlib import Path +from copienator import ( + EvaluationWorkspace, + ExitCode, + evaluation_parser, + execute, + read_json, + workspace_from_args, +) from platform_utils import replace_with_link_or_copy, safe_filename -def main(): - if len(sys.argv) < 3: - print("Usage: python giving_names.py ") - sys.exit(1) +ANNOTATION_CHOICES = ("BGnot", "Bnot", "Anot") - work_dir = os.path.abspath(sys.argv[1]) - copies_dir = Path(work_dir) / "Copies" - bnot_dir = sys.argv[2] - target_subdir = os.path.join(work_dir, "A Rendre") - os.makedirs(target_subdir, exist_ok=True) - # --- 1. Load the expected names list --- - expected_names = set() - names_path = os.path.join(work_dir, "names") - if not os.path.exists(names_path): - names_path = "names" # Fallback to current dir +def build_parser() -> argparse.ArgumentParser: + parser = evaluation_parser("Assign student names and prepare the return directory.") + parser.add_argument( + "annotation_dir", + choices=ANNOTATION_CHOICES, + help="Annotation directory to use", + ) + return parser - if os.path.exists(names_path): - with open(names_path, 'r', encoding='utf-8') as f: - expected_names = {line.strip() for line in f if line.strip()} - else: - print(f"Warning: 'names' file not found in {work_dir} or current directory.") - # --- 2. Existing Collection Logic --- +def _read_expected_names(workspace: EvaluationWorkspace) -> set[str]: + names_path = workspace.names_file() + if not names_path.exists(): + print( + f"Warning: names file not found in {workspace.root} or the current directory.", + file=sys.stderr, + ) + return set() + return { + line.strip() + for line in names_path.read_text(encoding="utf-8").splitlines() + if line.strip() + } + + +def prepare_named_returns( + workspace: EvaluationWorkspace, + annotation_dir_name: str, +) -> ExitCode: + workspace.require_directories("Copies", annotation_dir_name) + workspace.return_dir.mkdir(parents=True, exist_ok=True) + + expected_names = _read_expected_names(workspace) + copies_map: defaultdict[str, list[str]] = defaultdict(list) pattern = re.compile(r"^Copie(\d+)\.json$") - copies_map = defaultdict(list) - assigned_names = set() # To track which names were successfully linked + had_errors = False - for filename in os.listdir(copies_dir): - match = pattern.match(filename) - if match: - copie_id = match.group(1) - json_path = os.path.join(copies_dir, filename) - try: - with open(json_path, 'r', encoding='utf-8') as f: - data = json.load(f) - name = data.get("name", "Unknown").strip() - copies_map[name].append(copie_id) - except Exception as e: - print(f"Error processing {filename}: {e}") + for json_path in workspace.copies_dir.iterdir(): + match = pattern.match(json_path.name) + if not match: + continue + try: + data = read_json(json_path) + if not isinstance(data, dict): + raise TypeError("expected a JSON object") + name = str(data.get("name", "Unknown")).strip() + copies_map[name].append(match.group(1)) + except (OSError, TypeError, ValueError) as exc: + print(f"Error processing {json_path}: {exc}", file=sys.stderr) + had_errors = True - # --- 3. Process and Link --- - for name, ids in copies_map.items(): + assigned_names: set[str] = set() + selected_annotations = workspace.root / annotation_dir_name + fallback_annotations = workspace.annotation_dir("simple") + + for name, copy_ids in copies_map.items(): if name == "Unknown": - print(f"ALERT: 'Unknown' name found for IDs: {', '.join(ids)}") - elif len(ids) > 1: - print(f"ALERT: Name '{name}' assigned to multiple IDs: {', '.join(ids)}") + print( + f"Warning: unknown name for copies: {', '.join(copy_ids)}", + file=sys.stderr, + ) + elif len(copy_ids) > 1: + print( + f"Warning: name {name!r} is assigned to multiple copies: " + f"{', '.join(copy_ids)}", + file=sys.stderr, + ) safe_name = safe_filename(name) - - for copie_id in ids: - path_b = os.path.join(work_dir, f"{bnot_dir}/Copie{copie_id}") - path_a = os.path.join(work_dir, f"Anot/Copie{copie_id}") - + for copy_id in copy_ids: + selected = selected_annotations / f"Copie{copy_id}" + fallback = fallback_annotations / f"Copie{copy_id}" source_folder = None - if os.path.exists(os.path.join(path_b, "Concat.jpg")) and os.path.exists(os.path.join(path_b, "score.json")): - source_folder = path_b - elif os.path.exists(os.path.join(path_a, "Concat.jpg")) and os.path.exists(os.path.join(path_a, "score.json")): - source_folder = path_a - - if not source_folder: + for candidate in (selected, fallback): + if (candidate / "Concat.jpg").exists() and ( + candidate / "score.json" + ).exists(): + source_folder = candidate + break + if source_folder is None: continue - # If we reached here, the link is possible assigned_names.add(name) - - dest_folder_name = f"{safe_name} ({copie_id})" - dest_path = os.path.join(target_subdir, dest_folder_name) - os.makedirs(dest_path, exist_ok=True) - - links = [("Concat.jpg", f"{safe_name}.jpg"),("Concat_F.pdf", f"{safe_name}.pdf"), ("score.json", "score.json")] - for src_name, dst_name in links: - src_file = os.path.join(source_folder, src_name) - dst_link = os.path.join(dest_path, dst_name) + destination = workspace.return_dir / f"{safe_name} ({copy_id})" + destination.mkdir(parents=True, exist_ok=True) + links = ( + ("Concat.jpg", f"{safe_name}.jpg"), + ("Concat_F.pdf", f"{safe_name}.pdf"), + ("score.json", "score.json"), + ) + for source_name, destination_name in links: + source = source_folder / source_name + if not source.exists(): + continue try: - if os.path.exists(src_file): - method = replace_with_link_or_copy(src_file, dst_link, prefer="symlink") - if method == "copy": - print(f"Copied {src_name} for {dest_folder_name} (links unavailable)") - except OSError as e: - print(f"Error linking {src_name} for {dest_folder_name}: {e}") + method = replace_with_link_or_copy( + source, + destination / destination_name, + prefer="symlink", + ) + if method == "copy": + print( + f"Copied {source_name} for {destination.name} " + "(links unavailable)" + ) + except OSError as exc: + print( + f"Error linking {source} for {destination.name}: {exc}", + file=sys.stderr, + ) + had_errors = True - # --- 4. Print Unassigned Names --- unassigned = expected_names - assigned_names if unassigned: - print("\n" + "!" * 40) - print("NAMES FROM LIST NOT ASSIGNED:") - for n in sorted(unassigned): - print(f" - {n}") - print("!" * 40) + print("Names from the list that were not assigned:", file=sys.stderr) + for name in sorted(unassigned): + print(f" - {name}", file=sys.stderr) + + return ExitCode.PARTIAL if had_errors else ExitCode.SUCCESS + + +def run( + workspace: EvaluationWorkspace, + *, + annotation_dir: str, +) -> ExitCode: + return prepare_named_returns(workspace, annotation_dir) + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + return execute( + parser, + argv, + lambda args: run( + workspace_from_args(args, repository=Path.cwd()), + annotation_dir=args.annotation_dir, + ), + ) + if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/grouping.py b/grouping.py index 0008f04..f28c5e9 100644 --- a/grouping.py +++ b/grouping.py @@ -1,13 +1,22 @@ +import argparse import os -import json import re -import sys import shutil -from pathlib import Path from collections import defaultdict +from collections.abc import Sequence from concurrent.futures import ThreadPoolExecutor -from PIL import Image, ImageDraw, ImageFont + from pdf2image import convert_from_path, pdfinfo_from_path +from PIL import Image, ImageDraw, ImageFont + +from copienator import ( + EvaluationWorkspace, + ExitCode, + atomic_write_json, + evaluation_parser, + execute, + workspace_from_args, +) # Configuration DPI = 200 # Good balance for readability and size @@ -34,7 +43,7 @@ def get_pdf_height(path): # Return total height return single_page_px * num_pages - except Exception as e: + except Exception as e: # noqa: BLE001 - pdfinfo may raise backend-specific errors print(f"Error reading {path}: {e}") return 0 @@ -78,7 +87,7 @@ def group_files(file_list): groups = [] for item in sorted_files: - dd, path, height = item + _, _, height = item placed = False # 2. Try to fit item into an existing group (First Fit) @@ -140,7 +149,7 @@ def create_jpg(identifier, group_index, group, root_dir): combined_img = stitch_pdf_pages(imgs) if combined_img: images.append((dd, combined_img)) - except Exception as e: + except Exception as e: # noqa: BLE001 - PDF/image backends vary by platform print(f"Failed to convert {path}: {e}") if not images: @@ -159,7 +168,7 @@ def create_jpg(identifier, group_index, group, root_dir): # Try loading a font, fallback to default try: font = ImageFont.truetype("DejaVuSans.ttf", 40) - except IOError: + except OSError: print("font not found") font = ImageFont.load_default() @@ -193,8 +202,7 @@ def create_jpg(identifier, group_index, group, root_dir): # Save JSON metadata json_filename = f"Group_{group_index+1}.json" json_path = os.path.join(target_folder, json_filename) - with open(json_path, 'w') as f: - json.dump(metadata, f) + atomic_write_json(json_path, metadata, indent=None) # Save with size constraints output_filename = f"Group_{group_index+1}.jpg" @@ -227,18 +235,16 @@ def process_identifier(identifier, files_info, output_dir): for idx, group in enumerate(file_groups): create_jpg(identifier, idx, group, output_dir) -def main(): - if len(sys.argv) < 2: - print("Usage: python app.py ") - sys.exit(1) +def build_parser() -> argparse.ArgumentParser: + return evaluation_parser("Group copy extracts by question label.") - root_dir = Path(sys.argv[1]) - copies_dir = root_dir / "Copies" - par_label_dir = root_dir / "Par label" +def run(workspace: EvaluationWorkspace) -> ExitCode: + workspace.require_directories("Copies") + workspace.groups_dir.mkdir(parents=True, exist_ok=True) print("Scanning files...") - data = collect_files(copies_dir) + data = collect_files(workspace.copies_dir) print(f"Found {len(data)} identifiers. Processing...") @@ -247,11 +253,25 @@ def main(): # Process using 8 threads with ThreadPoolExecutor(max_workers=8) as executor: - for identifier in sorted_identifiers: - executor.submit(process_identifier, identifier, data[identifier], - par_label_dir) + futures = [ + executor.submit( + process_identifier, + identifier, + data[identifier], + workspace.groups_dir, + ) + for identifier in sorted_identifiers + ] + for future in futures: + future.result() print("Done.") + return ExitCode.SUCCESS + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + return execute(parser, argv, lambda args: run(workspace_from_args(args))) if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/import.py b/import.py index aa16106..8827562 100644 --- a/import.py +++ b/import.py @@ -1,43 +1,65 @@ -import sys +import argparse import shutil +import sys +from collections.abc import Sequence from pathlib import Path from config import IMPORT_DIR +from copienator import ( + EvaluationWorkspace, + ExitCode, + evaluation_parser, + execute, + workspace_from_args, +) -def sync_annotated(dir_arg, refaire): - if not refaire: - bgnot_dir = Path(dir_arg) / "BGnot" - else: - bgnot_dir = Path(dir_arg) / "BRnot" - - annotated_dir = IMPORT_DIR +def sync_annotated( + workspace: EvaluationWorkspace, + *, + refaire: bool = False, + import_dir: Path, +) -> ExitCode: + annotation_name = "BRnot" if refaire else "BGnot" + workspace.require_directories(annotation_name) + annotation_dir = workspace.root / annotation_name + annotated_dir = Path(import_dir).expanduser() if not annotated_dir.is_dir(): - print(f"Error: Directory {annotated_dir} does not exist.") - return + print(f"Error: directory does not exist: {annotated_dir}", file=sys.stderr) + return ExitCode.INVALID_WORKSPACE - # Iterate over all PDF files in the annotated directory + missing_targets = 0 for pdf_file in annotated_dir.glob("*.pdf"): - subdir_name = pdf_file.stem # 'f' from 'f.pdf' - target_subdir = bgnot_dir / subdir_name + target_subdir = annotation_dir / pdf_file.stem if not target_subdir.is_dir(): - print(f"Warning: Directory {target_subdir} not found.") + print(f"Warning: directory not found: {target_subdir}", file=sys.stderr) + missing_targets += 1 else: dest_file = target_subdir / "Concat_annotated.pdf" - print("copying ", pdf_file, " to ", dest_file) + print(f"Copying {pdf_file} to {dest_file}") shutil.copy2(pdf_file, dest_file) + return ExitCode.PARTIAL if missing_targets else ExitCode.SUCCESS -import argparse + +def build_parser() -> argparse.ArgumentParser: + parser = evaluation_parser("Import handwritten annotations from the tablet directory.") + parser.add_argument("--refaire", action="store_true", help="Process only copies/labels defined in refaire.json") + return parser + + +def run(workspace: EvaluationWorkspace, *, refaire: bool = False) -> ExitCode: + return sync_annotated(workspace, refaire=refaire, import_dir=Path(IMPORT_DIR)) + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + return execute( + parser, + argv, + lambda args: run(workspace_from_args(args), refaire=args.refaire), + ) if __name__ == "__main__": - - parser = argparse.ArgumentParser(description="Move to tablette folder.") - parser.add_argument("dir", help="The directory to process") - parser.add_argument("--refaire", action="store_true", help="Process only copies/labels defined in refaire.json") - - args = parser.parse_args() - root_dir = args.dir - - sync_annotated(root_dir, args.refaire) + raise SystemExit(main()) diff --git a/tests/test_gui_core.py b/tests/test_gui_core.py index 72dd914..7f62b48 100644 --- a/tests/test_gui_core.py +++ b/tests/test_gui_core.py @@ -1,5 +1,7 @@ from __future__ import annotations +import importlib.util +import io import os import queue import sys @@ -7,6 +9,7 @@ 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 @@ -20,6 +23,7 @@ from copienator import ( 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 @@ -36,6 +40,15 @@ from platform_utils import ( 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: @@ -73,6 +86,10 @@ class WorkspaceTests(unittest.TestCase): 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) @@ -134,6 +151,143 @@ class AtomicJsonTests(unittest.TestCase): 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)}