tests/unit/test_suite_a_token_consistency.py

"""Suite A — Token usage consistency (P0)."""

import pytest
from sqlalchemy import select

from app.models.client_access import ClientAccess
from app.services.analytics_core.tracker import track_turn
from app.services.token_budget import tokens_remaining, total_budget, usage_summary
from tests.conftest import FakePipelineResult


@pytest.mark.p0
def test_admin_token_usage_includes_bonus_tokens(client_access):
    """A-10: Admin token-usage includes bonus_tokens in budget."""
    response = usage_summary(client_access)
    expected_total = total_budget(client_access)
    assert response["budget"] == expected_total
    assert response["remaining"] == expected_total - client_access.tokens_used


@pytest.mark.p0
def test_admin_and_embed_remaining_match(client_access):
    """A-43: Token Usage page and Embed Token card must show same remaining."""
    admin = usage_summary(client_access)
    embed_remaining = tokens_remaining(client_access)
    assert admin["remaining"] == embed_remaining


@pytest.mark.p0
@pytest.mark.asyncio
async def test_track_turn_increments_client_access_tokens_used(
    db_session, mock_redis, author_user, client_access,
):
    """A-01: Chat increments client_access.tokens_used."""
    from app.models.base import generate_uuid
    from app.models.chat_session import ChatSession
    from datetime import datetime, timezone
    from tests.conftest import FakePipelineResult
    from app.services.analytics_core.tracker import track_turn

    db_session.add(
        ChatSession(
            id="session-1",
            author_id=author_user.id,
            visitor_fingerprint="fp-test",
            created_at=datetime.now(timezone.utc),
        )
    )
    await db_session.flush()

    tokens_before = client_access.tokens_used
    result = FakePipelineResult(prompt_tokens=100, completion_tokens=50)
    await track_turn(
        db=db_session,
        redis=mock_redis,
        session_id="session-1",
        author_id=author_user.id,
        book_id=None,
        user_message="Hello",
        result=result,
    )
    await db_session.commit()

    row = await db_session.scalar(
        select(ClientAccess).where(ClientAccess.id == client_access.id)
    )
    assert row.tokens_used == tokens_before + 150


@pytest.mark.p0
@pytest.mark.asyncio
async def test_track_turn_increments_redis_counter(
    db_session, mock_redis, author_user, client_access
):
    """A-02: Redis tokens:{author_id}:current is incremented on chat."""
    from app.models.chat_session import ChatSession
    from datetime import datetime, timezone
    db_session.add(ChatSession(
        id="session-redis",
        author_id=author_user.id,
        visitor_fingerprint="fp-redis",
        created_at=datetime.now(timezone.utc),
    ))
    await db_session.flush()

    result = FakePipelineResult(prompt_tokens=100, completion_tokens=50)
    await track_turn(
        db=db_session,
        redis=mock_redis,
        session_id="session-redis",
        author_id=author_user.id,
        book_id=None,
        user_message="Hello",
        result=result,
    )
    key = f"tokens:{author_user.id}:current"
    # Redis counter must have been incremented by the 150 tokens in this turn
    assert (mock_redis._counters.get(key) or 0) >= 150


@pytest.mark.p0
def test_superadmin_and_admin_budget_formula_match(client_access):
    """Consistency matrix: SuperAdmin total_budget vs Admin budget must match."""
    super_total = total_budget(client_access)
    admin_budget = usage_summary(client_access)["budget"]
    assert admin_budget == super_total


@pytest.mark.p0
def test_no_active_subscription_returns_zeros():
    """Authors without an active grant must see zero budget, not stale grants."""
    from app.services.token_budget import usage_summary_or_empty

    result = usage_summary_or_empty(None)
    assert result["budget"] == 0
    assert result["used"] == 0
    assert result["remaining"] == 0
    assert result["has_active_subscription"] is False
    assert result["subscription_status"] == "none"