feat(gui): câbler l'envoi de la télémétrie d'usage en fin de run

Le module usage_telemetry est maintenant réellement branché : la GUI V6
envoie les statistiques au portail après chaque run (les stats web
restaient vides sans cela).

- processing_runner : RunSummary porte une liste DocResult (ordinal,
  page_count via page_count_for, status, duration_ms, extension) — peuplée
  dans la boucle. Aucun nom/chemin de fichier.
- usage_telemetry : report_run_summary(summary, base_url, license_ref,
  machine_id, session, ...) construit le payload depuis le RunSummary et
  l'envoie (non bloquant). N'envoie RIEN sans license_ref. Spool JSONL si
  échec réseau.
- tab_usage : _finish() déclenche l'envoi en thread daemon (jamais bloquant
  pour l'UI ni le run).
- app : fournit le reporter à UsageTab avec le contexte licence (base_url du
  LicenseClient, license_ref via local_status, machine_id, app_version).

Tests : RunSummary.documents peuplé (0 chemin) ; report_run_summary (payload
correct, réseau KO → spool sans crash, pas d'envoi sans licence) ; _finish
appelle le reporter. 252 tests unit OK (0 régression), self-test OK.
V5/moteur/app_aivanov intacts, 0 dépendance. Aucun build/push sans GO Dom.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-15 21:24:43 +02:00
parent d30f7b74ef
commit 1bbe70a911
7 changed files with 260 additions and 0 deletions

View File

@@ -163,3 +163,29 @@ def test_no_double_run(tmp_path):
release.set()
worker.join(timeout=2)
assert runner.is_running is False
# -- détails par document (télémétrie) -------------------------------------
def test_run_records_per_document_details(tmp_path):
_touch(tmp_path / "a.pdf")
_touch(tmp_path / "b.pdf")
def fake(doc, out):
if doc.name == "b.pdf":
raise RuntimeError("boom")
return {}
runner = ProcessingRunner(process_fn=fake, extensions=_EXTS)
summary = runner.run(tmp_path)
assert len(summary.documents) == 2
statuses = {doc.ordinal: doc.status for doc in summary.documents}
assert statuses == {0: "success", 1: "failed"}
for doc in summary.documents:
assert doc.extension == "pdf"
assert isinstance(doc.duration_ms, int)
# RGPD : aucun nom/chemin de fichier dans les détails
assert not hasattr(doc, "path")
assert not hasattr(doc, "filename")
assert not hasattr(doc, "name")

View File

@@ -164,3 +164,26 @@ def test_reglages_labels_renamed_and_profile_readable(ctk_root, tmp_path, monkey
tab._open_terms_table()
tab.update_idletasks()
tab.destroy()
def test_usage_tab_finish_calls_reporter(ctk_root):
"""Câblage : la fin de run appelle le reporter de télémétrie (non bloquant)."""
import threading
from gui_v6.processing_runner import RunSummary
from gui_v6.tabs.tab_usage import UsageTab
called = threading.Event()
captured = {}
def reporter(summary):
captured["summary"] = summary
called.set()
tab = UsageTab(ctk_root, usage_reporter=reporter)
ctk_root.update_idletasks()
summary = RunSummary(total=1, succeeded=1)
tab._finish(summary)
assert called.wait(timeout=3.0) # reporter appelé en thread daemon
assert captured["summary"] is summary
tab.destroy()

View File

@@ -135,3 +135,67 @@ def test_flush_keeps_failures(tmp_path):
# l'échec reste en file pour un prochain essai
assert spool.exists()
assert "a" in spool.read_text(encoding="utf-8")
# --- report_run_summary (câblage fin de run) --------------------------------
class _FakeDoc:
def __init__(self, ordinal, page_count, status, duration_ms=None, extension=None):
self.ordinal = ordinal
self.page_count = page_count
self.status = status
self.duration_ms = duration_ms
self.extension = extension
class _FakeSummary:
def __init__(self, documents):
self.documents = documents
def test_report_run_summary_builds_and_sends():
from gui_v6.usage_telemetry import report_run_summary
sess = _FakeSession(status_code=200)
summary = _FakeSummary([
_FakeDoc(0, 5, "success", extension="pdf"),
_FakeDoc(1, None, "failed"),
])
ok = report_run_summary(
summary, base_url="http://localhost:8088", license_ref="LIC-1",
machine_id="machine-0001", session=sess, app_version="6.0.0-g1",
)
assert ok is True
payload = sess.calls[0]["json"]
assert payload["license_ref"] == "LIC-1"
assert payload["app_name"] == "gui_v6"
assert payload["document_count"] == 2
assert payload["total_pages"] == 5
blob = json.dumps(payload, ensure_ascii=False).lower()
assert "filename" not in blob and "path" not in blob
def test_report_run_summary_no_send_without_license():
from gui_v6.usage_telemetry import report_run_summary
sess = _FakeSession(status_code=200)
summary = _FakeSummary([_FakeDoc(0, 1, "success")])
ok = report_run_summary(
summary, base_url="http://x", license_ref=None, machine_id="m1", session=sess
)
assert ok is False
assert sess.calls == [] # aucun appel réseau sans licence
def test_report_run_summary_network_down_spools(tmp_path):
from gui_v6.usage_telemetry import report_run_summary
sess = _FakeSession(raise_exc=OSError("down"))
summary = _FakeSummary([_FakeDoc(0, 1, "success")])
spool = tmp_path / "spool.jsonl"
ok = report_run_summary(
summary, base_url="http://x", license_ref="LIC-1", machine_id="m1",
session=sess, spool_path=spool,
)
assert ok is False # ne lève pas
assert spool.exists() # conservé pour rejeu