102 lines
4.7 KiB
Python
102 lines
4.7 KiB
Python
"""Tests feature Conversations : archive chiffrée, sessions, purge, collect (RPC mocké)."""
|
|
|
|
import nacl.pwhash
|
|
|
|
from kazeia_central.core.adb import Adb
|
|
from kazeia_central.core.store import Store
|
|
from kazeia_central.features.medical.conversations import service as csvc
|
|
|
|
_OPS = nacl.pwhash.argon2id.OPSLIMIT_MIN
|
|
_MEM = nacl.pwhash.argon2id.MEMLIMIT_MIN
|
|
NOW = 1_700_000_000_000
|
|
|
|
|
|
def _store(tmp_path):
|
|
s = Store(tmp_path / "c.db", kdf_ops=_OPS, kdf_mem=_MEM)
|
|
s.unlock("pw")
|
|
return s
|
|
|
|
|
|
class _ConvAdb(Adb):
|
|
"""Faux adb : conversations_export (file-drop) + pull NDJSON + export_purge."""
|
|
def __init__(self, ndjson, count):
|
|
super().__init__()
|
|
self._nd = ndjson
|
|
self._count = count
|
|
self.purged = []
|
|
|
|
def content_call(self, method, *, arg_json=None, extras=None, serial=None):
|
|
if method == "conversations_export":
|
|
return {"ok": "true", "path": "/data/local/tmp/kazeia/exports/e.ndjson",
|
|
"count": str(self._count)}
|
|
if method == "export_purge":
|
|
self.purged.append(arg_json)
|
|
return {"ok": "true"}
|
|
return {"ok": "true"}
|
|
|
|
def pull(self, remote, local, *, serial=None):
|
|
with open(local, "w", encoding="utf-8") as f:
|
|
f.write(self._nd)
|
|
|
|
|
|
def test_collect_archives_richtext_and_purges(tmp_path):
|
|
s = _store(tmp_path)
|
|
nd = ('{"_meta":{"count":2}}\n'
|
|
'{"id":"t1","profile_id":"p1","session_id":"s1","timestamp":100,"role":"PATIENT","text":"bonjour: ca va ?"}\n'
|
|
'{"id":"t2","profile_id":"p1","session_id":"s1","timestamp":101,"role":"KAZEIA","text":"oui\\net toi ?"}\n')
|
|
adb = _ConvAdb(nd, 2)
|
|
r = csvc.collect(adb, s, "D1", now=NOW)
|
|
assert r["exported"] == 2 and r["archived"] == 2 and r["purged"] is True
|
|
assert adb.purged # export_purge appelé (plaintext MVP)
|
|
turns = s.session_turns("D1", "s1")
|
|
assert [t["text"] for t in turns] == ["bonjour: ca va ?", "oui\net toi ?"] # ':' et saut de ligne intacts
|
|
assert b"bonjour" not in (tmp_path / "c.db").read_bytes() # PII chiffrée au repos
|
|
|
|
|
|
def test_collect_dedupes_on_recollect(tmp_path):
|
|
s = _store(tmp_path)
|
|
nd = ('{"_meta":{}}\n{"id":"t1","profile_id":"p1","session_id":"s1","timestamp":1,"role":"PATIENT","text":"a"}\n')
|
|
csvc.collect(_ConvAdb(nd, 1), s, "D1", now=NOW)
|
|
r2 = csvc.collect(_ConvAdb(nd, 1), s, "D1", now=NOW) # re-pull → 0 nouveau
|
|
assert r2["exported"] == 1 and r2["archived"] == 0
|
|
|
|
|
|
def test_collect_empty_still_purges(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _ConvAdb("", 0)
|
|
r = csvc.collect(adb, s, "D1", now=NOW)
|
|
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", [
|
|
{"id": "t1", "profile_id": "p1", "session_id": "s1", "timestamp": 1, "role": "PATIENT", "text": "a"},
|
|
{"id": "t2", "profile_id": "p1", "session_id": "s1", "timestamp": 2, "role": "KAZEIA", "text": "b"}], now=NOW)
|
|
sess = csvc.sessions(s, "D1")
|
|
assert sess[0]["session_id"] == "s1" and sess[0]["turns"] == 2 and sess[0]["profile_id"] == "p1"
|
|
bundle = csvc.export_session(s, "D1", "s1")
|
|
assert bundle["turn_count"] == 2 and bundle["turns"][0]["text"] == "a"
|
|
assert csvc.purge(s, "D1", session_id="s1", now=NOW)["deleted"] == 2
|
|
assert csvc.sessions(s, "D1") == []
|