"""The platform's own LLM calls get booked, and the booking never breaks the call. The properties under test are the ones that make the ledger worth having rather than merely present: * a successful call and a FAILED call both land — the failures are the rows an admin most wants, because an expired key is a wall of one error code before anyone reports the feature is down; * the caller is unaffected when recording itself fails; * NULL tokens stay NULL rather than becoming a zero that sums wrong; * ``unattributed`` is a real bucket, not a dropped row. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta import pytest from sqlalchemy import select from app.domains.bailian import ledger from app.domains.bailian.ledger_read import overview from app.domains.bailian.models import LlmCallLedger pytestmark = pytest.mark.asyncio class TestPurposeNormalisation: """A purpose is a grouping key, so it must not fragment.""" def test_a_clean_slug_is_kept_verbatim(self): assert ledger.normalize_purpose("support.copilot") == "support.copilot" def test_messy_input_is_tidied_not_rejected(self): # Refusing to record would be worse than recording under a # tidied name: the call happened either way. assert ledger.normalize_purpose("Support Copilot!") == "support_copilot" def test_empty_becomes_the_visible_unattributed_bucket(self): assert ledger.normalize_purpose(None) == ledger.PURPOSE_UNKNOWN assert ledger.normalize_purpose(" ") == ledger.PURPOSE_UNKNOWN def test_it_never_exceeds_the_column(self): assert len(ledger.normalize_purpose("x" * 500)) <= 64 class TestTokenCoercion: """NULL is not zero, and a stored zero would sum wrong.""" @pytest.mark.parametrize("value", [None, 0, "0", -1, "", "abc", True, False]) def test_non_counts_become_null(self, value): assert ledger._as_positive_int(value) is None @pytest.mark.parametrize(("value", "expected"), [(1, 1), ("42", 42), (999999, 999999)]) def test_real_counts_survive(self, value, expected): assert ledger._as_positive_int(value) == expected class TestRecording: """These tests take the ``client`` fixture for its side effect. ``record`` deliberately opens its own session through the module-level ``async_session_factory`` — that independence is what keeps a bookkeeping write from poisoning a caller's transaction. The factory is bound to the APPLICATION engine, so without the redirect the ``client`` fixture installs (see its docstring: "any background-task code that opens its own session via the module-level factory therefore also lands on the test database"), these writes escape to the real dev database. That is not a test artefact worth working around — it is the same shape as a frontend test escaping jsdom to hit a live server, and it silently passed here until the table stopped existing in that other database. Read-back also uses a fresh session: the test's own transaction began earlier and cannot see the committed row under REPEATABLE READ. That is how every real reader sees these rows. """ @staticmethod async def _fetch(where): from app.core.database import async_session_factory async with async_session_factory() as session: return ( await session.execute(select(LlmCallLedger).where(where)) ).scalars().first() async def test_a_successful_call_is_booked(self, client, db_session): await ledger.record( purpose="automation_exception.triage", model="qwen3-max", region="cn-beijing", operation="chat_json", ok=True, input_tokens=120, output_tokens=45, duration_ms=800, ) row = await self._fetch( LlmCallLedger.purpose == "automation_exception.triage", ) assert row is not None assert row.ok is True assert row.input_tokens == 120 assert row.output_tokens == 45 assert row.operation == "chat_json" async def test_a_failed_call_is_booked_with_its_code(self, client, db_session): # The whole point: a key that expired shows up here long before # anyone reports the feature is down. await ledger.record( purpose="support.copilot", model="qwen3-max", region=None, operation="chat", ok=False, error_code="BAILIAN_LLM_HTTP_ERROR", ) row = await self._fetch(LlmCallLedger.purpose == "support.copilot") assert row is not None assert row.ok is False assert row.error_code == "BAILIAN_LLM_HTTP_ERROR" assert row.input_tokens is None async def test_recording_failure_never_reaches_the_caller(self, monkeypatch): """A bookkeeping row that will not save must not fail a user's finished work. This is the trade the module documents.""" import app.core.database as database_module def _explode(*_a, **_kw): raise RuntimeError("database is on fire") monkeypatch.setattr(database_module, "async_session_factory", _explode) # No exception escapes. await ledger.record( purpose="x.y", model="m", region=None, operation="chat", ok=True, ) async def test_an_undeclared_call_lands_in_the_visible_bucket(self, client, db_session): await ledger.record( purpose=None, model="qwen3-max", region=None, operation="chat_text", ok=True, input_tokens=5, ) row = await self._fetch(LlmCallLedger.purpose == ledger.PURPOSE_UNKNOWN) assert row is not None, "an unattributed call must be recorded, never dropped" class TestPurposeScope: def test_the_scope_sets_and_restores(self): assert ledger.current_purpose() is None with ledger.purpose_scope("cloud.efficiency"): assert ledger.current_purpose() == "cloud.efficiency" assert ledger.current_purpose() is None def test_the_scope_restores_even_when_the_body_raises(self): with pytest.raises(ValueError, match="boom"), \ ledger.purpose_scope("cloud.efficiency"): raise ValueError("boom") assert ledger.current_purpose() is None class TestOverviewRead: async def _seed(self, db_session, **kw): db_session.add(LlmCallLedger( id=uuid.uuid4(), purpose=kw.get("purpose", "a.b"), model=kw.get("model", "qwen3-max"), region=None, operation=kw.get("operation", "chat_json"), ok=kw.get("ok", True), input_tokens=kw.get("input_tokens"), output_tokens=kw.get("output_tokens"), error_code=kw.get("error_code"), called_at=kw.get("called_at", datetime.now(UTC)), )) await db_session.flush() async def test_calls_without_a_token_report_are_counted_separately( self, db_session, ): """The number that stops a NULL-heavy purpose reading as cheap.""" await self._seed(db_session, purpose="p.one", input_tokens=100, output_tokens=10) await self._seed(db_session, purpose="p.one") # vendor reported nothing data = await overview(db_session, days=7) row = next(r for r in data["by_purpose"] if r["purpose"] == "p.one") assert row["calls"] == 2 assert row["total_tokens"] == 110 assert row["calls_without_token_report"] == 1 async def test_a_row_outside_the_window_is_excluded(self, db_session): await self._seed( db_session, purpose="p.old", called_at=datetime.now(UTC) - timedelta(days=40), input_tokens=999, ) data = await overview(db_session, days=7) assert all(r["purpose"] != "p.old" for r in data["by_purpose"]) async def test_failures_surface_with_their_code(self, db_session): await self._seed( db_session, purpose="p.fail", ok=False, error_code="BAILIAN_LLM_HTTP_ERROR", ) data = await overview(db_session, days=7) assert any( e["error_code"] == "BAILIAN_LLM_HTTP_ERROR" and e["purpose"] == "p.fail" for e in data["top_errors"] ) row = next(r for r in data["by_purpose"] if r["purpose"] == "p.fail") assert row["failed_calls"] == 1 assert row["success_rate"] == 0.0 async def test_the_window_is_clamped_not_trusted(self, db_session): data = await overview(db_session, days=9999) assert data["window_days"] == 180 async def test_the_payload_says_it_is_not_money(self, db_session): """An MCP client reads this too and would otherwise be free to multiply tokens by a price it invented.""" data = await overview(db_session, days=7) assert "money" in data["money_note"].lower()