102 lines
3.0 KiB
Python
102 lines
3.0 KiB
Python
"""Tests du store chiffré. KDF en coût MIN (INTERACTIVE) pour la vitesse."""
|
|
|
|
import nacl.pwhash
|
|
|
|
from kazeia_central.core.store import BadPassword, Store, StoreLocked
|
|
|
|
_OPS = nacl.pwhash.argon2id.OPSLIMIT_MIN
|
|
_MEM = nacl.pwhash.argon2id.MEMLIMIT_MIN
|
|
|
|
NOW = 1_700_000_000_000
|
|
|
|
|
|
def _store(tmp_path):
|
|
return Store(tmp_path / "central.db", kdf_ops=_OPS, kdf_mem=_MEM)
|
|
|
|
|
|
def _turns():
|
|
return [
|
|
{"id": "t1", "profile_id": "p1", "session_id": "s1", "timestamp": NOW + 1,
|
|
"role": "PATIENT", "text": "j'ai mal dormi", "total_ms": 10},
|
|
{"id": "t2", "profile_id": "p1", "session_id": "s1", "timestamp": NOW + 2,
|
|
"role": "KAZEIA", "text": "tu te sens fatiguée ?", "ttft_ms": 5},
|
|
]
|
|
|
|
|
|
def test_setup_then_reopen_with_password(tmp_path):
|
|
s = _store(tmp_path)
|
|
assert not s.is_initialized and not s.is_unlocked
|
|
s.unlock("hunter2")
|
|
assert s.is_initialized and s.is_unlocked
|
|
s.set_device_label("ABC123", "Mme D.", now=NOW)
|
|
s.close()
|
|
|
|
# Ré-ouverture : bon mot de passe → OK et déchiffre
|
|
s2 = Store(tmp_path / "central.db") # défauts persistés (INTERACTIVE)
|
|
s2.unlock("hunter2")
|
|
assert s2.device_label("ABC123") == "Mme D."
|
|
s2.close()
|
|
|
|
|
|
def test_wrong_password_rejected(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.unlock("correct horse")
|
|
s.close()
|
|
s2 = Store(tmp_path / "central.db")
|
|
try:
|
|
s2.unlock("wrong")
|
|
assert False, "aurait dû lever BadPassword"
|
|
except BadPassword:
|
|
pass
|
|
finally:
|
|
s2.close()
|
|
|
|
|
|
def test_locked_guard(tmp_path):
|
|
s = _store(tmp_path)
|
|
try:
|
|
s.devices()
|
|
assert False, "aurait dû lever StoreLocked"
|
|
except StoreLocked:
|
|
pass
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_archive_decrypt_and_idempotent(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.unlock("pw")
|
|
n = s.archive_turns("ABC123", _turns(), now=NOW)
|
|
assert n == 2
|
|
# re-pull du même historique → 0 nouveau (dédoublonnage par clé)
|
|
assert s.archive_turns("ABC123", _turns(), now=NOW) == 0
|
|
rows = s.session_turns("ABC123", "s1")
|
|
assert [r["text"] for r in rows] == ["j'ai mal dormi", "tu te sens fatiguée ?"]
|
|
assert s.last_turn_ts("ABC123") == NOW + 2
|
|
assert s.last_turn_ts("ABC123", "s1") == NOW + 2
|
|
s.close()
|
|
|
|
|
|
def test_content_encrypted_at_rest(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.unlock("pw")
|
|
s.archive_turns("ABC123", _turns(), now=NOW)
|
|
s.set_device_label("ABC123", "Mme D.", now=NOW)
|
|
s.close()
|
|
# Lecture BRUTE du fichier : aucun texte PII ni label en clair
|
|
raw = (tmp_path / "central.db").read_bytes()
|
|
assert b"j'ai mal dormi" not in raw
|
|
assert b"Mme D." not in raw
|
|
# ... mais les métadonnées requêtables, oui (par construction)
|
|
assert b"s1" in raw
|
|
|
|
|
|
def test_audit_log(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.unlock("pw")
|
|
s.audit("export", actor="op1", target="ABC123/s1",
|
|
detail={"format": "ndjson", "turns": 2}, now=NOW)
|
|
log = s.audit_log()
|
|
assert log[0]["action"] == "export" and log[0]["actor"] == "op1"
|
|
s.close()
|