83 lines
3.5 KiB
Python
83 lines
3.5 KiB
Python
"""Tests feature Patients : fiche clinique chiffrée + service (profils via RPC mocké)."""
|
|
|
|
import base64
|
|
import json as _json
|
|
|
|
import nacl.pwhash
|
|
|
|
from kazeia_central.core.adb import Adb
|
|
from kazeia_central.core.store import Store
|
|
from kazeia_central.features.medical.patients import service as svc
|
|
|
|
_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 _PatAdb(Adb):
|
|
"""Faux adb : /profiles (query) + profile_dump/upsert (RPC base64)."""
|
|
def __init__(self, profiles):
|
|
super().__init__()
|
|
self._p = profiles
|
|
self.upserted = []
|
|
|
|
def content_query(self, path, *, serial=None, where=None, expect_rows=False, _retry=True):
|
|
if path == "profiles":
|
|
return [dict(p, has_pin=0, is_default=0, turn_count=0) for p in self._p]
|
|
return []
|
|
|
|
def content_call(self, method, *, arg_json=None, extras=None, serial=None):
|
|
if method == "profile_dump_json":
|
|
prof = next((p for p in self._p if p["id"] == arg_json["id"]), None)
|
|
if not prof:
|
|
return {"ok": "false", "error": "not_found"}
|
|
payload = {**prof, "language": prof.get("language", "fr")}
|
|
return {"ok": "true", "result_b64": base64.b64encode(_json.dumps(payload).encode()).decode()}
|
|
if method == "profile_upsert_json":
|
|
self.upserted.append(arg_json)
|
|
return {"ok": "true"}
|
|
return {"ok": "true"}
|
|
|
|
|
|
def test_fiche_roundtrip_encrypted(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.save_fiche("D1", "p1", {"civil": {"full_name": "Jean SECRET-NAME"}, "clinical": {"notes": "confidentiel"}}, "es", now=NOW)
|
|
r = s.fiche("D1", "p1")
|
|
assert r["fiche"]["civil"]["full_name"] == "Jean SECRET-NAME" and r["language"] == "es"
|
|
assert b"SECRET-NAME" not in (tmp_path / "c.db").read_bytes() # PII chiffrée au repos
|
|
assert s.fiches("D1")["p1"]["language"] == "es"
|
|
|
|
|
|
def test_list_patients_merges_fiche(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.save_fiche("D1", "p1", {"civil": {"full_name": "Mme D."}}, "fr", now=NOW)
|
|
adb = _PatAdb([{"id": "p1", "display_name": "Profil 1", "voice_id": "damien"},
|
|
{"id": "p2", "display_name": "Profil 2", "voice_id": None}])
|
|
out = {p["profile_id"]: p for p in svc.list_patients(adb, s, "D1")}
|
|
assert out["p1"]["has_fiche"] and out["p1"]["full_name"] == "Mme D." and out["p1"]["language"] == "fr"
|
|
assert out["p2"]["has_fiche"] is False and out["p2"]["language"] is None
|
|
|
|
|
|
def test_get_patient_combines_profile_and_fiche(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.save_fiche("D1", "p1", {"clinical": {"pathology": "anxiété"}}, "de", now=NOW)
|
|
adb = _PatAdb([{"id": "p1", "display_name": "P1", "voice_id": "damien", "language": "de"}])
|
|
d = svc.get_patient(adb, s, "D1", "p1")
|
|
assert d["profile"]["display_name"] == "P1" and d["fiche"]["clinical"]["pathology"] == "anxiété"
|
|
assert d["language"] == "de"
|
|
|
|
|
|
def test_set_language_pushes_and_mirrors(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _PatAdb([{"id": "p1", "display_name": "P1", "voice_id": "damien"}])
|
|
r = svc.set_language(adb, s, "D1", "p1", "es", now=NOW)
|
|
assert r["pushed"] and r["language"] == "es"
|
|
assert adb.upserted and adb.upserted[0]["id"] == "p1" and adb.upserted[0]["language"] == "es" # poussé sur device
|
|
assert s.fiche("D1", "p1")["language"] == "es" # miroir central
|