test_app.py

"""
Verifies the app ACTUALLY LAUNCHES under the pinned gradio, and that every
event handler is wired with a matching signature.

This is the check that was missing before: the previous app.py used
`stream_every`, which does not exist in gradio 4.44 — it would have crashed on
Space startup no matter how correct the rest of the code was.

Models are stubbed, so this tests the app wiring, not the ML.
"""
import os
import sys
import types
import numpy as np

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

# ── Stub the pipeline so no 3.5GB download happens ───────────────────────────
import pipeline as _pl


def _fake_transcribe(self, audio, sr=16000):
    return "duba asusuna sannan ka aika 35000 zuwa Amina"


def _fake_translate(self, text, src, tgt, **kw):
    table = {
        "duba asusuna sannan ka aika 35000 zuwa Amina":
            "check my balance and also send 35000 to amina",
        "2234567890": "2234567890",
        "eh": "yes",
    }
    return table.get(text, text)


def _fake_tts(self, text):
    return 16000, (np.sin(np.arange(8000) / 20) * 8000).astype(np.int16)


_pl.HausaVoiceAIPipeline.transcribe = _fake_transcribe
_pl.HausaVoiceAIPipeline.transcribe_partial = _fake_transcribe
_pl.HausaVoiceAIPipeline.translate = _fake_translate
_pl.HausaVoiceAIPipeline.synthesize = _fake_tts
_pl.HausaVoiceAIPipeline.hausa_text_to_audio = _fake_tts
_pl.HausaVoiceAIPipeline._load_asr = lambda self: None
_pl.HausaVoiceAIPipeline._load_asr_fast = lambda self: None
_pl.HausaVoiceAIPipeline._load_nllb = lambda self: None
_pl.HausaVoiceAIPipeline._load_tts = lambda self: None

os.environ.setdefault("SHOW_PARTIALS", "0")

PASS, FAIL = "\033[92mPASS\033[0m", "\033[91mFAIL\033[0m"
results = []


def check(name, cond, detail=""):
    results.append((name, cond))
    print(f"  [{PASS if cond else FAIL}] {name}" + (f"  — {detail}" if detail else ""))


print("\n══ App imports and builds under the pinned gradio ══")
import gradio as gr
print(f"  gradio {gr.__version__}")
import app as A
check("app module imported", True)
check("Blocks object built", isinstance(A.demo, gr.Blocks))

print("\n══ Streaming API exists in this gradio version ══")
import inspect
p = inspect.signature(gr.Audio(streaming=True).stream).parameters
check("Audio.stream accepts stream_every", "stream_every" in p)
check("gr.skip available", hasattr(gr, "skip"))

print("\n══ Event handlers are wired ══")
fns = A.demo.fns
check("events registered", len(fns) > 0, f"{len(fns)} handlers")

print("\n══ Text handler: full compound-request conversation ══")
st = None
audio, convo, status, st = A.on_text(
    "duba asusuna sannan ka aika 35000 zuwa Amina", st)
check("turn 1 returns audio", audio is not None and audio is not gr.skip())
check("turn 1 acknowledges BOTH tasks",
      "one at a time" in convo.lower() or "balance" in convo.lower(),
      status[:70])
audio, convo, status, st = A.on_text("2234567890", st)
check("turn 2 delivers balance + continues transfer",
      "balance is" in convo.lower() and "amina" in convo.lower())
audio, convo, status, st = A.on_text("eh", st)
check("turn 3 executes the transfer", "sent to amina" in convo.lower())
check("latency breakdown in status", "total" in status, status[-52:])

print("\n══ Empty input does not crash ══")
a, c, s, st2 = A.on_text("", None)
check("empty text handled", a == gr.skip() or a is None)
a, c, s, st2 = A.on_record(None, None)
check("no recording handled", a == gr.skip() or a is None)

print("\n══ Streaming handler with synthetic audio ══")
SR = 16000


def sil(ms, n=0.0005):
    k = int(SR * ms / 1000)
    return (np.random.randn(k) * n).astype(np.float32)


def sp(ms, amp=0.25):
    k = int(SR * ms / 1000)
    t = np.arange(k) / SR
    s = (np.sin(2*np.pi*120*t) + 0.5*np.sin(2*np.pi*700*t)
         + 0.3*np.sin(2*np.pi*1220*t))
    s = s * (0.6 + 0.4*np.sin(2*np.pi*4*t)) + np.random.randn(k)*0.01
    return (s / np.abs(s).max() * amp).astype(np.float32)


st = A.new_state()
stream = np.concatenate([sil(400), sp(1600), sil(1200)])
step = int(SR * 0.5)
got_reply = False
n_skips = 0
for i in range(0, len(stream), step):
    out = A.on_stream((SR, stream[i:i+step]), st)
    check_len = len(out) == 5
    audio_o, convo_o, status_o, partial_o, st = out
    if audio_o == gr.skip():
        n_skips += 1
    elif audio_o is not None:
        got_reply = True
check("stream handler returns 5 values", check_len)
check("VAD endpointed and produced a reply", got_reply)
check("idle ticks return gr.skip (no player restart)", n_skips > 0,
      f"{n_skips} skipped ticks")

print("\n══ Reset ══")
r = A.reset(st)
check("reset returns 5 values", len(r) == 5)
check("history cleared", "Press" in r[1] or "empty" in r[1])

print("\n══ Launch smoke test (real server, then shut down) ══")
try:
    A.demo.queue(max_size=4)
    _, url, _ = A.demo.launch(prevent_thread_lock=True, quiet=True,
                              server_port=7899, show_api=False)
    check("Gradio server started", True, url or "local")
    import urllib.request
    code = urllib.request.urlopen("http://127.0.0.1:7899/", timeout=15).getcode()
    check("HTTP 200 from the app root", code == 200, f"status {code}")
    A.demo.close()
except Exception as e:
    check("Gradio server started", False, str(e)[:110])

passed = sum(1 for _, ok in results if ok)
print(f"\n{'='*62}\n  {passed}/{len(results)} checks passed\n{'='*62}")
sys.exit(0 if passed == len(results) else 1)