"""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