206 lines
8.6 KiB
Python
206 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from PIL import Image
|
|
|
|
from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, read_json
|
|
from copienator.annotation_data import AnnotationLoadResult
|
|
from copienator.commands import annotating_with_checks as checks
|
|
from copienator.commands import export, import_annotations
|
|
from copienator.commands import reading_grouped_annotations as reader
|
|
from copienator_gui.refaire_sessions import begin_pass
|
|
|
|
|
|
class RefaireSessionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name) / "Exam"
|
|
self.root.mkdir()
|
|
self.workspace = EvaluationWorkspace(self.root)
|
|
atomic_write_json(self.workspace.refaire_file, [["Copie01", ["Ex 1"]]])
|
|
(self.root / "BRnot/Copie01").mkdir(parents=True)
|
|
(self.root / "BRnot/Copie01/Concat_annotated.pdf").write_bytes(b"legacy return")
|
|
self.state = {
|
|
"steps": {
|
|
"annotation": {"status": "success"},
|
|
"refaire_selection": {
|
|
"status": "success",
|
|
"values": {
|
|
"selection": [["Copie01", ["Ex 1"]]],
|
|
"annotation_dir": "Anot",
|
|
"layout": "grouped",
|
|
},
|
|
},
|
|
"refaire_merge": {"status": "success"},
|
|
},
|
|
"history": [],
|
|
}
|
|
atomic_write_json(self.workspace.gui_state_file, self.state)
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def test_new_then_repeat_preserve_each_pass_and_reset_only_redo_progress(self):
|
|
state, first = begin_pass(self.workspace, self.state, keep_selection=False)
|
|
self.assertEqual(
|
|
state["steps"]["annotation"], self.state["steps"]["annotation"]
|
|
)
|
|
self.assertNotIn("refaire_merge", state["steps"])
|
|
self.assertEqual(state["steps"]["refaire_selection"]["values"]["selection"], [])
|
|
self.assertEqual(
|
|
state["steps"]["refaire_selection"]["values"]["layout"], "grouped"
|
|
)
|
|
self.assertEqual(read_json(self.workspace.refaire_file), [])
|
|
archives = [
|
|
path for path in (self.root / "Reprises").iterdir() if path.name != first
|
|
]
|
|
self.assertEqual(len(archives), 1)
|
|
self.assertEqual(
|
|
(archives[0] / "BRnot/Copie01/Concat_annotated.pdf").read_bytes(),
|
|
b"legacy return",
|
|
)
|
|
selection = [["Copie02", ["Ex 2"]]]
|
|
atomic_write_json(self.workspace.refaire_file, selection)
|
|
first_output = self.workspace.annotation_dir("refaire")
|
|
first_output.mkdir(parents=True)
|
|
(first_output / "review.pdf").write_bytes(b"first review")
|
|
state["steps"]["refaire_merge"] = {"status": "success"}
|
|
updated, second = begin_pass(self.workspace, state, keep_selection=True)
|
|
self.assertNotEqual(first, second)
|
|
self.assertEqual(read_json(self.workspace.refaire_file), selection)
|
|
self.assertEqual(
|
|
updated["steps"]["refaire_selection"]["values"]["selection"], selection
|
|
)
|
|
self.assertEqual((first_output / "review.pdf").read_bytes(), b"first review")
|
|
self.assertEqual(read_json(first_output.parent / "refaire.json"), selection)
|
|
self.assertNotIn("status", updated["steps"]["refaire_selection"])
|
|
self.assertEqual(
|
|
self.workspace.annotation_dir("refaire"),
|
|
self.root / "Reprises" / second / "BRnot",
|
|
)
|
|
reloaded = EvaluationWorkspace(self.root)
|
|
self.assertEqual(reloaded.refaire_session_id, second)
|
|
|
|
def test_failed_activation_leaves_current_pass_and_state_unchanged(self):
|
|
before = copy.deepcopy(self.state)
|
|
with (
|
|
patch(
|
|
"copienator_gui.refaire_sessions.staged_files",
|
|
side_effect=OSError("disk unavailable"),
|
|
),
|
|
self.assertRaises(OSError),
|
|
):
|
|
begin_pass(self.workspace, self.state, keep_selection=False)
|
|
self.assertIsNone(self.workspace.refaire_session_id)
|
|
self.assertEqual(read_json(self.workspace.gui_state_file), before)
|
|
self.assertEqual(
|
|
read_json(self.workspace.refaire_file), [["Copie01", ["Ex 1"]]]
|
|
)
|
|
self.assertEqual(self.state, before)
|
|
self.assertEqual(
|
|
(self.root / "BRnot/Copie01/Concat_annotated.pdf").read_bytes(),
|
|
b"legacy return",
|
|
)
|
|
|
|
def test_exports_are_unique_and_old_returns_cannot_enter_new_pass(self):
|
|
export_root = Path(self.temp.name) / "Export"
|
|
import_root = Path(self.temp.name) / "Import"
|
|
import_root.mkdir()
|
|
state, first = begin_pass(self.workspace, self.state, keep_selection=True)
|
|
for expected_id in (first, None):
|
|
if expected_id is None:
|
|
state, second = begin_pass(self.workspace, state, keep_selection=True)
|
|
expected_id = second
|
|
output = self.workspace.annotation_dir("refaire") / "Ex 1 G1"
|
|
output.mkdir(parents=True)
|
|
(output / "Concat.pdf").write_bytes(expected_id.encode())
|
|
with patch.object(export, "EXPORT_DIR", export_root):
|
|
self.assertEqual(
|
|
export.run(self.workspace, refaire=True), ExitCode.SUCCESS
|
|
)
|
|
exported = (
|
|
export_root / "Exam" / expected_id / f"{expected_id}__Ex 1 G1.pdf"
|
|
)
|
|
self.assertEqual(exported.read_bytes(), expected_id.encode())
|
|
if expected_id == first:
|
|
(import_root / exported.name).write_bytes(b"old return")
|
|
current = self.workspace.annotation_dir("refaire") / "Ex 1 G1"
|
|
with patch.object(import_annotations, "IMPORT_DIR", import_root):
|
|
self.assertEqual(
|
|
import_annotations.run(self.workspace, refaire=True), ExitCode.PARTIAL
|
|
)
|
|
self.assertFalse((current / "Concat_annotated.pdf").exists())
|
|
(import_root / f"{second}__Ex 1 G1.pdf").write_bytes(b"new return")
|
|
self.assertEqual(
|
|
import_annotations.run(self.workspace, refaire=True), ExitCode.SUCCESS
|
|
)
|
|
self.assertEqual((current / "Concat_annotated.pdf").read_bytes(), b"new return")
|
|
self.workspace.require_directories("BRnot")
|
|
|
|
def test_generation_and_merge_follow_the_active_pass(self):
|
|
for directory in ("Copies", "Par label", "Anot"):
|
|
(self.root / directory).mkdir()
|
|
(self.root / "labels").write_text("Ex 1\n")
|
|
atomic_write_json(self.root / "correction.json", {})
|
|
answer = self.root / "answer.pdf"
|
|
answer.touch()
|
|
loaded = AnnotationLoadResult(
|
|
{
|
|
"01": {
|
|
"Ex 1": {
|
|
"pdf_path": answer,
|
|
"result": {"score": 2, "feedback": []},
|
|
"coordinates": (0, 0),
|
|
}
|
|
}
|
|
},
|
|
[],
|
|
)
|
|
begin_pass(self.workspace, self.state, keep_selection=True)
|
|
with (
|
|
patch.object(checks, "load_annotation_data", return_value=loaded),
|
|
patch.object(reader, "load_annotation_data", return_value=loaded),
|
|
patch.object(
|
|
checks.annotating,
|
|
"make_base_image",
|
|
return_value=(Image.new("RGB", (100, 100), "white"), 0, 0),
|
|
),
|
|
patch.object(
|
|
checks.annotating,
|
|
"compose_label_image",
|
|
return_value=(Image.new("RGB", (100, 100), "white"), 0),
|
|
),
|
|
):
|
|
self.assertEqual(
|
|
checks.run(self.workspace, self.root, refaire=True, overwrite=True),
|
|
ExitCode.SUCCESS,
|
|
)
|
|
output = self.workspace.annotation_dir("refaire") / "Copie01"
|
|
shutil.copy2(output / "Concat.pdf", output / "Concat_annotated.pdf")
|
|
self.assertEqual(
|
|
reader.run(self.workspace, refaire=True, annotation_dir="Anot"),
|
|
ExitCode.SUCCESS,
|
|
)
|
|
self.assertEqual(
|
|
read_json(self.root / "Anot/Copie01/score.json"), {"Ex 1": "2"}
|
|
)
|
|
self.assertEqual(
|
|
(self.root / "BRnot/Copie01/Concat_annotated.pdf").read_bytes(),
|
|
b"legacy return",
|
|
)
|
|
|
|
def test_invalid_session_id_cannot_escape_workspace(self):
|
|
atomic_write_json(self.root / "refaire-session.json", {"id": "../elsewhere"})
|
|
with self.assertRaises(ValueError):
|
|
self.workspace.annotation_dir("refaire")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|