tests/_helpers.py

"""Shared fixtures for the voice product suite. Stdlib only.

Top:  unit + download tests run offline:  python3 -m unittest discover -s tests
Live: VOICE_TEST_LIVE=1 enables test_live/test_voices/test_e2e (real HF fetches).
E2E:  also needs a llama-cli binary (LLAMA_CLI or PATH); skips otherwise.
"""
import argparse
import importlib.util
import os
import shutil
import sys
import tempfile
import unittest
from pathlib import Path

os.environ.setdefault("VOICE_NO_VENV", "1")

REPO_ROOT = Path(__file__).resolve().parent.parent
VOICE_PY = REPO_ROOT / "voice.py"

spec = importlib.util.spec_from_file_location("voice", str(VOICE_PY))
voice = importlib.util.module_from_spec(spec)
sys.modules["voice"] = voice
spec.loader.exec_module(voice)

# Live fixtures: small, stable, public.
CATGIRL_REPO = "xcx0902/Qwen3-1.7B-catgirl"  # sharded safetensors finetune (the head)
ABL_REPO = "DevQuasar/huihui-ai.Qwen3-1.7B-abliterated-GGUF"
ABL_FILE = "huihui-ai.Qwen3-1.7B-abliterated.Q8_0.gguf"  # quant head for passthrough
BASE_REPO = "unsloth/Qwen3-1.7B-GGUF"
BASE_FILE = "Qwen3-1.7B-UD-IQ2_XXS.gguf"  # smallest full GGUF: cast target + runner

LIVE = os.environ.get("VOICE_TEST_LIVE") == "1"


def find_llama_cli():
    # LLAMA_CLI env wins, then PATH. No machine paths: shippable anywhere.
    import shutil
    env = os.environ.get("LLAMA_CLI")
    if env and Path(env).is_file():
        return env
    return shutil.which("llama-cli")


LLAMA_CLI = find_llama_cli()
needs_llama = unittest.skipUnless(
    LIVE and LLAMA_CLI, "needs VOICE_TEST_LIVE=1 and a llama-cli binary")


def cache_dir():
    d = Path(os.environ.get("VOICE_TEST_CACHE") or (Path.home() / ".cache" / "voice-tests"))
    d.mkdir(parents=True, exist_ok=True)
    return d


def fetch_file(url, dest, label="", n_parts=8):
    """Full-file fetch through the product's own download_range (resumeable)."""
    dest = Path(dest)
    if dest.is_file() and dest.stat().st_size > 0:
        return dest
    import time
    import urllib.request
    last = None
    for attempt in range(6):
        req = urllib.request.Request(url, method="HEAD")
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                size = int(r.headers["Content-Length"])
                etag = r.headers.get("ETag")
            break
        except urllib.error.HTTPError as e:
            last = e
            if e.code in (429, 500, 502, 503) and attempt < 5:
                time.sleep(voice._bounded_retry_after(
                    getattr(e, "headers", None) or {}, 2.0 * (2 ** attempt), cap=60))
                continue
            raise
    else:
        raise last
    part = dest.parent / (dest.name + ".parts")
    combined = voice.download_range(url, 0, size - 1, part, etag, label=label or dest.name,
                                    n_parts=n_parts)
    import shutil
    shutil.move(str(combined), str(dest))
    return dest


class IsolatedVoices(unittest.TestCase):
    """Each test gets a fresh VOICES_DIR; restored afterwards."""

    def setUp(self):
        self._old = voice.VOICES_DIR
        self.vdir = Path(tempfile.mkdtemp(prefix="voices-"))
        voice.VOICES_DIR = self.vdir
        voice_ctx = sys.modules.get("vlib.ctx")
        if voice_ctx is not None:
            voice_ctx.VOICES_DIR = self.vdir
        # registry + reports + parts all derive from VOICES_DIR at call time,
        # except these module constants: repoint them too.
        self._old_reports = (voice.REPORTS_JSONL, voice.REPORTS_OUTBOX, voice.OPERATIONS_LOG,
                             voice.HF_TOKEN_PATH)
        voice.REPORTS_JSONL = self.vdir / "reports.jsonl"
        voice.REPORTS_OUTBOX = self.vdir / "outbox"
        voice.OPERATIONS_LOG = self.vdir / "operations.log"
        voice.HF_TOKEN_PATH = self.vdir / ".hf_token"
        if voice_ctx is not None:
            voice_ctx.REPORTS_JSONL = voice.REPORTS_JSONL
            voice_ctx.REPORTS_OUTBOX = voice.REPORTS_OUTBOX
            voice_ctx.OPERATIONS_LOG = voice.OPERATIONS_LOG
            voice_ctx.HF_TOKEN_PATH = voice.HF_TOKEN_PATH

    def tearDown(self):
        voice.VOICES_DIR = self._old
        (voice.REPORTS_JSONL, voice.REPORTS_OUTBOX, voice.OPERATIONS_LOG,
         voice.HF_TOKEN_PATH) = self._old_reports
        voice_ctx = sys.modules.get("vlib.ctx")
        if voice_ctx is not None:
            voice_ctx.VOICES_DIR = self._old
            (voice_ctx.REPORTS_JSONL, voice_ctx.REPORTS_OUTBOX, voice_ctx.OPERATIONS_LOG,
             voice_ctx.HF_TOKEN_PATH) = self._old_reports
        shutil.rmtree(self.vdir, ignore_errors=True)


def ns(**kw):
    a = dict(non_interactive=True, yes=True, verbose=False)
    a.update(kw)
    return argparse.Namespace(**a)


def run_guarded(argv, timeout, **kw):
    """subprocess.run with a kill guarantee: no orphan survives a timeout,
    a KeyboardInterrupt, or a runner kill. Returns CompletedProcess; raises
    subprocess.TimeoutExpired (after reaping) or KeyboardInterrupt (clean)."""
    import signal
    import subprocess
    kw.setdefault("stdin", subprocess.DEVNULL)
    proc = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                            text=True, **{k: v for k, v in kw.items()
                                           if k in ("stdin", "cwd", "env")})
    try:
        out, err = proc.communicate(timeout=timeout)
        return subprocess.CompletedProcess(argv, proc.returncode, out, err)
    except subprocess.TimeoutExpired:
        proc.kill()
        try:
            out, err = proc.communicate(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()
            out, err = proc.communicate()
        raise subprocess.TimeoutExpired(argv, timeout, out, err)
    except BaseException:
        # KeyboardInterrupt / runner death: take the child with us, then propagate
        try:
            proc.kill()
        except Exception:
            pass
        try:
            proc.wait(timeout=10)
        except Exception:
            pass
        raise