From 857efcef186f696971a19e1ca09d21c01536d45e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Miquel?= Date: Sat, 26 Sep 2026 23:01:47 +0200 Subject: [PATCH] =?UTF-8?q?fichiers=20oubli=C3=A9s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- copienator/answer_origins.py | 28 ++++ copienator/latex_json.py | 94 ++++++++++++ copienator_gui/giving_names.py | 228 ++++++++++++++++++++++++++++ copienator_gui/syncthing.py | 113 ++++++++++++++ tests/test_label_timeouts.py | 109 +++++++++++++ tests/test_latex_json.py | 49 ++++++ tests/test_submit_batches_resume.py | 70 +++++++++ tests/test_syncthing.py | 31 ++++ 8 files changed, 722 insertions(+) create mode 100644 copienator/answer_origins.py create mode 100644 copienator/latex_json.py create mode 100644 copienator_gui/giving_names.py create mode 100644 copienator_gui/syncthing.py create mode 100644 tests/test_label_timeouts.py create mode 100644 tests/test_latex_json.py create mode 100644 tests/test_submit_batches_resume.py create mode 100644 tests/test_syncthing.py diff --git a/copienator/answer_origins.py b/copienator/answer_origins.py new file mode 100644 index 0000000..5efb733 --- /dev/null +++ b/copienator/answer_origins.py @@ -0,0 +1,28 @@ +"""Provenance of shared answer content, independent of Gemini corrections.""" +from hashlib import sha256 +from pathlib import Path + +from .json_io import atomic_update_json, read_json + + +def _path(root): + return Path(root) / ".copienator" / "answer_origins.json" + + +def origins_for(root, copy_id, label, pdf): + entry = read_json(_path(root), default={}).get(str(copy_id), {}).get(label, {}) + if not entry or not Path(pdf).is_file(): + return [] + if entry.get("sha256") != sha256(Path(pdf).read_bytes()).hexdigest(): + return [] + return entry.get("labels", []) + + +def save_origins(root, copy_id, label, pdf, origins): + entry = {"labels": list(dict.fromkeys(origins)), + "sha256": sha256(Path(pdf).read_bytes()).hexdigest()} + + def update(data): + data.setdefault(str(copy_id), {})[label] = entry + + atomic_update_json(_path(root), update, default_factory=dict) diff --git a/copienator/latex_json.py b/copienator/latex_json.py new file mode 100644 index 0000000..fc2a158 --- /dev/null +++ b/copienator/latex_json.py @@ -0,0 +1,94 @@ +"""Conservative repair of LaTeX escaping in raw model-generated JSON. + +Call before JSON decoding, including for valid JSON: ``\\frac`` and ``\\not`` +can otherwise be silently decoded as control characters. This is heuristic; +unknown commands starting with a valid JSON escape are left alone. +""" +from __future__ import annotations + +import re + +# Only these content fields carry TeX. Never rewrite labels or structural keys. +_CONTENT_FIELDS = {'question_content', 'solution_content', 'context_content', 'rubric_content'} +_CONTENT_KEYS = {f'"{field}"' for field in _CONTENT_FIELDS} +_STRING = re.compile(r'"(?:\\.|[^"\\])*"', re.DOTALL) +_COMMAND = re.compile(r'[a-zA-Z]+') + + +def normalize_literal_newlines(text: str) -> str: + r"""Decode stray \n tokens in extracted TeX, preserving commands and \\. + + An odd backslash run ends in an escaped newline; any preceding pairs + remain LaTeX row breaks. Even runs are already legitimate LaTeX. + """ + def replace(match: re.Match[str]) -> str: + slashes = match.group(1) + return slashes[:-1] + "\n" if len(slashes) % 2 else match.group(0) + + return re.sub(r"(\\+)n(?![A-Za-z])", replace, text) +# Commands whose first letter is also a valid JSON escape. Other commands do +# not need a dictionary: their single backslash is invalid JSON already. +_AMBIGUOUS_COMMANDS = set(''' +begin beta bar backslash binom boxed boldsymbol big bigl bigr biggl biggr +bigskip blacktriangle blacksquare bmod breve bullet bra braket +frac fbox forall footnote footnotesize framebox +not neq neg nabla newline newcommand newenvironment noindent nonumber +nexists notin nmid nparallel nsubseteq nsupseteq normalfont normalsize +rightarrow right leftarrow rangle rbrace rceil rfloor rightarrowtail +Rightarrow rightharpoonup rightharpoondown rightleftharpoons rightleftarrows +rm rule ref rho rightskip raisebox require renewcommand +text textbf textit textrm textsf texttt textnormal textcolor textsc textsl +textup textwidth theta tau tan tanh times to top triangle triangleleft +triangleright tilde tfrac today tag tiny thickspace thinspace tbinom +underbrace underline underset uparrow updownarrow upsilon usepackage +'''.split()) + + +def _repair_string(token: str) -> str: + output = [] + index = 0 + while index < len(token): + char = token[index] + if char != '\\' or index + 1 == len(token): + output.append(char) + index += 1 + continue + following = token[index + 1] + # Consume already escaped pairs together, never double-escape them. + if following in '\\"/': + output.append(token[index:index + 2]) + index += 2 + continue + command_match = _COMMAND.match(token, index + 1) + command = command_match.group() if command_match else '' + unicode_escape = following == 'u' and re.match(r'[0-9a-fA-F]{4}', token[index + 2:]) + if not unicode_escape and (following not in 'bfnrtu' or command in _AMBIGUOUS_COMMANDS + or following == 'u'): + output.append('\\\\') + else: + output.append('\\') + index += 1 + return ''.join(output) + + +def repair_latex_json(raw: str) -> str: + """Repair single TeX backslashes without decoding or rewriting other JSON. + + Preserve standard escapes, Unicode escapes, and already escaped TeX. + A literal newline followed by e.g. 'ot' has the same wire representation + as a broken ``\\not`` command; known TeX command names take precedence. + Raw responses/caches should be retained to allow reviewing such cases. + """ + output = [] + end = 0 + previous = None + for match in _STRING.finditer(raw): + token = match.group() + if (previous is not None and raw[previous.end():match.start()].strip() == ':' + and previous.group() in _CONTENT_KEYS): + token = _repair_string(token) + output.extend((raw[end:match.start()], token)) + end = match.end() + previous = match + output.append(raw[end:]) + return ''.join(output) diff --git a/copienator_gui/giving_names.py b/copienator_gui/giving_names.py new file mode 100644 index 0000000..95219bf --- /dev/null +++ b/copienator_gui/giving_names.py @@ -0,0 +1,228 @@ +from __future__ import annotations + +import os +import re +import tkinter as tk +from collections import Counter +from dataclasses import dataclass +from pathlib import Path +from tkinter import messagebox, ttk +from typing import Callable + +from copienator import EvaluationWorkspace +from copienator.platform import open_path, safe_filename + + +RETURN_FOLDER_PATTERN = re.compile(r"^(?P.+) \((?P\d+)\)$") + + +@dataclass(frozen=True, slots=True) +class NameIssue: + folder: Path + name: str + copy_id: str + reasons: tuple[str, ...] + + +def find_name_issues(evaluation: Path) -> list[NameIssue]: + """Return unknown and duplicated names currently present in A Rendre.""" + return_dir = EvaluationWorkspace(evaluation).return_dir + parsed: list[tuple[Path, str, str]] = [] + if return_dir.is_dir(): + for folder in return_dir.iterdir(): + if not folder.is_dir() or folder.is_symlink(): + continue + match = RETURN_FOLDER_PATTERN.fullmatch(folder.name) + if match: + parsed.append( + (folder, match.group("name"), match.group("copy_id")) + ) + + counts = Counter(name.casefold() for _folder, name, _copy_id in parsed) + issues = [] + for folder, name, copy_id in parsed: + reasons = [] + if name.casefold() == "unknown": + reasons.append("nom inconnu") + if counts[name.casefold()] > 1: + reasons.append(f"nom attribué à {counts[name.casefold()]} copies") + if reasons: + issues.append(NameIssue(folder, name, copy_id, tuple(reasons))) + return sorted(issues, key=lambda issue: (int(issue.copy_id), issue.name.casefold())) + + +def original_copy_path(evaluation: Path, copy_id: str) -> Path | None: + workspace = EvaluationWorkspace(evaluation) + filename = f"Copie{copy_id}.pdf" + for candidate in ( + workspace.original_copies_dir / filename, + workspace.copies_dir / filename, + workspace.root / filename, + ): + if candidate.is_file(): + return candidate + return None + + +def rename_return_copy(folder: Path, requested_name: str) -> Path: + """Rename one return folder and its top-level JPG/PDF, preserving its ID.""" + match = RETURN_FOLDER_PATTERN.fullmatch(folder.name) + if not folder.is_dir() or folder.is_symlink() or match is None: + raise ValueError(f"Dossier de copie invalide : {folder}") + if folder.parent.name != "A Rendre": + raise ValueError("Le dossier à renommer doit se trouver dans A Rendre.") + + entered_name = requested_name.strip() + if not entered_name: + raise ValueError("Saisissez un nom.") + name = safe_filename(entered_name, fallback="") + if not name: + raise ValueError("Le nom ne contient aucun caractère utilisable.") + + copy_id = match.group("copy_id") + destination = folder.with_name(f"{name} ({copy_id})") + if destination != folder and os.path.lexists(destination): + raise FileExistsError(f"Le dossier {destination.name!r} existe déjà.") + + assets: dict[str, Path] = {} + for child in folder.iterdir(): + suffix = child.suffix.casefold() + if suffix not in {".jpg", ".pdf"}: + continue + if suffix in assets: + raise ValueError( + f"Plusieurs fichiers {suffix} sont présents dans {folder.name}." + ) + assets[suffix] = child + + renames: list[tuple[Path, Path]] = [] + for suffix, source in assets.items(): + target = source.with_name(f"{name}{suffix}") + if target != source and os.path.lexists(target): + raise FileExistsError(f"Le fichier {target.name!r} existe déjà.") + if target != source: + renames.append((source, target)) + + completed: list[tuple[Path, Path]] = [] + try: + for source, target in renames: + source.rename(target) + completed.append((source, target)) + if destination != folder: + folder.rename(destination) + except OSError: + for source, target in reversed(completed): + if os.path.lexists(target) and not os.path.lexists(source): + target.rename(source) + raise + return destination + + +class GivingNamesPanel(ttk.LabelFrame): + def __init__( + self, + master: tk.Misc, + evaluation: Path, + *, + opener: Callable[[Path], None] = open_path, + ) -> None: + super().__init__(master, text="Vérification des noms", padding=7) + self.evaluation = evaluation + self.opener = opener + self.columnconfigure(0, weight=1) + + actions = ttk.Frame(self) + actions.grid(row=0, column=0, sticky="w", pady=(0, 7)) + ttk.Button( + actions, + text="Ouvrir A Rendre", + command=self.open_return_folder, + ).pack(side="left") + ttk.Button(actions, text="Recharger", command=self.reload).pack( + side="left", padx=6 + ) + + self.issue_frame = ttk.Frame(self) + self.issue_frame.grid(row=1, column=0, sticky="ew") + self.issue_frame.columnconfigure(1, weight=1) + self.reload() + + def open_return_folder(self) -> None: + path = EvaluationWorkspace(self.evaluation).return_dir + if not path.is_dir(): + messagebox.showerror( + "Dossier introuvable", + "Exécutez d’abord l’attribution des noms pour créer A Rendre.", + ) + return + try: + self.opener(path) + except (OSError, RuntimeError) as exc: + messagebox.showerror("Ouverture impossible", str(exc)) + + def open_original(self, copy_id: str) -> None: + path = original_copy_path(self.evaluation, copy_id) + if path is None: + messagebox.showerror( + "Copie introuvable", f"Copie{copy_id}.pdf est introuvable." + ) + return + try: + self.opener(path) + except (OSError, RuntimeError) as exc: + messagebox.showerror("Ouverture impossible", str(exc)) + + def rename_issue(self, issue: NameIssue, variable: tk.StringVar) -> None: + try: + destination = rename_return_copy(issue.folder, variable.get()) + except (OSError, ValueError) as exc: + messagebox.showerror("Renommage impossible", str(exc)) + return + self.reload() + messagebox.showinfo("Copie renommée", f"Nouveau dossier : {destination.name}") + + def reload(self) -> None: + for child in self.issue_frame.winfo_children(): + child.destroy() + issues = find_name_issues(self.evaluation) + if not issues: + return_dir = EvaluationWorkspace(self.evaluation).return_dir + text = ( + "Aucun nom inconnu ou dupliqué." + if return_dir.is_dir() + else "A Rendre n’a pas encore été généré." + ) + ttk.Label(self.issue_frame, text=text).grid( + row=0, column=0, columnspan=4, sticky="w" + ) + return + + ttk.Label(self.issue_frame, text="Copie / problème").grid( + row=0, column=0, sticky="w", padx=(0, 6) + ) + ttk.Label(self.issue_frame, text="Nouveau nom").grid( + row=0, column=1, sticky="w" + ) + for row, issue in enumerate(issues, start=1): + ttk.Label( + self.issue_frame, + text=f"Copie{issue.copy_id} — {', '.join(issue.reasons)}", + ).grid(row=row, column=0, sticky="w", padx=(0, 6), pady=2) + variable = tk.StringVar() + ttk.Entry(self.issue_frame, textvariable=variable).grid( + row=row, column=1, sticky="ew", pady=2 + ) + original = original_copy_path(self.evaluation, issue.copy_id) + ttk.Button( + self.issue_frame, + text="Ouvrir l’originale", + command=lambda ident=issue.copy_id: self.open_original(ident), + state="normal" if original else "disabled", + ).grid(row=row, column=2, padx=6, pady=2) + ttk.Button( + self.issue_frame, + text="Renommer", + command=lambda item=issue, value=variable: self.rename_issue( + item, value + ), + ).grid(row=row, column=3, pady=2) diff --git a/copienator_gui/syncthing.py b/copienator_gui/syncthing.py new file mode 100644 index 0000000..7345313 --- /dev/null +++ b/copienator_gui/syncthing.py @@ -0,0 +1,113 @@ +"""Non-blocking controls for the system Syncthing instance.""" +import queue +import subprocess +import threading +from tkinter import ttk, messagebox + +SERVICE = "syncthing@sebastien.service" + + +def service_status(): + result = subprocess.run( + ["systemctl", "show", SERVICE, "--property=ActiveState", "--value"], + capture_output=True, text=True, timeout=10, + ) + if result.returncode: + raise RuntimeError(result.stderr.strip() or "Statut indisponible") + return result.stdout.strip() or "unknown" + + +def control_service(action): + if action == "ensure" and service_status() == "active": + return "active" + command = "start" if action == "ensure" else action + result = subprocess.run( + ["systemctl", "--no-ask-password", command, SERVICE], + capture_output=True, text=True, timeout=60, + ) + if result.returncode: + raise RuntimeError(result.stderr.strip() or f"Échec de systemctl {command}") + status = service_status() + if command == "start" and status != "active": + raise RuntimeError(f"Syncthing n’est pas actif : {status}") + return status + + +class SyncthingPanel(ttk.Frame): + def __init__(self, parent): + super().__init__(parent) + self.events = queue.Queue() + self.busy = False + self.pending_export = None + self.status = ttk.Label(self, text=f"{SERVICE} : vérification…") + self.status.pack(anchor="w") + controls = ttk.Frame(self) + controls.pack(anchor="w", pady=4) + self.start_button = ttk.Button(controls, text="Démarrer Syncthing", + command=lambda: self.request("start")) + self.start_button.pack(side="left") + self.stop_button = ttk.Button(controls, text="Arrêter Syncthing", + command=lambda: self.request("stop")) + self.stop_button.pack(side="left", padx=6) + self.timer = self.after(100, self.poll) + self.refresh_timer = None + self.bind("", self.cleanup, add=True) + self.request("status") + + def ensure_started(self, callback): + self.pending_export = callback + if not self.busy: + self.request("ensure") + + def request(self, action): + if self.busy: + return + self.busy = True + self.start_button.configure(state="disabled") + self.stop_button.configure(state="disabled") + if action != "status": + self.status.configure(text=f"{SERVICE} : opération en cours…") + + def work(): + try: + status = service_status() if action == "status" else control_service(action) + self.events.put((action, status, None)) + except (OSError, RuntimeError, subprocess.TimeoutExpired) as exc: + self.events.put((action, None, str(exc))) + + threading.Thread(target=work, daemon=True).start() + + def poll(self): + try: + action, status, error = self.events.get_nowait() + except queue.Empty: + pass + else: + self.busy = False + self.start_button.configure(state="normal") + self.stop_button.configure(state="normal") + self.status.configure(text=f"{SERVICE} : {error or status}") + if error and action != "status": + self.pending_export = None + messagebox.showerror("Syncthing", error) + elif action == "ensure": + callback, self.pending_export = self.pending_export, None + if callback: + callback() + elif self.pending_export: + self.request("ensure") + if self.winfo_exists(): + self.timer = self.after(100, self.poll) + if self.refresh_timer is None: + self.refresh_timer = self.after(5000, self.refresh) + + def refresh(self): + self.refresh_timer = None + self.request("status") + + def cleanup(self, event): + if event.widget == self: + self.pending_export = None + for timer in (self.timer, self.refresh_timer): + if timer: + self.after_cancel(timer) diff --git a/tests/test_label_timeouts.py b/tests/test_label_timeouts.py new file mode 100644 index 0000000..5d49d37 --- /dev/null +++ b/tests/test_label_timeouts.py @@ -0,0 +1,109 @@ +import io +import tempfile +import unittest +from concurrent.futures import wait as real_wait +from contextlib import redirect_stdout +from pathlib import Path +from threading import Event +from unittest.mock import Mock, patch + +from copienator import EvaluationWorkspace, ExitCode +from copienator.commands import gemini_for_labels as labels + + +class LabelTimeoutTests(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.workspace = EvaluationWorkspace(Path(directory.name)) + (self.workspace.root / "labels").write_text("Ex 1\n") + (self.workspace.root / "names").write_text("Student\n") + self.workspace.copies_dir.mkdir() + (self.workspace.root / "Cutleft").mkdir() + self.images = [] + for name in ("Copie01_01", "Copie01_02", "Copie02_01"): + image = self.workspace.root / "Cutleft" / f"{name}.jpg" + image.write_bytes(b"image") + self.images.append(image) + self.client = Mock() + self.response = Mock(text='{"name":"Student","list":[]}') + + def group(self, images): + return labels.process_copy_group( + self.workspace, "Copie01", images, client=self.client, + labels_text="Ex 1\n", names_text="Student\n", + valid_labels={"Ex 1"}, valid_names={"Student"}, overwrite=False, + sleep=lambda _: None, target_interval=0, + ) + + def test_timeout_recovery_uses_bounded_requests_and_saves_result(self): + self.client.models.generate_content.side_effect = [TimeoutError(), self.response] + self.assertEqual(self.group(self.images[:1]), 1) + self.assertEqual(self.client.models.generate_content.call_count, 2) + for call in self.client.models.generate_content.call_args_list: + options = call.kwargs["config"].http_options + self.assertEqual(options.timeout, 120_000) + self.assertEqual(options.retry_options.attempts, 1) + self.assertTrue((self.workspace.copies_dir / "Copie01_01.json").exists()) + + def test_exhaustion_defers_dependent_pages_but_other_copies_finish_and_resume(self): + self.client.models.generate_content.side_effect = [ + TimeoutError(), TimeoutError(), TimeoutError(), self.response, + ] + output = io.StringIO() + with patch.object(labels, "selected_images", return_value=(self.images, [])), redirect_stdout(output): + result = labels.run(self.workspace, [], client=self.client, + sleep=lambda _: None, max_workers=1) + self.assertEqual(result, ExitCode.PARTIAL) + self.assertEqual(self.client.models.generate_content.call_count, 4) + self.assertFalse((self.workspace.copies_dir / "Copie01_01.json").exists()) + self.assertFalse((self.workspace.copies_dir / "Copie01_02.json").exists()) + self.assertTrue((self.workspace.copies_dir / "Copie02_01.json").exists()) + self.assertIn("Incomplete images", output.getvalue()) + self.assertIn("Copie01_02.jpg", output.getvalue()) + self.client.models.generate_content.reset_mock(side_effect=True) + self.client.models.generate_content.return_value = self.response + with patch.object(labels, "selected_images", return_value=(self.images, [])): + result = labels.run(self.workspace, [], client=self.client, sleep=lambda _: None) + self.assertEqual(result, ExitCode.SUCCESS) + self.assertEqual(self.client.models.generate_content.call_count, 2) + + def test_changing_invalid_labels_cannot_retry_forever(self): + self.client.models.generate_content.side_effect = [Mock(text=( + '{"name":"Student","list":[{"box_2d":[1,2,3,4],"label":"bad%d"}]}' % index + )) for index in range(3)] + with self.assertRaises(labels.IncompleteCopyError): + self.group(self.images[:1]) + self.assertEqual(self.client.models.generate_content.call_count, 3) + + def test_progress_reports_waiting_image_and_elapsed_time(self): + entered, release = Event(), Event() + self.addCleanup(release.set) + + def request(**kwargs): + entered.set() + if not release.wait(2): + raise AssertionError("Test did not release request") + return self.response + + first_wait = True + + def wait(futures, timeout): + nonlocal first_wait + self.assertEqual(timeout, 30) + if first_wait: + first_wait = False + self.assertTrue(entered.wait(2)) + return set(), set(futures) + release.set() + return real_wait(futures, timeout=2) + + self.client.models.generate_content.side_effect = request + output = io.StringIO() + with patch.object(labels, "selected_images", return_value=(self.images[:1], [])), \ + patch.object(labels, "wait", side_effect=wait), redirect_stdout(output): + result = labels.run(self.workspace, [], client=self.client, sleep=lambda _: None) + self.assertEqual(result, ExitCode.SUCCESS) + self.assertIn("1 copies outstanding", output.getvalue()) + self.assertIn("Copie01_01.jpg, attempt 1/3", output.getvalue()) + self.assertIn("elapsed for image", output.getvalue()) diff --git a/tests/test_latex_json.py b/tests/test_latex_json.py new file mode 100644 index 0000000..7ba805c --- /dev/null +++ b/tests/test_latex_json.py @@ -0,0 +1,49 @@ +import json +import unittest + +from copienator.latex_json import repair_latex_json, normalize_literal_newlines +from copienator.commands.gemini_for_enonce import ExamQuestions, ExamSolutions, ExamContext, GroupRubrics + + +class LatexJsonTests(unittest.TestCase): + def test_literal_newlines_preserve_tex_commands_and_row_breaks(self): + content = r"Paragraph\n\n Next $\neq \notin \newline$ \\ next \\\n row" + expected = "Paragraph\n\n Next " + r"$\neq \notin \newline$ \\ next \\" + "\n row" + self.assertEqual(normalize_literal_newlines(content), expected) + self.assertEqual(normalize_literal_newlines(expected), expected) + raw = json.dumps({'solutions': [{'label': '1)', 'solution_content': content}]}) + parsed = ExamSolutions.model_validate_json(repair_latex_json(raw)) + self.assertEqual(parsed.solutions[0].solution_content, expected) + + def test_mixed_correct_and_broken_escapes_in_all_extraction_fields(self): + cases = [ + (ExamQuestions, 'questions', 'question_content', {'label': '1)'}), + (ExamSolutions, 'solutions', 'solution_content', {'label': '1)'}), + (ExamContext, 'contexts', 'context_content', + {'target_question_label': '1)', 'last_question_label': '2)'}), + (GroupRubrics, 'rubrics', 'rubric_content', {'label': '1)'})] + raw_content = r'$\circ \rightarrow \frac{1}{2} \not= \theta \begin{x} \unknown \underline{x}$\nSuite : $\\sqrt{x}$' + expected = r'$\circ \rightarrow \frac{1}{2} \not= \theta \begin{x} \unknown \underline{x}$' + '\n' + r'Suite : $\sqrt{x}$' + for model, collection, field, labels in cases: + with self.subTest(field=field): + raw = json.dumps({collection: [{**labels, field: 'PLACEHOLDER'}]}).replace('PLACEHOLDER', raw_content) + repaired = repair_latex_json(raw) + parsed = model.model_validate_json(repaired) + self.assertEqual(getattr(getattr(parsed, collection)[0], field), expected) + self.assertEqual(repair_latex_json(repaired), repaired) + + def test_valid_json_is_unchanged(self): + content = 'Énoncé "cité"\nNouvelle ligne\ntexte\nnext\ni \ne \nu \tindent\r\n' + r'$\not= \frac{1}{2} \\ \begin{array}$' + raw = json.dumps({'question_content': content, 'label': r'\not'}) + self.assertEqual(repair_latex_json(raw), raw) + self.assertEqual(repair_latex_json(r'{"rubric_content":"\u00e9 \/ \"ok\""}'), + r'{"rubric_content":"\u00e9 \/ \"ok\""}') + + def test_only_content_values_are_repaired(self): + raw = r'{"label":"\not", "other":"\frac", "rubric_content":"\frac"}' + self.assertEqual(repair_latex_json(raw), + r'{"label":"\not", "other":"\frac", "rubric_content":"\\frac"}') + + def test_does_not_repair_structural_errors(self): + with self.assertRaises(json.JSONDecodeError): + json.loads(repair_latex_json(r'{"question_content":"\frac", broken}')) diff --git a/tests/test_submit_batches_resume.py b/tests/test_submit_batches_resume.py new file mode 100644 index 0000000..ba0b24e --- /dev/null +++ b/tests/test_submit_batches_resume.py @@ -0,0 +1,70 @@ +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +from copienator import CliError, EvaluationWorkspace, ExitCode, atomic_write_json, read_json +from copienator.commands.submit_batches import run + + +class BatchSubmissionResumeTests(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.workspace = EvaluationWorkspace(Path(directory.name)) + for tier in ("flash", "pro"): + (self.workspace.root / f"batch_requests_{tier}.jsonl").write_text("{}\n") + self.client = Mock() + self.client.files.upload.return_value = SimpleNamespace(name="files/input") + self.client.batches.create.return_value = SimpleNamespace(name="batches/flash") + + def test_restart_after_second_upload_failure_preserves_first_job(self): + self.client.files.upload.side_effect = [ + SimpleNamespace(name="files/input"), ConnectionResetError("reset"), + ] + with self.assertRaises(ConnectionResetError): + run(self.workspace, client=self.client) + original = read_json(self.workspace.batch_jobs_file)["jobs"]["flash"] + self.assertEqual(original["name"], "batches/flash") + + self.client.reset_mock() + self.client.files.upload.side_effect = None + self.client.batches.create.return_value = SimpleNamespace(name="batches/pro") + self.assertEqual(run(self.workspace, client=self.client), ExitCode.SUCCESS) + self.client.files.upload.assert_called_once() + self.assertEqual(self.client.files.upload.call_args.kwargs["file"], + str(self.workspace.root / "batch_requests_pro.jsonl")) + self.client.batches.create.assert_called_once() + jobs = read_json(self.workspace.batch_jobs_file)["jobs"] + self.assertEqual(jobs["flash"], original) + self.assertEqual(jobs["pro"]["name"], "batches/pro") + + self.client.reset_mock() + self.assertEqual(run(self.workspace, client=self.client), ExitCode.SUCCESS) + self.client.files.upload.assert_not_called() + self.client.batches.create.assert_not_called() + + def test_legacy_manifest_is_skipped_and_explicit_overwrite_resubmits(self): + atomic_write_json(self.workspace.batch_jobs_file, {"jobs": { + "flash": {"name": "batches/old-flash"}, + "pro": {"name": "batches/old-pro"}, + }}) + self.assertEqual(run(self.workspace, client=self.client), ExitCode.SUCCESS) + self.client.files.upload.assert_not_called() + self.assertEqual(run(self.workspace, client=self.client, overwrite=True), ExitCode.SUCCESS) + self.assertEqual(self.client.batches.create.call_count, 2) + + def test_invalid_manifest_does_not_trigger_duplicate_submission(self): + for manifest in ({"jobs": []}, {"jobs": {"flash": {}}}): + with self.subTest(manifest=manifest): + atomic_write_json(self.workspace.batch_jobs_file, manifest) + with self.assertRaises(CliError): + run(self.workspace, client=self.client) + self.client.files.upload.assert_not_called() + + def test_empty_inputs_report_partial(self): + for tier in ("flash", "pro"): + (self.workspace.root / f"batch_requests_{tier}.jsonl").write_text("") + self.assertEqual(run(self.workspace, client=self.client), ExitCode.PARTIAL) + self.client.files.upload.assert_not_called() diff --git a/tests/test_syncthing.py b/tests/test_syncthing.py new file mode 100644 index 0000000..cee73d2 --- /dev/null +++ b/tests/test_syncthing.py @@ -0,0 +1,31 @@ +import subprocess +import unittest +from unittest.mock import patch + +from copienator_gui.syncthing import SERVICE, control_service + + +class SyncthingTests(unittest.TestCase): + def test_active_service_is_not_restarted(self): + with patch("copienator_gui.syncthing.service_status", return_value="active"), \ + patch("copienator_gui.syncthing.subprocess.run") as run: + self.assertEqual(control_service("ensure"), "active") + run.assert_not_called() + + def test_inactive_service_is_started_and_verified(self): + with patch("copienator_gui.syncthing.service_status", side_effect=["inactive", "active"]), \ + patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run: + self.assertEqual(control_service("ensure"), "active") + self.assertEqual(run.call_args.args[0], ["systemctl", "--no-ask-password", "start", SERVICE]) + + def test_failed_start_reports_error(self): + with patch("copienator_gui.syncthing.service_status", return_value="inactive"), \ + patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 1, stderr="Access denied")): + with self.assertRaisesRegex(RuntimeError, "Access denied"): + control_service("ensure") + + def test_stop_targets_system_instance(self): + with patch("copienator_gui.syncthing.service_status", return_value="inactive"), \ + patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run: + self.assertEqual(control_service("stop"), "inactive") + self.assertEqual(run.call_args.args[0], ["systemctl", "--no-ask-password", "stop", SERVICE])