148 lines
5.8 KiB
Python
148 lines
5.8 KiB
Python
"""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=()):
|
|
super().__init__()
|
|
self._voices = voices # [(id, wav_path, wav_exists)]
|
|
self._deployed = list(deployed)
|
|
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]
|
|
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
|