From d450b747607b308bd26db6aedc0485dd78053d12 Mon Sep 17 00:00:00 2001 From: alf Date: Wed, 1 Jul 2026 19:26:40 +0200 Subject: [PATCH] =?UTF-8?q?test(conversations):=20verrouille=20le=20sch?= =?UTF-8?q?=C3=A9ma=20NDJSON=20exact=20de=20l'app=20(schema:1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Test dédié : _meta en 1ʳᵉ ligne skippée, tours à 10 clés (id int→str, role enum PATIENT/KAZEIA, ttft_ms/total_ms/voice_used/model_used en null explicite acceptés). Garantit qu'une dérive de schéma ne passe pas en silence. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_conversations.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/test_conversations.py b/tests/test_conversations.py index bb441d1..1df9bd2 100644 --- a/tests/test_conversations.py +++ b/tests/test_conversations.py @@ -68,6 +68,26 @@ def test_collect_empty_still_purges(tmp_path): assert r["exported"] == 0 and r["archived"] == 0 and r["purged"] is True +def test_collect_matches_app_ndjson_schema(tmp_path): + # Schéma EXACT figé par l'app (KazeiaTelemetryProvider.conversationsExport, schema:1) : + # _meta en 1ʳᵉ ligne + tours à 10 clés, id=int, role enum, 4 champs nullable en `null`. + s = _store(tmp_path) + nd = ('{"_meta":{"schema":1,"exported_at":1782920549886,"count":2,"filter":{}}}\n' + '{"id":123,"profile_id":"p_ec7aba19","session_id":"s_1","timestamp":1782920500000,' + '"role":"PATIENT","text":"j\'ai mal : dormi","ttft_ms":210,"total_ms":1840,' + '"voice_used":"richard","model_used":"qwen3.5-4b"}\n' + '{"id":124,"profile_id":"p_ec7aba19","session_id":"s_1","timestamp":1782920501000,' + '"role":"KAZEIA","text":"tu te sens fatiguée ?","ttft_ms":null,"total_ms":null,' + '"voice_used":null,"model_used":null}\n') + r = csvc.collect(_ConvAdb(nd, 2), s, "D1", now=NOW) + assert r["exported"] == 2 and r["archived"] == 2 and r["purged"] is True + turns = s.session_turns("D1", "s_1") + assert [t["turn_id"] for t in turns] == ["123", "124"] # id int → clé str, pas de collision + assert [t["role"] for t in turns] == ["PATIENT", "KAZEIA"] + assert turns[1]["ttft_ms"] is None and turns[1]["voice_used"] is None # nullables acceptés, pas de crash + assert turns[0]["voice_used"] == "richard" + + def test_sessions_and_purge(tmp_path): s = _store(tmp_path) s.archive_turns("D1", [