265 lines
11 KiB
Python
265 lines
11 KiB
Python
"""Tests du sous-système voix — modèle GLOBAL (catalogue flotte + déploiement).
|
|
|
|
Le moteur lourd (OmniVoice) n'est PAS testé ici (validé live) — faux bridge + faux adb
|
|
pour prouver la COMPOSITION (catalogue, déploiement scope-conscient, ingestion).
|
|
"""
|
|
|
|
import json as _json
|
|
import os as _os
|
|
import tempfile
|
|
|
|
import nacl.pwhash
|
|
|
|
from kazeia_central.core.adb import Adb
|
|
from kazeia_central.core.adb.client import AdbDevice
|
|
from kazeia_central.core.store import Store
|
|
from kazeia_central.features.technical.voices import admin_ingest
|
|
from kazeia_central.features.technical.voices import orchestrator as vo
|
|
from kazeia_central.features.technical.voices import transfer
|
|
|
|
_OPS = nacl.pwhash.argon2id.OPSLIMIT_MIN
|
|
_MEM = nacl.pwhash.argon2id.MEMLIMIT_MIN
|
|
NOW = 1_700_000_000_000
|
|
_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 _store(tmp_path):
|
|
s = Store(tmp_path / "c.db", kdf_ops=_OPS, kdf_mem=_MEM)
|
|
s.unlock("pw")
|
|
return s
|
|
|
|
|
|
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"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 _Adb(Adb):
|
|
"""Faux adb : manifestes admin, profils, dossier omnivoice (déployés), push/rm."""
|
|
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 "):
|
|
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 _Bridge:
|
|
def enroll(self, wav, text, out):
|
|
open(out, "wb").write(b"OVSP-bytes")
|
|
return {"bytes": 9, "t_ref": 62, "ref_text": "asr du segment"}
|
|
|
|
|
|
# ---- chemins -------------------------------------------------------------
|
|
def test_omnivoice_dir_marker():
|
|
assert transfer.omnivoice_dir_for("/data/local/tmp/kazeia/models/../voix/voix/a.wav") \
|
|
== "/data/local/tmp/kazeia/models/omnivoice/voices"
|
|
|
|
|
|
# ---- store : catalogue global + déploiements -----------------------------
|
|
def test_store_global_catalog_and_deployments(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("marie_1", b"WAV", origin_serial="DA", source_wav_path="/x", name="Marie", now=NOW)
|
|
s.store_voice_ovsp("marie_1", b"OVSP", now=NOW)
|
|
s.lock_voice("marie_1", "p1", now=NOW)
|
|
assert s.voice_wav_bytes("marie_1") == b"WAV" and s.voice_ovsp_bytes("marie_1") == b"OVSP"
|
|
s.add_deployment("marie_1", "DA", now=NOW)
|
|
s.add_deployment("marie_1", "DB", now=NOW)
|
|
rec = s.voice_record("marie_1")
|
|
assert rec["origin_serial"] == "DA" and rec["locked_profile_id"] == "p1"
|
|
assert rec["scope"] == "exclusive" and sorted(rec["deployed_serials"]) == ["DA", "DB"]
|
|
assert s.voices_on("DB") == {"marie_1"}
|
|
s.remove_deployment("marie_1", "DB")
|
|
assert s.deployments_of("marie_1") == ["DA"]
|
|
|
|
|
|
def test_store_ovsp_encrypted_at_rest(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("v", b"x", origin_serial="DA", source_wav_path=None, now=NOW)
|
|
s.set_voice_manifest("v", scope="exclusive", owner_name="P", consent_text="SECRET-consent", now=NOW)
|
|
raw = (tmp_path / "c.db").read_bytes()
|
|
assert b"SECRET-consent" not in raw # consentement chiffré
|
|
|
|
|
|
# ---- politique de déployabilité ------------------------------------------
|
|
def test_deployable_policy():
|
|
assert vo._deployable({"scope": "global"}, set()) == (True, None)
|
|
assert vo._deployable({"scope": "pending"}, {"p1"}) == (False, "pending")
|
|
assert vo._deployable({"scope": "exclusive", "locked_profile_id": "p1"}, {"p1"}) == (True, None)
|
|
ok, reason = vo._deployable({"scope": "exclusive", "locked_profile_id": "p1"}, {"p2"})
|
|
assert ok is False and reason == "verrou_profil_absent"
|
|
|
|
|
|
# ---- déploiement par tablette --------------------------------------------
|
|
def test_deploy_voice_pushes_and_records(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("paul", b"WAV", origin_serial="D1", source_wav_path=None, now=NOW)
|
|
s.store_voice_ovsp("paul", b"OVSP", now=NOW)
|
|
s.set_voice_manifest("paul", scope="global", owner_name=None, consent_text=None, now=NOW)
|
|
adb = _Adb()
|
|
r = vo.deploy_voice(adb, s, _Bridge(), "D1", "paul", now=NOW)
|
|
assert r["deployed"] and adb.pushed[0][1] == _os.path.join(_OVDIR, "paul.ovsp")
|
|
assert "D1" in s.deployments_of("paul")
|
|
|
|
|
|
def test_deploy_voice_enrolls_if_no_ovsp(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("paul", b"WAV", origin_serial="D1", source_wav_path=None, now=NOW)
|
|
s.set_voice_manifest("paul", scope="global", owner_name=None, consent_text=None, now=NOW)
|
|
adb = _Adb()
|
|
r = vo.deploy_voice(adb, s, _Bridge(), "D1", "paul", now=NOW) # pas d'ovsp → enrôle
|
|
assert r["deployed"] and s.voice_ovsp_bytes("paul") == b"OVSP-bytes"
|
|
|
|
|
|
def test_deploy_voice_refuses_exclusive_absent_owner(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("marie", b"WAV", origin_serial="D1", source_wav_path=None, now=NOW)
|
|
s.store_voice_ovsp("marie", b"OVSP", now=NOW)
|
|
s.lock_voice("marie", "p_absent", now=NOW)
|
|
adb = _Adb(profiles=["p1"])
|
|
r = vo.deploy_voice(adb, s, _Bridge(), "D1", "marie", now=NOW)
|
|
assert r["deployed"] is False and r["reason"] == "verrou_profil_absent"
|
|
assert adb.pushed == []
|
|
|
|
|
|
def test_reenroll_regenerates_and_redeploys(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("paul", b"WAV", origin_serial="D1", source_wav_path=None, now=NOW)
|
|
s.store_voice_ovsp("paul", b"OLD-pollu", now=NOW) # artefact pollué (ancien cap)
|
|
s.set_voice_manifest("paul", scope="global", owner_name=None, consent_text=None, now=NOW)
|
|
s.add_deployment("paul", "D1", now=NOW)
|
|
adb = _Adb()
|
|
r = vo.reenroll_voice(adb, s, _Bridge(), "paul", now=NOW)
|
|
assert r["reenrolled"] and r["redeployed"] == ["D1"]
|
|
assert s.voice_ovsp_bytes("paul") == b"OVSP-bytes" # régénéré (≠ ancien)
|
|
assert adb.pushed[-1][1] == _os.path.join(_OVDIR, "paul.ovsp")
|
|
|
|
|
|
def test_reenroll_requires_archived_wav(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.set_voice_manifest("ghost", scope="global", owner_name=None, consent_text=None, now=NOW)
|
|
try:
|
|
vo.reenroll_voice(_Adb(), s, _Bridge(), "ghost", now=NOW)
|
|
assert False, "aurait dû lever (WAV non archivé)"
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
def test_undeploy_voice(tmp_path):
|
|
s = _store(tmp_path)
|
|
s.archive_voice_wav("paul", b"x", origin_serial="D1", source_wav_path=None, now=NOW)
|
|
s.add_deployment("paul", "D1", now=NOW)
|
|
adb = _Adb()
|
|
vo.undeploy_voice(adb, s, "D1", "paul", now=NOW)
|
|
assert s.deployments_of("paul") == []
|
|
assert any("paul.ovsp" in r for r in adb.removed)
|
|
|
|
|
|
def test_list_for_tablet_split(tmp_path):
|
|
s = _store(tmp_path)
|
|
# paul déployé sur D1 ; marie exclusive p2 (absente) ; léa globale non déployée
|
|
for vid, scope in (("paul", "global"), ("lea", "global")):
|
|
s.archive_voice_wav(vid, b"x", origin_serial="D1", source_wav_path=None, now=NOW)
|
|
s.set_voice_manifest(vid, scope=scope, owner_name=None, consent_text=None, now=NOW)
|
|
s.archive_voice_wav("marie", b"x", origin_serial="D1", source_wav_path=None, now=NOW)
|
|
s.lock_voice("marie", "p2", now=NOW)
|
|
adb = _Adb(profiles=["p1"], deployed=["paul"])
|
|
view = vo.list_for_tablet(adb, s, "D1")
|
|
dep = {v["voice_id"] for v in view["deployed"]}
|
|
able = {v["voice_id"]: v for v in view["deployable"]}
|
|
assert "paul" in dep
|
|
assert able["lea"]["deployable"] is True
|
|
assert able["marie"]["deployable"] is False and able["marie"]["reason"] == "verrou_profil_absent"
|
|
|
|
|
|
# ---- ingestion admin → catalogue global ----------------------------------
|
|
def test_ingest_global_archives_and_deploys(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _Adb([_manifest("paul", "global")])
|
|
man = admin_ingest.list_manifests(adb, "D1")[0]
|
|
r = vo.ingest_voice(adb, s, _Bridge(), "D1", man, delete_source=False, now=NOW)
|
|
assert r["deployed"] and r["scope"] == "global"
|
|
rec = s.voice_record("paul")
|
|
assert rec["origin_serial"] == "D1" and "D1" in rec["deployed_serials"]
|
|
|
|
|
|
def test_ingest_pending_no_deploy(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _Adb([_manifest("anon", "pending")])
|
|
man = admin_ingest.list_manifests(adb, "D1")[0]
|
|
r = vo.ingest_voice(adb, s, _Bridge(), "D1", man, delete_source=False, now=NOW)
|
|
assert r["deployed"] is False and r["reason"] == "pending"
|
|
assert adb.pushed == [] and s.voice_record("anon")["archived_at"] == NOW
|
|
|
|
|
|
def test_ingest_exclusive_locks(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _Adb([_manifest("marie", "exclusive", owner="p1")], profiles=["p1"])
|
|
man = admin_ingest.list_manifests(adb, "D1")[0]
|
|
r = vo.ingest_voice(adb, s, _Bridge(), "D1", man, delete_source=False, now=NOW)
|
|
assert r["deployed"] and s.voice_record("marie")["locked_profile_id"] == "p1"
|
|
|
|
|
|
def test_sync_admin_delete_cleans(tmp_path):
|
|
s = _store(tmp_path)
|
|
adb = _Adb([_manifest("paul", "global")])
|
|
rep = vo.sync_admin(adb, s, _Bridge(), "D1", delete_source=True, now=NOW)
|
|
assert rep[0]["ok"] and rep[0]["deployed"] and rep[0]["wav_deleted"]
|
|
assert any(p.endswith("paul.wav") for p in adb.removed)
|
|
|
|
|
|
# ---- déclencheur auto -----------------------------------------------------
|
|
def test_autosync_blocked_when_locked(tmp_path):
|
|
from kazeia_central.features.technical.voices.autosync import VoiceAutoSync
|
|
locked = Store(tmp_path / "l.db", kdf_ops=_OPS, kdf_mem=_MEM) # pas unlock
|
|
a = VoiceAutoSync(_Adb([_manifest("paul", "global")]), locked, _Bridge())
|
|
assert a.sync_now("D1")["state"] == "store_verrouille"
|
|
|
|
|
|
def test_autosync_enabled_ingests(tmp_path):
|
|
from kazeia_central.features.technical.voices.autosync import VoiceAutoSync
|
|
s = _store(tmp_path)
|
|
a = VoiceAutoSync(_Adb([_manifest("paul", "global")]), s, _Bridge())
|
|
a.set_enabled(True, delete_source=False)
|
|
st = a.sync_now("D1")
|
|
assert st["state"] == "termine" and st["enrolled"] == 1
|