nihalashetty
ae67bff5a3
Self-hosted platform for building, testing, and shipping LangChain/LangGraph agents. Deep-agent sub-agents on the canvas, a live tracing/observability timeline, auto-provisioned built-in tools with import/export, per-environment tool variables, streamed evaluations, and per-user auth token forwarding.
208 lines
10 KiB
Python
208 lines
10 KiB
Python
"""Characterization tests for the stats rollups (dashboard + project_stats).
|
|
|
|
These pin the EXACT output of the two endpoints for a controlled dataset so the
|
|
in-memory -> SQL-aggregate refactor can be proven behaviour-preserving. Every expected
|
|
number here is hand-computed from the seeded traces below.
|
|
|
|
Dataset (tenant t_stats), 6 traces across 2 projects:
|
|
A P1 W1 run done tok 10 cost 0.10 lat 100 -1h (in window)
|
|
B P1 W1 run error tok 20 cost 0.20 lat 300 -2h (in window)
|
|
C P1 -- assistant done tok 5 cost 0.05 lat 50 -3h (in window)
|
|
D P2 -- adhoc done tok 8 cost 0.08 lat 80 -4h (in window)
|
|
E P2 wf_gone run done tok 2 cost 0.02 lat 20 -5h (in window)
|
|
F P1 W1 run done tok100 cost 1.00 lat1000 -10d (OUT of window)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
from forge.db.base import SessionLocal
|
|
from forge.models import Project, Span, Tool, Trace, Workflow
|
|
from forge.routers.stats import dashboard, project_analytics, project_stats
|
|
|
|
TENANT = "t_stats"
|
|
|
|
|
|
async def _seed() -> tuple[str, str]:
|
|
now = datetime.utcnow()
|
|
|
|
def h(n): # n hours ago
|
|
return now - timedelta(hours=n)
|
|
|
|
async with SessionLocal() as s:
|
|
p1 = Project(tenant_id=TENANT, name="Alpha", slug="alpha")
|
|
p2 = Project(tenant_id=TENANT, name="Beta", slug="beta")
|
|
s.add_all([p1, p2])
|
|
await s.flush()
|
|
w1 = Workflow(tenant_id=TENANT, project_id=p1.id, name="Support")
|
|
s.add(w1)
|
|
s.add(Tool(tenant_id=TENANT, project_id=p1.id, name="t", kind="builtin", config={}))
|
|
await s.flush()
|
|
rows = [
|
|
Trace(tenant_id=TENANT, project_id=p1.id, workflow_id=w1.id, run_id="rA", name="run", status="done", total_tokens=10, total_cost_usd=0.10, latency_ms=100, started_at=h(1)),
|
|
Trace(tenant_id=TENANT, project_id=p1.id, workflow_id=w1.id, run_id="rB", name="run", status="error", total_tokens=20, total_cost_usd=0.20, latency_ms=300, started_at=h(2)),
|
|
Trace(tenant_id=TENANT, project_id=p1.id, workflow_id=None, run_id="rC", name="assistant", status="done", total_tokens=5, total_cost_usd=0.05, latency_ms=50, started_at=h(3)),
|
|
Trace(tenant_id=TENANT, project_id=p2.id, workflow_id=None, run_id="rD", name="adhoc", status="done", total_tokens=8, total_cost_usd=0.08, latency_ms=80, started_at=h(4)),
|
|
Trace(tenant_id=TENANT, project_id=p2.id, workflow_id="wf_gone", run_id="rE", name="run", status="done", total_tokens=2, total_cost_usd=0.02, latency_ms=20, started_at=h(5)),
|
|
Trace(tenant_id=TENANT, project_id=p1.id, workflow_id=w1.id, run_id="rF", name="run", status="done", total_tokens=100, total_cost_usd=1.00, latency_ms=1000, started_at=now - timedelta(days=10)),
|
|
]
|
|
s.add_all(rows)
|
|
await s.commit()
|
|
return p1.id, p2.id
|
|
|
|
|
|
async def test_dashboard_rollups():
|
|
p1, p2 = await _seed()
|
|
async with SessionLocal() as s:
|
|
d = await dashboard(session=s, tenant_id=TENANT)
|
|
|
|
assert d["total_runs"] == 6
|
|
assert d["runs_7d"] == 5
|
|
assert d["success_rate"] == 80.0 # 4 done of 5 in-window
|
|
assert d["avg_latency_ms"] == 110 # int((100+300+50+80+20)/5)
|
|
assert d["spend_7d"] == 0.45
|
|
|
|
# per-project counts for the dashboard cards
|
|
assert d["projects"][p1] == {"workflows": 1, "tools": 1, "runs_7d": 3}
|
|
assert d["projects"][p2] == {"workflows": 0, "tools": 0, "runs_7d": 2}
|
|
|
|
# recent = 8 most recent all-time, newest first (A,B,C,D,E,F)
|
|
recent = [(r["workflow"], r["project"], r["status"], r["tokens"]) for r in d["recent"]]
|
|
assert recent == [
|
|
("Support", "Alpha", "done", 10),
|
|
("Support", "Alpha", "error", 20),
|
|
("run", "Alpha", "done", 5),
|
|
("run", "Beta", "done", 8),
|
|
("run", "Beta", "done", 2),
|
|
("Support", "Alpha", "done", 100),
|
|
]
|
|
|
|
# totals (all-time)
|
|
assert d["totals"] == {
|
|
"runs": 6, "tokens": 145, "cost_usd": 1.45, "avg_latency_ms": 258,
|
|
"errors": 1, "error_rate": 16.7,
|
|
}
|
|
|
|
# reports (per project, all-time), cost desc
|
|
reps = {r["project_id"]: r for r in d["reports"]}
|
|
assert [r["project_id"] for r in d["reports"]] == [p1, p2]
|
|
assert reps[p1]["project"] == "Alpha" and reps[p1]["runs"] == 4 and reps[p1]["tokens"] == 135
|
|
assert reps[p1]["cost_usd"] == 1.35 and reps[p1]["avg_latency_ms"] == 362
|
|
assert reps[p1]["errors"] == 1 and reps[p1]["error_rate"] == 25.0
|
|
assert reps[p1]["assistant_cost_usd"] == 0.05 and reps[p1]["assistant_turns"] == 1
|
|
assert reps[p2]["runs"] == 2 and reps[p2]["cost_usd"] == 0.1 and reps[p2]["assistant_turns"] == 0
|
|
|
|
|
|
async def test_project_stats_rollups():
|
|
p1, p2 = await _seed()
|
|
async with SessionLocal() as s:
|
|
d1 = await project_stats(project_id=p1, session=s, tenant_id=TENANT)
|
|
d2 = await project_stats(project_id=p2, session=s, tenant_id=TENANT)
|
|
|
|
assert d1["totals"] == {"runs": 4, "tokens": 135, "cost_usd": 1.35, "avg_latency_ms": 362, "errors": 1, "error_rate": 25.0}
|
|
assert d1["last_7d"] == {"runs": 3, "tokens": 35, "cost_usd": 0.35, "avg_latency_ms": 150, "errors": 1, "error_rate": 33.3}
|
|
assert d1["assistant"] == {"runs": 1, "tokens": 5, "cost_usd": 0.05, "avg_latency_ms": 50, "errors": 0, "error_rate": 0.0, "turns": 1}
|
|
|
|
# report rows for P1: workflow "Support" (A,B,F) then assistant (C), cost desc
|
|
r1 = d1["reports"]
|
|
assert [(r["kind"], r["label"], r["runs"], r["cost_usd"]) for r in r1] == [
|
|
("workflow", "Support", 3, 1.3),
|
|
("assistant", "Forge Assistant", 1, 0.05),
|
|
]
|
|
|
|
# P2 exercises the 'other' (name) group and the deleted-workflow label
|
|
r2 = d2["reports"]
|
|
assert [(r["kind"], r["label"], r["runs"], r["cost_usd"]) for r in r2] == [
|
|
("other", "adhoc", 1, 0.08),
|
|
("workflow", "(deleted workflow)", 1, 0.02),
|
|
]
|
|
|
|
|
|
# --- analytics endpoint (time-series + breakdowns) ----------------------------------------
|
|
# Isolated in its own tenant so it can't collide with the rollup fixtures above regardless of
|
|
# test order (the suite shares one SQLite file; init_db only create_all's, it never truncates).
|
|
ATENANT = "t_analytics"
|
|
|
|
|
|
async def _seed_analytics() -> str:
|
|
now = datetime.utcnow()
|
|
|
|
def h(n):
|
|
return now - timedelta(hours=n)
|
|
|
|
async with SessionLocal() as s:
|
|
p = Project(tenant_id=ATENANT, name="Alpha", slug="alpha")
|
|
s.add(p)
|
|
await s.flush()
|
|
w = Workflow(tenant_id=ATENANT, project_id=p.id, name="Support")
|
|
s.add(w)
|
|
await s.flush()
|
|
t1 = Trace(tenant_id=ATENANT, project_id=p.id, workflow_id=w.id, run_id="rA1", name="run", status="done", source="playground", total_tokens=10, total_cost_usd=0.10, latency_ms=100, started_at=h(1))
|
|
t2 = Trace(tenant_id=ATENANT, project_id=p.id, workflow_id=w.id, run_id="rA2", name="run", status="error", source="playground", total_tokens=20, total_cost_usd=0.20, latency_ms=3000, started_at=h(2))
|
|
t3 = Trace(tenant_id=ATENANT, project_id=p.id, workflow_id=None, run_id="rA3", name="run", status="done", source="api", total_tokens=5, total_cost_usd=0.05, latency_ms=50, started_at=h(25))
|
|
t4 = Trace(tenant_id=ATENANT, project_id=p.id, workflow_id=None, run_id="rA4", name="assistant", status="done", source="assistant", total_tokens=8, total_cost_usd=0.08, latency_ms=5500, started_at=h(3))
|
|
# Out of the 30-day window but inside the previous (30-60d) window -> feeds prev_totals.
|
|
t5 = Trace(tenant_id=ATENANT, project_id=p.id, workflow_id=w.id, run_id="rA5", name="run", status="done", source="playground", total_tokens=100, total_cost_usd=1.00, latency_ms=200, started_at=now - timedelta(days=40))
|
|
s.add_all([t1, t2, t3, t4, t5])
|
|
await s.flush()
|
|
s.add_all([
|
|
Span(tenant_id=ATENANT, trace_id=t1.id, name="get_order", kind="tool", latency_ms=300, input_tokens=0, output_tokens=0, cost_usd=0.0),
|
|
Span(tenant_id=ATENANT, trace_id=t1.id, name="model", kind="llm", model="claude", latency_ms=800, input_tokens=4, output_tokens=6, cost_usd=0.09),
|
|
Span(tenant_id=ATENANT, trace_id=t2.id, name="get_order", kind="tool", latency_ms=200, input_tokens=0, output_tokens=0, cost_usd=0.0, error="boom"),
|
|
])
|
|
await s.commit()
|
|
return p.id
|
|
|
|
|
|
async def test_project_analytics():
|
|
pid = await _seed_analytics()
|
|
async with SessionLocal() as s:
|
|
a = await project_analytics(project_id=pid, days=30, session=s, tenant_id=ATENANT)
|
|
|
|
assert a["range"]["days"] == 30 and a["range"]["bucket"] == "day"
|
|
|
|
# Windowed totals (T1..T4); T5 is out of window.
|
|
assert a["totals"]["runs"] == 4
|
|
assert a["totals"]["tokens"] == 43
|
|
assert a["totals"]["cost_usd"] == 0.43
|
|
assert a["totals"]["errors"] == 1
|
|
assert a["totals"]["avg_latency_ms"] == 2162 # int((100+3000+50+5500)/4)
|
|
|
|
# Previous 30-60d window holds only T5.
|
|
assert a["prev_totals"]["runs"] == 1 and a["prev_totals"]["tokens"] == 100
|
|
|
|
# 30 days back -> 31 daily points (inclusive), continuous, summing to the 4 in-window runs.
|
|
assert len(a["timeseries"]) == 31
|
|
assert sum(p["runs"] for p in a["timeseries"]) == 4
|
|
assert sum(p["errors"] for p in a["timeseries"]) == 1
|
|
|
|
by_src = {r["source"]: r for r in a["by_source"]}
|
|
assert by_src["playground"]["runs"] == 2
|
|
assert by_src["api"]["runs"] == 1
|
|
assert by_src["assistant"]["runs"] == 1
|
|
|
|
by_wf = {(r["kind"], r["label"]): r for r in a["by_workflow"]}
|
|
assert by_wf[("workflow", "Support")]["runs"] == 2
|
|
assert by_wf[("assistant", "Forge Assistant")]["runs"] == 1
|
|
assert by_wf[("other", "run")]["runs"] == 1
|
|
|
|
tools = {t["name"]: t for t in a["tools"]}
|
|
assert tools["get_order"]["calls"] == 2
|
|
assert tools["get_order"]["errors"] == 1
|
|
assert tools["get_order"]["avg_latency_ms"] == 250 # (300+200)/2
|
|
|
|
models = {m["model"]: m for m in a["models"]}
|
|
assert models["claude"]["calls"] == 1
|
|
assert models["claude"]["tokens"] == 10
|
|
assert models["claude"]["cost_usd"] == 0.09
|
|
|
|
hist = {b["label"]: b["count"] for b in a["latency_histogram"]}
|
|
assert hist["<250ms"] == 2 # T1 (100), T3 (50)
|
|
assert hist["2-5s"] == 1 # T2 (3000)
|
|
assert hist["5-10s"] == 1 # T4 (5500)
|
|
|
|
# Recent activity feed: the 4 in-window runs, newest first.
|
|
assert len(a["recent"]) == 4
|
|
assert a["recent"][0]["status"] == "done" # T1, most recent
|