diff --git a/backend/src/control_center/analytics/aggregator.py b/backend/src/control_center/analytics/aggregator.py index 1ed8970..17249e7 100644 --- a/backend/src/control_center/analytics/aggregator.py +++ b/backend/src/control_center/analytics/aggregator.py @@ -119,6 +119,18 @@ def apply_interaction_event(event: AnalyticsEvent) -> bool: pipe.expire(key, AGG_TTL_SECONDS) pipe.execute() + if event.org_id is not None: + # A small, low-cardinality index of "which services has this org + # actually seen traffic from" -- read_known_services below is + # what /analytics/services (a later PR) enumerates, since there + # is no other way to discover which analytics:agg:{org}:{service}: + # {date} keys exist without this index (Redis has no efficient + # "list keys matching a pattern" primitive worth using here). + # No TTL: a service name, once seen for an org, stays a legitimate + # thing to report zero-for on a quiet day, same reasoning + # AGG_TTL_SECONDS's own 90-day retention already applies to. + _redis.sadd(f"analytics:services:{event.org_id}", event.service) + if event.user_id is not None: active_keys = [f"analytics:active_users:{date_str}"] if event.org_id is not None: @@ -271,6 +283,34 @@ def read_daily_active_users(date_strs: list[str], org_id: Optional[int] = None) return [{"date": d, "count": read_active_user_count([d], org_id=org_id)} for d in date_strs] +def read_active_user_ids(date_strs: list[str], org_id: Optional[int] = None) -> set[str]: + """Raw user_id set (union across date_strs) -- INTERNAL ONLY. Never + returned directly by any API response (task brief: "do not expose + raw user IDs through normal analytics endpoints") -- the one + legitimate caller is a later PR's team-roster intersection, which + only ever surfaces a resulting *count*, never this set itself.""" + prefix = f"analytics:active_users:{org_id}:" if org_id is not None else "analytics:active_users:" + keys = [f"{prefix}{d}" for d in date_strs] + if not keys: + return set() + try: + if len(keys) == 1: + return set(_redis.smembers(keys[0])) + return set(_redis.sunion(*keys)) + except Exception: + return set() + + +def read_known_services(org_id: int) -> list[str]: + """The service names apply_interaction_event has ever recorded + traffic from for this org -- see that function's own comment on the + `analytics:services:{org_id}` index this reads.""" + try: + return sorted(_redis.smembers(f"analytics:services:{org_id}")) + except Exception: + return [] + + def read_user_activity(org_id: int, date_str: str) -> dict[str, int]: """user_id -> query_count for one org/day -- never returned directly by any API response (task brief: "do not expose raw user IDs"); only diff --git a/backend/src/control_center/analytics/cache.py b/backend/src/control_center/analytics/cache.py index ec10e5c..8b9dc9f 100644 --- a/backend/src/control_center/analytics/cache.py +++ b/backend/src/control_center/analytics/cache.py @@ -14,7 +14,7 @@ import json import os -from typing import Any, Callable +from typing import Any, Awaitable, Callable import redis @@ -65,6 +65,43 @@ def get_or_set( return value +async def get_or_set_async( + key: str, + endpoint: str, + compute_coro: Callable[[], Awaitable[Any]], + ttl: int = DEFAULT_TTL_SECONDS, +) -> Any: + """Async twin of `get_or_set` -- PR-C's service layer talks to other + upstreams (billing/TES/team-roster) via `httpx.AsyncClient` from + inside async FastAPI route handlers, so the compute side has to be + awaitable. Redis itself is still touched synchronously (redis-py's + sync client, same as the rest of this package) -- fine here since + those calls are fast, in-network, and already wrapped in their own + try/except; only `compute_coro` needs to be awaited. + """ + from control_center.analytics.metrics import CACHE_HITS, CACHE_MISSES + + try: + cached = _redis.get(key) + except Exception: + cached = None + + if cached is not None: + CACHE_HITS.labels(endpoint=endpoint).inc() + try: + return json.loads(cached) + except (TypeError, ValueError): + pass + + CACHE_MISSES.labels(endpoint=endpoint).inc() + value = await compute_coro() + try: + _redis.setex(key, ttl, json.dumps(value, default=str)) + except Exception: + pass + return value + + def invalidate(key: str) -> None: """Best-effort explicit invalidation. Not required for correctness (every cache entry has a TTL and naturally expires -- Section 9's own diff --git a/backend/src/control_center/analytics/router.py b/backend/src/control_center/analytics/router.py new file mode 100644 index 0000000..08bbb81 --- /dev/null +++ b/backend/src/control_center/analytics/router.py @@ -0,0 +1,189 @@ +"""The /analytics/* HTTP surface (task brief Section 7). Every route: +1. resolves+enforces scope via `Depends(require_analytics_scope)` (401/403 + handled entirely by that dependency -- no authorization decision is + made in this file); +2. resolves the date range (`service.resolve_date_range`, default last + 30 days); +3. reads-through a 5-minute Redis cache (`cache.get_or_set_async`) keyed + by endpoint+scope+range, never computing twice for the same request + shape within the TTL; +4. delegates all computation to service.py -- no route function contains + aggregation logic itself. + +No new authentication path: the same `require_analytics_scope` +dependency (built on `core.jwt_verify.verify_token`, the one JWT +verifier this whole service already uses) gates every route here. +""" +from __future__ import annotations + +import csv +import io +from datetime import date +from typing import Any, Callable, Coroutine, Optional + +from fastapi import APIRouter, Depends, Header, Query +from fastapi.responses import JSONResponse, StreamingResponse + +from control_center.analytics import cache, service +from control_center.analytics.metrics import API_ERRORS, API_REQUESTS +from control_center.analytics.permissions import AnalyticsScope, require_analytics_scope + +router = APIRouter(prefix="/analytics", tags=["analytics"]) + + +def _cache_key(endpoint: str, scope: AnalyticsScope, from_date: date, to_date: date) -> str: + return f"analytics:{endpoint}:{scope.org_id}:{scope.team_id}:{from_date.isoformat()}:{to_date.isoformat()}" + + +async def _run(endpoint: str, cache_key: str, compute_coro: Callable[[], Coroutine[Any, Any, dict]]) -> JSONResponse: + API_REQUESTS.labels(endpoint=endpoint).inc() + try: + data = await cache.get_or_set_async(cache_key, endpoint, compute_coro, ttl=cache.DEFAULT_TTL_SECONDS) + except Exception: + API_ERRORS.labels(endpoint=endpoint, status_code="500").inc() + raise + return JSONResponse(data) + + +@router.get("/overview") +async def overview( + from_date: Optional[date] = Query(default=None), + to_date: Optional[date] = Query(default=None), + authorization: Optional[str] = Header(default=None), + scope: AnalyticsScope = Depends(require_analytics_scope), +) -> JSONResponse: + resolved_from, resolved_to = service.resolve_date_range(from_date, to_date) + key = _cache_key("overview", scope, resolved_from, resolved_to) + return await _run("overview", key, lambda: service.get_overview(scope, resolved_from, resolved_to, authorization)) + + +@router.get("/queries") +async def queries( + from_date: Optional[date] = Query(default=None), + to_date: Optional[date] = Query(default=None), + authorization: Optional[str] = Header(default=None), + scope: AnalyticsScope = Depends(require_analytics_scope), +) -> JSONResponse: + resolved_from, resolved_to = service.resolve_date_range(from_date, to_date) + key = _cache_key("queries", scope, resolved_from, resolved_to) + return await _run("queries", key, lambda: service.get_queries(scope, resolved_from, resolved_to, authorization)) + + +@router.get("/users") +async def users( + from_date: Optional[date] = Query(default=None), + to_date: Optional[date] = Query(default=None), + authorization: Optional[str] = Header(default=None), + scope: AnalyticsScope = Depends(require_analytics_scope), +) -> JSONResponse: + resolved_from, resolved_to = service.resolve_date_range(from_date, to_date) + key = _cache_key("users", scope, resolved_from, resolved_to) + return await _run("users", key, lambda: service.get_users(scope, resolved_from, resolved_to, authorization)) + + +@router.get("/services") +async def services( + from_date: Optional[date] = Query(default=None), + to_date: Optional[date] = Query(default=None), + scope: AnalyticsScope = Depends(require_analytics_scope), +) -> JSONResponse: + resolved_from, resolved_to = service.resolve_date_range(from_date, to_date) + key = _cache_key("services", scope, resolved_from, resolved_to) + return await _run("services", key, lambda: service.get_services(scope, resolved_from, resolved_to)) + + +@router.get("/workflows") +async def workflows( + from_date: Optional[date] = Query(default=None), + to_date: Optional[date] = Query(default=None), + authorization: Optional[str] = Header(default=None), + scope: AnalyticsScope = Depends(require_analytics_scope), +) -> JSONResponse: + resolved_from, resolved_to = service.resolve_date_range(from_date, to_date) + key = _cache_key("workflows", scope, resolved_from, resolved_to) + return await _run("workflows", key, lambda: service.get_workflows(scope, resolved_from, resolved_to, authorization)) + + +@router.get("/performance") +async def performance( + from_date: Optional[date] = Query(default=None), + to_date: Optional[date] = Query(default=None), + # Still gated by require_analytics_scope (any admin role -- 403 for a + # regular user) even though the response itself is platform-wide and + # ignores scope.org_id/team_id entirely -- an authenticated admin of + # any organization may see platform-wide operational health, but a + # regular user still may not reach analytics at all. + _scope: AnalyticsScope = Depends(require_analytics_scope), +) -> JSONResponse: + resolved_from, resolved_to = service.resolve_date_range(from_date, to_date) + key = f"analytics:performance:platform:{resolved_from.isoformat()}:{resolved_to.isoformat()}" + return await _run("performance", key, lambda: service.get_performance(resolved_from, resolved_to)) + + +@router.get("/usage") +async def usage( + authorization: Optional[str] = Header(default=None), + scope: AnalyticsScope = Depends(require_analytics_scope), +) -> JSONResponse: + key = f"analytics:usage:{scope.org_id}:{scope.team_id}" + return await _run("usage", key, lambda: service.get_usage(scope, authorization)) + + +_EXPORTABLE: dict[str, Callable[[AnalyticsScope, date, date, Optional[str]], Coroutine[Any, Any, dict]]] = { + "overview": lambda scope, f, t, auth: service.get_overview(scope, f, t, auth), + "queries": lambda scope, f, t, auth: service.get_queries(scope, f, t, auth), + "users": lambda scope, f, t, auth: service.get_users(scope, f, t, auth), + "services": lambda scope, f, t, auth: service.get_services(scope, f, t), + "workflows": lambda scope, f, t, auth: service.get_workflows(scope, f, t, auth), + "performance": lambda scope, f, t, auth: service.get_performance(f, t), +} + + +def _rows_for_export(export_type: str, data: dict) -> tuple[list[str], list[list[Any]]]: + """Turns one service.py response dict into (header, rows) for CSV -- + the "daily trend" shape (queries/users) becomes one row per date, the + "services" shape becomes one row per service, everything else + (overview/workflows/performance) becomes one summary row.""" + if export_type in ("queries", "users") and isinstance(data.get("daily"), list): + header = ["date", "count"] + rows = [[row.get("date"), row.get("count")] for row in data["daily"]] + return header, rows + if export_type == "services": + header = ["service", "total_calls", "errors", "error_rate", "avg_latency_ms"] + rows = [ + [r.get("service"), r.get("total_calls"), r.get("errors"), r.get("error_rate"), r.get("avg_latency_ms")] + for r in data.get("services", []) + ] + return header, rows + scalar_items = [(k, v) for k, v in data.items() if not isinstance(v, (list, dict))] + return [k for k, _ in scalar_items], [[v for _, v in scalar_items]] + + +@router.get("/export") +async def export( + type: str = Query(...), + from_date: Optional[date] = Query(default=None), + to_date: Optional[date] = Query(default=None), + authorization: Optional[str] = Header(default=None), + scope: AnalyticsScope = Depends(require_analytics_scope), +) -> StreamingResponse: + builder = _EXPORTABLE.get(type) + if builder is None: + return JSONResponse({"error": f"unknown export type: {type!r}"}, status_code=400) + + resolved_from, resolved_to = service.resolve_date_range(from_date, to_date) + API_REQUESTS.labels(endpoint="export").inc() + data = await builder(scope, resolved_from, resolved_to, authorization) + + header, rows = _rows_for_export(type, data) + buffer = io.StringIO() + writer = csv.writer(buffer) + writer.writerow(header) + writer.writerows(rows) + buffer.seek(0) + + return StreamingResponse( + iter([buffer.getvalue()]), + media_type="text/csv", + headers={"Content-Disposition": f'attachment; filename="analytics_{type}.csv"'}, + ) diff --git a/backend/src/control_center/analytics/service.py b/backend/src/control_center/analytics/service.py new file mode 100644 index 0000000..f5d0897 --- /dev/null +++ b/backend/src/control_center/analytics/service.py @@ -0,0 +1,346 @@ +"""Orchestration layer for every /analytics/* endpoint (router.py, this +same PR). Assembles a response from aggregator.py's pre-aggregated Redis +reads plus prometheus.py/billing_client.py/tes_client.py -- never +computes anything by replaying a stream, never duplicates billing math. + +Team-level attribution (Section 8's `AnalyticsScope.team_id`) is +implemented via roster intersection against `analytics:user_activity`/ +`analytics:active_users` -- confirmed safe by this feature's own hard +verification gate before this PR started: omnibioai-auth's +`GET /orgs/{org_id}/teams/{team_id}/members` is unpaginated (a single +`.all()` query, see `team_service.list_team_members`) and returns the +complete roster in one call, so a single fetch is always complete. Any +call site below that needs it treats a *failed* roster fetch (auth-service +unreachable) as "team scope unavailable" and returns null fields with +`team_scope_available: false`, never silently falling back to the org- +level number mislabeled as team-scoped. + +`/analytics/performance` and the `latency_source="events"` fallback used +elsewhere are PLATFORM-WIDE ONLY -- `audit:events` carries no +organization_id (see aggregator.py's own module docstring) -- and are +always returned with `org_id`/`team_id` set to null, never echoing the +caller's own requested scope, per this feature's explicit requirement +that these numbers are never presented as tenant-scoped. + +`/analytics/workflows` (and the `workflows_run` field on +`/analytics/overview`) is null for a platform_admin -- confirmed via the +same hard verification gate: TES's `GET /api/runs` is scoped to the +*forwarded token's own* organization_id/team_id (Mode B Phase 1B's +`store.list(organization_id=identity.organization_id, team_id= +identity.team_id, filtered=True)`), not by a query parameter a caller can +aim at an arbitrary org. That's exactly right for org_admin/team_admin +(their own AnalyticsScope.org_id is, by permissions.py's own +construction, always their token's own org_id) but cannot honestly +answer "org X's workflows" for a platform_admin whose own token identity +may not even belong to org X -- returned as null rather than silently +describing the wrong organization. +""" +from __future__ import annotations + +import os +from datetime import date, datetime, timedelta +from typing import Any, Optional + +import httpx + +from control_center.analytics import aggregator, billing_client, cache, prometheus, tes_client +from control_center.analytics.permissions import AnalyticsScope + +IAM_URL = os.environ.get("IAM_URL", "http://auth-service:8001") +_TEAM_ROSTER_TIMEOUT_SECONDS = 5.0 +_TEAM_ROSTER_CACHE_TTL_SECONDS = 300 + +DEFAULT_RANGE_DAYS = 30 + + +def resolve_date_range(from_date: Optional[date], to_date: Optional[date]) -> tuple[date, date]: + """Defaults to the last 30 days (task brief Section 7) when either + bound is omitted. `to_date` defaults to today even if only `from_date` + was supplied, and vice versa -- there is no partial-range state.""" + resolved_to = to_date or date.today() + resolved_from = from_date or (resolved_to - timedelta(days=DEFAULT_RANGE_DAYS - 1)) + return resolved_from, resolved_to + + +async def _team_roster(org_id: int, team_id: int, authorization: Optional[str]) -> Optional[set[str]]: + """Fetches team-membership via the same IAM endpoint + routes_team_proxy.py already relays (GET /orgs/{org}/teams/{team}/members), + called directly here since service.py already talks to other + upstreams (billing/TES) the same way. Cached 5 minutes -- team + membership changes rarely enough that a short lag is acceptable, the + same TTL every other analytics cache entry uses. Returns None on any + failure (unreachable, non-200, malformed body) -- callers treat that + as "team scope unavailable", never as an empty roster.""" + cache_key = f"analytics:team_roster:{org_id}:{team_id}" + + async def _compute() -> Optional[list[str]]: + headers = {"Authorization": authorization} if authorization else {} + try: + async with httpx.AsyncClient() as client: + r = await client.get( + f"{IAM_URL}/orgs/{org_id}/teams/{team_id}/members", + headers=headers, timeout=_TEAM_ROSTER_TIMEOUT_SECONDS, + ) + if r.status_code != 200: + return None + body = r.json() + except (httpx.HTTPError, ValueError): + return None + if not isinstance(body, list): + return None + return [str(m["user_id"]) for m in body if isinstance(m, dict) and "user_id" in m] + + roster = await cache.get_or_set_async(cache_key, "team_roster", _compute, ttl=_TEAM_ROSTER_CACHE_TTL_SECONDS) + return set(roster) if roster is not None else None + + +async def _resolve_team_roster_if_needed( + scope: AnalyticsScope, authorization: Optional[str], +) -> tuple[Optional[set[str]], bool]: + """Returns (roster, applicable). applicable=False means this request + isn't team-scoped at all (org- or platform-level) -- roster is always + None in that case and callers must use the org/platform aggregate + directly, not treat a None roster as a failed lookup.""" + if scope.team_id is None or scope.org_id is None: + return None, False + roster = await _team_roster(scope.org_id, scope.team_id, authorization) + return roster, True + + +def _daily_query_counts(org_id: Optional[int], date_strs: list[str], roster: Optional[set[str]]) -> list[int]: + counts = [] + for date_str in date_strs: + if roster is not None: + activity = aggregator.read_user_activity(org_id, date_str) + counts.append(sum(v for uid, v in activity.items() if uid in roster)) + else: + counts.append(aggregator.read_agg(date_str, org_id=org_id)["query_count"]) + return counts + + +def _active_user_count(org_id: Optional[int], date_strs: list[str], roster: Optional[set[str]]) -> int: + if roster is not None: + return len(aggregator.read_active_user_ids(date_strs, org_id=org_id) & roster) + return aggregator.read_active_user_count(date_strs, org_id=org_id) + + +async def _workflow_run_count(scope: AnalyticsScope, from_date: date, to_date: date, authorization: Optional[str]) -> Optional[int]: + if scope.is_platform_admin: + return None + runs = await tes_client.get_runs(authorization) + if runs is None: + return None + from_epoch, to_epoch = _epoch_bounds(from_date, to_date) + return sum( + 1 for run in runs + if isinstance(run.get("created_epoch"), (int, float)) and from_epoch <= run["created_epoch"] <= to_epoch + ) + + +def _epoch_bounds(from_date: date, to_date: date) -> tuple[float, float]: + return ( + datetime.combine(from_date, datetime.min.time()).timestamp(), + datetime.combine(to_date, datetime.max.time()).timestamp(), + ) + + +async def get_overview(scope: AnalyticsScope, from_date: date, to_date: date, authorization: Optional[str]) -> dict[str, Any]: + date_strs = aggregator.date_range(from_date, to_date) + roster, team_applicable = await _resolve_team_roster_if_needed(scope, authorization) + + result: dict[str, Any] = { + "from_date": from_date.isoformat(), "to_date": to_date.isoformat(), + "org_id": scope.org_id, "team_id": scope.team_id, + } + + if team_applicable and roster is None: + result.update(total_queries=None, active_users=None, team_scope_available=False) + else: + result["total_queries"] = sum(_daily_query_counts(scope.org_id, date_strs, roster)) + result["active_users"] = _active_user_count(scope.org_id, date_strs, roster) + if team_applicable: + result["team_scope_available"] = True + + # error_rate always reflects the org/platform-level counters (no + # per-user error tracking exists to roster-intersect -- see this + # module's own docstring on team-level limitations). + org_agg = aggregator.read_agg_range(date_strs, org_id=scope.org_id) + result["error_rate"] = round(org_agg["query_error_count"] / org_agg["query_count"], 4) if org_agg["query_count"] else 0.0 + result["workflows_run"] = await _workflow_run_count(scope, from_date, to_date, authorization) + return result + + +async def get_queries(scope: AnalyticsScope, from_date: date, to_date: date, authorization: Optional[str]) -> dict[str, Any]: + date_strs = aggregator.date_range(from_date, to_date) + roster, team_applicable = await _resolve_team_roster_if_needed(scope, authorization) + + result: dict[str, Any] = { + "from_date": from_date.isoformat(), "to_date": to_date.isoformat(), + "org_id": scope.org_id, "team_id": scope.team_id, + } + if team_applicable and roster is None: + result.update(total_queries=None, daily=[{"date": d, "count": None} for d in date_strs], team_scope_available=False) + return result + + counts = _daily_query_counts(scope.org_id, date_strs, roster) + result["daily"] = [{"date": d, "count": c} for d, c in zip(date_strs, counts)] + result["total_queries"] = sum(counts) + if team_applicable: + result["team_scope_available"] = True + return result + + +async def get_users(scope: AnalyticsScope, from_date: date, to_date: date, authorization: Optional[str]) -> dict[str, Any]: + date_strs = aggregator.date_range(from_date, to_date) + roster, team_applicable = await _resolve_team_roster_if_needed(scope, authorization) + + result: dict[str, Any] = {"org_id": scope.org_id, "team_id": scope.team_id} + if team_applicable and roster is None: + result.update(daily=[{"date": d, "count": None} for d in date_strs], dau=None, wau=None, mau=None, team_scope_available=False) + return result + + daily = [{"date": d, "count": _active_user_count(scope.org_id, [d], roster)} for d in date_strs] + + wau_dates = aggregator.date_range(to_date - timedelta(days=6), to_date) + mau_dates = aggregator.date_range(to_date - timedelta(days=29), to_date) + + result["daily"] = daily + result["dau"] = _active_user_count(scope.org_id, [to_date.isoformat()], roster) + result["wau"] = _active_user_count(scope.org_id, wau_dates, roster) + result["mau"] = _active_user_count(scope.org_id, mau_dates, roster) + if team_applicable: + result["team_scope_available"] = True + return result + + +async def get_services(scope: AnalyticsScope, from_date: date, to_date: date) -> dict[str, Any]: + """Always org-level (never roster-intersected to a team) -- no + per-user-per-service granularity is tracked, see this module's own + docstring. A team_admin still sees their own org's full service + breakdown, same as an org_admin would.""" + result: dict[str, Any] = { + "from_date": from_date.isoformat(), "to_date": to_date.isoformat(), + "org_id": scope.org_id, "team_id": scope.team_id, + } + if scope.org_id is None: + result["services"] = [] + result["note"] = "service breakdown requires an organization scope" + return result + + date_strs = aggregator.date_range(from_date, to_date) + rows = [] + for svc in aggregator.read_known_services(scope.org_id): + totals = aggregator.read_agg_range(date_strs, org_id=scope.org_id, service=svc) + total_calls = int(totals["query_count"]) + errors = int(totals["query_error_count"]) + rows.append({ + "service": svc, + "total_calls": total_calls, + "errors": errors, + "error_rate": round(errors / total_calls, 4) if total_calls else 0.0, + # No per-service latency source exists today (interactions:events + # carries no duration_ms -- see schemas.py's own docstring); + # P95/P99 remain the primary performance indicator via + # /analytics/performance (platform-wide), not this field. + "avg_latency_ms": None, + }) + rows.sort(key=lambda r: -r["total_calls"]) + result["services"] = rows + return result + + +async def get_performance(from_date: date, to_date: date) -> dict[str, Any]: + """PLATFORM-WIDE ONLY. See this module's own docstring.""" + hours = aggregator.hours_for_range( + datetime.combine(from_date, datetime.min.time()), + datetime.combine(to_date, datetime.max.time()), + ) + date_strs = aggregator.date_range(from_date, to_date) + agg = aggregator.read_agg_range(date_strs) + + request_count = int(agg["request_count"]) + error_count = int(agg["request_error_count"]) + error_rate = round(error_count / request_count, 4) if request_count else 0.0 + days = max((to_date - from_date).days + 1, 1) + throughput_per_day = round(request_count / days, 2) + + prom = await prometheus.query_latency_quantiles(job="api-gateway") + if prom.get("available"): + latency_source = "prometheus" + p50 = prom["p50"] * 1000 if prom["p50"] is not None else None + p95 = prom["p95"] * 1000 if prom["p95"] is not None else None + p99 = prom["p99"] * 1000 if prom["p99"] is not None else None + else: + latency_source = "events" + estimated = aggregator.read_platform_latency_percentiles(hours) + p50, p95, p99 = estimated["p50"], estimated["p95"], estimated["p99"] + + return { + "scope": "platform", + "org_id": None, + "team_id": None, + "p50_latency_ms": p50, "p95_latency_ms": p95, "p99_latency_ms": p99, + "error_rate": error_rate, + "throughput_per_day": throughput_per_day, + "latency_source": latency_source, + "from_date": from_date.isoformat(), "to_date": to_date.isoformat(), + } + + +async def get_workflows(scope: AnalyticsScope, from_date: date, to_date: date, authorization: Optional[str]) -> dict[str, Any]: + result: dict[str, Any] = { + "from_date": from_date.isoformat(), "to_date": to_date.isoformat(), + "org_id": scope.org_id, "team_id": scope.team_id, + } + if scope.is_platform_admin: + result.update( + workflows_run=None, daily=None, success_rate=None, + note="TES's run-listing API is scoped to the caller's own identity, not by an arbitrary org_id -- unavailable for a platform_admin request", + ) + return result + + runs = await tes_client.get_runs(authorization) + if runs is None: + result.update(workflows_run=None, daily=None, success_rate=None) + return result + + from_epoch, to_epoch = _epoch_bounds(from_date, to_date) + by_day: dict[str, int] = {} + completed = failed = total = 0 + for run in runs: + created = run.get("created_epoch") + if not isinstance(created, (int, float)) or not (from_epoch <= created <= to_epoch): + continue + total += 1 + day = datetime.fromtimestamp(created).date().isoformat() + by_day[day] = by_day.get(day, 0) + 1 + state = run.get("state") + if state == "COMPLETED": + completed += 1 + elif state == "FAILED": + failed += 1 + + finished = completed + failed + result["workflows_run"] = total + result["daily"] = [{"date": d, "count": c} for d, c in sorted(by_day.items())] + result["success_rate"] = round(completed / finished, 4) if finished else None + return result + + +async def get_usage(scope: AnalyticsScope, authorization: Optional[str]) -> dict[str, Any]: + """Billing/entitlement consumption -- passes omnibioai-billing's own + numbers through unmodified (task brief: "do not modify billing + rating logic"). org_id=None (platform-wide/no explicit org) has no + billing counterpart -- billing is always per-organization.""" + if scope.org_id is None: + return {"org_id": None, "team_id": scope.team_id, "billing_available": False, "usage": None, "limits": None, "note": "billing usage requires an organization scope"} + + usage_available, usage = await billing_client.get_usage(scope.org_id, authorization) + limits_available, limits = await billing_client.get_usage_limits(scope.org_id, authorization) + + return { + "org_id": scope.org_id, "team_id": scope.team_id, + "billing_available": usage_available and limits_available, + "usage": usage, + "limits": limits, + } diff --git a/backend/src/control_center/main.py b/backend/src/control_center/main.py index 2080532..d3367f0 100644 --- a/backend/src/control_center/main.py +++ b/backend/src/control_center/main.py @@ -49,6 +49,7 @@ from control_center.api.routes_rag_proxy import router as rag_proxy_router from control_center.api.routes_platform_config_proxy import router as platform_config_proxy_router from control_center.api.routes_platform_interactions_proxy import router as platform_interactions_proxy_router +from control_center.analytics.router import router as analytics_router from control_center.api.routes_cloud import router as cloud_router from control_center.api.routes_integrations import router as integrations_router from control_center.core.auth import require_permission @@ -180,6 +181,14 @@ def _setup_logging() -> logging.Logger: # that data, or (Infrastructure/Operations only) via its own in-process # platform.manage_infra check. app.include_router(dashboard_router) +# Usage Analytics v1: every route here is individually gated by its own +# Depends(require_analytics_scope) (401/403 per-request, platform_admin/ +# org_admin/team_admin/deny) -- no blanket router-level permission +# dependency, same "each route owns its own authorization" posture +# dashboard_router immediately above already established, for the same +# reason: a single blanket permission here would either over- or +# under-restrict across the different roles analytics needs to serve. +app.include_router(analytics_router) # ============================================================================== diff --git a/backend/tests/_fake_redis.py b/backend/tests/_fake_redis.py index 54177e2..475d6f3 100644 --- a/backend/tests/_fake_redis.py +++ b/backend/tests/_fake_redis.py @@ -95,6 +95,10 @@ def sunion(self, *keys) -> set: result |= self._sets.get(key, set()) return result + def smembers(self, key: str) -> set: + self._maybe_raise("smembers") + return set(self._sets.get(key, set())) + def expire(self, key: str, seconds: int) -> bool: self._maybe_raise("expire") self._ttls[key] = seconds diff --git a/backend/tests/test_analytics_aggregator.py b/backend/tests/test_analytics_aggregator.py index 53afcbc..248882d 100644 --- a/backend/tests/test_analytics_aggregator.py +++ b/backend/tests/test_analytics_aggregator.py @@ -8,7 +8,7 @@ from __future__ import annotations import unittest -from datetime import datetime, timedelta +from datetime import datetime from unittest.mock import patch from control_center.analytics import aggregator @@ -238,6 +238,43 @@ def test_redis_error_returns_empty_dict(self) -> None: self.assertEqual(aggregator.read_user_activity(1, "2026-01-15"), {}) +class TestReadActiveUserIds(AggregatorTestCase): + def test_empty_dates_returns_empty_set(self) -> None: + self.assertEqual(aggregator.read_active_user_ids([]), set()) + + def test_single_date_returns_smembers(self) -> None: + aggregator.apply_interaction_event(_event()) + self.assertEqual(aggregator.read_active_user_ids(["2026-01-15"], org_id=1), {"42"}) + + def test_multiple_dates_unions(self) -> None: + aggregator.apply_interaction_event(_event(event_id="a", user_id=1, timestamp=datetime(2026, 1, 1))) + aggregator.apply_interaction_event(_event(event_id="b", user_id=2, timestamp=datetime(2026, 1, 2))) + result = aggregator.read_active_user_ids(["2026-01-01", "2026-01-02"], org_id=1) + self.assertEqual(result, {"1", "2"}) + + def test_redis_error_returns_empty_set(self) -> None: + self.fake.raise_on = {"smembers"} + self.assertEqual(aggregator.read_active_user_ids(["2026-01-15"]), set()) + + def test_redis_error_on_multi_date_union_returns_empty_set(self) -> None: + self.fake.raise_on = {"sunion"} + self.assertEqual(aggregator.read_active_user_ids(["2026-01-01", "2026-01-02"]), set()) + + +class TestReadKnownServices(AggregatorTestCase): + def test_no_events_returns_empty_list(self) -> None: + self.assertEqual(aggregator.read_known_services(1), []) + + def test_records_and_sorts_service_names(self) -> None: + aggregator.apply_interaction_event(_event(event_id="a", service="rag")) + aggregator.apply_interaction_event(_event(event_id="b", service="workflow-bundles", event_type="workflow.completed")) + self.assertEqual(aggregator.read_known_services(1), ["rag", "workflow-bundles"]) + + def test_redis_error_returns_empty_list(self) -> None: + self.fake.raise_on = {"smembers"} + self.assertEqual(aggregator.read_known_services(1), []) + + class TestPlatformLatencyPercentiles(AggregatorTestCase): def test_no_data_returns_all_none(self) -> None: result = aggregator.read_platform_latency_percentiles(["2026-01-15T10"]) diff --git a/backend/tests/test_analytics_cache.py b/backend/tests/test_analytics_cache.py index 71093ce..b8c015a 100644 --- a/backend/tests/test_analytics_cache.py +++ b/backend/tests/test_analytics_cache.py @@ -12,6 +12,64 @@ from _fake_redis import FakeRedis +class GetOrSetAsyncTestCase(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.fake = FakeRedis() + self._patcher = patch.object(cache, "_redis", self.fake) + self._patcher.start() + self.addCleanup(self._patcher.stop) + + async def test_cache_miss_awaits_compute_and_stores(self) -> None: + calls = [] + + async def compute(): + calls.append(1) + return {"total": 5} + + result = await cache.get_or_set_async("ak1", "overview", compute) + self.assertEqual(result, {"total": 5}) + self.assertEqual(len(calls), 1) + + async def test_cache_hit_skips_compute(self) -> None: + async def compute(): + return {"total": 1} + + await cache.get_or_set_async("ak2", "overview", compute) + + async def should_not_run(): + raise AssertionError("should not compute") + + result = await cache.get_or_set_async("ak2", "overview", should_not_run) + self.assertEqual(result, {"total": 1}) + + async def test_corrupted_cache_entry_falls_back_to_compute(self) -> None: + self.fake._strings["ak3"] = "not-json{" + + async def compute(): + return {"total": 9} + + result = await cache.get_or_set_async("ak3", "overview", compute) + self.assertEqual(result, {"total": 9}) + + async def test_redis_get_failure_falls_back_to_compute(self) -> None: + self.fake.raise_on = {"get"} + + async def compute(): + return {"total": 2} + + result = await cache.get_or_set_async("ak4", "overview", compute) + self.assertEqual(result, {"total": 2}) + + async def test_redis_setex_failure_still_returns_value(self) -> None: + self.fake.raise_on = {"setex"} + + async def compute(): + return {"total": 3} + + result = await cache.get_or_set_async("ak5", "overview", compute) + self.assertEqual(result, {"total": 3}) + + class GetOrSetTestCase(unittest.TestCase): def setUp(self) -> None: self.fake = FakeRedis() diff --git a/backend/tests/test_analytics_router.py b/backend/tests/test_analytics_router.py new file mode 100644 index 0000000..b5fe923 --- /dev/null +++ b/backend/tests/test_analytics_router.py @@ -0,0 +1,298 @@ +""" +tests/test_analytics_router.py + +End-to-end tests for the /analytics/* endpoints via FastAPI's TestClient, +matching test_routes_dashboard.py's own conventions. Covers the task +brief's own API test list (Section 12): date filtering, org filtering, +team filtering, grouping, empty results, cache behavior, Prometheus +unavailable, billing unavailable -- plus RBAC-through-HTTP for the full +platform_admin/org_admin/team_admin/regular-user matrix. +""" +from __future__ import annotations + +import unittest +from unittest.mock import AsyncMock, patch + +import jwt +from fastapi.testclient import TestClient + +from control_center.analytics import aggregator, tes_client +from control_center.analytics.permissions import MANAGE_ALL_ORGS +from control_center.core import jwt_verify as jwt_verify_module +from control_center.main import app +from _fake_redis import FakeRedis + +client = TestClient(app) +_no_raise_client = TestClient(app, raise_server_exceptions=False) +SECRET = "test-secret" + + +def _token(**claims) -> str: + return jwt.encode(claims, SECRET, algorithm="HS256") + + +def _auth(**claims) -> dict: + return {"Authorization": f"Bearer {_token(**claims)}"} + + +class AnalyticsRouterTestCase(unittest.TestCase): + def setUp(self) -> None: + jwt_patcher = patch.object(jwt_verify_module, "JWT_SECRET", SECRET) + jwt_patcher.start() + self.addCleanup(jwt_patcher.stop) + + self.fake = FakeRedis() + redis_patcher = patch.object(aggregator, "_redis", self.fake) + redis_patcher.start() + self.addCleanup(redis_patcher.stop) + + from control_center.analytics import cache as cache_module + cache_patcher = patch.object(cache_module, "_redis", self.fake) + cache_patcher.start() + self.addCleanup(cache_patcher.stop) + + tes_patcher = patch.object(tes_client, "get_runs", AsyncMock(return_value=None)) + tes_patcher.start() + self.addCleanup(tes_patcher.stop) + + +class AuthenticationTestCase(AnalyticsRouterTestCase): + def test_missing_token_returns_401(self) -> None: + r = client.get("/analytics/overview") + self.assertEqual(r.status_code, 401) + + def test_invalid_token_returns_401(self) -> None: + r = client.get("/analytics/overview", headers={"Authorization": "Bearer garbage"}) + self.assertEqual(r.status_code, 401) + + +class RbacMatrixTestCase(AnalyticsRouterTestCase): + def test_platform_admin_allowed(self) -> None: + r = client.get("/analytics/overview", headers=_auth(sub="1", permissions=[MANAGE_ALL_ORGS])) + self.assertEqual(r.status_code, 200) + + def test_org_admin_own_org_allowed(self) -> None: + headers = _auth(sub="1", permissions=[], org_id=5, org_role=["org_admin"]) + r = client.get("/analytics/overview", params={"org_id": 5}, headers=headers) + self.assertEqual(r.status_code, 200) + self.assertEqual(r.json()["org_id"], 5) + + def test_org_admin_other_org_denied(self) -> None: + headers = _auth(sub="1", permissions=[], org_id=5, org_role=["org_admin"]) + r = client.get("/analytics/overview", params={"org_id": 6}, headers=headers) + self.assertEqual(r.status_code, 403) + + def test_team_admin_permitted_team_allowed(self) -> None: + headers = _auth(sub="1", permissions=[], org_id=5, org_role=[], team_id=10, team_role="admin") + r = client.get("/analytics/overview", params={"org_id": 5, "team_id": 10}, headers=headers) + self.assertEqual(r.status_code, 200) + self.assertEqual(r.json()["team_id"], 10) + + def test_team_admin_unauthorized_team_denied(self) -> None: + headers = _auth(sub="1", permissions=[], org_id=5, org_role=[], team_id=10, team_role="admin") + r = client.get("/analytics/overview", params={"org_id": 5, "team_id": 11}, headers=headers) + self.assertEqual(r.status_code, 403) + + def test_regular_user_denied(self) -> None: + headers = _auth(sub="1", permissions=[], org_id=5, org_role=["member"]) + r = client.get("/analytics/overview", headers=headers) + self.assertEqual(r.status_code, 403) + + +class OverviewEndpointTestCase(AnalyticsRouterTestCase): + def test_empty_result_shape(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get("/analytics/overview", params={"org_id": 1}, headers=headers) + self.assertEqual(r.status_code, 200) + body = r.json() + self.assertEqual(body["total_queries"], 0) + self.assertEqual(body["active_users"], 0) + self.assertEqual(body["error_rate"], 0.0) + + def test_date_filtering(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get( + "/analytics/overview", + params={"org_id": 1, "from_date": "2026-01-01", "to_date": "2026-01-05"}, + headers=headers, + ) + self.assertEqual(r.status_code, 200) + self.assertEqual(r.json()["from_date"], "2026-01-01") + self.assertEqual(r.json()["to_date"], "2026-01-05") + + def test_cache_hit_skips_recomputation(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + params = {"org_id": 1, "from_date": "2026-01-01", "to_date": "2026-01-01"} + r1 = client.get("/analytics/overview", params=params, headers=headers) + self.assertEqual(r1.status_code, 200) + with patch("control_center.analytics.service.get_overview", AsyncMock(side_effect=AssertionError("should not recompute"))): + r2 = client.get("/analytics/overview", params=params, headers=headers) + self.assertEqual(r2.status_code, 200) + self.assertEqual(r1.json(), r2.json()) + + +class RunHelperTestCase(AnalyticsRouterTestCase): + def test_compute_failure_increments_error_metric_and_propagates(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + with patch("control_center.analytics.service.get_overview", AsyncMock(side_effect=RuntimeError("boom"))): + r = _no_raise_client.get("/analytics/overview", params={"org_id": 1}, headers=headers) + self.assertEqual(r.status_code, 500) + + +class QueriesEndpointTestCase(AnalyticsRouterTestCase): + def test_returns_daily_trend(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get( + "/analytics/queries", + params={"org_id": 1, "from_date": "2026-01-01", "to_date": "2026-01-02"}, + headers=headers, + ) + self.assertEqual(r.status_code, 200) + body = r.json() + self.assertEqual(len(body["daily"]), 2) + self.assertEqual(body["total_queries"], 0) + + +class WorkflowsEndpointTestCase(AnalyticsRouterTestCase): + def test_platform_admin_gets_null_with_note(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get("/analytics/workflows", headers=headers) + self.assertEqual(r.status_code, 200) + body = r.json() + self.assertIsNone(body["workflows_run"]) + self.assertIn("note", body) + + def test_org_admin_gets_counted_runs(self) -> None: + runs = [{"created_epoch": 1768000000, "state": "COMPLETED"}] + headers = _auth(sub="1", permissions=[], org_id=5, org_role=["org_admin"]) + with patch.object(tes_client, "get_runs", AsyncMock(return_value=runs)): + r = client.get( + "/analytics/workflows", + params={"org_id": 5, "from_date": "2020-01-01", "to_date": "2030-01-01"}, + headers=headers, + ) + self.assertEqual(r.status_code, 200) + self.assertEqual(r.json()["workflows_run"], 1) + + +class UsersEndpointTestCase(AnalyticsRouterTestCase): + def test_never_exposes_raw_user_ids(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get("/analytics/users", params={"org_id": 1}, headers=headers) + self.assertEqual(r.status_code, 200) + body = r.json() + self.assertIn("dau", body) + self.assertIn("wau", body) + self.assertIn("mau", body) + self.assertIn("daily", body) + self.assertNotIn("user_ids", body) + self.assertNotIn("users", body) + + +class PerformanceEndpointTestCase(AnalyticsRouterTestCase): + def test_prometheus_unavailable_still_returns_200(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get("/analytics/performance", headers=headers) + self.assertEqual(r.status_code, 200) + body = r.json() + self.assertEqual(body["scope"], "platform") + self.assertIsNone(body["org_id"]) + self.assertIsNone(body["team_id"]) + self.assertEqual(body["latency_source"], "events") + + def test_regular_user_still_denied(self) -> None: + headers = _auth(sub="1", permissions=[], org_id=5, org_role=["member"]) + r = client.get("/analytics/performance", headers=headers) + self.assertEqual(r.status_code, 403) + + +class UsageEndpointTestCase(AnalyticsRouterTestCase): + def test_billing_unavailable_degrades_gracefully(self) -> None: + # billing_client.get_usage/get_usage_limits already have their own + # dedicated unreachable-upstream tests (test_analytics_billing_client.py); + # this proves the router surfaces that (False, None) contract as a + # 200 with billing_available=False, not a 5xx. + from control_center.analytics import billing_client + + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + with ( + patch.object(billing_client, "get_usage", AsyncMock(return_value=(False, None))), + patch.object(billing_client, "get_usage_limits", AsyncMock(return_value=(False, None))), + ): + r = client.get("/analytics/usage", params={"org_id": 1}, headers=headers) + self.assertEqual(r.status_code, 200) + self.assertFalse(r.json()["billing_available"]) + + def test_no_org_id_for_platform_admin_returns_unavailable(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get("/analytics/usage", headers=headers) + self.assertEqual(r.status_code, 200) + self.assertFalse(r.json()["billing_available"]) + + +class ServicesEndpointTestCase(AnalyticsRouterTestCase): + def test_grouping_by_service(self) -> None: + from datetime import datetime + from control_center.analytics.schemas import AnalyticsEvent + aggregator.apply_interaction_event(AnalyticsEvent( + event_id="e1", event_type="query.completed", timestamp=datetime(2026, 1, 15, 10), + org_id=1, service="rag", action="rag.query", status="success", + )) + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get("/analytics/services", params={"org_id": 1, "from_date": "2026-01-15", "to_date": "2026-01-15"}, headers=headers) + self.assertEqual(r.status_code, 200) + services = r.json()["services"] + self.assertEqual(len(services), 1) + self.assertEqual(services[0]["service"], "rag") + + +class ExportEndpointTestCase(AnalyticsRouterTestCase): + def test_export_overview_returns_csv(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get("/analytics/export", params={"type": "overview", "org_id": 1}, headers=headers) + self.assertEqual(r.status_code, 200) + self.assertEqual(r.headers["content-type"], "text/csv; charset=utf-8") + self.assertIn("total_queries", r.text) + + def test_export_queries_returns_daily_rows(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get( + "/analytics/export", + params={"type": "queries", "org_id": 1, "from_date": "2026-01-01", "to_date": "2026-01-02"}, + headers=headers, + ) + self.assertEqual(r.status_code, 200) + lines = r.text.strip().splitlines() + self.assertEqual(lines[0], "date,count") + self.assertEqual(len(lines), 3) # header + 2 days + + def test_export_services_returns_service_rows(self) -> None: + from datetime import datetime + from control_center.analytics.schemas import AnalyticsEvent + aggregator.apply_interaction_event(AnalyticsEvent( + event_id="e1", event_type="query.completed", timestamp=datetime(2026, 1, 15, 10), + org_id=1, service="rag", action="rag.query", status="success", + )) + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get( + "/analytics/export", + params={"type": "services", "org_id": 1, "from_date": "2026-01-15", "to_date": "2026-01-15"}, + headers=headers, + ) + self.assertEqual(r.status_code, 200) + lines = r.text.strip().splitlines() + self.assertEqual(lines[0], "service,total_calls,errors,error_rate,avg_latency_ms") + self.assertIn("rag", lines[1]) + + def test_unknown_export_type_returns_400(self) -> None: + headers = _auth(sub="1", permissions=[MANAGE_ALL_ORGS]) + r = client.get("/analytics/export", params={"type": "bogus"}, headers=headers) + self.assertEqual(r.status_code, 400) + + def test_export_requires_auth(self) -> None: + r = client.get("/analytics/export", params={"type": "overview"}) + self.assertEqual(r.status_code, 401) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_analytics_service.py b/backend/tests/test_analytics_service.py new file mode 100644 index 0000000..02151aa --- /dev/null +++ b/backend/tests/test_analytics_service.py @@ -0,0 +1,349 @@ +""" +tests/test_analytics_service.py + +Unit tests for control_center.analytics.service. Uses FakeRedis for the +real aggregator behind these functions (so counts/percentiles are real, +not mocked), and mocks httpx for the team-roster/TES/billing/Prometheus +upstreams. +""" +from __future__ import annotations + +import unittest +from datetime import date, datetime +from unittest.mock import AsyncMock, MagicMock, patch + +from control_center.analytics import aggregator, billing_client, prometheus, service, tes_client +from control_center.analytics.permissions import AnalyticsScope +from _fake_redis import FakeRedis + + +def _event(**overrides): + from control_center.analytics.schemas import AnalyticsEvent + fields = dict( + event_id="evt-1", event_type="query.completed", timestamp=datetime(2026, 1, 15, 10), + org_id=1, team_id=None, user_id=42, service="rag", action="rag.query", + status="success", duration_ms=None, request_id=None, metadata={}, + ) + fields.update(overrides) + return AnalyticsEvent(**fields) + + +def _resp(status_code: int, json_body=None) -> MagicMock: + r = MagicMock() + r.status_code = status_code + r.json.return_value = json_body + return r + + +def _mock_ctx(response): + mock_client = MagicMock() + mock_client.get = AsyncMock(return_value=response) + mock_ctx = MagicMock() + mock_ctx.__aenter__ = AsyncMock(return_value=mock_client) + mock_ctx.__aexit__ = AsyncMock(return_value=False) + return mock_ctx + + +class ServiceTestCase(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.fake = FakeRedis() + self._agg_patcher = patch.object(aggregator, "_redis", self.fake) + self._agg_patcher.start() + self.addCleanup(self._agg_patcher.stop) + self._cache_patcher = patch.object(service.cache, "_redis", self.fake) + self._cache_patcher.start() + self.addCleanup(self._cache_patcher.stop) + + +class ResolveDateRangeTestCase(unittest.TestCase): + def test_both_none_defaults_to_last_30_days(self) -> None: + with patch("control_center.analytics.service.date") as mock_date: + mock_date.today.return_value = date(2026, 1, 30) + mock_date.side_effect = lambda *a, **kw: date(*a, **kw) + frm, to = service.resolve_date_range(None, None) + self.assertEqual(to, date(2026, 1, 30)) + self.assertEqual(frm, date(2026, 1, 1)) + + def test_both_given_passed_through(self) -> None: + frm, to = service.resolve_date_range(date(2026, 1, 1), date(2026, 1, 5)) + self.assertEqual((frm, to), (date(2026, 1, 1), date(2026, 1, 5))) + + +class TeamRosterTestCase(ServiceTestCase): + async def test_success_returns_user_id_set(self) -> None: + body = [{"user_id": 1, "role": "admin"}, {"user_id": 2, "role": "member"}] + with patch("control_center.analytics.service.httpx.AsyncClient", return_value=_mock_ctx(_resp(200, body))): + roster = await service._team_roster(1, 10, "Bearer tok") + self.assertEqual(roster, {"1", "2"}) + + async def test_non_200_returns_none(self) -> None: + with patch("control_center.analytics.service.httpx.AsyncClient", return_value=_mock_ctx(_resp(404))): + roster = await service._team_roster(1, 10, "Bearer tok") + self.assertIsNone(roster) + + async def test_unreachable_returns_none(self) -> None: + import httpx + mock_client = MagicMock() + mock_client.get = AsyncMock(side_effect=httpx.ConnectError("refused")) + mock_ctx = MagicMock() + mock_ctx.__aenter__ = AsyncMock(return_value=mock_client) + mock_ctx.__aexit__ = AsyncMock(return_value=False) + with patch("control_center.analytics.service.httpx.AsyncClient", return_value=mock_ctx): + roster = await service._team_roster(1, 10, "Bearer tok") + self.assertIsNone(roster) + + async def test_non_list_body_returns_none(self) -> None: + with patch("control_center.analytics.service.httpx.AsyncClient", return_value=_mock_ctx(_resp(200, {"not": "a list"}))): + roster = await service._team_roster(1, 10, "Bearer tok") + self.assertIsNone(roster) + + async def test_member_missing_user_id_is_skipped(self) -> None: + body = [{"user_id": 1}, {"role": "member"}] + with patch("control_center.analytics.service.httpx.AsyncClient", return_value=_mock_ctx(_resp(200, body))): + roster = await service._team_roster(1, 10, "Bearer tok") + self.assertEqual(roster, {"1"}) + + async def test_result_is_cached(self) -> None: + body = [{"user_id": 1}] + with patch("control_center.analytics.service.httpx.AsyncClient", return_value=_mock_ctx(_resp(200, body))) as mock_cls: + await service._team_roster(1, 10, "Bearer tok") + await service._team_roster(1, 10, "Bearer tok") + self.assertEqual(mock_cls.call_count, 1) + + +class ResolveTeamRosterIfNeededTestCase(unittest.IsolatedAsyncioTestCase): + async def test_no_team_id_not_applicable(self) -> None: + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + roster, applicable = await service._resolve_team_roster_if_needed(scope, "Bearer tok") + self.assertIsNone(roster) + self.assertFalse(applicable) + + async def test_no_org_id_not_applicable(self) -> None: + scope = AnalyticsScope(is_platform_admin=True, org_id=None, team_id=5, user_id="1") + roster, applicable = await service._resolve_team_roster_if_needed(scope, "Bearer tok") + self.assertIsNone(roster) + self.assertFalse(applicable) + + async def test_team_scoped_delegates_to_team_roster(self) -> None: + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=10, user_id="1") + with patch.object(service, "_team_roster", AsyncMock(return_value={"1", "2"})): + roster, applicable = await service._resolve_team_roster_if_needed(scope, "Bearer tok") + self.assertEqual(roster, {"1", "2"}) + self.assertTrue(applicable) + + +class GetOverviewTestCase(ServiceTestCase): + async def test_org_scoped_success(self) -> None: + aggregator.apply_interaction_event(_event()) + aggregator.apply_interaction_event(_event(event_id="evt-2", event_type="query.failed", status="error")) + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + with patch.object(tes_client, "get_runs", AsyncMock(return_value=None)): + result = await service.get_overview(scope, date(2026, 1, 15), date(2026, 1, 15), "Bearer tok") + self.assertEqual(result["total_queries"], 2) + self.assertEqual(result["active_users"], 1) + self.assertEqual(result["error_rate"], 0.5) + self.assertIsNone(result["workflows_run"]) + self.assertEqual(result["org_id"], 1) + + async def test_platform_admin_workflows_run_always_none(self) -> None: + scope = AnalyticsScope(is_platform_admin=True, org_id=None, team_id=None, user_id="1") + result = await service.get_overview(scope, date(2026, 1, 15), date(2026, 1, 15), "Bearer tok") + self.assertIsNone(result["workflows_run"]) + + async def test_team_scoped_roster_unavailable_returns_none_and_flag(self) -> None: + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=10, user_id="1") + with patch.object(service, "_team_roster", AsyncMock(return_value=None)): + result = await service.get_overview(scope, date(2026, 1, 15), date(2026, 1, 15), "Bearer tok") + self.assertIsNone(result["total_queries"]) + self.assertIsNone(result["active_users"]) + self.assertFalse(result["team_scope_available"]) + + async def test_team_scoped_roster_available_sums_only_roster_members(self) -> None: + aggregator.apply_interaction_event(_event(event_id="a", user_id=1)) + aggregator.apply_interaction_event(_event(event_id="b", user_id=2)) + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=10, user_id="1") + with patch.object(service, "_team_roster", AsyncMock(return_value={"1"})): + result = await service.get_overview(scope, date(2026, 1, 15), date(2026, 1, 15), "Bearer tok") + self.assertEqual(result["total_queries"], 1) + self.assertEqual(result["active_users"], 1) + self.assertTrue(result["team_scope_available"]) + + async def test_zero_queries_gives_zero_error_rate(self) -> None: + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + result = await service.get_overview(scope, date(2026, 1, 15), date(2026, 1, 15), "Bearer tok") + self.assertEqual(result["error_rate"], 0.0) + + async def test_org_admin_workflow_count_from_tes(self) -> None: + runs = [{"created_epoch": int(datetime(2026, 1, 15, 10).timestamp()), "state": "COMPLETED"}] + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + with patch.object(tes_client, "get_runs", AsyncMock(return_value=runs)): + result = await service.get_overview(scope, date(2026, 1, 15), date(2026, 1, 15), "Bearer tok") + self.assertEqual(result["workflows_run"], 1) + + +class GetQueriesTestCase(ServiceTestCase): + async def test_daily_breakdown_and_total(self) -> None: + aggregator.apply_interaction_event(_event(event_id="a", timestamp=datetime(2026, 1, 1))) + aggregator.apply_interaction_event(_event(event_id="b", timestamp=datetime(2026, 1, 2))) + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + result = await service.get_queries(scope, date(2026, 1, 1), date(2026, 1, 2), "Bearer tok") + self.assertEqual(result["total_queries"], 2) + self.assertEqual(len(result["daily"]), 2) + + async def test_team_scope_unavailable_returns_null_daily(self) -> None: + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=10, user_id="1") + with patch.object(service, "_team_roster", AsyncMock(return_value=None)): + result = await service.get_queries(scope, date(2026, 1, 1), date(2026, 1, 1), "Bearer tok") + self.assertIsNone(result["total_queries"]) + self.assertIsNone(result["daily"][0]["count"]) + self.assertFalse(result["team_scope_available"]) + + async def test_team_scope_available_sums_roster_members_only(self) -> None: + aggregator.apply_interaction_event(_event(event_id="a", user_id=1, timestamp=datetime(2026, 1, 1))) + aggregator.apply_interaction_event(_event(event_id="b", user_id=2, timestamp=datetime(2026, 1, 1))) + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=10, user_id="1") + with patch.object(service, "_team_roster", AsyncMock(return_value={"1"})): + result = await service.get_queries(scope, date(2026, 1, 1), date(2026, 1, 1), "Bearer tok") + self.assertEqual(result["total_queries"], 1) + self.assertTrue(result["team_scope_available"]) + + +class GetUsersTestCase(ServiceTestCase): + async def test_dau_wau_mau(self) -> None: + aggregator.apply_interaction_event(_event(event_id="a", user_id=1, timestamp=datetime(2026, 1, 30))) + aggregator.apply_interaction_event(_event(event_id="b", user_id=2, timestamp=datetime(2026, 1, 25))) + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + result = await service.get_users(scope, date(2026, 1, 30), date(2026, 1, 30), "Bearer tok") + self.assertEqual(result["dau"], 1) + self.assertEqual(result["wau"], 2) + self.assertEqual(result["mau"], 2) + + async def test_team_scope_unavailable(self) -> None: + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=10, user_id="1") + with patch.object(service, "_team_roster", AsyncMock(return_value=None)): + result = await service.get_users(scope, date(2026, 1, 30), date(2026, 1, 30), "Bearer tok") + self.assertIsNone(result["dau"]) + self.assertFalse(result["team_scope_available"]) + + async def test_team_scope_available(self) -> None: + aggregator.apply_interaction_event(_event(user_id=1, timestamp=datetime(2026, 1, 30))) + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=10, user_id="1") + with patch.object(service, "_team_roster", AsyncMock(return_value={"1"})): + result = await service.get_users(scope, date(2026, 1, 30), date(2026, 1, 30), "Bearer tok") + self.assertEqual(result["dau"], 1) + self.assertTrue(result["team_scope_available"]) + + async def test_never_returns_raw_user_ids(self) -> None: + aggregator.apply_interaction_event(_event()) + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + result = await service.get_users(scope, date(2026, 1, 15), date(2026, 1, 15), "Bearer tok") + dumped = str(result) + self.assertNotIn("42", dumped) # the user_id from _event() + + +class GetServicesTestCase(ServiceTestCase): + async def test_no_org_id_returns_empty_with_note(self) -> None: + scope = AnalyticsScope(is_platform_admin=True, org_id=None, team_id=None, user_id="1") + result = await service.get_services(scope, date(2026, 1, 15), date(2026, 1, 15)) + self.assertEqual(result["services"], []) + self.assertIn("note", result) + + async def test_breaks_down_by_service(self) -> None: + aggregator.apply_interaction_event(_event(event_id="a", service="rag")) + aggregator.apply_interaction_event(_event(event_id="b", service="rag", event_type="query.failed", status="error")) + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + result = await service.get_services(scope, date(2026, 1, 15), date(2026, 1, 15)) + self.assertEqual(len(result["services"]), 1) + row = result["services"][0] + self.assertEqual(row["service"], "rag") + self.assertEqual(row["total_calls"], 2) + self.assertEqual(row["errors"], 1) + self.assertEqual(row["error_rate"], 0.5) + self.assertIsNone(row["avg_latency_ms"]) + + +class GetPerformanceTestCase(ServiceTestCase): + async def test_events_fallback_when_prometheus_unavailable(self) -> None: + aggregator.apply_audit_event(event_id="a1", timestamp=datetime(2026, 1, 15, 10), is_request=True, is_error=False, latency_ms=100) + with patch.object(prometheus, "query_latency_quantiles", AsyncMock(return_value={"available": False, "result": None})): + result = await service.get_performance(date(2026, 1, 15), date(2026, 1, 15)) + self.assertEqual(result["latency_source"], "events") + self.assertEqual(result["scope"], "platform") + self.assertIsNone(result["org_id"]) + self.assertIsNone(result["team_id"]) + + async def test_prometheus_used_when_available(self) -> None: + prom_result = {"available": True, "p50": 0.05, "p95": 0.2, "p99": 0.5} + with patch.object(prometheus, "query_latency_quantiles", AsyncMock(return_value=prom_result)): + result = await service.get_performance(date(2026, 1, 15), date(2026, 1, 15)) + self.assertEqual(result["latency_source"], "prometheus") + self.assertEqual(result["p50_latency_ms"], 50.0) + self.assertEqual(result["p95_latency_ms"], 200.0) + self.assertEqual(result["p99_latency_ms"], 500.0) + + async def test_error_rate_and_throughput_computed(self) -> None: + aggregator.apply_audit_event(event_id="a1", timestamp=datetime(2026, 1, 15, 10), is_request=True, is_error=True, latency_ms=100) + aggregator.apply_audit_event(event_id="a2", timestamp=datetime(2026, 1, 15, 11), is_request=True, is_error=False, latency_ms=100) + with patch.object(prometheus, "query_latency_quantiles", AsyncMock(return_value={"available": False, "result": None})): + result = await service.get_performance(date(2026, 1, 15), date(2026, 1, 15)) + self.assertEqual(result["error_rate"], 0.5) + self.assertEqual(result["throughput_per_day"], 2.0) + + +class GetWorkflowsTestCase(ServiceTestCase): + async def test_platform_admin_gets_none_with_note(self) -> None: + scope = AnalyticsScope(is_platform_admin=True, org_id=None, team_id=None, user_id="1") + result = await service.get_workflows(scope, date(2026, 1, 1), date(2026, 1, 31), "Bearer tok") + self.assertIsNone(result["workflows_run"]) + self.assertIn("note", result) + + async def test_tes_unavailable_returns_none(self) -> None: + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + with patch.object(tes_client, "get_runs", AsyncMock(return_value=None)): + result = await service.get_workflows(scope, date(2026, 1, 1), date(2026, 1, 31), "Bearer tok") + self.assertIsNone(result["workflows_run"]) + + async def test_counts_runs_in_range_and_success_rate(self) -> None: + runs = [ + {"created_epoch": int(datetime(2026, 1, 15, 10).timestamp()), "state": "COMPLETED"}, + {"created_epoch": int(datetime(2026, 1, 16, 10).timestamp()), "state": "FAILED"}, + {"created_epoch": int(datetime(2025, 1, 1, 10).timestamp()), "state": "COMPLETED"}, # out of range + ] + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + with patch.object(tes_client, "get_runs", AsyncMock(return_value=runs)): + result = await service.get_workflows(scope, date(2026, 1, 1), date(2026, 1, 31), "Bearer tok") + self.assertEqual(result["workflows_run"], 2) + self.assertEqual(result["success_rate"], 0.5) + self.assertEqual(len(result["daily"]), 2) + + +class GetUsageTestCase(ServiceTestCase): + async def test_no_org_id_returns_unavailable(self) -> None: + scope = AnalyticsScope(is_platform_admin=True, org_id=None, team_id=None, user_id="1") + result = await service.get_usage(scope, "Bearer tok") + self.assertFalse(result["billing_available"]) + self.assertIn("note", result) + + async def test_success_passes_through_billing_data(self) -> None: + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + with ( + patch.object(billing_client, "get_usage", AsyncMock(return_value=(True, {"services": []}))), + patch.object(billing_client, "get_usage_limits", AsyncMock(return_value=(True, {"limit": 100}))), + ): + result = await service.get_usage(scope, "Bearer tok") + self.assertTrue(result["billing_available"]) + self.assertEqual(result["usage"], {"services": []}) + self.assertEqual(result["limits"], {"limit": 100}) + + async def test_billing_unavailable(self) -> None: + scope = AnalyticsScope(is_platform_admin=False, org_id=1, team_id=None, user_id="1") + with ( + patch.object(billing_client, "get_usage", AsyncMock(return_value=(False, None))), + patch.object(billing_client, "get_usage_limits", AsyncMock(return_value=(True, {}))), + ): + result = await service.get_usage(scope, "Bearer tok") + self.assertFalse(result["billing_available"]) + + +if __name__ == "__main__": + unittest.main()