praxis/tests/test_pending_keys.py

"""Auto-generated tests for Phase 1 exit criteria that require live API keys.

Per the VERIFY stage directive ("For unverifiable items: auto-generate test
scripts that WOULD verify them when keys are present"), these tests exercise the
two Phase 1 exit criteria that are pending voice-service key provisioning:

  - Exit criterion #1 (live audio session): a real WebRTC voice turn completes.
  - Exit criterion #2 (live latency measurement): R1-R4 probes produce real
    numbers and the TTS decision is finalized.

At v0.1 VERIFY time, only GITEA_TOKEN (operational) is guaranteed; the three
voice-service keys (DEEPGRAM_API_KEY, CARTESIA_API_KEY, OLLAMA_API_KEY) are NOT
provisioned in this environment. These tests are therefore SKIPPED when the keys
are absent, and will run automatically once the keys are provided via `.env` /
`.ciagent/.env.secrets` / the environment.

Run:
    pytest tests/test_pending_keys.py -rs          # shows skip reasons
    DEEPGRAM_API_KEY=... pytest tests/test_pending_keys.py   # runs the live tests

These tests are intentionally network-bound and are NOT part of the default
fast suite. They are gated behind the key presence checks so CI without keys
stays green.
"""

from __future__ import annotations

import asyncio
import os
import socket
import time

import pytest

# Keys that must be present for the live verifications.
REQUIRED_KEYS = ("DEEPGRAM_API_KEY", "CARTESIA_API_KEY", "OLLAMA_API_KEY")


def _have_live_keys() -> bool:
    """True iff all three voice-service keys are non-empty in the environment."""
    return all(os.environ.get(k, "").strip() for k in REQUIRED_KEYS)


pytestmark = pytest.mark.skipif(
    not _have_live_keys(),
    reason=(
        "Live voice-service keys (DEEPGRAM_API_KEY, CARTESIA_API_KEY, "
        "OLLAMA_API_KEY) are not provisioned in this environment. "
        "Set them in .env / .ciagent/.env.secrets and re-run to exercise "
        "Phase 1 exit criteria #1 (live audio session) and #2 (live latency)."
    ),
)


# ─── Exit criterion #2: live latency measurement (R1-R4) ────────────────────


def test_r1_deepgram_first_partial_latency():
    """R1: Deepgram Nova-3 first-partial-transcript latency is measured (not
    vendor-claimed) and recorded. Runs scripts/probe_deepgram.py end-to-end."""
    import subprocess
    import sys

    proc = subprocess.run(
        [sys.executable, "scripts/probe_deepgram.py", "--iterations", "5"],
        capture_output=True, text=True, timeout=120,
    )
    assert proc.returncode == 0, f"probe failed: {proc.stderr[:500]}"
    assert "KEY_MISSING" not in proc.stdout, "probe did not detect a key (unexpected)"


def test_r2_cartesia_first_audio_latency():
    """R2: Cartesia Sonic first-audio-byte latency is measured and recorded."""
    import subprocess
    import sys

    proc = subprocess.run(
        [sys.executable, "scripts/probe_cartesia.py", "--iterations", "5"],
        capture_output=True, text=True, timeout=120,
    )
    assert proc.returncode == 0, f"probe failed: {proc.stderr[:500]}"
    assert "KEY_MISSING" not in proc.stdout


def test_r3_ollama_ttft_both_models():
    """R3: Ollama Cloud TTFT for gemma4:cloud + deepseek-v4-flash:cloud no-think
    is measured. Also confirms R6 (Pipecat Ollama direct-API integration)."""
    import subprocess
    import sys

    proc = subprocess.run(
        [sys.executable, "scripts/probe_ollama.py", "--iterations", "5"],
        capture_output=True, text=True, timeout=120,
    )
    assert proc.returncode == 0, f"probe failed: {proc.stderr[:500]}"
    assert "KEY_MISSING" not in proc.stdout


def test_r4_integrated_e2e_latency_within_or_documented():
    """R4: the integrated three-hop e2e (transcript → Ollama → Cartesia) is
    measured against the 600ms budget. Per G-003, if OVER budget with Cartesia,
    the Piper leg must be measured and the TTS decision finalized. This test
    asserts the probe runs and produces a median; it does NOT hard-assert
    <600ms (the G-003 no-go actions handle an over-budget result)."""
    import subprocess
    import sys

    proc = subprocess.run(
        [sys.executable, "scripts/probe_e2e.py", "--iterations", "5"],
        capture_output=True, text=True, timeout=180,
    )
    assert proc.returncode == 0, f"probe failed: {proc.stderr[:500]}"
    assert "KEY_MISSING" not in proc.stdout
    # The probe prints a median line when it collects samples.
    assert "median" in proc.stdout.lower(), "no median reported (probe did not collect samples)"


# ─── Exit criterion #1: live audio session (R6 + full loop) ─────────────────


def test_ollama_gemma4_cloud_returns_first_token():
    """R6 / REQ-LLM-01: a real call to gemma4:cloud via Ollama Cloud direct API
    returns at least one token. Confirms the LLM adapter + bearer auth work
    against the live endpoint."""
    from server.llm.ollama_cloud import OllamaCloudLLM

    llm = OllamaCloudLLM()

    async def _run():
        out = []
        async for chunk in llm.chat(
            [
                {"role": "system", "content": "You are Jordan, a frustrated customer."},
                {"role": "user", "content": "Hi, I want a refund."},
            ],
            stream=True,
        ):
            out.append(chunk.content)
            if len(out) >= 1:
                break
        return out

    chunks = asyncio.run(_run())
    assert len(chunks) > 0, "gemma4:cloud returned no tokens (auth or endpoint issue)"


def test_ollama_deepseek_debrief_no_think_returns_text():
    """REQ-LLM-02: deepseek-v4-flash:cloud in no-think mode returns a debrief-
    style response. Confirms the debrief model + no-think flag work live."""
    from server.llm.ollama_cloud import OllamaCloudLLM

    llm = OllamaCloudLLM()

    async def _run():
        text, _usage = await llm.chat_full(
            [
                {"role": "system", "content": "Reply with one word."},
                {"role": "user", "content": "Say hello."},
            ],
            model=llm.debrief_model,
            no_think=True,
        )
        return text

    text = asyncio.run(_run())
    assert len(text.strip()) > 0, "deepseek-v4-flash:cloud no_think returned empty"


def test_cartesia_tts_streams_audio():
    """REQ-VOICE-02: Cartesia Sonic TTS streams real PCM audio for a sample
    line. Confirms the TTS adapter + WebSocket auth work live."""
    from server.tts.cartesia_tts import CartesiaTTS

    tts = CartesiaTTS()

    async def _run():
        chunks = [c async for c in tts.synthesize("Hi, I want my money back.")]
        return chunks

    chunks = asyncio.run(_run())
    assert len(chunks) > 0, "Cartesia returned no audio (auth or endpoint issue)"
    assert sum(len(c) for c in chunks) > 0


def test_deepgram_stt_service_constructs_with_live_key():
    """REQ-VOICE-01 / REQ-ORCH-01: the Deepgram Nova-3 STT service constructs
    with a live key (the pipeline wiring is verified separately; this confirms
    the key is accepted by the Deepgram client)."""
    from server.pipeline import _build_stt

    stt = _build_stt()
    assert stt is not None


def test_live_latency_report_has_real_numbers(tmp_path):
    """Exit criterion #2: after running the probes, docs/latency-report.md (or a
    generated report) contains real measured numbers, not vendor claims. This
    test re-runs the e2e probe and checks the structured output has a non-zero
    median."""
    import json
    import subprocess
    import sys

    out = tmp_path / "r4.json"
    proc = subprocess.run(
        [sys.executable, "scripts/probe_e2e.py", "--iterations", "3", "--out", str(out)],
        capture_output=True, text=True, timeout=180,
    )
    assert proc.returncode == 0, f"probe failed: {proc.stderr[:500]}"
    data = json.loads(out.read_text())
    e2e = data.get("e2e_cartesia", {})
    assert e2e.get("n", 0) > 0, "no e2e samples collected"
    assert e2e.get("median_ms", 0) > 0, "median latency is not a real positive number"