awoooi/apps/api/tests/test_ai_control_status_summary.py

from __future__ import annotations

import json
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest

from src.core import config as config_module
from src.services import ai_control
from src.services import ai_rate_limiter as limiter_module
from src.services.ollama_health_monitor import make_cache_key


class _ReceiptRedis:
    def __init__(self, receipts: dict[str, str]) -> None:
        self._receipts = receipts

    async def get(self, key: str):
        return self._receipts.get(key)


class _AuthenticationLimiter:
    async def get_provider_authentication_status(self, provider: str):
        if provider == "claude":
            return {
                "configured": True,
                "authenticated": True,
                "authentication_verified": True,
                "verification_status": "authenticated",
            }
        return {
            "configured": True,
            "authenticated": False,
            "authentication_verified": True,
            "verification_status": "authentication_rejected",
        }

    async def get_usage_stats(self, provider: str):
        return {
            "provider": provider,
            "readback_status": "verified",
            "total_cost_usd": {
                "current": 0.011802 if provider == "claude" else 0.000297,
            },
            "accounting": {
                "complete": True,
                "status": "receipt_backed",
                "degraded": False,
            },
        }


def _settings() -> SimpleNamespace:
    return SimpleNamespace(
        USE_AI_ROUTER=True,
        OLLAMA_URL="http://gcp-a.internal:11434",
        OLLAMA_SECONDARY_URL="http://gcp-b.internal:11434",
        OLLAMA_FALLBACK_URL="http://111.internal:11434",
        CLAUDE_EXECUTION_MODE="canary",
        CLAUDE_CANARY_PERCENT=5,
        GEMINI_EXECUTION_MODE="canary",
        GEMINI_CANARY_PERCENT=5,
    )


@pytest.mark.asyncio
async def test_ai_status_uses_receipts_and_run_owned_canary_semantics(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    settings = _settings()
    receipts = {
        make_cache_key(settings.OLLAMA_URL): json.dumps({"status": "healthy"}),
        make_cache_key(settings.OLLAMA_SECONDARY_URL): json.dumps(
            {"status": "offline"}
        ),
        make_cache_key(settings.OLLAMA_FALLBACK_URL): json.dumps(
            {"status": "degraded"}
        ),
    }
    redis = _ReceiptRedis(receipts)
    disabled_readback = AsyncMock(return_value=False)

    monkeypatch.setattr(config_module, "get_settings", lambda: settings)
    monkeypatch.setattr(ai_control, "_get_redis", AsyncMock(return_value=redis))
    monkeypatch.setattr(ai_control, "get_ai_router_enabled", AsyncMock(return_value=True))
    monkeypatch.setattr(ai_control, "get_primary_provider", AsyncMock(return_value=None))
    monkeypatch.setattr(ai_control, "is_provider_disabled", disabled_readback)
    monkeypatch.setattr(
        ai_control,
        "get_paid_provider_canary_control_readback",
        AsyncMock(
            return_value={
                "lock_present": False,
                "disabled_verified": True,
                "legacy_persistent_state_detected": False,
            }
        ),
    )
    monkeypatch.setattr(
        limiter_module,
        "get_ai_rate_limiter",
        lambda: _AuthenticationLimiter(),
    )

    summary = await ai_control.handle_ai_command("/ai status")

    assert "🟢 GCP-A | health=healthy | control=enabled" in summary
    assert "🔴 GCP-B | health=offline | control=enabled" in summary
    assert "🟠 111 | health=degraded | control=enabled" in summary
    assert "🧪 Claude | configured=true | authenticated=true" in summary
    assert "verification=authenticated | canary=canary 5%" in summary
    assert "⚠️ Gemini | configured=true | authenticated=false" in summary
    assert "verification=authentication_rejected | canary=canary 5%" in summary
    assert summary.count("route_lease=待單次 run-owned lease") == 2
    assert "claude: $0.011802 | receipt=verified" in summary
    assert "gemini: $0.000297 | receipt=verified" in summary
    assert "❌ Claude" not in summary
    assert "❌ Gemini" not in summary
    assert "硬鎖" not in summary
    assert [call.args[0] for call in disabled_readback.await_args_list] == [
        "ollama_gcp_a",
        "ollama_gcp_b",
        "ollama_local",
    ]


@pytest.mark.asyncio
async def test_ai_status_marks_active_paid_route_as_run_owned_lease(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    settings = _settings()
    redis = _ReceiptRedis({})

    monkeypatch.setattr(config_module, "get_settings", lambda: settings)
    monkeypatch.setattr(ai_control, "_get_redis", AsyncMock(return_value=redis))
    monkeypatch.setattr(ai_control, "get_ai_router_enabled", AsyncMock(return_value=True))
    monkeypatch.setattr(ai_control, "get_primary_provider", AsyncMock(return_value=None))
    monkeypatch.setattr(
        ai_control,
        "is_provider_disabled",
        AsyncMock(return_value=False),
    )
    monkeypatch.setattr(
        ai_control,
        "get_paid_provider_canary_control_readback",
        AsyncMock(
            return_value={
                "lock_present": True,
                "lock_ttl_seconds": 120,
                "claude_disabled": False,
                "gemini_disabled": False,
                "disabled_verified": False,
                "legacy_persistent_state_detected": False,
            }
        ),
    )
    monkeypatch.setattr(
        limiter_module,
        "get_ai_rate_limiter",
        lambda: _AuthenticationLimiter(),
    )

    summary = await ai_control.handle_ai_command("/ai status")

    assert "🟢 Claude" in summary
    assert "⚠️ Gemini" in summary
    assert summary.count("route_lease=active_run_owned_lease") == 2
    assert "❌ Claude" not in summary
    assert "❌ Gemini" not in summary


@pytest.mark.asyncio
async def test_ai_cost_uses_receipt_backed_accounting_not_legacy_keys(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        limiter_module,
        "get_ai_rate_limiter",
        lambda: _AuthenticationLimiter(),
    )

    summary = await ai_control.handle_ai_command("/ai cost")

    assert "claude: $0.011802 | receipt=verified" in summary
    assert "gemini: $0.000297 | receipt=verified" in summary
    assert "總計: $0.012099" in summary
    assert "receipt-backed accounting" in summary