better faulty encoding detection

This commit is contained in:
sebastien committed 2026-09-29 08:03:47 +02:00
1 parent ec8e79bb1a
commit 2a4c5f4f7c
12 files changed
+299 -25

No files matched your search

+13 -2
View File
@@ -21,6 +21,7 @@ from copienator.commands import grouping
from copienator.answer_origins import save_origins from copienator.answer_origins import save_origins
from copienator import prompting from copienator import prompting
from copienator.feedback_boxes import valid_feedback_box from copienator.feedback_boxes import valid_feedback_box
from copienator.latex_json import repair_latex_json
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
@@ -100,6 +101,16 @@ def report_group_progress(completed: int, total: int) -> None:
) )
def parse_correction_response(response_text: str):
"""Decode model JSON without consuming known LaTeX command prefixes."""
return json.loads(
repair_latex_json(
response_text,
additional_content_fields={"text"},
)
)
def request_graceful_stop(_signum=None, _frame=None) -> None: def request_graceful_stop(_signum=None, _frame=None) -> None:
"""Stop scheduling Gemini calls while allowing in-flight calls to finish.""" """Stop scheduling Gemini calls while allowing in-flight calls to finish."""
if not stop_requested.is_set(): if not stop_requested.is_set():
@@ -319,7 +330,7 @@ def correct_boxes_with_gemini(pid, label, pdf_path, original_feedbacks,
contents, config = prompting.request_for_box_correction(pdf_path, original_feedbacks) contents, config = prompting.request_for_box_correction(pdf_path, original_feedbacks)
response_text = call_gemini_with_retries(MODEL_ID_flash, contents, config) response_text = call_gemini_with_retries(MODEL_ID_flash, contents, config)
corrected_feedbacks = json.loads(response_text) corrected_feedbacks = parse_correction_response(response_text)
global_feedbacks = [f for f in original_feedbacks if not f["box_2d"]] global_feedbacks = [f for f in original_feedbacks if not f["box_2d"]]
@@ -507,7 +518,7 @@ def process_single_task(task_tuple, precomputed_response=None):
tprint(f"Asking Gemini {'Flash' if use_flash else 'Pro '}: {label} {group_name}") tprint(f"Asking Gemini {'Flash' if use_flash else 'Pro '}: {label} {group_name}")
full_response_text = call_gemini_with_retries(model_to_use, contents, config) full_response_text = call_gemini_with_retries(model_to_use, contents, config)
json_data = json.loads(full_response_text) json_data = parse_correction_response(full_response_text)
# Ensure consistency of answer placements # Ensure consistency of answer placements
for p in json_data: for p in json_data:
+9
View File
@@ -554,11 +554,18 @@ def run(
client = genai.Client(api_key=api_key) client = genai.Client(api_key=api_key)
groups = group_images(images) groups = group_images(images)
total_groups = len(groups)
active: dict[str, tuple[str, float]] = {} active: dict[str, tuple[str, float]] = {}
progress_lock = Lock() progress_lock = Lock()
incomplete: list[Path] = [] incomplete: list[Path] = []
run_started = time.monotonic() run_started = time.monotonic()
def print_group_progress(completed: int) -> None:
print(
f"Label detection progress: {completed}/{total_groups} copy groups completed.",
flush=True,
)
def report_progress(group: str, description: str, started: float) -> None: def report_progress(group: str, description: str, started: float) -> None:
with progress_lock: with progress_lock:
active[group] = (description, started) active[group] = (description, started)
@@ -582,6 +589,7 @@ def run(
for group_key, files in groups.items() for group_key, files in groups.items()
} }
pending = set(futures) pending = set(futures)
print_group_progress(0)
while pending: while pending:
done, pending = wait(pending, timeout=PROGRESS_INTERVAL) done, pending = wait(pending, timeout=PROGRESS_INTERVAL)
for future in done: for future in done:
@@ -591,6 +599,7 @@ def run(
future.result() future.result()
except IncompleteCopyError as exc: except IncompleteCopyError as exc:
incomplete.extend(exc.images) incomplete.extend(exc.images)
print_group_progress(total_groups - len(pending))
if pending: if pending:
now = time.monotonic() now = time.monotonic()
with progress_lock: with progress_lock:
+1
View File
@@ -313,6 +313,7 @@ class ImageViewer:
# Bindings # Bindings
self.root.bind(PLOTTING_KB["OK"], self.on_enter) self.root.bind(PLOTTING_KB["OK"], self.on_enter)
self.root.bind(PLOTTING_KB["previous"], self.on_previous) self.root.bind(PLOTTING_KB["previous"], self.on_previous)
self.root.bind("P", self.on_previous)
self.root.bind(PLOTTING_KB["edit"], self.on_edit) self.root.bind(PLOTTING_KB["edit"], self.on_edit)
self.root.bind(PLOTTING_KB["open pdf"], self.on_open_pdf) self.root.bind(PLOTTING_KB["open pdf"], self.on_open_pdf)
self.root.bind(PLOTTING_KB["open original pdf"], self.on_open_ori_pdf) self.root.bind(PLOTTING_KB["open original pdf"], self.on_open_ori_pdf)
+36
View File
@@ -16,6 +16,7 @@ from copienator import (
read_json, read_json,
workspace_from_args, workspace_from_args,
) )
from copienator.latex_json import repair_decoded_latex_controls
WORD_LIST_FILE = Path(__file__).resolve().parents[1] / "data" / "liste_francais.txt" WORD_LIST_FILE = Path(__file__).resolve().parents[1] / "data" / "liste_francais.txt"
ACCENT_PATTERN = re.compile(r"[éèêëàâäîïôöùûüçœÉÈÊËÀÂÄÎÏÔÖÙÛÜÇŒ]") ACCENT_PATTERN = re.compile(r"[éèêëàâäîïôöùûüçœÉÈÊËÀÂÄÎÏÔÖÙÛÜÇŒ]")
@@ -89,6 +90,7 @@ def fix_hex_corruption_safe(text: str) -> str:
def repair_json_escape_corruption(text: str) -> str: def repair_json_escape_corruption(text: str) -> str:
r"""Restore observed LaTeX commands consumed as JSON control escapes.""" r"""Restore observed LaTeX commands consumed as JSON control escapes."""
text = repair_decoded_latex_controls(text)
replacements = ( replacements = (
("\x0crac", r"\frac"), ("\x0crac", r"\frac"),
("\x0ceuille", r"\equiv"), ("\x0ceuille", r"\equiv"),
@@ -132,6 +134,33 @@ def clean_obj(value: Any, lookup: dict[str, str]) -> Any:
return value return value
def unresolved_math_control_paths(value: Any, path: str = "$") -> list[str]:
"""Locate destructive JSON control escapes left in mathematical text."""
if isinstance(value, str):
return (
[path]
if any(
control in {"\b", "\f"}
for match in MATH_PATTERN.finditer(value)
for control in match.group(0)
)
else []
)
if isinstance(value, list):
return [
issue
for index, item in enumerate(value)
for issue in unresolved_math_control_paths(item, f"{path}[{index}]")
]
if isinstance(value, dict):
return [
issue
for key, item in value.items()
for issue in unresolved_math_control_paths(item, f"{path}[{key!r}]")
]
return []
def run( def run(
workspace: EvaluationWorkspace, workspace: EvaluationWorkspace,
*, *,
@@ -141,11 +170,18 @@ def run(
lookup = build_lookup_map(word_list_path) lookup = build_lookup_map(word_list_path)
data = read_json(workspace.correction_file) data = read_json(workspace.correction_file)
cleaned = clean_obj(data, lookup) cleaned = clean_obj(data, lookup)
unresolved = unresolved_math_control_paths(cleaned)
backup = workspace.root / "correction_precleanup.json" backup = workspace.root / "correction_precleanup.json"
atomic_write_bytes(backup, workspace.correction_file.read_bytes()) atomic_write_bytes(backup, workspace.correction_file.read_bytes())
print(f"Original JSON backed up to {backup}") print(f"Original JSON backed up to {backup}")
atomic_write_json(workspace.correction_file, cleaned) atomic_write_json(workspace.correction_file, cleaned)
print(f"Fixed JSON saved to {workspace.correction_file}") print(f"Fixed JSON saved to {workspace.correction_file}")
if unresolved:
print(
"Warning: unresolved JSON control characters remain in mathematical "
"text at: " + ", ".join(unresolved)
)
return ExitCode.PARTIAL
return ExitCode.SUCCESS return ExitCode.SUCCESS
+40 -4
View File
@@ -13,6 +13,9 @@ _CONTENT_FIELDS = {'question_content', 'solution_content', 'context_content', 'r
_CONTENT_KEYS = {f'"{field}"' for field in _CONTENT_FIELDS} _CONTENT_KEYS = {f'"{field}"' for field in _CONTENT_FIELDS}
_STRING = re.compile(r'"(?:\\.|[^"\\])*"', re.DOTALL) _STRING = re.compile(r'"(?:\\.|[^"\\])*"', re.DOTALL)
_COMMAND = re.compile(r'[a-zA-Z]+') _COMMAND = re.compile(r'[a-zA-Z]+')
_MATH = re.compile(r"(\$\$.*?\$\$|\$.*?\$|\\\(.*?\\\)|\\\[.*?\\\])", re.DOTALL)
_DECODED_COMMAND = re.compile(r"[\b\f\n\r\t]([A-Za-z]+)")
_JSON_CONTROL_PREFIX = {"\b": "b", "\f": "f", "\n": "n", "\r": "r", "\t": "t"}
def normalize_literal_newlines(text: str) -> str: def normalize_literal_newlines(text: str) -> str:
@@ -44,6 +47,29 @@ underbrace underline underset uparrow updownarrow upsilon usepackage
'''.split()) '''.split())
def repair_decoded_latex_controls(text: str) -> str:
r"""Restore known TeX commands after JSON decoded their first letter.
Only math spans and whitelisted command names are changed. This avoids
treating ordinary paragraph newlines or tabs as damaged LaTeX.
"""
def repair_math(math_match: re.Match[str]) -> str:
def repair_command(command_match: re.Match[str]) -> str:
candidate = (
_JSON_CONTROL_PREFIX[command_match.group(0)[0]]
+ command_match.group(1)
)
return (
f"\\{candidate}"
if candidate in _AMBIGUOUS_COMMANDS
else command_match.group(0)
)
return _DECODED_COMMAND.sub(repair_command, math_match.group(0))
return _MATH.sub(repair_math, text)
def _repair_string(token: str) -> str: def _repair_string(token: str) -> str:
output = [] output = []
index = 0 index = 0
@@ -71,7 +97,9 @@ def _repair_string(token: str) -> str:
return ''.join(output) return ''.join(output)
def repair_latex_json(raw: str) -> str: def repair_latex_json(
raw: str, *, additional_content_fields: set[str] | None = None
) -> str:
"""Repair single TeX backslashes without decoding or rewriting other JSON. """Repair single TeX backslashes without decoding or rewriting other JSON.
Preserve standard escapes, Unicode escapes, and already escaped TeX. Preserve standard escapes, Unicode escapes, and already escaped TeX.
@@ -79,14 +107,22 @@ def repair_latex_json(raw: str) -> str:
as a broken ``\\not`` command; known TeX command names take precedence. as a broken ``\\not`` command; known TeX command names take precedence.
Raw responses/caches should be retained to allow reviewing such cases. Raw responses/caches should be retained to allow reviewing such cases.
""" """
additional_content_keys = {
f'"{field}"' for field in (additional_content_fields or set())
}
output = [] output = []
end = 0 end = 0
previous = None previous = None
for match in _STRING.finditer(raw): for match in _STRING.finditer(raw):
token = match.group() token = match.group()
if (previous is not None and raw[previous.end():match.start()].strip() == ':' if previous is not None and raw[previous.end():match.start()].strip() == ':':
and previous.group() in _CONTENT_KEYS): if previous.group() in _CONTENT_KEYS:
token = _repair_string(token) token = _repair_string(token)
elif previous.group() in additional_content_keys:
token = _MATH.sub(
lambda math_match: _repair_string(math_match.group(0)),
token,
)
output.extend((raw[end:match.start()], token)) output.extend((raw[end:match.start()], token))
end = match.end() end = match.end()
previous = match previous = match
+15 -5
View File
@@ -325,7 +325,7 @@ class CopienatorApp(tk.Tk):
top = ttk.Frame(self, padding=(10, 8)) top = ttk.Frame(self, padding=(10, 8))
top.grid(row=0, column=0, sticky="ew") top.grid(row=0, column=0, sticky="ew")
top.columnconfigure(5, weight=1) top.columnconfigure(6, weight=1)
ttk.Button(top, text="Charger", command=self._load_evaluation).grid( ttk.Button(top, text="Charger", command=self._load_evaluation).grid(
row=0, column=0, padx=(0, 8) row=0, column=0, padx=(0, 8)
) )
@@ -333,14 +333,19 @@ class CopienatorApp(tk.Tk):
self.evaluation_entry = ttk.Entry(top, textvariable=self.evaluation_var, width=42) self.evaluation_entry = ttk.Entry(top, textvariable=self.evaluation_var, width=42)
self.evaluation_entry.grid(row=0, column=2, sticky="w") self.evaluation_entry.grid(row=0, column=2, sticky="w")
self.evaluation_entry.bind("<Return>", lambda _event: self._load_evaluation()) self.evaluation_entry.bind("<Return>", lambda _event: self._load_evaluation())
self.browse_evaluation_button = ttk.Button(
top, text="Parcourir…", command=self._browse_evaluation
)
self.browse_evaluation_button.grid(row=0, column=3, padx=(6, 0))
self.open_evaluation_button = ttk.Button( self.open_evaluation_button = ttk.Button(
top, text="Ouvrir le dossier", command=self._open_evaluation_folder top, text="Ouvrir le dossier", command=self._open_evaluation_folder
) )
self.open_evaluation_button.grid(row=0, column=3, padx=(6, 0)) self.open_evaluation_button.grid(row=0, column=4, padx=(12, 0))
ttk.Button(top, text="Parcourir…", command=self._browse_evaluation).grid( self.open_statement_button = ttk.Button(
row=0, column=4, padx=6 top, text="Ouvrir l’énoncé", command=self._open_statement_pdf
) )
ttk.Button(top, text="Diagnostic…", command=self._show_diagnostics).grid(row=0, column=6, padx=(6, 0)) self.open_statement_button.grid(row=0, column=5, padx=(6, 0))
ttk.Button(top, text="Diagnostic…", command=self._show_diagnostics).grid(row=0, column=7, padx=(6, 0))
ttk.Label(top, text="Clé Gemini").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=(7, 0)) ttk.Label(top, text="Clé Gemini").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=(7, 0))
ttk.Entry(top, textvariable=self.api_key_var, show="•", width=32).grid( ttk.Entry(top, textvariable=self.api_key_var, show="•", width=32).grid(
@@ -560,6 +565,11 @@ class CopienatorApp(tk.Tk):
def _open_evaluation_folder(self) -> None: def _open_evaluation_folder(self) -> None:
self._open_desktop_path(self.evaluation, "dossier d’évaluation") self._open_desktop_path(self.evaluation, "dossier d’évaluation")
def _open_statement_pdf(self) -> None:
evaluation = self.evaluation
statement = EvaluationWorkspace(evaluation).statement_pdf_file if evaluation else None
self._open_desktop_path(statement, "fichier enonce.pdf")
def _toggle_proxy(self) -> None: def _toggle_proxy(self) -> None:
self.proxy_entry.configure(state="normal" if self.use_proxy_var.get() else "disabled") self.proxy_entry.configure(state="normal" if self.use_proxy_var.get() else "disabled")
+13 -7
View File
@@ -4,12 +4,13 @@ import subprocess
import threading import threading
from tkinter import ttk, messagebox from tkinter import ttk, messagebox
SERVICE = "syncthing@sebastien.service" SERVICE = "syncthing.service"
SYSTEMCTL = ("systemctl", "--user", "--no-ask-password")
def service_status(): def service_status():
result = subprocess.run( result = subprocess.run(
["systemctl", "show", SERVICE, "--property=ActiveState", "--value"], [*SYSTEMCTL, "show", SERVICE, "--property=ActiveState", "--value"],
capture_output=True, text=True, timeout=10, capture_output=True, text=True, timeout=10,
) )
if result.returncode: if result.returncode:
@@ -22,7 +23,7 @@ def control_service(action):
return "active" return "active"
command = "start" if action == "ensure" else action command = "start" if action == "ensure" else action
result = subprocess.run( result = subprocess.run(
["systemctl", "--no-ask-password", command, SERVICE], [*SYSTEMCTL, command, SERVICE],
capture_output=True, text=True, timeout=60, capture_output=True, text=True, timeout=60,
) )
if result.returncode: if result.returncode:
@@ -87,13 +88,18 @@ class SyncthingPanel(ttk.Frame):
self.start_button.configure(state="normal") self.start_button.configure(state="normal")
self.stop_button.configure(state="normal") self.stop_button.configure(state="normal")
self.status.configure(text=f"{SERVICE} : {error or status}") self.status.configure(text=f"{SERVICE} : {error or status}")
if error and action != "status": if action == "ensure":
self.pending_export = None
messagebox.showerror("Syncthing", error)
elif action == "ensure":
callback, self.pending_export = self.pending_export, None callback, self.pending_export = self.pending_export, None
if error:
messagebox.showwarning(
"Syncthing",
f"Impossible de démarrer Syncthing : {error}\n\n"
"L’export va continuer.",
)
if callback: if callback:
callback() callback()
elif error and action != "status":
messagebox.showerror("Syncthing", error)
elif self.pending_export: elif self.pending_export:
self.request("ensure") self.request("ensure")
if self.winfo_exists(): if self.winfo_exists():
+17 -3
View File
@@ -99,15 +99,29 @@ class GuiConvenienceTests(unittest.TestCase):
self.assertEqual(self.app.state_store.step("statement_persp")["status"], expected) self.assertEqual(self.app.state_store.step("statement_persp")["status"], expected)
self.assertEqual(self.app.state_store.step("statement_groups")["status"], "ready") self.assertEqual(self.app.state_store.step("statement_groups")["status"], "ready")
def test_compact_evaluation_input_and_open_folder_button(self): def test_evaluation_buttons_are_ordered_and_open_folder_and_statement(self):
self.assertEqual(int(self.app.evaluation_entry.cget("width")), 42) self.assertEqual(int(self.app.evaluation_entry.cget("width")), 42)
self.assertEqual( self.assertEqual(
int(self.app.open_evaluation_button.grid_info()["column"]), int(self.app.browse_evaluation_button.grid_info()["column"]),
int(self.app.evaluation_entry.grid_info()["column"]) + 1, int(self.app.evaluation_entry.grid_info()["column"]) + 1,
) )
self.assertEqual(
int(self.app.open_evaluation_button.grid_info()["column"]),
int(self.app.browse_evaluation_button.grid_info()["column"]) + 1,
)
self.assertEqual(
int(self.app.open_statement_button.grid_info()["column"]),
int(self.app.open_evaluation_button.grid_info()["column"]) + 1,
)
statement = self.evaluation / "enonce.pdf"
statement.touch()
with patch("copienator_gui.app.open_path") as opened: with patch("copienator_gui.app.open_path") as opened:
self.app.open_evaluation_button.invoke() self.app.open_evaluation_button.invoke()
opened.assert_called_once_with(self.evaluation) self.app.open_statement_button.invoke()
self.assertEqual(
opened.call_args_list,
[unittest.mock.call(self.evaluation), unittest.mock.call(statement)],
)
def test_manual_resolution_panel_follows_command_target(self): def test_manual_resolution_panel_follows_command_target(self):
manual = self.evaluation / "manual_resolutions.txt" manual = self.evaluation / "manual_resolutions.txt"
+61
View File
@@ -21,6 +21,7 @@ from pypdf import PdfReader, PdfWriter
from copienator import ( from copienator import (
CliError, CliError,
EvaluationWorkspace, EvaluationWorkspace,
ExitCode,
WorkspaceNotFoundError, WorkspaceNotFoundError,
WorkspaceValidationError, WorkspaceValidationError,
atomic_update_json, atomic_update_json,
@@ -1058,6 +1059,26 @@ class StandardCliTests(unittest.TestCase):
previous_image, previous_json, previous_meta previous_image, previous_json, previous_meta
) )
def test_plotting_accepts_lowercase_and_uppercase_previous_shortcuts(self) -> None:
module = self.modules["plotting"]
root = Mock()
root.winfo_screenwidth.return_value = 1200
root.winfo_screenheight.return_value = 800
label = Mock()
with patch.object(module.tk, "Label", return_value=label), patch.object(
module.ImageViewer, "poll_queue"
):
viewer = module.ImageViewer(
root,
SimpleNamespace(root=Path("/evaluation")),
set(),
queue.Queue(),
)
root.bind.assert_any_call("p", viewer.on_previous)
root.bind.assert_any_call("P", viewer.on_previous)
def test_plotting_validates_plain_and_directional_labels(self) -> None: def test_plotting_validates_plain_and_directional_labels(self) -> None:
module = self.modules["plotting"] module = self.modules["plotting"]
self.assertEqual( self.assertEqual(
@@ -1317,6 +1338,16 @@ class StandardCliTests(unittest.TestCase):
self.assertEqual(result["list"][0]["label"], "??Ex 10 : 1)a)") self.assertEqual(result["list"][0]["label"], "??Ex 10 : 1)a)")
self.assertIn("keeping them with a ?? prefix", output.getvalue()) self.assertIn("keeping them with a ?? prefix", output.getvalue())
def test_correction_response_preserves_ambiguous_latex_commands(self) -> None:
module = self.modules["correction"]
parsed = module.parse_correction_response(
r'[{"result":{"feedback":[{"text":"$\forall x, f \big(x\big) \to x$"}]}}]'
)
self.assertEqual(
parsed[0]["result"]["feedback"][0]["text"],
r"$\forall x, f \big(x\big) \to x$",
)
def test_correction_overwrite_keeps_previous_state_until_a_commit(self) -> None: def test_correction_overwrite_keeps_previous_state_until_a_commit(self) -> None:
module = self.modules["correction"] module = self.modules["correction"]
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
@@ -1774,6 +1805,23 @@ class StandardCliTests(unittest.TestCase):
r"Formulas $\frac{1}{2}$, $\theta \equiv 0 [2\pi]$, " r"Formulas $\frac{1}{2}$, $\theta \equiv 0 [2\pi]$, "
r"$e \times x$, and [\negthinspace[0,n]", r"$e \times x$, and [\negthinspace[0,n]",
) )
decoded_math_commands = (
"$"
+ "\x0corall x, f "
+ "\x08ig(x\x08ig) "
+ "\to "
+ "\text{false}$"
)
self.assertEqual(
module.clean_string(decoded_math_commands, {}),
r"$\forall x, f \big(x\big) \to \text{false}$",
)
self.assertEqual(
module.unresolved_math_control_paths(
{"feedback": ["Plain\fprose", "$unknown\fform$"]}
),
["$['feedback'][1]"],
)
self.assertEqual( self.assertEqual(
module.clean_string("x ∈ A and B ⊂ C", {}), module.clean_string("x ∈ A and B ⊂ C", {}),
r"x \ensuremath{\in} A and B \ensuremath{\subset} C", r"x \ensuremath{\in} A and B \ensuremath{\subset} C",
@@ -1800,6 +1848,19 @@ class StandardCliTests(unittest.TestCase):
self.assertEqual((evaluation / "correction.json").read_bytes(), cleaned) self.assertEqual((evaluation / "correction.json").read_bytes(), cleaned)
self.assertEqual(backup.read_bytes(), original) self.assertEqual(backup.read_bytes(), original)
with tempfile.TemporaryDirectory() as directory:
evaluation = Path(directory) / "Exam"
evaluation.mkdir()
atomic_write_json(
evaluation / "correction.json",
{"text": "$unknown\fcommand$"},
)
output = io.StringIO()
with redirect_stdout(output):
result = module.run(EvaluationWorkspace(evaluation))
self.assertEqual(result, ExitCode.PARTIAL)
self.assertIn("$['text']", output.getvalue())
def test_manual_resolution_with_no_actions_is_safe(self) -> None: def test_manual_resolution_with_no_actions_is_safe(self) -> None:
module = self.modules["resolve_manual"] module = self.modules["resolve_manual"]
with tempfile.TemporaryDirectory() as directory: with tempfile.TemporaryDirectory() as directory:
+2
View File
@@ -104,6 +104,8 @@ class LabelTimeoutTests(unittest.TestCase):
patch.object(labels, "wait", side_effect=wait), redirect_stdout(output): patch.object(labels, "wait", side_effect=wait), redirect_stdout(output):
result = labels.run(self.workspace, [], client=self.client, sleep=lambda _: None) result = labels.run(self.workspace, [], client=self.client, sleep=lambda _: None)
self.assertEqual(result, ExitCode.SUCCESS) self.assertEqual(result, ExitCode.SUCCESS)
self.assertIn("Label detection progress: 0/1 copy groups completed.", output.getvalue())
self.assertIn("Label detection progress: 1/1 copy groups completed.", output.getvalue())
self.assertIn("1 copies outstanding", output.getvalue()) self.assertIn("1 copies outstanding", output.getvalue())
self.assertIn("Copie01_01.jpg, attempt 1/3", output.getvalue()) self.assertIn("Copie01_01.jpg, attempt 1/3", output.getvalue())
self.assertIn("elapsed for image", output.getvalue()) self.assertIn("elapsed for image", output.getvalue())
+49 -1
View File
@@ -1,7 +1,11 @@
import json import json
import unittest import unittest
from copienator.latex_json import repair_latex_json, normalize_literal_newlines from copienator.latex_json import (
normalize_literal_newlines,
repair_decoded_latex_controls,
repair_latex_json,
)
from copienator.commands.gemini_for_enonce import ExamQuestions, ExamSolutions, ExamContext, GroupRubrics from copienator.commands.gemini_for_enonce import ExamQuestions, ExamSolutions, ExamContext, GroupRubrics
@@ -44,6 +48,50 @@ class LatexJsonTests(unittest.TestCase):
self.assertEqual(repair_latex_json(raw), self.assertEqual(repair_latex_json(raw),
r'{"label":"\not", "other":"\frac", "rubric_content":"\\frac"}') r'{"label":"\not", "other":"\frac", "rubric_content":"\\frac"}')
def test_feedback_text_can_be_repaired_before_json_decoding(self):
raw = (
r'{"text":"$\forall x, f \big(x\big) \neq x \to \text{false}$",'
r'"other":"\forall"}'
)
repaired = repair_latex_json(
raw,
additional_content_fields={"text"},
)
parsed = json.loads(repaired)
self.assertEqual(
parsed["text"],
r"$\forall x, f \big(x\big) \neq x \to \text{false}$",
)
self.assertEqual(parsed["other"], "\x0corall")
valid_prose = json.dumps({"text": "First line\ntext outside math"})
self.assertEqual(
repair_latex_json(
valid_prose,
additional_content_fields={"text"},
),
valid_prose,
)
def test_decoded_control_repair_is_limited_to_known_commands_in_math(self):
corrupted = (
"$"
+ "\x0corall x, f "
+ "\x08ig(x\x08ig) "
+ "\neq x "
+ "\to "
+ "\text{false}$"
)
self.assertEqual(
repair_decoded_latex_controls(corrupted),
r"$\forall x, f \big(x\big) \neq x \to \text{false}$",
)
ordinary_whitespace = "First paragraph\nnext line\tindented"
self.assertEqual(
repair_decoded_latex_controls(ordinary_whitespace),
ordinary_whitespace,
)
def test_does_not_repair_structural_errors(self): def test_does_not_repair_structural_errors(self):
with self.assertRaises(json.JSONDecodeError): with self.assertRaises(json.JSONDecodeError):
json.loads(repair_latex_json(r'{"question_content":"\frac", broken}')) json.loads(repair_latex_json(r'{"question_content":"\frac", broken}'))
+43 -3
View File
@@ -1,8 +1,16 @@
import subprocess import subprocess
import os
import tkinter as tk
import unittest import unittest
from unittest.mock import patch from unittest.mock import patch
from copienator_gui.syncthing import SERVICE, control_service from copienator_gui.syncthing import (
SERVICE,
SYSTEMCTL,
SyncthingPanel,
control_service,
service_status,
)
class SyncthingTests(unittest.TestCase): class SyncthingTests(unittest.TestCase):
@@ -16,7 +24,7 @@ class SyncthingTests(unittest.TestCase):
with patch("copienator_gui.syncthing.service_status", side_effect=["inactive", "active"]), \ with patch("copienator_gui.syncthing.service_status", side_effect=["inactive", "active"]), \
patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run: patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run:
self.assertEqual(control_service("ensure"), "active") self.assertEqual(control_service("ensure"), "active")
self.assertEqual(run.call_args.args[0], ["systemctl", "--no-ask-password", "start", SERVICE]) self.assertEqual(run.call_args.args[0], [*SYSTEMCTL, "start", SERVICE])
def test_failed_start_reports_error(self): def test_failed_start_reports_error(self):
with patch("copienator_gui.syncthing.service_status", return_value="inactive"), \ with patch("copienator_gui.syncthing.service_status", return_value="inactive"), \
@@ -28,4 +36,36 @@ class SyncthingTests(unittest.TestCase):
with patch("copienator_gui.syncthing.service_status", return_value="inactive"), \ with patch("copienator_gui.syncthing.service_status", return_value="inactive"), \
patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run: patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run:
self.assertEqual(control_service("stop"), "inactive") self.assertEqual(control_service("stop"), "inactive")
self.assertEqual(run.call_args.args[0], ["systemctl", "--no-ask-password", "stop", SERVICE]) self.assertEqual(run.call_args.args[0], [*SYSTEMCTL, "stop", SERVICE])
def test_status_uses_the_non_privileged_user_service(self):
with patch(
"copienator_gui.syncthing.subprocess.run",
return_value=subprocess.CompletedProcess([], 0, stdout="active\n"),
) as run:
self.assertEqual(service_status(), "active")
self.assertEqual(
run.call_args.args[0],
[*SYSTEMCTL, "show", SERVICE, "--property=ActiveState", "--value"],
)
@unittest.skipUnless(os.environ.get("DISPLAY"), "Tk requires a display")
class SyncthingPanelTests(unittest.TestCase):
def test_failed_automatic_start_warns_and_continues_export(self):
root = tk.Tk()
self.addCleanup(root.destroy)
with patch("copienator_gui.syncthing.threading.Thread.start"):
panel = SyncthingPanel(root)
panel.pack()
continued = []
panel.pending_export = lambda: continued.append(True)
panel.busy = True
panel.events.put(("ensure", None, "Access denied"))
with patch("copienator_gui.syncthing.messagebox.showwarning") as warning:
panel.poll()
warning.assert_called_once()
self.assertIn("L’export va continuer", warning.call_args.args[1])
self.assertEqual(continued, [True])