1783 lines
78 KiB
Python
1783 lines
78 KiB
Python
from __future__ import annotations
|
||
|
||
import importlib
|
||
import io
|
||
import json
|
||
import os
|
||
import queue
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from contextlib import redirect_stderr
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest.mock import Mock, patch
|
||
|
||
from PIL import Image
|
||
from pypdf import PdfReader, PdfWriter
|
||
|
||
from copienator import (
|
||
EvaluationWorkspace,
|
||
WorkspaceNotFoundError,
|
||
WorkspaceValidationError,
|
||
atomic_update_json,
|
||
atomic_write_bytes,
|
||
atomic_write_json,
|
||
read_json,
|
||
workspace_from_target,
|
||
)
|
||
from copienator.annotation_actions import apply_checkbox_actions, apply_score_overrides
|
||
from copienator.annotation_data import AnnotationLoadResult, load_annotation_data
|
||
from copienator.filesystem import staged_directory, staged_files
|
||
from copienator.dispatcher import COMMANDS, main as dispatcher_main
|
||
from copienator_gui.app import (
|
||
CopienatorApp,
|
||
DEFAULT_HTTPS_PROXY,
|
||
build_runner_environment,
|
||
copy_pdf_paths,
|
||
detected_annotation_directories,
|
||
has_manual_conflicts,
|
||
plotting_shortcut_lines,
|
||
process_status,
|
||
)
|
||
from copienator_gui.diagnostics import collect_diagnostics
|
||
from copienator_gui.runner import ProcessRunner
|
||
from copienator_gui.state import StateStore
|
||
from copienator_gui.workflow import build_command, build_workflow, evaluation_argument
|
||
from copienator.commands.copies_tools import rename_all, rotate_all
|
||
from copienator.platform import (
|
||
WindowsLabelError,
|
||
replace_with_link_or_copy,
|
||
safe_filename,
|
||
validate_windows_labels,
|
||
windows_filename_problems,
|
||
)
|
||
|
||
REPOSITORY = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
def load_script_module(filename: str, _module_name: str):
|
||
stem = Path(filename).stem.replace("-", "_")
|
||
if stem == "import":
|
||
stem = "import_annotations"
|
||
return importlib.import_module(f"copienator.commands.{stem}")
|
||
|
||
|
||
def command_arguments(command: list[str]) -> list[str]:
|
||
assert command[2:4] == ["-m", "copienator"]
|
||
return command[5:]
|
||
|
||
|
||
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.label_groups_file, root / "label_groups")
|
||
self.assertEqual(
|
||
workspace.gemini_exam_items_file, root / "Tmp" / "exam_items.txt"
|
||
)
|
||
self.assertEqual(workspace.copies_dir, root / "Copies")
|
||
self.assertEqual(workspace.annotation_dir("grouped"), root / "BGnot")
|
||
self.assertEqual(workspace.state_database, root / ".copienator" / "state.sqlite3")
|
||
self.assertEqual(workspace.command_argument(), "Exam")
|
||
|
||
def test_discover_from_nested_artifact(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
root = Path(directory) / "Exam"
|
||
nested = root / "Copies" / "Copie01"
|
||
nested.mkdir(parents=True)
|
||
(root / "labels").write_text("Ex 1\n", encoding="utf-8")
|
||
artifact = nested / "Ex 1.pdf"
|
||
artifact.touch()
|
||
|
||
workspace = EvaluationWorkspace.discover(artifact)
|
||
self.assertEqual(workspace.root, root)
|
||
|
||
def test_discover_and_require_report_clear_errors(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
root = Path(directory)
|
||
with self.assertRaises(WorkspaceNotFoundError):
|
||
EvaluationWorkspace.discover(root)
|
||
|
||
workspace = EvaluationWorkspace(root)
|
||
with self.assertRaises(WorkspaceValidationError) as context:
|
||
workspace.require("enonce.pdf", "labels")
|
||
self.assertEqual(context.exception.missing, ["enonce.pdf", "labels"])
|
||
|
||
(root / "Copies").write_text("not a directory", encoding="utf-8")
|
||
with self.assertRaises(WorkspaceValidationError):
|
||
workspace.require_directories("Copies")
|
||
|
||
def test_names_file_prefers_evaluation_then_repository(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
repository = Path(directory)
|
||
root = repository / "Exam"
|
||
root.mkdir()
|
||
(repository / "names").write_text("Global\n", encoding="utf-8")
|
||
workspace = EvaluationWorkspace(root, repository)
|
||
self.assertEqual(workspace.names_file(), repository / "names")
|
||
|
||
(root / "names").write_text("Local\n", encoding="utf-8")
|
||
self.assertEqual(workspace.names_file(), root / "names")
|
||
|
||
|
||
class DispatcherTests(unittest.TestCase):
|
||
def test_unknown_command_has_argument_error_code(self) -> None:
|
||
with redirect_stderr(io.StringIO()):
|
||
self.assertEqual(dispatcher_main(["does-not-exist"]), 2)
|
||
|
||
def test_every_gui_python_command_is_registered(self) -> None:
|
||
for step in build_workflow(True):
|
||
for variant in step.variants:
|
||
if variant.kind == "python":
|
||
with self.subTest(step=step.id, command=variant.program):
|
||
self.assertIn(variant.program, COMMANDS)
|
||
|
||
def test_root_contains_only_configuration_python_modules(self) -> None:
|
||
root_modules = {path.name for path in REPOSITORY.glob("*.py")}
|
||
self.assertIn("default_config.py", root_modules)
|
||
self.assertLessEqual(root_modules, {"config.py", "default_config.py"})
|
||
|
||
|
||
class AtomicJsonTests(unittest.TestCase):
|
||
def test_atomic_binary_round_trip(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
path = Path(directory) / "result.jsonl"
|
||
atomic_write_bytes(path, b'{"one":1}\n')
|
||
self.assertEqual(path.read_bytes(), b'{"one":1}\n')
|
||
|
||
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")
|
||
|
||
def test_invalidation_preserves_first_visit_marker(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
store = StateStore()
|
||
store.load(Path(directory))
|
||
store.update_step("splitting", status="success", visited=True)
|
||
|
||
store.invalidate_after(["plotting", "splitting"], "plotting")
|
||
|
||
self.assertEqual(store.step("splitting")["status"], "stale")
|
||
self.assertTrue(store.step("splitting")["visited"])
|
||
|
||
|
||
class AnnotationDataTests(unittest.TestCase):
|
||
def test_loader_indexes_coordinates_without_mutating_correction(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_dir = evaluation / "Copies" / "Copie01"
|
||
group_dir = evaluation / "Par label" / "Ex 1"
|
||
copy_dir.mkdir(parents=True)
|
||
group_dir.mkdir(parents=True)
|
||
(copy_dir / "Ex 1_new.pdf").write_bytes(b"pdf")
|
||
correction = {
|
||
"Ex 1": [
|
||
[
|
||
{
|
||
"id": "01",
|
||
"result": {
|
||
"suffix": "_new",
|
||
"feedback": [{"text": "x", "box_2d": [100, 200, 300, 400]}],
|
||
},
|
||
}
|
||
]
|
||
]
|
||
}
|
||
atomic_write_json(evaluation / "correction.json", correction)
|
||
atomic_write_json(
|
||
group_dir / "Group_1.json",
|
||
[["01", 10, 90, 1.0, "Ex 1"]],
|
||
)
|
||
Image.new("RGB", (100, 200), "white").save(group_dir / "Group_1.jpg")
|
||
|
||
loaded = load_annotation_data(EvaluationWorkspace(evaluation))
|
||
item = loaded.data["01"]["Ex 1"]
|
||
self.assertEqual(item["pdf_path"], copy_dir / "Ex 1_new.pdf")
|
||
self.assertEqual(item["coordinates"], (10, 90))
|
||
self.assertEqual(item["result"]["feedback"][0]["box_2d"], [20, 20, 60, 40])
|
||
self.assertEqual(read_json(evaluation / "correction.json"), correction)
|
||
self.assertEqual(loaded.warnings, [])
|
||
|
||
def test_refaire_filter_adds_a_missing_correction_entry(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_dir = evaluation / "Copies" / "Copie01"
|
||
copy_dir.mkdir(parents=True)
|
||
(evaluation / "Par label").mkdir()
|
||
(copy_dir / "Ex 2.pdf").write_bytes(b"pdf")
|
||
atomic_write_json(evaluation / "correction.json", {})
|
||
|
||
loaded = load_annotation_data(
|
||
EvaluationWorkspace(evaluation),
|
||
refaire_list=[["Copie01", ["Ex 2"]]],
|
||
)
|
||
item = loaded.data["01"]["Ex 2"]
|
||
self.assertEqual(item["pdf_path"], copy_dir / "Ex 2.pdf")
|
||
self.assertEqual(item["result"]["error"], "non traité")
|
||
|
||
def test_staged_directory_preserves_then_replaces_destination(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
destination = Path(directory) / "output"
|
||
destination.mkdir()
|
||
(destination / "state.txt").write_text("old", encoding="utf-8")
|
||
|
||
with self.assertRaises(RuntimeError), staged_directory(
|
||
destination
|
||
) as staging:
|
||
(staging / "state.txt").write_text("broken", encoding="utf-8")
|
||
raise RuntimeError("rendering failed")
|
||
self.assertEqual(
|
||
(destination / "state.txt").read_text(encoding="utf-8"), "old"
|
||
)
|
||
|
||
with staged_directory(destination) as staging:
|
||
(staging / "state.txt").write_text("new", encoding="utf-8")
|
||
self.assertEqual(
|
||
(destination / "state.txt").read_text(encoding="utf-8"), "new"
|
||
)
|
||
leftovers = [path for path in destination.parent.iterdir() if path.name.startswith(".output.")]
|
||
self.assertEqual(leftovers, [])
|
||
|
||
def test_staged_files_preserve_inputs_and_roll_back_outputs(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
destination = Path(directory) / "Copie01"
|
||
destination.mkdir()
|
||
(destination / "bnote.json").write_text("input", encoding="utf-8")
|
||
(destination / "score.json").write_text("old", encoding="utf-8")
|
||
|
||
with self.assertRaises(RuntimeError), staged_files(destination) as staging:
|
||
(staging / "score.json").write_text("broken", encoding="utf-8")
|
||
(staging / "Concat.jpg").write_text("partial", encoding="utf-8")
|
||
raise RuntimeError("rendering failed")
|
||
self.assertEqual((destination / "score.json").read_text(), "old")
|
||
self.assertFalse((destination / "Concat.jpg").exists())
|
||
|
||
with staged_files(destination) as staging:
|
||
(staging / "score.json").write_text("new", encoding="utf-8")
|
||
(staging / "Concat.jpg").write_text("complete", encoding="utf-8")
|
||
self.assertEqual((destination / "score.json").read_text(), "new")
|
||
self.assertEqual((destination / "bnote.json").read_text(), "input")
|
||
|
||
def test_annotation_actions_are_shared_and_do_not_touch_json_sources(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
score_path = Path(directory) / "score.json"
|
||
atomic_write_json(score_path, {"Ex 1": "3"})
|
||
labels_data = {
|
||
"Ex 1": {
|
||
"result": {
|
||
"score": 1,
|
||
"feedback": [{"text": "global"}],
|
||
}
|
||
}
|
||
}
|
||
logs = []
|
||
dirty = apply_checkbox_actions(
|
||
labels_data,
|
||
[{"label": "Ex 1", "type": "del_global", "index": 0}],
|
||
logs.append,
|
||
)
|
||
dirty |= apply_score_overrides(labels_data, score_path, logs.append)
|
||
self.assertEqual(dirty, {"Ex 1"})
|
||
self.assertTrue(
|
||
labels_data["Ex 1"]["result"]["feedback"][0]["to_delete"]
|
||
)
|
||
self.assertEqual(labels_data["Ex 1"]["result"]["score"], "3")
|
||
self.assertEqual(read_json(score_path), {"Ex 1": "3"})
|
||
|
||
|
||
class StandardCliTests(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
cls.modules = {
|
||
"annotating": load_script_module("annotating.py", "annotating"),
|
||
"annotating_with_checks": load_script_module(
|
||
"annotating_with_checks.py", "annotating_with_checks"
|
||
),
|
||
"annotating_by_label": load_script_module(
|
||
"annotating_by_label.py", "annotating_by_label"
|
||
),
|
||
"reading_annotations": load_script_module(
|
||
"reading_annotations.py", "reading_annotations"
|
||
),
|
||
"reading_grouped_annotations": load_script_module(
|
||
"reading_grouped_annotations.py", "reading_grouped_annotations"
|
||
),
|
||
"cutleft": load_script_module("cutleft.py", "cutleft"),
|
||
"splitting_int": load_script_module(
|
||
"splitting_int.py", "splitting_int"
|
||
),
|
||
"page_splitter": load_script_module(
|
||
"page_splitter.py", "page_splitter"
|
||
),
|
||
"plotting": load_script_module("plotting.py", "plotting"),
|
||
"gemini_for_labels": load_script_module(
|
||
"gemini_for_labels.py", "gemini_for_labels"
|
||
),
|
||
"gemini_for_enonce": load_script_module(
|
||
"gemini_for_enonce.py", "gemini_for_enonce"
|
||
),
|
||
"enonce_info": load_script_module("enonce_info.py", "enonce_info"),
|
||
"correction": load_script_module("correction.py", "correction"),
|
||
"submit_batches": load_script_module(
|
||
"submit_batches.py", "submit_batches"
|
||
),
|
||
"batch_status": load_script_module("batch_status.py", "batch_status"),
|
||
"fetch_batched_results": load_script_module(
|
||
"fetch_batched_results.py", "fetch_batched_results"
|
||
),
|
||
"copies_tools": load_script_module(
|
||
"copies_tools.py", "copienator_copies_tools_test"
|
||
),
|
||
"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"),
|
||
"post_correction": load_script_module(
|
||
"post-correction.py", "copienator_post_correction_test"
|
||
),
|
||
"resolve_manual": load_script_module(
|
||
"resolve_manual.py", "copienator_resolve_manual_test"
|
||
),
|
||
"verify_groups": load_script_module(
|
||
"verify_groups.py", "copienator_verify_groups_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],
|
||
"post_correction": [missing],
|
||
"resolve_manual": [missing],
|
||
"verify_groups": [missing],
|
||
"copies_tools": ["rotate", missing],
|
||
"annotating": [missing],
|
||
"annotating_with_checks": [missing],
|
||
"annotating_by_label": [missing],
|
||
"reading_annotations": [missing],
|
||
"reading_grouped_annotations": [missing],
|
||
"cutleft": [missing],
|
||
"splitting_int": [missing],
|
||
"page_splitter": [missing],
|
||
"plotting": [missing],
|
||
"gemini_for_labels": [missing],
|
||
"gemini_for_enonce": [missing],
|
||
"enonce_info": [missing],
|
||
"correction": [missing],
|
||
"submit_batches": [missing],
|
||
"fetch_batched_results": [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 = {
|
||
"gemini_for_enonce": (
|
||
"statement",
|
||
"gemini",
|
||
{"target": evaluation, "restart": True},
|
||
),
|
||
"enonce_info": (
|
||
"statement",
|
||
"personal",
|
||
{"target": evaluation},
|
||
),
|
||
"export": (
|
||
"export",
|
||
"default",
|
||
{"target": evaluation, "annotation_dir": "Bnot", "refaire": True},
|
||
),
|
||
"import": (
|
||
"import",
|
||
"default",
|
||
{"target": evaluation, "annotation_dir": "Anot", "refaire": True},
|
||
),
|
||
"giving_names": (
|
||
"giving_names",
|
||
"default",
|
||
{"target": evaluation, "annotation_dir": "BGnot"},
|
||
),
|
||
"grouping": ("grouping", "default", {"target": evaluation}),
|
||
"post_correction": (
|
||
"post_correction",
|
||
"default",
|
||
{"target": evaluation},
|
||
),
|
||
"resolve_manual": (
|
||
"manual_resolution",
|
||
"default",
|
||
{"target": evaluation},
|
||
),
|
||
"verify_groups": (
|
||
"verify_groups",
|
||
"default",
|
||
{"target": evaluation},
|
||
),
|
||
"annotating": (
|
||
"annotation",
|
||
"simple",
|
||
{"target": evaluation, "overwrite": True},
|
||
),
|
||
"annotating_with_checks": (
|
||
"annotation",
|
||
"checks",
|
||
{"target": evaluation, "overwrite": True, "refaire": True},
|
||
),
|
||
"annotating_by_label": (
|
||
"annotation",
|
||
"grouped",
|
||
{"target": evaluation, "overwrite": True},
|
||
),
|
||
"reading_annotations": (
|
||
"read_annotations",
|
||
"standard",
|
||
{"target": evaluation, "update_score": True},
|
||
),
|
||
"reading_grouped_annotations": (
|
||
"read_annotations",
|
||
"grouped",
|
||
{"target": evaluation, "update_score": True, "refaire": True},
|
||
),
|
||
"cutleft": (
|
||
"cutleft",
|
||
"default",
|
||
{"target": evaluation, "fullpage": True},
|
||
),
|
||
"splitting_int": (
|
||
"splitting",
|
||
"default",
|
||
{"target": evaluation},
|
||
),
|
||
"page_splitter": (
|
||
"page_splitter",
|
||
"default",
|
||
{"target": evaluation},
|
||
),
|
||
"plotting": (
|
||
"plotting",
|
||
"default",
|
||
{"target": evaluation},
|
||
),
|
||
"gemini_for_labels": (
|
||
"labels",
|
||
"default",
|
||
{"target": evaluation, "overwrite": True},
|
||
),
|
||
"correction": (
|
||
"correction",
|
||
"live",
|
||
{"target": evaluation, "overwrite": True, "limit": 5},
|
||
),
|
||
"submit_batches": (
|
||
"submit_batches",
|
||
"default",
|
||
{"target": evaluation},
|
||
),
|
||
"fetch_batched_results": (
|
||
"fetch_batches",
|
||
"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_arguments(command)
|
||
)
|
||
parsed_path = getattr(parsed, "evaluation", None) or parsed.target
|
||
self.assertEqual(str(parsed_path), evaluation)
|
||
|
||
for step_id in ("rotate", "rename"):
|
||
step = steps[step_id]
|
||
variant = step.variants[0]
|
||
command = build_command(
|
||
REPOSITORY,
|
||
step,
|
||
variant,
|
||
{"target": evaluation},
|
||
evaluation,
|
||
)
|
||
with self.subTest(script=f"copies_tools:{step_id}"):
|
||
parsed = self.modules["copies_tools"].build_parser().parse_args(
|
||
command_arguments(command)
|
||
)
|
||
self.assertEqual(parsed.operation, step_id)
|
||
self.assertEqual(str(parsed.evaluation), evaluation)
|
||
|
||
def test_enonce_info_preserves_labels_when_no_blocks_are_found(self) -> None:
|
||
module = self.modules["enonce_info"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
evaluation.mkdir()
|
||
(evaluation / "source.tex").write_text(
|
||
"No SHEETINFO blocks\n", encoding="utf-8"
|
||
)
|
||
(evaluation / "labels").write_text("Existing\n", encoding="utf-8")
|
||
self.assertEqual(
|
||
module.process_directory(EvaluationWorkspace(evaluation)), 4
|
||
)
|
||
self.assertEqual((evaluation / "labels").read_text(), "Existing\n")
|
||
self.assertFalse(list(evaluation.glob(".labels.*.tmp")))
|
||
|
||
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_export_accepts_each_annotation_directory(self) -> None:
|
||
module = self.modules["export"]
|
||
for annotation_dir in ("Anot", "Bnot"):
|
||
with self.subTest(annotation_dir=annotation_dir), tempfile.TemporaryDirectory() as directory:
|
||
base = Path(directory)
|
||
evaluation = base / "Exam"
|
||
source = evaluation / annotation_dir / "Copie01"
|
||
source.mkdir(parents=True)
|
||
suffix = ".jpg" if annotation_dir == "Anot" else ".pdf"
|
||
(source / f"Concat{suffix}").write_bytes(annotation_dir.encode())
|
||
with patch.object(module, "EXPORT_DIR", base / "Export"):
|
||
self.assertEqual(module.main([str(evaluation), annotation_dir]), 0)
|
||
self.assertEqual(
|
||
(base / "Export" / "Exam" / f"Copie01{suffix}").read_bytes(),
|
||
annotation_dir.encode(),
|
||
)
|
||
|
||
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_import_accepts_each_annotation_directory(self) -> None:
|
||
module = self.modules["import"]
|
||
for annotation_dir in ("Anot", "Bnot"):
|
||
with self.subTest(annotation_dir=annotation_dir), tempfile.TemporaryDirectory() as directory:
|
||
base = Path(directory)
|
||
evaluation = base / "Exam"
|
||
target = evaluation / annotation_dir / "Copie01"
|
||
target.mkdir(parents=True)
|
||
import_dir = base / "Import"
|
||
import_dir.mkdir()
|
||
suffix = ".jpg" if annotation_dir == "Anot" else ".pdf"
|
||
(import_dir / f"Copie01{suffix}").write_bytes(b"handwritten")
|
||
with patch.object(module, "IMPORT_DIR", import_dir):
|
||
self.assertEqual(module.main([str(evaluation), annotation_dir]), 0)
|
||
self.assertEqual(
|
||
(target / f"Concat_annotated{suffix}").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())
|
||
|
||
def test_cutleft_can_target_one_copy_without_rendering_at_import(self) -> None:
|
||
module = self.modules["cutleft"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_pdf = evaluation / "Copies" / "Copie01.pdf"
|
||
copy_pdf.parent.mkdir(parents=True)
|
||
copy_pdf.write_bytes(b"pdf")
|
||
with patch.object(module, "ImageReviewer") as reviewer:
|
||
self.assertEqual(module.main([str(copy_pdf), "--fullpage"]), 0)
|
||
files, output_dir = reviewer.call_args.args[:2]
|
||
self.assertEqual(files, [copy_pdf])
|
||
self.assertEqual(output_dir, evaluation / "Cutleft")
|
||
self.assertEqual(reviewer.call_args.kwargs["default_max_per_file"], 1)
|
||
|
||
def test_cutleft_atomic_save_removes_only_obsolete_copy_outputs(self) -> None:
|
||
module = self.modules["cutleft"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
output_dir = Path(directory) / "Cutleft"
|
||
output_dir.mkdir()
|
||
(output_dir / "Copie01_01.jpg").write_bytes(b"old-one")
|
||
(output_dir / "Copie01_02.jpg").write_bytes(b"old-two")
|
||
(output_dir / "Copie02_01.jpg").write_bytes(b"other-copy")
|
||
(output_dir / "Copie01_schema.json").write_text(
|
||
"{}", encoding="utf-8"
|
||
)
|
||
image = Image.new("RGB", (5, 5), "white")
|
||
result = (
|
||
image,
|
||
[image],
|
||
{
|
||
"original_filename": "Copie01.pdf",
|
||
"total_pages": 1,
|
||
"number_of_files": 1,
|
||
"columns_per_file": [1],
|
||
},
|
||
)
|
||
module.save_results(result, Path("Copie01.pdf"), output_dir)
|
||
self.assertTrue((output_dir / "Copie01_01.jpg").is_file())
|
||
self.assertFalse((output_dir / "Copie01_02.jpg").exists())
|
||
self.assertEqual(
|
||
(output_dir / "Copie02_01.jpg").read_bytes(), b"other-copy"
|
||
)
|
||
self.assertEqual(
|
||
read_json(output_dir / "Copie01_schema.json")["total_pages"], 1
|
||
)
|
||
|
||
def test_splitting_missing_json_is_partial(self) -> None:
|
||
module = self.modules["splitting_int"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_pdf = evaluation / "Copies" / "Copie01.pdf"
|
||
copy_pdf.parent.mkdir(parents=True)
|
||
copy_pdf.write_bytes(b"pdf")
|
||
(evaluation / "labels").write_text("Ex 1\n", encoding="utf-8")
|
||
self.assertEqual(module.main([str(evaluation)]), 4)
|
||
|
||
def test_splitting_failure_preserves_previous_copy_outputs(self) -> None:
|
||
module = self.modules["splitting_int"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_pdf = evaluation / "Copies" / "Copie01.pdf"
|
||
output_dir = evaluation / "Copies" / "Copie01"
|
||
output_dir.mkdir(parents=True)
|
||
copy_pdf.write_bytes(b"pdf")
|
||
(output_dir / "sentinel.pdf").write_bytes(b"old")
|
||
workspace = EvaluationWorkspace(evaluation)
|
||
with patch.object(
|
||
module,
|
||
"_render_split_outputs",
|
||
side_effect=RuntimeError("render failed"),
|
||
), self.assertRaises(RuntimeError):
|
||
module.split_an_interro(workspace, copy_pdf, [])
|
||
self.assertEqual((output_dir / "sentinel.pdf").read_bytes(), b"old")
|
||
|
||
def test_splitting_moves_obsolete_outputs_to_missing_on_commit(self) -> None:
|
||
module = self.modules["splitting_int"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_pdf = evaluation / "Copies" / "Copie01.pdf"
|
||
output_dir = evaluation / "Copies" / "Copie01"
|
||
missing_dir = output_dir / "Missing"
|
||
missing_dir.mkdir(parents=True)
|
||
copy_pdf.write_bytes(b"pdf")
|
||
(output_dir / "Old.pdf").write_bytes(b"obsolete")
|
||
(missing_dir / "Earlier.pdf").write_bytes(b"earlier")
|
||
|
||
def render(_pdf, _coordinates, staging):
|
||
(staging / "Ex 1.pdf").write_bytes(b"new")
|
||
return {"Ex 1.pdf"}
|
||
|
||
with patch.object(module, "_render_split_outputs", side_effect=render):
|
||
module.split_an_interro(EvaluationWorkspace(evaluation), copy_pdf, [])
|
||
self.assertEqual((output_dir / "Ex 1.pdf").read_bytes(), b"new")
|
||
self.assertEqual(
|
||
(output_dir / "Missing" / "Old.pdf").read_bytes(), b"obsolete"
|
||
)
|
||
self.assertEqual(
|
||
(output_dir / "Missing" / "Earlier.pdf").read_bytes(), b"earlier"
|
||
)
|
||
|
||
def test_splitting_renders_a_real_one_page_answer(self) -> None:
|
||
module = self.modules["splitting_int"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_pdf = evaluation / "Copies" / "Copie01.pdf"
|
||
copy_pdf.parent.mkdir(parents=True)
|
||
document = module.pymupdf.open()
|
||
page = document.new_page(width=600, height=800)
|
||
page.insert_text((100, 200), "Student answer")
|
||
document.save(copy_pdf)
|
||
document.close()
|
||
(evaluation / "labels").write_text("Ex 1\n", encoding="utf-8")
|
||
atomic_write_json(
|
||
copy_pdf.with_suffix(".json"),
|
||
{
|
||
"name": "Copie01",
|
||
"list": [{"label": "Ex 1", "box_2d": [100, 100, 300, 300]}],
|
||
},
|
||
)
|
||
|
||
self.assertEqual(module.main([str(evaluation)]), 0)
|
||
answer = evaluation / "Copies" / "Copie01" / "Ex 1.pdf"
|
||
self.assertTrue(answer.is_file())
|
||
self.assertEqual(len(PdfReader(answer).pages), 1)
|
||
|
||
def test_page_splitter_commits_original_and_generated_copy(self) -> None:
|
||
module = self.modules["page_splitter"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
evaluation.mkdir()
|
||
original = evaluation / "Copie01.pdf"
|
||
generated = evaluation / ".generated.pdf"
|
||
original.write_bytes(b"original")
|
||
generated.write_bytes(b"processed")
|
||
workspace = EvaluationWorkspace(evaluation)
|
||
|
||
output = module.commit_processed_pdf(workspace, original, generated)
|
||
self.assertEqual(output, evaluation / "Copies" / "Copie01.pdf")
|
||
self.assertEqual(output.read_bytes(), b"processed")
|
||
self.assertEqual(
|
||
(evaluation / "Copies Originales" / "Copie01.pdf").read_bytes(),
|
||
b"original",
|
||
)
|
||
self.assertFalse(original.exists())
|
||
|
||
def test_page_splitter_commit_failure_rolls_back_both_files(self) -> None:
|
||
module = self.modules["page_splitter"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
original = evaluation / "Copie01.pdf"
|
||
backup = evaluation / "Copies Originales" / "Copie01.pdf"
|
||
output = evaluation / "Copies" / "Copie01.pdf"
|
||
backup.parent.mkdir(parents=True)
|
||
output.parent.mkdir()
|
||
original.write_bytes(b"new original")
|
||
backup.write_bytes(b"previous original")
|
||
output.write_bytes(b"previous output")
|
||
missing_generated = evaluation / "missing-generated.pdf"
|
||
|
||
with self.assertRaises(FileNotFoundError):
|
||
module.commit_processed_pdf(
|
||
EvaluationWorkspace(evaluation), original, missing_generated
|
||
)
|
||
self.assertEqual(original.read_bytes(), b"new original")
|
||
self.assertEqual(backup.read_bytes(), b"previous original")
|
||
self.assertEqual(output.read_bytes(), b"previous output")
|
||
|
||
def test_page_splitter_reprocesses_from_preserved_original(self) -> None:
|
||
module = self.modules["page_splitter"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
generated = evaluation / "Copies" / "Copie01.pdf"
|
||
original = evaluation / "Copies Originales" / "Copie01.pdf"
|
||
generated.parent.mkdir(parents=True)
|
||
original.parent.mkdir()
|
||
generated.write_bytes(b"processed")
|
||
original.write_bytes(b"original")
|
||
workspace = EvaluationWorkspace(evaluation)
|
||
self.assertEqual(
|
||
module._selected_inputs(workspace, generated),
|
||
[original],
|
||
)
|
||
|
||
def test_plotting_batch_save_is_atomic(self) -> None:
|
||
module = self.modules["plotting"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
(evaluation / "Copies").mkdir(parents=True)
|
||
viewer = module.ImageViewer.__new__(module.ImageViewer)
|
||
viewer.base_dir = evaluation
|
||
viewer.active_copie_name = "Copie01"
|
||
viewer.accumulated_results = {
|
||
"name": "Test",
|
||
"list": [{"label": "Ex 1"}],
|
||
}
|
||
viewer.save_current_batch()
|
||
self.assertEqual(
|
||
read_json(evaluation / "Copies" / "Copie01.json"),
|
||
{"name": "Test", "list": [{"label": "Ex 1"}]},
|
||
)
|
||
self.assertIsNone(viewer.accumulated_results)
|
||
|
||
def test_plotting_validates_plain_and_directional_labels(self) -> None:
|
||
module = self.modules["plotting"]
|
||
self.assertEqual(
|
||
module.normalized_labels(
|
||
[
|
||
{"label": "Ex 1"},
|
||
{"label": "|Ex 2"},
|
||
{"label": "Ex 3|"},
|
||
{"label": "_"},
|
||
]
|
||
),
|
||
["Ex 1", "Ex 2", "Ex 3"],
|
||
)
|
||
|
||
def test_plotting_worker_failure_still_terminates_its_queue(self) -> None:
|
||
module = self.modules["plotting"]
|
||
output_queue = queue.Queue()
|
||
with patch.object(
|
||
module, "_worker_items", side_effect=RuntimeError("worker failed")
|
||
):
|
||
module.worker_thread(Path("."), [], [], output_queue)
|
||
image, json_path, metadata = output_queue.get_nowait()
|
||
self.assertIsNone(image)
|
||
self.assertIsNone(json_path)
|
||
self.assertEqual(metadata, {"worker_error": "worker failed"})
|
||
|
||
def test_label_detection_resolves_copy_and_cutleft_targets(self) -> None:
|
||
module = self.modules["gemini_for_labels"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_pdf = evaluation / "Copies" / "Copie01.pdf"
|
||
image_one = evaluation / "Cutleft" / "Copie01_01.jpg"
|
||
image_two = evaluation / "Cutleft" / "Copie01_02.jpg"
|
||
copy_pdf.parent.mkdir(parents=True)
|
||
image_one.parent.mkdir()
|
||
copy_pdf.write_bytes(b"pdf")
|
||
image_one.write_bytes(b"image")
|
||
image_two.write_bytes(b"image")
|
||
workspace = EvaluationWorkspace(evaluation)
|
||
|
||
images, warnings = module.selected_images(workspace, [copy_pdf])
|
||
self.assertEqual(images, [image_one, image_two])
|
||
self.assertEqual(warnings, [])
|
||
images, warnings = module.selected_images(workspace, [image_two])
|
||
self.assertEqual(images, [image_two])
|
||
self.assertEqual(warnings, [])
|
||
|
||
def test_label_detection_preserves_context_and_writes_atomically(self) -> None:
|
||
module = self.modules["gemini_for_labels"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copies = evaluation / "Copies"
|
||
cutleft = evaluation / "Cutleft"
|
||
copies.mkdir(parents=True)
|
||
cutleft.mkdir()
|
||
first = cutleft / "Copie01_01.jpg"
|
||
second = cutleft / "Copie01_02.jpg"
|
||
first.write_bytes(b"first image")
|
||
second.write_bytes(b"second image")
|
||
atomic_write_json(
|
||
copies / "Copie01_01.json",
|
||
{
|
||
"name": "Student",
|
||
"list": [{"box_2d": [1, 2, 3, 4], "label": "Ex 1"}],
|
||
},
|
||
)
|
||
response = Mock(
|
||
text=(
|
||
'{"name":"Continued","list":'
|
||
'[{"box_2d":[10,20,30,40],"label":"Ex 2"}]}'
|
||
)
|
||
)
|
||
client = Mock()
|
||
client.models.generate_content.return_value = response
|
||
|
||
generated = module.process_copy_group(
|
||
EvaluationWorkspace(evaluation),
|
||
"Copie01",
|
||
[first, second],
|
||
client=client,
|
||
labels_text="Ex 1\nEx 2\n",
|
||
names_text="Student\n",
|
||
valid_labels={"Ex 1", "Ex 2"},
|
||
valid_names={"Student", "Continued", "Unknown"},
|
||
overwrite=False,
|
||
sleep=lambda _seconds: None,
|
||
target_interval=0,
|
||
)
|
||
self.assertEqual(generated, 1)
|
||
self.assertEqual(client.models.generate_content.call_count, 1)
|
||
self.assertEqual(
|
||
read_json(copies / "Copie01_02.json"),
|
||
{
|
||
"name": "Continued",
|
||
"list": [{"box_2d": [10, 20, 30, 40], "label": "Ex 2"}],
|
||
},
|
||
)
|
||
|
||
def test_label_detection_retries_unknown_labels(self) -> None:
|
||
module = self.modules["gemini_for_labels"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
(evaluation / "Copies").mkdir(parents=True)
|
||
image = evaluation / "Cutleft" / "Copie01_01.jpg"
|
||
image.parent.mkdir()
|
||
image.write_bytes(b"image")
|
||
client = Mock()
|
||
client.models.generate_content.side_effect = [
|
||
Mock(
|
||
text=(
|
||
'{"name":"Student","list":'
|
||
'[{"box_2d":[1,2,3,4],"label":"Wrong"}]}'
|
||
)
|
||
),
|
||
Mock(
|
||
text=(
|
||
'{"name":"Student","list":'
|
||
'[{"box_2d":[1,2,3,4],"label":"Ex 1"}]}'
|
||
)
|
||
),
|
||
]
|
||
sleeps = []
|
||
module.process_copy_group(
|
||
EvaluationWorkspace(evaluation),
|
||
"Copie01",
|
||
[image],
|
||
client=client,
|
||
labels_text="Ex 1\n",
|
||
names_text="Student\n",
|
||
valid_labels={"Ex 1"},
|
||
valid_names={"Student", "Unknown", "Continued"},
|
||
overwrite=True,
|
||
sleep=sleeps.append,
|
||
target_interval=0,
|
||
)
|
||
self.assertEqual(client.models.generate_content.call_count, 2)
|
||
self.assertIn(10, sleeps)
|
||
self.assertEqual(
|
||
read_json(evaluation / "Copies" / "Copie01_01.json")["list"][0][
|
||
"label"
|
||
],
|
||
"Ex 1",
|
||
)
|
||
|
||
def test_correction_overwrite_keeps_previous_state_until_a_commit(self) -> None:
|
||
module = self.modules["correction"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
evaluation.mkdir()
|
||
correction = evaluation / "correction.json"
|
||
progress = evaluation / "correction_progress.json"
|
||
atomic_write_json(correction, {"Ex 1": [[{"id": "01"}]]})
|
||
atomic_write_json(progress, [["old.jpg", "Ex 1"]])
|
||
args = module.build_parser().parse_args(
|
||
[str(evaluation), "--overwrite"]
|
||
)
|
||
module.configure_runtime(
|
||
EvaluationWorkspace(evaluation),
|
||
[("new.jpg", "Ex 1")],
|
||
args,
|
||
api_client=Mock(),
|
||
)
|
||
self.assertEqual(read_json(correction), {"Ex 1": [[{"id": "01"}]]})
|
||
self.assertEqual(read_json(progress), [["old.jpg", "Ex 1"]])
|
||
self.assertEqual(module.results, {"Ex 1": []})
|
||
self.assertEqual(module.tasks_to_process, [("new.jpg", "Ex 1")])
|
||
|
||
def test_correction_reset_restores_old_and_deletes_new_files(self) -> None:
|
||
module = self.modules["correction"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_dir = evaluation / "Copies" / "Copie01"
|
||
copy_dir.mkdir(parents=True)
|
||
atomic_write_json(evaluation / "correction.json", {"old": True})
|
||
atomic_write_json(evaluation / "correction_progress.json", ["old"])
|
||
(copy_dir / "Ex 1.pdf").write_bytes(b"current")
|
||
(copy_dir / "Ex 1_old.pdf").write_bytes(b"original")
|
||
(copy_dir / "Ex 2_new.pdf").write_bytes(b"generated")
|
||
|
||
self.assertEqual(module.main([str(evaluation), "--reset"]), 0)
|
||
self.assertFalse((evaluation / "correction.json").exists())
|
||
self.assertFalse((evaluation / "correction_progress.json").exists())
|
||
self.assertEqual((copy_dir / "Ex 1.pdf").read_bytes(), b"original")
|
||
self.assertFalse((copy_dir / "Ex 1_old.pdf").exists())
|
||
self.assertFalse((copy_dir / "Ex 2_new.pdf").exists())
|
||
|
||
def test_correction_batch_request_files_are_written_atomically(self) -> None:
|
||
module = self.modules["correction"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
group_dir = evaluation / "Par label" / "Ex 1"
|
||
(evaluation / "Copies").mkdir(parents=True)
|
||
group_dir.mkdir(parents=True)
|
||
(evaluation / "labels").write_text("Ex 1\n", encoding="utf-8")
|
||
image = group_dir / "Group_1.jpg"
|
||
image.write_bytes(b"image")
|
||
atomic_write_json(
|
||
image.with_suffix(".json"),
|
||
[["01", 0, 400, 1.0, "Ex 1"]],
|
||
)
|
||
args = module.build_parser().parse_args([str(evaluation), "--batch"])
|
||
module.configure_runtime(
|
||
EvaluationWorkspace(evaluation),
|
||
[(str(image), "Ex 1")],
|
||
args,
|
||
api_client=Mock(),
|
||
)
|
||
with patch.object(module.prompting, "make_prompt", return_value="prompt"):
|
||
self.assertEqual(module.run_configured(args), 0)
|
||
lines = (evaluation / "batch_requests_flash.jsonl").read_text().splitlines()
|
||
self.assertEqual(len(lines), 1)
|
||
self.assertEqual(json.loads(lines[0])["key"], str(image))
|
||
self.assertEqual(
|
||
(evaluation / "batch_requests_pro.jsonl").read_text(), ""
|
||
)
|
||
|
||
def test_batch_helpers_use_mocked_api_and_atomic_combination(self) -> None:
|
||
submit = self.modules["submit_batches"]
|
||
fetch = self.modules["fetch_batched_results"]
|
||
status = self.modules["batch_status"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
evaluation.mkdir()
|
||
(evaluation / "batch_requests_flash.jsonl").write_text(
|
||
"{}\n", encoding="utf-8"
|
||
)
|
||
client = Mock()
|
||
client.files.upload.return_value = SimpleNamespace(name="files/input")
|
||
client.batches.create.return_value = SimpleNamespace(name="batches/1")
|
||
self.assertEqual(
|
||
submit.run(EvaluationWorkspace(evaluation), client=client), 0
|
||
)
|
||
self.assertEqual(client.batches.create.call_count, 1)
|
||
manifest = json.loads((evaluation / "batch_jobs.json").read_text())
|
||
self.assertEqual(manifest["jobs"]["flash"]["name"], "batches/1")
|
||
|
||
jobs = [
|
||
SimpleNamespace(
|
||
name="batches/1",
|
||
display_name=f"flash-correction-{evaluation.name}",
|
||
state=SimpleNamespace(name="JOB_STATE_SUCCEEDED"),
|
||
dest=SimpleNamespace(file_name="files/flash-result"),
|
||
),
|
||
SimpleNamespace(
|
||
name="batches/2",
|
||
display_name=f"pro-correction-{evaluation.name}",
|
||
state=SimpleNamespace(name="JOB_STATE_SUCCEEDED"),
|
||
dest=SimpleNamespace(file_name="files/pro-result"),
|
||
),
|
||
]
|
||
client.batches.get.return_value = jobs[0]
|
||
client.files.download.side_effect = [b'{"flash":1}\n', b'{"pro":1}']
|
||
self.assertEqual(
|
||
fetch.run(EvaluationWorkspace(evaluation), client=client), 0
|
||
)
|
||
self.assertEqual(
|
||
(evaluation / "batched_correction_result.jsonl").read_bytes(),
|
||
b'{"flash":1}\n',
|
||
)
|
||
|
||
job = SimpleNamespace(
|
||
state=SimpleNamespace(name="JOB_STATE_SUCCEEDED"),
|
||
dest=SimpleNamespace(file_name="files/result"),
|
||
)
|
||
client.batches.get.return_value = job
|
||
client.files.download.side_effect = None
|
||
client.files.download.return_value = b'{"downloaded":true}\n'
|
||
output = evaluation / "one-result.jsonl"
|
||
self.assertEqual(
|
||
status.download_job("batches/1", output=output, client=client), 0
|
||
)
|
||
self.assertEqual(output.read_bytes(), b'{"downloaded":true}\n')
|
||
|
||
def test_post_correction_main_cleans_json_atomically(self) -> None:
|
||
module = self.modules["post_correction"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
evaluation.mkdir()
|
||
atomic_write_json(
|
||
evaluation / "correction.json",
|
||
{"text": "outside_name and $math_name$", "suffix": "_new"},
|
||
)
|
||
self.assertEqual(module.main([str(evaluation)]), 0)
|
||
self.assertEqual(
|
||
read_json(evaluation / "correction.json"),
|
||
{"text": r"outside\_name and $math_name$", "suffix": "_new"},
|
||
)
|
||
|
||
def test_manual_resolution_with_no_actions_is_safe(self) -> None:
|
||
module = self.modules["resolve_manual"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
(evaluation / "Copies").mkdir(parents=True)
|
||
atomic_write_json(evaluation / "correction.json", {})
|
||
manual = evaluation / "manual_resolutions.txt"
|
||
manual.write_text("### Nothing to do\n", encoding="utf-8")
|
||
|
||
self.assertEqual(module.main([str(evaluation)]), 0)
|
||
self.assertFalse(manual.exists())
|
||
self.assertEqual(read_json(evaluation / "correction.json"), {})
|
||
|
||
def test_malformed_manual_resolution_is_not_deleted(self) -> None:
|
||
module = self.modules["resolve_manual"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
(evaluation / "Copies").mkdir(parents=True)
|
||
correction = evaluation / "correction.json"
|
||
atomic_write_json(correction, {})
|
||
manual = evaluation / "manual_resolutions.txt"
|
||
manual.write_text("this is malformed\n", encoding="utf-8")
|
||
before = correction.read_bytes()
|
||
|
||
with redirect_stderr(io.StringIO()):
|
||
self.assertEqual(module.main([str(evaluation)]), 1)
|
||
self.assertTrue(manual.exists())
|
||
self.assertEqual(correction.read_bytes(), before)
|
||
|
||
def test_documented_manual_resolution_operators_are_parsed(self) -> None:
|
||
module = self.modules["resolve_manual"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
path = Path(directory) / "manual_resolutions.txt"
|
||
path.write_text(
|
||
"\n".join(
|
||
f"Copie01 Old label {operator} New label"
|
||
for operator in module.OPERATORS
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
instructions = module.parse_instructions(path)
|
||
self.assertEqual(
|
||
[instruction.operator for instruction in instructions],
|
||
list(module.OPERATORS),
|
||
)
|
||
|
||
def test_replace_manual_resolution_creates_refaire_state(self) -> None:
|
||
module = self.modules["resolve_manual"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_dir = evaluation / "Copies" / "Copie01"
|
||
copy_dir.mkdir(parents=True)
|
||
(copy_dir / "Old.pdf").write_bytes(b"old answer")
|
||
(copy_dir / "New.pdf").write_bytes(b"replaced answer")
|
||
atomic_write_json(
|
||
evaluation / "correction.json",
|
||
{
|
||
"Old": [
|
||
[
|
||
{
|
||
"id": "01",
|
||
"result": {"error": "wrg-lbl:New?delayed"},
|
||
}
|
||
]
|
||
],
|
||
"New": [[{"id": "01", "result": {"error": ""}}]],
|
||
},
|
||
)
|
||
(evaluation / "manual_resolutions.txt").write_text(
|
||
"Copie01 Old -x New\n", encoding="utf-8"
|
||
)
|
||
|
||
self.assertEqual(module.main([str(evaluation)]), 0)
|
||
self.assertEqual((copy_dir / "New_old.pdf").read_bytes(), b"replaced answer")
|
||
self.assertEqual((copy_dir / "New_new.pdf").read_bytes(), b"old answer")
|
||
self.assertEqual(
|
||
read_json(evaluation / "refaire.json"),
|
||
[["Copie01", ["New"]]],
|
||
)
|
||
correction = read_json(evaluation / "correction.json")
|
||
self.assertEqual(
|
||
correction["Old"][0][0]["result"]["error"],
|
||
"wrg-lbl-moved-to:New",
|
||
)
|
||
self.assertEqual(correction["New"][0][0]["result"]["suffix"], "_new")
|
||
|
||
def test_verify_groups_reports_success_and_missing_answers(self) -> None:
|
||
module = self.modules["verify_groups"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_dir = evaluation / "Copies" / "Copie01"
|
||
group_dir = evaluation / "Par label" / "Ex 1"
|
||
copy_dir.mkdir(parents=True)
|
||
group_dir.mkdir(parents=True)
|
||
(copy_dir / "Ex 1.pdf").write_bytes(b"pdf")
|
||
|
||
self.assertEqual(module.main([str(evaluation)]), 1)
|
||
atomic_write_json(
|
||
group_dir / "Group_1.json",
|
||
[["01", 0, 100, 1.0, "Ex 1"]],
|
||
)
|
||
self.assertEqual(module.main([str(evaluation)]), 0)
|
||
|
||
def test_checked_annotation_discovers_workspace_from_copy_pdf(self) -> None:
|
||
module = self.modules["annotating_with_checks"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_pdf = evaluation / "Copies" / "Copie40.pdf"
|
||
copy_pdf.parent.mkdir(parents=True)
|
||
copy_pdf.write_bytes(b"pdf")
|
||
args = module.build_parser().parse_args([str(copy_pdf)])
|
||
workspace, target = workspace_from_target(args)
|
||
self.assertEqual(workspace.root, evaluation)
|
||
self.assertEqual(target, copy_pdf)
|
||
self.assertEqual(module._copy_id_from_target(workspace, target), "40")
|
||
|
||
def test_checked_refaire_requires_refaire_file(self) -> None:
|
||
module = self.modules["annotating_with_checks"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
(evaluation / "Copies").mkdir(parents=True)
|
||
(evaluation / "Par label").mkdir()
|
||
(evaluation / "labels").write_text("Ex 1\n", encoding="utf-8")
|
||
atomic_write_json(evaluation / "correction.json", {})
|
||
with redirect_stderr(io.StringIO()):
|
||
self.assertEqual(module.main([str(evaluation), "--refaire"]), 3)
|
||
|
||
def test_grouped_reader_refaire_requires_refaire_file(self) -> None:
|
||
module = self.modules["reading_grouped_annotations"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
(evaluation / "Copies").mkdir(parents=True)
|
||
(evaluation / "Par label").mkdir()
|
||
(evaluation / "BGnot").mkdir()
|
||
(evaluation / "labels").write_text("Ex 1\n", encoding="utf-8")
|
||
atomic_write_json(evaluation / "correction.json", {})
|
||
with redirect_stderr(io.StringIO()):
|
||
self.assertEqual(module.main([str(evaluation), "--refaire"]), 3)
|
||
|
||
def test_note_detection_accepts_a_missing_note_layer(self) -> None:
|
||
module = self.modules["reading_annotations"]
|
||
self.assertFalse(module.has_significant_notes(None))
|
||
|
||
def test_grouped_reader_reports_worker_failures(self) -> None:
|
||
module = self.modules["reading_grouped_annotations"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
(evaluation / "Copies").mkdir(parents=True)
|
||
(evaluation / "Par label").mkdir()
|
||
(evaluation / "BGnot" / "Ex 1").mkdir(parents=True)
|
||
(evaluation / "labels").write_text("Ex 1\n", encoding="utf-8")
|
||
atomic_write_json(evaluation / "correction.json", {})
|
||
loaded = AnnotationLoadResult({"01": {"Ex 1": {}}}, [])
|
||
with (
|
||
patch.object(module, "load_annotation_data", return_value=loaded),
|
||
patch.object(
|
||
module,
|
||
"_scan_annotation_directory",
|
||
side_effect=RuntimeError("worker failed"),
|
||
),
|
||
redirect_stderr(io.StringIO()) as errors,
|
||
):
|
||
self.assertEqual(module.main([str(evaluation)]), 1)
|
||
self.assertIn("worker failed", errors.getvalue())
|
||
|
||
def test_checked_render_failure_preserves_previous_student_output(self) -> None:
|
||
module = self.modules["annotating_with_checks"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
copy_dir = evaluation / "Copies" / "Copie01"
|
||
copy_dir.mkdir(parents=True)
|
||
(evaluation / "Par label").mkdir()
|
||
(evaluation / "labels").write_text("Ex 1\n", encoding="utf-8")
|
||
atomic_write_json(evaluation / "correction.json", {})
|
||
answer = copy_dir / "Ex 1.pdf"
|
||
answer.write_bytes(b"pdf")
|
||
previous = evaluation / "Bnot" / "Copie01"
|
||
previous.mkdir(parents=True)
|
||
(previous / "sentinel.txt").write_text("old", encoding="utf-8")
|
||
loaded = AnnotationLoadResult(
|
||
{
|
||
"01": {
|
||
"Ex 1": {
|
||
"pdf_path": answer,
|
||
"result": {"feedback": [], "score": 1},
|
||
"coordinates": (0, 0),
|
||
}
|
||
}
|
||
},
|
||
[],
|
||
)
|
||
with patch.object(
|
||
module, "load_annotation_data", return_value=loaded
|
||
), patch.object(
|
||
self.modules["annotating"],
|
||
"make_base_image",
|
||
side_effect=RuntimeError("render failed"),
|
||
), redirect_stderr(io.StringIO()):
|
||
self.assertEqual(module.main([str(evaluation), "--overwrite"]), 1)
|
||
self.assertEqual(
|
||
(previous / "sentinel.txt").read_text(encoding="utf-8"), "old"
|
||
)
|
||
|
||
def test_grouped_overwrite_preserves_previous_output_when_incomplete(self) -> None:
|
||
module = self.modules["annotating_by_label"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
(evaluation / "Copies").mkdir(parents=True)
|
||
(evaluation / "Par label").mkdir()
|
||
(evaluation / "labels").write_text("Ex 1\n", encoding="utf-8")
|
||
(evaluation / "label_groups").write_text("Ex 1\n", encoding="utf-8")
|
||
atomic_write_json(evaluation / "correction.json", {})
|
||
previous = evaluation / "BGnot"
|
||
previous.mkdir()
|
||
(previous / "sentinel.txt").write_text("old", encoding="utf-8")
|
||
loaded = AnnotationLoadResult(
|
||
{"01": {"Ex 1": {"pdf_path": Path("missing")}}},
|
||
[],
|
||
)
|
||
with patch.object(
|
||
module, "load_annotation_data", return_value=loaded
|
||
), patch.object(module, "_generate_groups", return_value=(1, True)):
|
||
self.assertEqual(module.main([str(evaluation), "--overwrite"]), 4)
|
||
self.assertEqual(
|
||
(previous / "sentinel.txt").read_text(encoding="utf-8"), "old"
|
||
)
|
||
|
||
def test_grouped_annotations_default_to_gemini_question_groups(self) -> None:
|
||
module = self.modules["annotating_by_label"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory) / "Exam"
|
||
items = evaluation / "Tmp" / "exam_items.txt"
|
||
items.parent.mkdir(parents=True)
|
||
items.write_text(
|
||
"# edited Gemini groups\n"
|
||
"Ex 1 ### First question\n"
|
||
"CONTEXT ### Shared context\n"
|
||
"Ex 2 ### Second question\n"
|
||
"\n---\n\n"
|
||
"Ex 3 ### Third question\n",
|
||
encoding="utf-8",
|
||
)
|
||
workspace = EvaluationWorkspace(evaluation)
|
||
|
||
with patch.object(module.utils, "edit_file_and_enter") as editor:
|
||
groups = module._load_label_groups(
|
||
workspace, ["Ex 1", "Ex 2", "Ex 3"]
|
||
)
|
||
|
||
self.assertEqual(groups, [["Ex 1", "Ex 2"], ["Ex 3"]])
|
||
self.assertEqual(
|
||
workspace.label_groups_file.read_text(encoding="utf-8"),
|
||
"Ex 1,Ex 2\nEx 3\n",
|
||
)
|
||
editor.assert_called_once_with(workspace.label_groups_file)
|
||
|
||
def test_grouped_annotations_fall_back_when_gemini_was_not_run(self) -> None:
|
||
module = self.modules["annotating_by_label"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
workspace = EvaluationWorkspace(Path(directory) / "Exam")
|
||
workspace.root.mkdir()
|
||
labels = ["Ex 1 : a", "Ex 1 : b", "Ex 2"]
|
||
|
||
with patch.object(module.utils, "edit_file_and_enter"):
|
||
groups = module._load_label_groups(workspace, labels)
|
||
|
||
self.assertEqual(groups, [["Ex 1 : a", "Ex 1 : b"], ["Ex 2"]])
|
||
|
||
def test_existing_label_groups_override_gemini_defaults(self) -> None:
|
||
module = self.modules["annotating_by_label"]
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
workspace = EvaluationWorkspace(Path(directory) / "Exam")
|
||
workspace.gemini_exam_items_file.parent.mkdir(parents=True)
|
||
workspace.gemini_exam_items_file.write_text(
|
||
"Ex 1 ### First\nEx 2 ### Second\n", encoding="utf-8"
|
||
)
|
||
workspace.label_groups_file.write_text(
|
||
"Ex 1\nEx 2\n", encoding="utf-8"
|
||
)
|
||
|
||
with patch.object(module.utils, "edit_file_and_enter") as editor:
|
||
groups = module._load_label_groups(workspace, ["Ex 1", "Ex 2"])
|
||
|
||
self.assertEqual(groups, [["Ex 1"], ["Ex 2"]])
|
||
editor.assert_not_called()
|
||
|
||
def test_grouped_batching_does_not_split_one_student(self) -> None:
|
||
module = self.modules["annotating_by_label"]
|
||
image = Image.new("RGB", (10, 60), "white")
|
||
rendered = [
|
||
("01", "Ex 1", image, 0, []),
|
||
("01", "Ex 2", image, 0, []),
|
||
("02", "Ex 1", image, 0, []),
|
||
]
|
||
with patch.object(module, "MAX_HEIGHT_PX", 100):
|
||
batches = module.split_batches(rendered)
|
||
self.assertEqual([len(batch) for batch in batches], [2, 1])
|
||
|
||
|
||
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_first_visit_automation_is_declared_on_expected_steps(self) -> None:
|
||
auto_start = {
|
||
step.id for step in self.steps.values() if step.auto_start_first_visit
|
||
}
|
||
skip_for_live = {
|
||
step.id for step in self.steps.values() if step.skip_for_live_correction
|
||
}
|
||
skip_without_conflicts = {
|
||
step.id
|
||
for step in self.steps.values()
|
||
if step.skip_without_manual_conflicts
|
||
}
|
||
|
||
self.assertEqual(auto_start, {"splitting", "grouping"})
|
||
self.assertEqual(
|
||
skip_for_live, {"submit_batches", "batch_status", "fetch_batches"}
|
||
)
|
||
self.assertEqual(skip_without_conflicts, {"manual_resolution"})
|
||
|
||
def test_review_persp_has_shorter_title(self) -> None:
|
||
self.assertEqual(self.steps["review_persp"].title, "Relire les barèmes")
|
||
|
||
def test_export_has_shorter_title_and_annotation_argument(self) -> None:
|
||
self.assertEqual(self.steps["export"].title, "Exporter")
|
||
command = self.command(
|
||
"export",
|
||
"default",
|
||
{"target": self.evaluation, "annotation_dir": "Bnot"},
|
||
)
|
||
self.assertEqual(command_arguments(command), [self.evaluation, "Bnot"])
|
||
|
||
def test_copy_listing_prefers_processed_copies(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory)
|
||
(evaluation / "enonce.pdf").touch()
|
||
(evaluation / "scan.pdf").touch()
|
||
copies = evaluation / "Copies"
|
||
copies.mkdir()
|
||
(copies / "Copie02.pdf").touch()
|
||
(copies / "Copie01.pdf").touch()
|
||
|
||
self.assertEqual(
|
||
[path.name for path in copy_pdf_paths(evaluation)],
|
||
["Copie01.pdf", "Copie02.pdf"],
|
||
)
|
||
|
||
def test_annotation_directory_detection_uses_known_directories(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory)
|
||
(evaluation / "Bnot").mkdir()
|
||
(evaluation / "Anot").mkdir()
|
||
(evaluation / "Other").mkdir()
|
||
self.assertEqual(
|
||
detected_annotation_directories(evaluation), ("Bnot", "Anot")
|
||
)
|
||
|
||
def test_export_and_import_defaults_follow_previous_runs(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
evaluation = Path(directory)
|
||
(evaluation / "Bnot").mkdir()
|
||
(evaluation / "Anot").mkdir()
|
||
store = StateStore()
|
||
store.load(evaluation)
|
||
store.update_step("annotation", last_run_variant="simple")
|
||
app = SimpleNamespace(evaluation=evaluation, state_store=store)
|
||
|
||
choices, default = CopienatorApp._annotation_directory_choices(
|
||
app, "export"
|
||
)
|
||
self.assertEqual(choices, ("Bnot", "Anot"))
|
||
self.assertEqual(default, "Anot")
|
||
|
||
store.update_step(
|
||
"export", last_run_values={"annotation_dir": "Bnot"}
|
||
)
|
||
_choices, default = CopienatorApp._annotation_directory_choices(
|
||
app, "import"
|
||
)
|
||
self.assertEqual(default, "Bnot")
|
||
|
||
def test_plotting_shortcuts_describe_open_actions(self) -> None:
|
||
rendered = "\n".join(plotting_shortcut_lines())
|
||
self.assertIn("ouvrir l’énoncé", rendered)
|
||
self.assertIn("ouvrir la copie traitée", rendered)
|
||
self.assertIn("ouvrir la copie originale", rendered)
|
||
|
||
def test_proxy_is_opt_in_and_prefilled(self) -> None:
|
||
base = {"HTTPS_PROXY": "http://system-proxy", "OTHER": "kept"}
|
||
without_proxy = build_runner_environment(
|
||
base, " secret ", DEFAULT_HTTPS_PROXY, False
|
||
)
|
||
with_proxy = build_runner_environment(base, "", DEFAULT_HTTPS_PROXY, True)
|
||
|
||
self.assertNotIn("HTTPS_PROXY", without_proxy)
|
||
self.assertEqual(without_proxy["GEMINI_API_KEY"], "secret")
|
||
self.assertEqual(without_proxy["OTHER"], "kept")
|
||
self.assertEqual(with_proxy["HTTPS_PROXY"], DEFAULT_HTTPS_PROXY)
|
||
|
||
def test_manual_conflicts_ignore_blank_and_comment_lines(self) -> None:
|
||
with tempfile.TemporaryDirectory() as directory:
|
||
path = Path(directory) / "manual_resolutions.txt"
|
||
self.assertFalse(has_manual_conflicts(path))
|
||
path.write_text("\n### Instructions\n ### exemple\n", encoding="utf-8")
|
||
self.assertFalse(has_manual_conflicts(path))
|
||
path.write_text(
|
||
"### Instructions\nCopie01 Ex 1 -> Ex 2\n", encoding="utf-8"
|
||
)
|
||
self.assertTrue(has_manual_conflicts(path))
|
||
|
||
def test_live_correction_arguments(self) -> None:
|
||
command = self.command(
|
||
"correction",
|
||
"live",
|
||
{"target": self.evaluation, "overwrite": True, "limit": "0"},
|
||
)
|
||
self.assertEqual(
|
||
command_arguments(command),
|
||
[self.evaluation, "--overwrite", "--limit", "0"],
|
||
)
|
||
self.assertEqual(command[4], "correct")
|
||
|
||
def test_hybrid_correction_arguments(self) -> None:
|
||
command = self.command(
|
||
"correction", "hybrid", {"target": self.evaluation, "batch_from": "Ex 4"}
|
||
)
|
||
self.assertEqual(
|
||
command_arguments(command), [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.assertEqual(command[4], "annotate-grouped")
|
||
|
||
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:], ["-m", "copienator", "copies", "rotate", self.evaluation])
|
||
self.assertEqual(rename[2:], ["-m", "copienator", "copies", "rename", self.evaluation])
|
||
|
||
def test_clean_is_the_last_dangerous_gui_step(self) -> None:
|
||
ordered_steps = build_workflow(True)
|
||
clean = ordered_steps[-1]
|
||
self.assertEqual(clean.id, "clean")
|
||
self.assertTrue(clean.optional)
|
||
self.assertTrue(clean.variants[0].dangerous)
|
||
self.assertIn("progression", clean.variants[0].danger_warning)
|
||
self.assertEqual(
|
||
command_arguments(
|
||
self.command("clean", "default", {"target": self.evaluation})
|
||
),
|
||
[self.evaluation, "--yes"],
|
||
)
|
||
|
||
|
||
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("copienator.platform.os.link", side_effect=OSError), patch(
|
||
"copienator.platform.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()
|