File size: 6,036 Bytes
9f5fbb5 4a1430c 9f5fbb5 91d026f 9f5fbb5 91d026f 9f5fbb5 91d026f 4a1430c 9f5fbb5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | """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
|