sebastien/Copies
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

EvaluationWorkspace

Browse Source
This commit is contained in:
sebastien
2026-08-20 12:44:58 +02:00
parent ac4ab782b2
commit 2f1cd00e32
11 changed files with 746 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_gui_core.py
+294
View File
@@ -0,0 +1,294 @@
from __future__ import annotations
import os
import queue
import sys
import tempfile
import time
import unittest
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from unittest.mock import patch
from pypdf import PdfReader, PdfWriter
from copienator import (
EvaluationWorkspace,
WorkspaceNotFoundError,
WorkspaceValidationError,
atomic_update_json,
atomic_write_json,
read_json,
)
from copienator_gui.diagnostics import collect_diagnostics
from copienator_gui.runner import ProcessRunner
from copienator_gui.state import StateStore
from copienator_gui.workflow import build_command, build_workflow, evaluation_argument
from copies_tools import rename_all, rotate_all
from platform_utils import (
WindowsLabelError,
replace_with_link_or_copy,
safe_filename,
validate_windows_labels,
windows_filename_problems,
)
REPOSITORY = Path(__file__).resolve().parents[1]
class WorkspaceTests(unittest.TestCase):
def test_canonical_paths_and_no_constructor_side_effects(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory) / "Exam"
root.mkdir()
workspace = EvaluationWorkspace(root, Path(directory))
self.assertFalse(workspace.metadata_dir.exists())
self.assertEqual(workspace.labels_file, root / "labels")
self.assertEqual(workspace.copies_dir, root / "Copies")
self.assertEqual(workspace.annotation_dir("grouped"), root / "BGnot")
self.assertEqual(workspace.state_database, root / ".copienator" / "state.sqlite3")
self.assertEqual(workspace.command_argument(), "Exam")
def test_discover_from_nested_artifact(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory) / "Exam"
nested = root / "Copies" / "Copie01"
nested.mkdir(parents=True)
(root / "labels").write_text("Ex 1\n", encoding="utf-8")
artifact = nested / "Ex 1.pdf"
artifact.touch()
workspace = EvaluationWorkspace.discover(artifact)
self.assertEqual(workspace.root, root)
def test_discover_and_require_report_clear_errors(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
with self.assertRaises(WorkspaceNotFoundError):
EvaluationWorkspace.discover(root)
workspace = EvaluationWorkspace(root)
with self.assertRaises(WorkspaceValidationError) as context:
workspace.require("enonce.pdf", "labels")
self.assertEqual(context.exception.missing, ["enonce.pdf", "labels"])
def test_names_file_prefers_evaluation_then_repository(self) -> None:
with tempfile.TemporaryDirectory() as directory:
repository = Path(directory)
root = repository / "Exam"
root.mkdir()
(repository / "names").write_text("Global\n", encoding="utf-8")
workspace = EvaluationWorkspace(root, repository)
self.assertEqual(workspace.names_file(), repository / "names")
(root / "names").write_text("Local\n", encoding="utf-8")
self.assertEqual(workspace.names_file(), root / "names")
class AtomicJsonTests(unittest.TestCase):
def test_atomic_round_trip_and_unicode(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "nested" / "state.json"
value = {"name": "Élève", "steps": [1, 2]}
atomic_write_json(path, value)
self.assertEqual(read_json(path), value)
self.assertEqual(list(path.parent.glob(f".{path.name}.*.tmp")), [])
def test_serialization_failure_preserves_previous_file(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "state.json"
atomic_write_json(path, {"version": 1})
before = path.read_bytes()
with self.assertRaises(TypeError):
atomic_write_json(path, {"invalid": object()}) # type: ignore[dict-item]
self.assertEqual(path.read_bytes(), before)
def test_locked_updates_do_not_lose_concurrent_changes(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "state.json"
def increment() -> None:
def update(value):
value["count"] += 1
atomic_update_json(
path,
update,
default_factory=lambda: {"count": 0},
)
with ThreadPoolExecutor(max_workers=8) as executor:
list(executor.map(lambda _index: increment(), range(40)))
self.assertEqual(read_json(path), {"count": 40})
def test_gui_state_uses_workspace_and_atomic_writer(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
store = StateStore()
store.load(root)
store.update_step("labels", status="success")
self.assertEqual(store.workspace, EvaluationWorkspace(root))
persisted = read_json(root / ".copienator-gui.json")
self.assertEqual(persisted["steps"]["labels"]["status"], "success")
class WorkflowTests(unittest.TestCase):
def setUp(self) -> None:
self.steps = {step.id: step for step in build_workflow(True)}
self.evaluation = evaluation_argument(REPOSITORY, REPOSITORY / "Example")
def command(self, step_id: str, variant_id: str, values: dict[str, object]) -> list[str]:
step = self.steps[step_id]
variant = next(item for item in step.variants if item.id == variant_id)
return build_command(REPOSITORY, step, variant, values, self.evaluation)
def test_personal_steps_are_configurable(self) -> None:
standard_ids = {step.id for step in build_workflow(False)}
personal_ids = {step.id for step in build_workflow(True)}
self.assertNotIn("update_ods", standard_ids)
self.assertIn("update_ods", personal_ids)
def test_live_correction_arguments(self) -> None:
command = self.command(
"correction",
"live",
{"target": self.evaluation, "overwrite": True, "limit": "0"},
)
self.assertEqual(command[3:], [self.evaluation, "--overwrite", "--limit", "0"])
self.assertIn("correction.py", command[2])
def test_hybrid_correction_arguments(self) -> None:
command = self.command(
"correction", "hybrid", {"target": self.evaluation, "batch_from": "Ex 4"}
)
self.assertEqual(command[3:], [self.evaluation, "--batch-from", "Ex 4"])
def test_annotation_variants_are_exclusive_commands(self) -> None:
command = self.command(
"annotation", "grouped", {"target": self.evaluation, "overwrite": True}
)
self.assertIn("annotating_by_label.py", command[2])
self.assertNotIn("annotating.py", command[2])
def test_copy_preparation_commands_are_python(self) -> None:
rotate = self.command("rotate", "rotate", {"target": self.evaluation})
rename = self.command("rename", "rename", {"target": self.evaluation})
self.assertEqual(rotate[2:], [str(REPOSITORY / "copies_tools.py"), "rotate", self.evaluation])
self.assertEqual(rename[2:], [str(REPOSITORY / "copies_tools.py"), "rename", self.evaluation])
class CrossPlatformFileTests(unittest.TestCase):
def test_rotate_all_skips_statement(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
for filename in ("Copie.pdf", "enonce.pdf"):
writer = PdfWriter()
writer.add_blank_page(width=100, height=200)
with (root / filename).open("wb") as output:
writer.write(output)
self.assertEqual(rotate_all(root), 1)
self.assertEqual(PdfReader(root / "Copie.pdf").pages[0].rotation, 180)
self.assertEqual(PdfReader(root / "enonce.pdf").pages[0].rotation, 0)
def test_rename_all_is_collision_safe(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
(root / "z.pdf").write_bytes(b"z")
(root / "a.pdf").write_bytes(b"a")
(root / "Copie01.pdf").write_bytes(b"existing")
(root / "enonce.pdf").write_bytes(b"statement")
rename_all(root)
self.assertEqual((root / "Copie01.pdf").read_bytes(), b"a")
self.assertEqual((root / "Copie02.pdf").read_bytes(), b"existing")
self.assertEqual((root / "Copie03.pdf").read_bytes(), b"z")
self.assertEqual((root / "enonce.pdf").read_bytes(), b"statement")
def test_link_falls_back_to_copy(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "source.txt"
destination = root / "destination.txt"
source.write_text("content", encoding="utf-8")
with patch("platform_utils.os.link", side_effect=OSError), patch(
"platform_utils.os.symlink", side_effect=OSError
):
method = replace_with_link_or_copy(source, destination)
self.assertEqual(method, "copy")
self.assertEqual(destination.read_text(encoding="utf-8"), "content")
def test_windows_safe_student_filename(self) -> None:
self.assertEqual(safe_filename('Jean: Dupont? '), "Jean Dupont")
self.assertEqual(safe_filename("CON"), "_CON")
self.assertEqual(safe_filename("name..."), "name")
def test_windows_labels_are_rejected_without_modification(self) -> None:
labels = ["Ex 1", "Ex 2 : a)", "AUX", "Ex 3."]
with self.assertRaises(WindowsLabelError) as context:
validate_windows_labels(labels, platform_name="nt")
message = str(context.exception)
self.assertIn("Ex 2 : a)", message)
self.assertIn("AUX", message)
self.assertIn("Ex 3.", message)
self.assertEqual(labels[1], "Ex 2 : a)")
def test_windows_safe_labels_are_accepted(self) -> None:
validate_windows_labels(["Ex 1 - a)", "Question 2"], platform_name="nt")
validate_windows_labels(["Ex 1 : a)"], platform_name="posix")
self.assertTrue(windows_filename_problems("Ex 1 : a)"))
def test_diagnostics_include_supported_platform(self) -> None:
checks = collect_diagnostics(False, "test-key")
system = next(check for check in checks if check.name == "Système")
key = next(check for check in checks if check.name == "Clé Gemini")
self.assertTrue(system.ok)
self.assertTrue(key.ok)
def test_diagnostics_detect_windows_unsafe_labels(self) -> None:
with tempfile.TemporaryDirectory() as directory:
evaluation = Path(directory)
(evaluation / "labels").write_text("Ex 1 : a)\nEx 2\n", encoding="utf-8")
checks = collect_diagnostics(False, "test-key", evaluation)
labels = next(check for check in checks if check.name == "Labels compatibles Windows")
self.assertFalse(labels.ok)
self.assertIn("Ex 1 : a)", labels.detail)
class RunnerTests(unittest.TestCase):
def wait_for_finish(self, runner: ProcessRunner, timeout: float = 5) -> tuple[int, bool]:
deadline = time.time() + timeout
while time.time() < deadline:
try:
event, payload = runner.events.get(timeout=0.2)
except queue.Empty:
continue
if event == "finished":
return payload
self.fail("Le sous-processus ne s’est pas terminé")
def test_interactive_input_and_output(self) -> None:
with tempfile.TemporaryDirectory() as directory:
runner = ProcessRunner()
runner.start(
[
sys.executable,
"-u",
"-c",
"value=input('réponse ? '); print('reçu:', value)",
],
REPOSITORY,
os.environ.copy(),
Path(directory) / "runner.log",
)
time.sleep(0.1)
runner.send_input("oui")
self.assertEqual(self.wait_for_finish(runner), (0, False))
self.assertIn("reçu: oui", (Path(directory) / "runner.log").read_text(encoding="utf-8"))
if __name__ == "__main__":
unittest.main()