"""Tests du sous-système voix : dérivation chemins, archive store, orchestration. Le moteur lourd (CosyVoice/Whisper) 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_cosyvoice_dir_marker(): wp = "/data/local/tmp/kazeia/models/../voix/voix/amir.wav" assert transfer.cosyvoice_dir_for(wp) == "/data/local/tmp/kazeia/models/cosyvoice" def test_cosyvoice_dir_fallback_prod(): wp = "/sdcard/Android/data/com.kazeia/files/kazeia/voix/voix/amir.wav" assert transfer.cosyvoice_dir_for(wp) == \ "/sdcard/Android/data/com.kazeia/files/kazeia/models/cosyvoice" # ---- 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["cvps_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}.cvps" 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"GGUF-CVPS-" + text.encode()[:5]) return {"bytes": 242000, "feat_shape": [750, 80], "n_tokens": 375, "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["cvps_bytes"] == 242000 and r["wav_deleted"] is False assert adb.pushed and adb.pushed[0][1] == \ "/data/local/tmp/kazeia/models/cosyvoice/amir.cvps" 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): from kazeia_central.voice.autosync import VoiceAutoSync s = _store(tmp_path) a = VoiceAutoSync(_VoiceAdb([("amir", _WP, 1)]), s, _FakeBridge()) 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