348 lines
14 KiB
Python
348 lines
14 KiB
Python
"""Tests du sous-système voix : dérivation chemins, archive store, orchestration.
|
|
|
|
Le moteur lourd (OmniVoice) n'est PAS testé ici (validé live) — on injecte un faux
|
|
bridge et un faux adb pour prouver la COMPOSITION (workflow) sans le modèle.
|
|
"""
|
|
|
|
import nacl.pwhash
|
|
|
|
from kazeia_central.adb import Adb
|
|
from kazeia_central.adb.client import AdbDevice
|
|
from kazeia_central.store import Store
|
|
from kazeia_central.voice import orchestrator as vo
|
|
from kazeia_central.voice import transfer
|
|
|
|
_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
|
|
|
|
|
|
# ---- dérivation des chemins device ---------------------------------------
|
|
def test_omnivoice_dir_marker():
|
|
wp = "/data/local/tmp/kazeia/models/../voix/voix/amir.wav"
|
|
assert transfer.omnivoice_dir_for(wp) == "/data/local/tmp/kazeia/models/omnivoice/voices"
|
|
|
|
|
|
def test_omnivoice_dir_fallback_prod():
|
|
wp = "/sdcard/Android/data/com.kazeia/files/kazeia/voix/voix/amir.wav"
|
|
assert transfer.omnivoice_dir_for(wp) == \
|
|
"/sdcard/Android/data/com.kazeia/files/kazeia/models/omnivoice/voices"
|
|
|
|
|
|
# ---- archive voix dans le store ------------------------------------------
|
|
def test_voice_wav_archive_roundtrip_and_encrypted(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("D1", "amir", b"FAKEWAVDATA-amir", source_wav_path="/dev/amir.wav", now=NOW)
|
|
assert s.voice_wav_bytes("D1", "amir") == b"FAKEWAVDATA-amir"
|
|
# chiffré au repos : le blob brut ne contient pas le contenu
|
|
enc = (tmp_path / "voices" / "D1" / "amir.wav.enc").read_bytes()
|
|
assert b"FAKEWAVDATA" not in enc
|
|
rec = s.voice_record("D1", "amir")
|
|
assert rec["has_wav"] and rec["wav_sha256"] and rec["archived_at"] == NOW
|
|
|
|
|
|
def test_voice_transcription_encrypted(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("D1", "amir", b"x", source_wav_path=None, now=NOW)
|
|
s.set_voice_transcription("D1", "amir", "texte parlé confidentiel", "fr", now=NOW)
|
|
assert s.voice_record("D1", "amir")["transcription"] == "texte parlé confidentiel"
|
|
raw = (tmp_path / "c.db").read_bytes()
|
|
assert b"texte parl" not in raw # PII jamais en clair en base
|
|
|
|
|
|
def test_voice_status_marks(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("D1", "amir", b"x", source_wav_path=None, now=NOW)
|
|
s.mark_voice_enrolled("D1", "amir", 242000, now=NOW)
|
|
s.mark_voice_deployed("D1", "amir", now=NOW)
|
|
r = s.voice_record("D1", "amir")
|
|
assert r["ovsp_bytes"] == 242000 and r["enrolled_at"] and r["deployed_at"]
|
|
|
|
|
|
# ---- orchestration (faux adb + faux bridge) ------------------------------
|
|
class _VoiceAdb(Adb):
|
|
def __init__(self, voices, deployed=(), profiles=()):
|
|
super().__init__()
|
|
self._voices = voices # [(id, wav_path, wav_exists)]
|
|
self._deployed = list(deployed)
|
|
self._profiles = list(profiles) # [(profile_id, voice_id)]
|
|
self.pushed = []
|
|
self.deleted = []
|
|
|
|
def ready_devices(self):
|
|
return [AdbDevice("D1", "device", {})]
|
|
|
|
def content_query(self, path, *, serial=None, where=None, expect_rows=False, _retry=True):
|
|
if path == "voices":
|
|
return [{"id": i, "wav_path": w, "wav_exists": 1} for i, w, _ in self._voices]
|
|
if path == "profiles":
|
|
return [{"id": pid, "voice_id": vid} for pid, vid in self._profiles]
|
|
return []
|
|
|
|
def shell(self, command, *, serial=None, timeout=None):
|
|
if command.startswith("ls "):
|
|
return "\n".join(f"{d}.ovsp" for d in self._deployed)
|
|
if "rm -f" in command:
|
|
self.deleted.append(command.split("rm -f ")[1].split(" &&")[0])
|
|
return "OK"
|
|
return ""
|
|
|
|
def pull(self, remote, local, *, serial=None):
|
|
open(local, "wb").write(b"WAV:" + remote.encode())
|
|
|
|
def push(self, local, remote, *, serial=None):
|
|
self.pushed.append((open(local, "rb").read(), remote))
|
|
|
|
|
|
class _FakeBridge:
|
|
def transcribe(self, wav, model=None):
|
|
return {"text": "bonjour ceci est ma voix", "language": "fr", "model": "small"}
|
|
|
|
def enroll(self, wav, text, out):
|
|
open(out, "wb").write(b"OVSP" + text.encode()[:5])
|
|
return {"bytes": 242000, "t_ref": 364, "ref_seconds": 14.6, "text_bytes": len(text)}
|
|
|
|
|
|
_WP = "/data/local/tmp/kazeia/models/../voix/voix/amir.wav"
|
|
|
|
|
|
def test_list_voices_status(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _VoiceAdb([("amir", _WP, 1), ("damien",
|
|
"/data/local/tmp/kazeia/models/../voix/voix/damien.wav", 1)],
|
|
deployed=["damien"])
|
|
out = {v["voice_id"]: v for v in vo.list_voices(adb, s, "D1")}
|
|
assert out["damien"]["status"] == "deployed"
|
|
assert out["amir"]["status"] == "pending"
|
|
|
|
|
|
def test_prepare_archives_and_transcribes(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _VoiceAdb([("amir", _WP, 1)])
|
|
res = vo.prepare(adb, s, _FakeBridge(), "D1", "amir", now=NOW)
|
|
assert res["transcription"].startswith("bonjour")
|
|
# WAV archivé (chiffré) + transcription posée
|
|
assert s.voice_wav_bytes("D1", "amir") == b"WAV:" + _WP.encode()
|
|
assert s.voice_record("D1", "amir")["transcription"].startswith("bonjour")
|
|
|
|
|
|
def test_enroll_deploy_pushes_and_optional_delete(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _VoiceAdb([("amir", _WP, 1)])
|
|
# sans suppression
|
|
r = vo.enroll_deploy(adb, s, _FakeBridge(), "D1", "amir", "ma voix",
|
|
delete_source=False, now=NOW)
|
|
assert r["ovsp_bytes"] == 242000 and r["wav_deleted"] is False
|
|
assert adb.pushed and adb.pushed[0][1] == \
|
|
"/data/local/tmp/kazeia/models/omnivoice/voices/amir.ovsp"
|
|
assert s.voice_record("D1", "amir")["deployed_at"] == NOW
|
|
assert adb.deleted == []
|
|
# avec suppression
|
|
r2 = vo.enroll_deploy(adb, s, _FakeBridge(), "D1", "amir", "ma voix",
|
|
delete_source=True, now=NOW)
|
|
assert r2["wav_deleted"] is True and adb.deleted == [_WP]
|
|
assert s.voice_record("D1", "amir")["wav_deleted_at"] == NOW
|
|
|
|
|
|
# ---- verrou voix ↔ profil (assignation exclusive + généralistes) ---------
|
|
def test_lock_unlock_voice(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("D1", "amir", b"x", source_wav_path=None, now=NOW)
|
|
s.lock_voice("D1", "amir", "patient_07", now=NOW)
|
|
assert s.voice_record("D1", "amir")["locked_profile_id"] == "patient_07"
|
|
s.unlock_voice("D1", "amir", now=NOW)
|
|
assert s.voice_record("D1", "amir")["locked_profile_id"] is None
|
|
|
|
|
|
def test_list_voices_shows_lock_and_users(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _VoiceAdb([("amir", _WP, 1)], profiles=[("p1", "amir"), ("p2", "amir")])
|
|
s.archive_voice_wav("D1", "amir", b"x", source_wav_path=_WP, now=NOW)
|
|
s.lock_voice("D1", "amir", "p1", now=NOW)
|
|
v = vo.list_voices(adb, s, "D1")[0]
|
|
assert v["locked_profile_id"] == "p1"
|
|
assert set(v["used_by_profiles"]) == {"p1", "p2"}
|
|
|
|
|
|
def test_sync_skips_voice_locked_to_absent_profile(tmp_path):
|
|
s = _store(tmp_path)
|
|
# voix verrouillée sur un profil ABSENT de la tablette → non déployée
|
|
adb = _VoiceAdb([("amir", _WP, 1)], profiles=[("p_present", "x")])
|
|
s.archive_voice_wav("D1", "amir", b"WAV", source_wav_path=_WP, now=NOW)
|
|
s.lock_voice("D1", "amir", "p_absent", now=NOW)
|
|
rep = vo.sync_device(adb, s, _FakeBridge(), "D1", delete_source=False, now=NOW)
|
|
assert rep[0]["ok"] is False and rep[0]["skipped"] == "verrou_profil_absent"
|
|
assert adb.pushed == [] # rien poussé
|
|
|
|
|
|
def test_sync_deploys_general_voice(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _VoiceAdb([("amir", _WP, 1)]) # généraliste (non verrouillée)
|
|
rep = vo.sync_device(adb, s, _FakeBridge(), "D1", delete_source=False, now=NOW)
|
|
assert rep[0]["ok"] is True and adb.pushed
|
|
assert s.voice_record("D1", "amir")["deployed_at"] == NOW
|
|
|
|
|
|
# ---- déclencheur auto -----------------------------------------------------
|
|
def test_autosync_blocked_when_store_locked(tmp_path):
|
|
from kazeia_central.store import Store
|
|
from kazeia_central.voice.autosync import VoiceAutoSync
|
|
locked_store = Store(tmp_path / "locked.db", kdf_ops=_OPS, kdf_mem=_MEM) # PAS unlock
|
|
a = VoiceAutoSync(_VoiceAdb([("amir", _WP, 1)]), locked_store, _FakeBridge())
|
|
st = a.sync_now("D1")
|
|
assert st["state"] == "store_verrouille"
|
|
|
|
|
|
def test_autosync_enabled_runs_sync(tmp_path):
|
|
# autosync ingère depuis le stockage admin (manifestes) → adb admin + bridge ASR.
|
|
from kazeia_central.voice.autosync import VoiceAutoSync
|
|
s = _store(tmp_path)
|
|
a = VoiceAutoSync(_AdminAdb([_manifest("paul", "global")]), s, _BridgeASR())
|
|
a.set_enabled(True, delete_source=False)
|
|
assert a.status()["enabled"] is True
|
|
st = a.sync_now("D1")
|
|
assert st["state"] == "termine" and st["enrolled"] == 1
|
|
|
|
|
|
# ---- ingestion depuis le stockage ADMIN (manifestes record-time, §4.2) ----
|
|
import json as _json
|
|
import os as _os
|
|
|
|
from kazeia_central.voice import admin_ingest
|
|
|
|
_ADMIN = admin_ingest.ADMIN_VOIX_DIR
|
|
_PWP = "/data/local/tmp/kazeia/models/../voix/voix/x.wav"
|
|
_OVDIR = "/data/local/tmp/kazeia/models/omnivoice/voices"
|
|
|
|
|
|
def _manifest(vid, scope, owner=None):
|
|
m = {"id": vid, "name": vid.title(), "state": "recorded", "wav": f"{vid}.wav",
|
|
"duration_seconds": 16.0, "created_at": 1,
|
|
"reference_text": f"Phrase de consentement de {vid} — SECRET-{vid}",
|
|
"source_device": "tablet-admin-recording", "scope": scope}
|
|
if owner:
|
|
m["owner_profile_id"] = owner
|
|
m["owner_name"] = "Patient " + owner
|
|
return m
|
|
|
|
|
|
class _AdminAdb(Adb):
|
|
def __init__(self, manifests, profiles=(), deployed=()):
|
|
super().__init__()
|
|
self._manifests = {m["id"]: m for m in manifests}
|
|
self._profiles = list(profiles)
|
|
self._deployed = list(deployed)
|
|
self.pushed = []
|
|
self.removed = []
|
|
|
|
def ready_devices(self):
|
|
return [AdbDevice("D1", "device", {})]
|
|
|
|
def content_query(self, path, *, serial=None, where=None, expect_rows=False, _retry=True):
|
|
if path == "voices":
|
|
return [{"id": "x", "wav_path": _PWP, "wav_exists": 0}]
|
|
if path == "profiles":
|
|
return [{"id": p, "voice_id": None} for p in self._profiles]
|
|
return []
|
|
|
|
def shell(self, command, *, serial=None, timeout=None):
|
|
if command.startswith(f"ls {_ADMIN}"):
|
|
return "\n".join(f"{i}.json" for i in self._manifests)
|
|
if command.startswith("ls "): # dossier omnivoice (déployés)
|
|
return "\n".join(f"{d}.ovsp" for d in self._deployed)
|
|
if "rm -f" in command:
|
|
self.removed.append(command.split("rm -f ")[1].split(" ")[0])
|
|
return ""
|
|
|
|
def pull(self, remote, local, *, serial=None):
|
|
if remote.endswith(".json"):
|
|
vid = _os.path.basename(remote)[:-5]
|
|
with open(local, "w", encoding="utf-8") as f:
|
|
_json.dump(self._manifests[vid], f)
|
|
else:
|
|
open(local, "wb").write(b"WAV:" + remote.encode())
|
|
|
|
def push(self, local, remote, *, serial=None):
|
|
self.pushed.append((open(local, "rb").read(), remote))
|
|
|
|
|
|
class _BridgeASR:
|
|
def enroll(self, wav, text, out):
|
|
open(out, "wb").write(b"OVSP-asr")
|
|
return {"bytes": 1234, "t_ref": 62, "ref_text": "transcription asr du segment"}
|
|
|
|
|
|
def test_admin_list_manifests_parse():
|
|
adb = _AdminAdb([_manifest("marie", "exclusive", owner="p1"), _manifest("paul", "global")])
|
|
out = {v["id"]: v for v in admin_ingest.list_manifests(adb, "D1")}
|
|
assert out["marie"]["scope"] == "exclusive" and out["marie"]["owner_profile_id"] == "p1"
|
|
assert out["marie"]["wav_device_path"] == _os.path.join(_ADMIN, "marie.wav")
|
|
assert out["paul"]["scope"] == "global" and out["paul"]["owner_profile_id"] is None
|
|
|
|
|
|
def test_set_voice_manifest_consent_encrypted(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.set_voice_manifest("D1", "marie", scope="exclusive", owner_name="Patient p1",
|
|
consent_text="Phrase SECRET-marie", now=NOW)
|
|
rec = s.voice_record("D1", "marie")
|
|
assert rec["scope"] == "exclusive" and rec["consent_text"] == "Phrase SECRET-marie"
|
|
raw = (tmp_path / "c.db").read_bytes()
|
|
assert b"SECRET-marie" not in raw # consentement (PII) chiffré au repos
|
|
|
|
|
|
def test_ingest_pending_archives_no_deploy(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _AdminAdb([_manifest("anon", "pending")])
|
|
man = admin_ingest.list_manifests(adb, "D1")[0]
|
|
r = vo.ingest_voice(adb, s, _BridgeASR(), "D1", man, delete_source=False, now=NOW)
|
|
assert r["deployed"] is False and r["reason"] == "pending"
|
|
assert adb.pushed == [] # rien déployé
|
|
assert s.voice_record("D1", "anon")["archived_at"] == NOW # mais WAV archivé
|
|
assert s.voice_record("D1", "anon")["scope"] == "pending"
|
|
|
|
|
|
def test_ingest_global_deploys(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _AdminAdb([_manifest("paul", "global")])
|
|
man = admin_ingest.list_manifests(adb, "D1")[0]
|
|
r = vo.ingest_voice(adb, s, _BridgeASR(), "D1", man, delete_source=False, now=NOW)
|
|
assert r["deployed"] is True and r["scope"] == "global"
|
|
assert adb.pushed and adb.pushed[0][1] == _os.path.join(_OVDIR, "paul.ovsp")
|
|
assert s.voice_record("D1", "paul")["locked_profile_id"] is None
|
|
|
|
|
|
def test_ingest_exclusive_locks_and_deploys(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _AdminAdb([_manifest("marie", "exclusive", owner="p1")], profiles=["p1", "p2"])
|
|
man = admin_ingest.list_manifests(adb, "D1")[0]
|
|
r = vo.ingest_voice(adb, s, _BridgeASR(), "D1", man, delete_source=False, now=NOW)
|
|
assert r["deployed"] is True
|
|
assert s.voice_record("D1", "marie")["locked_profile_id"] == "p1"
|
|
|
|
|
|
def test_ingest_exclusive_owner_absent_skips_deploy(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _AdminAdb([_manifest("marie", "exclusive", owner="p_absent")], profiles=["p1"])
|
|
man = admin_ingest.list_manifests(adb, "D1")[0]
|
|
r = vo.ingest_voice(adb, s, _BridgeASR(), "D1", man, delete_source=False, now=NOW)
|
|
assert r["deployed"] is False and r["reason"] == "verrou_profil_absent"
|
|
assert adb.pushed == []
|
|
# archivée + verrouillée quand même (juste pas déployée ici)
|
|
assert s.voice_record("D1", "marie")["locked_profile_id"] == "p_absent"
|
|
|
|
|
|
def test_sync_admin_delete_source_cleans_admin(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _AdminAdb([_manifest("paul", "global")])
|
|
rep = vo.sync_admin(adb, s, _BridgeASR(), "D1", delete_source=True, now=NOW)
|
|
assert rep[0]["ok"] and rep[0]["deployed"] and rep[0]["wav_deleted"]
|
|
# WAV + manifeste supprimés du stockage admin
|
|
assert any(p.endswith("paul.wav") for p in adb.removed)
|
|
assert any(p.endswith("paul.json") for p in adb.removed)
|