feat(gui): client diagnostics non bloquant + spool best-effort (E3)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-30 10:39:25 +02:00
parent 4b7a31b9df
commit 8eb8cf9999
2 changed files with 175 additions and 0 deletions

View File

@@ -44,3 +44,67 @@ def test_build_payload_counts_and_no_pii_leak():
assert forbidden not in blob, f"fuite RGPD : {forbidden}"
for item in payload["items"]:
assert set(item) <= {"ordinal", "status", "error_type", "error_code", "duration_ms"}
class _FakeResp:
def __init__(self, status_code):
self.status_code = status_code
class _FakeSession:
def __init__(self, status_code=200, raise_exc=None):
self.status_code = status_code
self.raise_exc = raise_exc
self.calls = []
def post(self, url, json=None, timeout=None):
self.calls.append((url, json, timeout))
if self.raise_exc:
raise self.raise_exc
return _FakeResp(self.status_code)
def test_client_report_ok_on_2xx():
sess = _FakeSession(status_code=200)
client = diagnostics.DiagnosticsClient("https://app.aivanov.eu/", session=sess)
assert client.report({"run_id": "r"}) is True
assert sess.calls[0][0] == "https://app.aivanov.eu/api/v1/diagnostics/report"
def test_client_report_false_on_network_error_without_raising():
sess = _FakeSession(raise_exc=RuntimeError("no network"))
client = diagnostics.DiagnosticsClient("https://app.aivanov.eu", session=sess)
assert client.report({"run_id": "r"}) is False # ne lève pas
def test_report_run_diagnostics_no_send_without_license(tmp_path):
sess = _FakeSession()
ok = diagnostics.report_run_diagnostics(
SimpleNamespace(documents=[]), base_url="https://app.aivanov.eu",
license_ref=None, machine_id="m" * 12, session=sess,
spool_path=tmp_path / "spool.jsonl",
)
assert ok is False and sess.calls == []
def test_report_run_diagnostics_network_down_spools(tmp_path):
sess = _FakeSession(raise_exc=RuntimeError("down"))
spool = tmp_path / "spool.jsonl"
summary = SimpleNamespace(documents=[_doc(ordinal=0, status="failed",
error_type="ValueError", error_code="processing_error")])
ok = diagnostics.report_run_diagnostics(
summary, base_url="https://app.aivanov.eu", license_ref="LIC-1",
machine_id="m" * 12, session=sess, spool_path=spool,
)
assert ok is False and spool.exists()
line = json.loads(spool.read_text(encoding="utf-8").splitlines()[0])
assert line["failed_count"] == 1
def test_flush_spool_sends_and_clears(tmp_path):
spool = tmp_path / "spool.jsonl"
diagnostics.spool_payload(spool, {"run_id": "r1"})
diagnostics.spool_payload(spool, {"run_id": "r2"})
sent = diagnostics.flush_spool(spool, diagnostics.DiagnosticsClient(
"https://app.aivanov.eu", session=_FakeSession(status_code=200)))
assert sent == 2 and not spool.exists()