Praxis CI
128 lines
4.5 KiB
Python
128 lines
4.5 KiB
Python
"""Tests for cost logging (TASK-04-04) + session recorder (TASK-04-03)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from server.cost import CostBreakdown, derive_cost, load_rates
|
|
from server.session_recorder import SessionRecorder
|
|
from db.store import PraxisStore, HARDCODED_LEARNER_ID
|
|
|
|
|
|
# ─── TASK-04-04: cost logging ────────────────────────────────────────────────
|
|
|
|
|
|
def test_derive_cost_basic():
|
|
"""derive_cost produces a non-null cents value from usage inputs."""
|
|
b = derive_cost(
|
|
llm_input_tokens=500,
|
|
llm_output_tokens=200,
|
|
deepgram_audio_minutes=2.0,
|
|
tts_characters=800,
|
|
debrief_input_tokens=300,
|
|
debrief_output_tokens=150,
|
|
tts_provider="cartesia",
|
|
)
|
|
assert b.derived_cents > 0
|
|
assert b.llm_input_tokens == 500
|
|
assert b.tts_characters == 800
|
|
|
|
|
|
def test_derive_cost_piper_zero_tts():
|
|
"""Piper self-hosted TTS is $0 marginal cost (R4 mitigation, post-pilot path)."""
|
|
b = derive_cost(tts_characters=10000, tts_provider="piper")
|
|
# Piper rate is 0.0 per 1k chars → TTS contributes 0.
|
|
assert b.derived_cents == 0
|
|
|
|
|
|
def test_derive_cost_breakdown_dict():
|
|
b = derive_cost(llm_input_tokens=1000, llm_output_tokens=500)
|
|
d = b.as_dict()
|
|
assert d["llm_input_tokens"] == 1000
|
|
assert d["derived_cents"] > 0
|
|
assert "rates" in d
|
|
|
|
|
|
def test_load_rates_from_yaml():
|
|
"""cost_rates.yaml is present and loadable."""
|
|
rates = load_rates()
|
|
assert "gemma4_cloud_per_1k_tokens_cents" in rates
|
|
assert rates["piper_per_1k_chars_cents"] == 0.0
|
|
|
|
|
|
def test_cost_no_enforced_ceiling():
|
|
"""D-012: v0.1 has no enforced cost ceiling (pilot). A high-cost session
|
|
is still logged, not rejected."""
|
|
b = derive_cost(
|
|
llm_input_tokens=1_000_000,
|
|
llm_output_tokens=500_000,
|
|
deepgram_audio_minutes=600.0,
|
|
tts_characters=2_000_000,
|
|
)
|
|
# No ceiling — just a (large) number.
|
|
assert b.derived_cents > 0
|
|
|
|
|
|
# ─── TASK-04-03: session recorder wiring ─────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_db(tmp_path: Path) -> Path:
|
|
return tmp_path / "test_recorder.db"
|
|
|
|
|
|
def test_session_recorder_full_lifecycle(tmp_db: Path):
|
|
"""TASK-04-03: start → log turns → set branch → end → DB has session + turns + progress."""
|
|
store = PraxisStore(tmp_db)
|
|
|
|
async def _run():
|
|
await store.init()
|
|
rec = SessionRecorder(store, scenario_id="cs_refund_ca_v01")
|
|
sid = await rec.start()
|
|
await rec.log_turn("assistant", tts_text="Hi, I want a refund.", latency_ms=None)
|
|
await rec.log_turn("user", asr_text="I'm sorry, I can offer a refund.", latency_ms=450.0)
|
|
await rec.log_turn("assistant", tts_text="Okay, what's the issue?", latency_ms=520.0)
|
|
rec.add_audio_minutes(1.5)
|
|
rec.add_debrief_tokens(input_tokens=200, output_tokens=100)
|
|
rec.set_branch_path(["accept_resolution"])
|
|
breakdown = await rec.end(outcome="success", debrief_text="You did well.")
|
|
sess = await store.get_session(sid)
|
|
turns = await store.get_turns(sid)
|
|
return sess, turns, breakdown
|
|
|
|
sess, turns, breakdown = asyncio.run(_run())
|
|
assert sess is not None
|
|
assert sess.outcome == "success"
|
|
assert sess.branch_path == ["accept_resolution"]
|
|
assert sess.cost_estimated_cents is not None and sess.cost_estimated_cents > 0
|
|
assert sess.debrief_text == "You did well."
|
|
assert len(turns) == 3
|
|
assert breakdown.derived_cents > 0
|
|
|
|
|
|
def test_session_recorder_progress_updated(tmp_db: Path):
|
|
"""TASK-04-03: end() updates the progress table (attempts + last_outcome)."""
|
|
store = PraxisStore(tmp_db)
|
|
|
|
async def _run():
|
|
await store.init()
|
|
rec = SessionRecorder(store, scenario_id="cs_refund_ca_v01")
|
|
await rec.start()
|
|
await rec.log_turn("user", asr_text="policy says no refunds")
|
|
rec.set_branch_path(["escalate"])
|
|
await rec.end(outcome="failure")
|
|
|
|
async with store._connect() as db:
|
|
cur = await db.execute(
|
|
"SELECT attempts, last_outcome FROM progress WHERE learner_id = ? AND scenario_id = ?",
|
|
(HARDCODED_LEARNER_ID, "cs_refund_ca_v01"),
|
|
)
|
|
return await cur.fetchone()
|
|
|
|
row = asyncio.run(_run())
|
|
assert row is not None
|
|
assert row[0] == 1
|
|
assert row[1] == "failure" |