Compare commits

...
2 Commits
Author SHA1 Message Date
sebastien a56fc6e680 utilities to fix LaTeX issues 2026-09-29 10:53:32 +02:00
sebastien 2a4c5f4f7c better faulty encoding detection 2026-09-29 08:03:47 +02:00
15 changed files with 463 additions and 27 deletions

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 import prompting
from copienator.feedback_boxes import valid_feedback_box
from copienator.latex_json import repair_latex_json
from copienator import (
CliError,
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:
"""Stop scheduling Gemini calls while allowing in-flight calls to finish."""
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)
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"]]
@@ -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}")
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
for p in json_data:
+9
View File
@@ -554,11 +554,18 @@ def run(
client = genai.Client(api_key=api_key)
groups = group_images(images)
total_groups = len(groups)
active: dict[str, tuple[str, float]] = {}
progress_lock = Lock()
incomplete: list[Path] = []
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:
with progress_lock:
active[group] = (description, started)
@@ -582,6 +589,7 @@ def run(
for group_key, files in groups.items()
}
pending = set(futures)
print_group_progress(0)
while pending:
done, pending = wait(pending, timeout=PROGRESS_INTERVAL)
for future in done:
@@ -591,6 +599,7 @@ def run(
future.result()
except IncompleteCopyError as exc:
incomplete.extend(exc.images)
print_group_progress(total_groups - len(pending))
if pending:
now = time.monotonic()
with progress_lock:
+1
View File
@@ -313,6 +313,7 @@ class ImageViewer:
# Bindings
self.root.bind(PLOTTING_KB["OK"], self.on_enter)
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["open pdf"], self.on_open_pdf)
self.root.bind(PLOTTING_KB["open original pdf"], self.on_open_ori_pdf)
+46 -1
View File
@@ -16,6 +16,7 @@ from copienator import (
read_json,
workspace_from_args,
)
from copienator.latex_json import repair_decoded_latex_controls
WORD_LIST_FILE = Path(__file__).resolve().parents[1] / "data" / "liste_francais.txt"
ACCENT_PATTERN = re.compile(r"[éèêëàâäîïôöùûüçœÉÈÊËÀÂÄÎÏÔÖÙÛÜÇŒ]")
@@ -47,6 +48,15 @@ def escape_latex_underscores(text: str) -> str:
return "".join(parts)
def escape_latex_percents(text: str) -> str:
r"""Escape percent signs before annotation text is passed to LaTeX.
Unlike underscores, percent signs are comments both inside and outside
math mode. Collapsing an existing escape run keeps cleanup idempotent.
"""
return re.sub(r"\\*%", lambda _match: r"\%", text)
def normalize_overescaped_latex_commands(text: str) -> str:
r"""Collapse doubled command escapes inside LaTeX math environments.
@@ -89,6 +99,7 @@ def fix_hex_corruption_safe(text: str) -> str:
def repair_json_escape_corruption(text: str) -> str:
r"""Restore observed LaTeX commands consumed as JSON control escapes."""
text = repair_decoded_latex_controls(text)
replacements = (
("\x0crac", r"\frac"),
("\x0ceuille", r"\equiv"),
@@ -116,7 +127,7 @@ def clean_string(text: str, lookup: dict[str, str]) -> str:
text = fast_fix(text, lookup).replace("\x00", "")
text = repair_json_escape_corruption(text)
text = normalize_overescaped_latex_commands(text)
return escape_latex_underscores(text)
return escape_latex_percents(escape_latex_underscores(text))
def clean_obj(value: Any, lookup: dict[str, str]) -> Any:
@@ -132,6 +143,33 @@ def clean_obj(value: Any, lookup: dict[str, str]) -> Any:
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(
workspace: EvaluationWorkspace,
*,
@@ -141,11 +179,18 @@ def run(
lookup = build_lookup_map(word_list_path)
data = read_json(workspace.correction_file)
cleaned = clean_obj(data, lookup)
unresolved = unresolved_math_control_paths(cleaned)
backup = workspace.root / "correction_precleanup.json"
atomic_write_bytes(backup, workspace.correction_file.read_bytes())
print(f"Original JSON backed up to {backup}")
atomic_write_json(workspace.correction_file, cleaned)
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
+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}
_STRING = re.compile(r'"(?:\\.|[^"\\])*"', re.DOTALL)
_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:
@@ -44,6 +47,29 @@ underbrace underline underset uparrow updownarrow upsilon usepackage
'''.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:
output = []
index = 0
@@ -71,7 +97,9 @@ def _repair_string(token: str) -> str:
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.
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.
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 = []
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)
if previous is not None and raw[previous.end():match.start()].strip() == ':':
if previous.group() in _CONTENT_KEYS:
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))
end = match.end()
previous = match
+10
View File
@@ -141,6 +141,13 @@ import subprocess
from .configuration import LATEX_AFTER, LATEX_BEFORE
LATEX_WARNING_MARKER = "[Copienator:latex-warning] "
def report_latex_warning(output_pdf_path) -> None:
"""Emit a stable, machine-readable path alongside human diagnostics."""
print(f"{LATEX_WARNING_MARKER}{Path(output_pdf_path).resolve()}")
def compile_to_pdf(text, output_pdf_path, *, strict=False):
"""Wraps text in standalone header/footer templates and compiles it to PDF."""
@@ -185,6 +192,7 @@ def compile_to_pdf(text, output_pdf_path, *, strict=False):
shutil.move(generated_pdf, output_pdf_path)
if result.returncode != 0:
report_latex_warning(output_pdf_path)
lines = result.stdout.splitlines()
first = next((i for i, line in enumerate(lines) if line.lstrip().startswith("!")), None)
detail = "\n" + "\n".join(lines[first:first + 5]) if first is not None else ""
@@ -196,10 +204,12 @@ def compile_to_pdf(text, output_pdf_path, *, strict=False):
if strict:
raise RuntimeError(f"LaTeX failed for {output_pdf_path}{detail}")
elif not pdf_generated:
report_latex_warning(output_pdf_path)
print(f"Warning: LaTeX compilation produced no PDF for {output_pdf_path}")
if strict:
raise RuntimeError(f"No PDF generated for {output_pdf_path}")
except Exception as e:
if strict:
raise
report_latex_warning(output_pdf_path)
print(f"Compilation error for {output_pdf_path}: {e}")
+121 -5
View File
@@ -20,6 +20,7 @@ from copienator.platform import (
open_path,
validate_windows_labels,
)
from copienator.utils import LATEX_WARNING_MARKER
from .diagnostics import collect_diagnostics
from .batch_monitor import BatchMonitor
@@ -281,6 +282,8 @@ class CopienatorApp(tk.Tk):
self.pending_refaire_commands: list[list[str]] = []
self.refaire_command_index = 0
self._correction_progress_buffer = ""
self._latex_warning_buffer = ""
self._latex_warning_paths: list[Path] = []
self.title("Copienator — assistant de correction")
self.geometry("1180x820")
@@ -325,7 +328,7 @@ class CopienatorApp(tk.Tk):
top = ttk.Frame(self, padding=(10, 8))
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(
row=0, column=0, padx=(0, 8)
)
@@ -333,14 +336,19 @@ class CopienatorApp(tk.Tk):
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.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(
top, text="Ouvrir le dossier", command=self._open_evaluation_folder
)
self.open_evaluation_button.grid(row=0, column=3, padx=(6, 0))
ttk.Button(top, text="Parcourir…", command=self._browse_evaluation).grid(
row=0, column=4, padx=6
self.open_evaluation_button.grid(row=0, column=4, padx=(12, 0))
self.open_statement_button = ttk.Button(
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.Entry(top, textvariable=self.api_key_var, show="•", width=32).grid(
@@ -488,6 +496,13 @@ class CopienatorApp(tk.Tk):
clipboard_actions.grid(row=1, column=0, columnspan=6, sticky="w", pady=(5, 0))
ttk.Button(clipboard_actions, text="Copier la sélection", command=self._copy_console_selection).pack(side="left")
ttk.Button(clipboard_actions, text="Copier toute la sortie", command=self._copy_console_all).pack(side="left", padx=6)
self.latex_warning_button = ttk.Button(
clipboard_actions,
text="Problèmes LaTeX…",
command=self._show_current_latex_warnings,
)
self.latex_warning_button.pack(side="left", padx=6)
self.latex_warning_button.pack_forget()
ttk.Label(console_actions, text="Réponse au script").grid(row=0, column=0, padx=(0, 6))
self.stdin_var = tk.StringVar()
self.stdin_entry = ttk.Entry(console_actions, textvariable=self.stdin_var)
@@ -560,6 +575,11 @@ class CopienatorApp(tk.Tk):
def _open_evaluation_folder(self) -> None:
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:
self.proxy_entry.configure(state="normal" if self.use_proxy_var.get() else "disabled")
@@ -572,6 +592,7 @@ class CopienatorApp(tk.Tk):
messagebox.showerror("Dossier invalide", "Choisissez un dossier d’évaluation existant.")
return
self.batch_monitor.stop()
self._reset_latex_warnings()
self._save_current_form()
self.state_store.load(evaluation)
for step in self.steps:
@@ -1522,6 +1543,8 @@ class CopienatorApp(tk.Tk):
last_run_values=run_values,
)
self.active_step_id = step.id
if step.id == "statement":
self._reset_latex_warnings()
if step.id == "correction":
self._correction_progress_buffer = ""
self.info_var.set("Correction : préparation des groupes…")
@@ -1632,6 +1655,7 @@ class CopienatorApp(tk.Tk):
if event == "output":
output = str(payload)
self._track_correction_progress(output)
self._track_latex_warnings(output)
self._append_console(output)
elif event == "input_echo":
self._append_console(f"> {payload}")
@@ -1660,6 +1684,88 @@ class CopienatorApp(tk.Tk):
f"({percent} %)."
)
def _reset_latex_warnings(self) -> None:
self._latex_warning_buffer = ""
self._latex_warning_paths = []
if hasattr(self, "latex_warning_button"):
self.latex_warning_button.pack_forget()
def _track_latex_warnings(self, output: str, *, flush: bool = False) -> None:
if self.active_step_id != "statement":
return
combined = self._latex_warning_buffer + output
lines = combined.splitlines(keepends=True)
self._latex_warning_buffer = ""
if lines and not lines[-1].endswith(("\n", "\r")) and not flush:
self._latex_warning_buffer = lines.pop()
for line in lines:
marker_at = line.find(LATEX_WARNING_MARKER)
if marker_at < 0:
continue
raw_path = line[marker_at + len(LATEX_WARNING_MARKER):].strip()
if not raw_path:
continue
path = Path(raw_path).expanduser()
if not path.is_absolute() and self.evaluation:
path = self.evaluation / path
path = path.resolve()
if path not in self._latex_warning_paths:
self._latex_warning_paths.append(path)
if self._latex_warning_paths:
self.latex_warning_button.pack(side="left", padx=6)
def _show_current_latex_warnings(self) -> None:
if not self._latex_warning_paths:
messagebox.showinfo("Compilation LaTeX", "Aucun problème de compilation détecté.")
return
dialog = tk.Toplevel(self)
dialog.title("Problèmes de compilation LaTeX")
dialog.geometry("900x380")
dialog.transient(self)
dialog.columnconfigure(0, weight=1)
dialog.rowconfigure(1, weight=1)
ttk.Label(
dialog,
text=(
f"{len(self._latex_warning_paths)} fichier(s) n’ont pas été compilés "
"correctement. Sélectionnez-en un pour ouvrir ses éléments."
),
padding=10,
wraplength=850,
justify="left",
).grid(row=0, column=0, sticky="ew")
paths = tk.Listbox(dialog, exportselection=False)
for pdf_path in self._latex_warning_paths:
try:
label = str(pdf_path.relative_to(self.evaluation)) if self.evaluation else str(pdf_path)
except ValueError:
label = str(pdf_path)
paths.insert("end", label)
paths.grid(row=1, column=0, sticky="nsew", padx=10)
paths.selection_set(0)
actions = ttk.Frame(dialog, padding=10)
actions.grid(row=2, column=0, sticky="ew")
def selected_pdf() -> Path:
selection = paths.curselection()
index = selection[0] if selection else 0
return self._latex_warning_paths[index]
def open_folder() -> None:
self._open_desktop_path(selected_pdf().parent, "dossier de compilation")
def open_source() -> None:
self._open_desktop_path(selected_pdf().with_suffix(".tex"), "fichier source LaTeX")
def open_pdf() -> None:
self._open_desktop_path(selected_pdf(), "fichier PDF compilé")
ttk.Button(actions, text="Ouvrir le dossier", command=open_folder).pack(side="left")
ttk.Button(actions, text="Ouvrir le .tex", command=open_source).pack(side="left", padx=6)
ttk.Button(actions, text="Ouvrir le PDF", command=open_pdf).pack(side="left")
ttk.Button(actions, text="Fermer", command=dialog.destroy).pack(side="right")
def _batch_watch_status_changed(self, message: str) -> None:
self.batch_watch_status.set(message)
self._update_controls()
@@ -1700,6 +1806,8 @@ class CopienatorApp(tk.Tk):
step_id = self.active_step_id
if not step_id:
return
if step_id == "statement":
self._track_latex_warnings("", flush=True)
status = process_status(return_code, interrupted)
if status == "success" and self.pending_refaire_commands:
command = self.pending_refaire_commands.pop(0)
@@ -1780,9 +1888,17 @@ class CopienatorApp(tk.Tk):
title = self.step_by_id[step_id].title
self._append_console(f"\n[Terminé — code {return_code} — {STATUS_LABELS[status]}]\n")
self.info_var.set(f"{title} : {STATUS_LABELS[status].lower()} (code {return_code}).")
if step_id == "statement" and self._latex_warning_paths:
count = len(self._latex_warning_paths)
self.info_var.set(
f"{title} : {STATUS_LABELS[status].lower()} (code {return_code}) — "
f"{count} problème(s) LaTeX."
)
self.active_step_id = None
self._populate_tree()
self._update_controls()
if step_id == "statement" and self._latex_warning_paths:
self.after_idle(self._show_current_latex_warnings)
if step_id == "personal_read":
if status == "success":
self.info_var.set("Commande terminée. Poursuivez la relecture, puis cliquez sur « Marquer réussie ».")
+13 -7
View File
@@ -4,12 +4,13 @@ import subprocess
import threading
from tkinter import ttk, messagebox
SERVICE = "syncthing@sebastien.service"
SERVICE = "syncthing.service"
SYSTEMCTL = ("systemctl", "--user", "--no-ask-password")
def service_status():
result = subprocess.run(
["systemctl", "show", SERVICE, "--property=ActiveState", "--value"],
[*SYSTEMCTL, "show", SERVICE, "--property=ActiveState", "--value"],
capture_output=True, text=True, timeout=10,
)
if result.returncode:
@@ -22,7 +23,7 @@ def control_service(action):
return "active"
command = "start" if action == "ensure" else action
result = subprocess.run(
["systemctl", "--no-ask-password", command, SERVICE],
[*SYSTEMCTL, command, SERVICE],
capture_output=True, text=True, timeout=60,
)
if result.returncode:
@@ -87,13 +88,18 @@ class SyncthingPanel(ttk.Frame):
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":
if action == "ensure":
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:
callback()
elif error and action != "status":
messagebox.showerror("Syncthing", error)
elif self.pending_export:
self.request("ensure")
if self.winfo_exists():
+1 -1
View File
@@ -706,7 +706,7 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
"personal_deploy",
"Étapes personnelles",
"Déployer et publier les copies",
"Étape manuelle : déployer miqmacs-copies-assets puis mettre à jour les copies depuis l’administration.",
"Étape manuelle : déployer miqmacs-copies-assets et la section `Interro` (inutile : mettre à jour les copies depuis l’administration).",
(manual("deploy"),),
personal=True,
),
+40 -3
View File
@@ -9,6 +9,7 @@ from copienator_gui.app import CopienatorApp
from copienator_gui.workflow import command_display
from copienator.commands.page_splitter import _selected_inputs
from copienator import EvaluationWorkspace
from copienator.utils import LATEX_WARNING_MARKER
@unittest.skipUnless(os.environ.get("DISPLAY"), "Tk requires a display")
@@ -99,15 +100,51 @@ class GuiConvenienceTests(unittest.TestCase):
self.assertEqual(self.app.state_store.step("statement_persp")["status"], expected)
self.assertEqual(self.app.state_store.step("statement_groups")["status"], "ready")
def test_compact_evaluation_input_and_open_folder_button(self):
def test_statement_latex_warnings_are_tracked_for_both_variants(self):
for variant in ("gemini", "personal"):
with self.subTest(variant=variant):
self.app._reset_latex_warnings()
self.app.state_store.update_step("statement", last_run_variant=variant)
self.app.active_step_id = "statement"
first = self.evaluation / "Text2" / "Ex 1.pdf"
second = self.evaluation / "Sol2" / "Ex 2.pdf"
marker = LATEX_WARNING_MARKER
self.app._track_latex_warnings(f"noise\n{marker}{first}")
self.assertEqual(self.app._latex_warning_paths, [])
self.app._track_latex_warnings(f"\n{marker}{second}\n{marker}{first}\n")
self.assertEqual(self.app._latex_warning_paths, [first, second])
self.assertTrue(self.app.latex_warning_button.winfo_manager())
with patch.object(self.app, "after_idle") as after_idle:
self.app._finish_process(4, False)
self.assertTrue(
any(call.args[0] == self.app._show_current_latex_warnings
for call in after_idle.call_args_list)
)
self.assertIn("2 problème(s) LaTeX", self.app.info_var.get())
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.open_evaluation_button.grid_info()["column"]),
int(self.app.browse_evaluation_button.grid_info()["column"]),
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:
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):
manual = self.evaluation / "manual_resolutions.txt"
+73
View File
@@ -21,6 +21,7 @@ from pypdf import PdfReader, PdfWriter
from copienator import (
CliError,
EvaluationWorkspace,
ExitCode,
WorkspaceNotFoundError,
WorkspaceValidationError,
atomic_update_json,
@@ -1058,6 +1059,26 @@ class StandardCliTests(unittest.TestCase):
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:
module = self.modules["plotting"]
self.assertEqual(
@@ -1317,6 +1338,16 @@ class StandardCliTests(unittest.TestCase):
self.assertEqual(result["list"][0]["label"], "??Ex 10 : 1)a)")
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:
module = self.modules["correction"]
with tempfile.TemporaryDirectory() as directory:
@@ -1754,6 +1785,18 @@ class StandardCliTests(unittest.TestCase):
r"valid $\begin{cases} A=0 \\ B=1 \end{cases}$",
)
self.assertEqual(module.clean_string(cleaned_text, {}), cleaned_text)
percent_text = r"Code `n % p`, prose 50%, math $n \% p$ and $50%$."
escaped_percent_text = (
r"Code `n \% p`, prose 50\%, math $n \% p$ and $50\%$."
)
self.assertEqual(
module.clean_string(percent_text, {}),
escaped_percent_text,
)
self.assertEqual(
module.clean_string(escaped_percent_text, {}),
escaped_percent_text,
)
corrupted_json_escapes = (
"Formulas $"
+ "\x0c"
@@ -1774,6 +1817,23 @@ class StandardCliTests(unittest.TestCase):
r"Formulas $\frac{1}{2}$, $\theta \equiv 0 [2\pi]$, "
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(
module.clean_string("x ∈ A and B ⊂ C", {}),
r"x \ensuremath{\in} A and B \ensuremath{\subset} C",
@@ -1800,6 +1860,19 @@ class StandardCliTests(unittest.TestCase):
self.assertEqual((evaluation / "correction.json").read_bytes(), cleaned)
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:
module = self.modules["resolve_manual"]
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):
result = labels.run(self.workspace, [], client=self.client, sleep=lambda _: None)
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("Copie01_01.jpg, attempt 1/3", output.getvalue())
self.assertIn("elapsed for image", output.getvalue())
+49 -1
View File
@@ -1,7 +1,11 @@
import json
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
@@ -44,6 +48,50 @@ class LatexJsonTests(unittest.TestCase):
self.assertEqual(repair_latex_json(raw),
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):
with self.assertRaises(json.JSONDecodeError):
json.loads(repair_latex_json(r'{"question_content":"\frac", broken}'))
+43 -3
View File
@@ -1,8 +1,16 @@
import subprocess
import os
import tkinter as tk
import unittest
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):
@@ -16,7 +24,7 @@ class SyncthingTests(unittest.TestCase):
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])
self.assertEqual(run.call_args.args[0], [*SYSTEMCTL, "start", SERVICE])
def test_failed_start_reports_error(self):
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"), \
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])
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])
+2
View File
@@ -38,6 +38,7 @@ class CompileToPdfTests(unittest.TestCase):
self.assertEqual(output.read_bytes(), b"partial")
self.assertIn("Warning: LaTeX compilation reported errors", console.getvalue())
self.assertIn(utils.LATEX_WARNING_MARKER + str(output.resolve()), console.getvalue())
self.assertIn("a PDF was generated", console.getvalue())
self.assertIn("missing \\item", console.getvalue())
@@ -69,6 +70,7 @@ class CompileToPdfTests(unittest.TestCase):
self.assertFalse(output.exists())
self.assertIn("produced no PDF", console.getvalue())
self.assertIn(utils.LATEX_WARNING_MARKER + str(output.resolve()), console.getvalue())
if __name__ == "__main__":