awoooi/apps/api/tests/test_ai_provider_status_readback.py

from __future__ import annotations

from unittest.mock import AsyncMock

import pytest

from src.api.v1.ai import get_ai_status
from src.api.v1.health import get_ai_usage
from src.core import config as config_module
from src.services import ai_control as ai_control_module
from src.services import ai_rate_limiter as limiter_module
from src.services.ai_provider_policy import production_route_readback

PRODUCTION_PROVIDER_ORDER = [
    "ollama_gcp_a",
    "ollama_gcp_b",
    "ollama_local",
    "gemini",
]


def _usage(*, authenticated: bool | None = None) -> dict:
    return {
        "provider": "gemini",
        "limited": True,
        "readback_status": "verified",
        "cost_exceeded": False,
        "accounting": {
            "complete": True,
            "status": "receipt_backed",
        },
        "pricing_policy": {
            "model": "gemini-2.5-flash-lite",
            "supported": True,
        },
        "total_cost_usd": {
            "current": 0.0,
            "reserved": 0.0,
            "limit": 5.0,
            "alert_threshold": 4.0,
        },
        "authentication": {
            "provider": "gemini",
            "configured": True,
            "authenticated": authenticated,
            "authentication_verified": authenticated is True,
            "verification_status": (
                "authenticated_verified_by_generation"
                if authenticated is True
                else "configured_not_verified"
            ),
            "verification_source": (
                "durable_generation_receipt" if authenticated is True else None
            ),
            "verified_at": "2026-07-14T12:00:00+08:00" if authenticated else None,
        },
    }


class _Limiter:
    def __init__(self, usage: dict | None = None, error: Exception | None = None) -> None:
        self.usage = usage
        self.error = error

    async def get_usage_stats(self, provider: str) -> dict:
        assert provider == "gemini"
        if self.error:
            raise self.error
        assert self.usage is not None
        return dict(self.usage)


def _configure(
    monkeypatch: pytest.MonkeyPatch,
    *,
    limiter: _Limiter,
    disabled: bool = False,
) -> None:
    monkeypatch.setattr(
        config_module.settings,
        "GEMINI_API_KEY",
        "configured-test-value",
    )
    monkeypatch.setenv("ENABLE_GEMINI", "true")
    monkeypatch.setattr(limiter_module, "get_ai_rate_limiter", lambda: limiter)
    monkeypatch.setattr(
        ai_control_module,
        "is_provider_disabled",
        AsyncMock(return_value=disabled),
    )


@pytest.mark.asyncio
async def test_ai_status_separates_configured_authenticated_and_cost_guard(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _configure(monkeypatch, limiter=_Limiter(_usage(authenticated=None)))

    result = await get_ai_status()

    route = result["production_route"]
    assert result["fallback_order"] == PRODUCTION_PROVIDER_ORDER
    assert [hop["identity"] for hop in route["hops"]] == [
        "GCP-A",
        "GCP-B",
        "host111",
        "Gemini API",
    ]
    assert result["gemini_configured"] is True
    assert result["gemini_enabled"] is True
    assert result["gemini_authenticated"] is None
    assert result["gemini_authentication_verified"] is False
    assert result["gemini_production_executable"] is False
    assert "configured-test-value" not in repr(result)


@pytest.mark.asyncio
async def test_disabled_gemini_remains_visible_but_non_executable(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _configure(
        monkeypatch,
        limiter=_Limiter(_usage(authenticated=True)),
        disabled=True,
    )

    result = await get_ai_status()

    assert result["fallback_order"] == PRODUCTION_PROVIDER_ORDER
    assert result["gemini_enabled"] is False
    assert result["gemini_production_executable"] is False
    assert result["production_route"]["hops"][3]["production_executable"] is False


@pytest.mark.asyncio
async def test_verified_auth_and_complete_accounting_enable_cost_guard_readback(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _configure(monkeypatch, limiter=_Limiter(_usage(authenticated=True)))

    result = await get_ai_status()

    assert result["production_route"]["gemini"]["cost_guard_ready"] is True
    assert result["gemini_production_executable"] is True


@pytest.mark.parametrize(
    "accounting,readback_status",
    [
        ({}, "verified"),
        ({"complete": True}, "verified"),
        ({"complete": False, "status": "receipt_backed"}, "verified"),
        ({"complete": True, "status": "degraded_missing_receipt"}, "verified"),
        (
            {"complete": True, "status": "receipt_backed", "degraded": True},
            "verified",
        ),
        ({"complete": True, "status": "receipt_backed"}, "degraded"),
    ],
)
def test_cost_guard_static_readback_fails_closed_without_complete_nondegraded_accounting(
    accounting: dict,
    readback_status: str,
) -> None:
    cost_guard = _usage(authenticated=True)
    cost_guard["accounting"] = accounting
    cost_guard["readback_status"] = readback_status

    route = production_route_readback(
        gemini_configured=True,
        gemini_enabled=True,
        gemini_authentication=cost_guard["authentication"],
        cost_guard=cost_guard,
    )

    assert route["gemini"]["cost_guard_ready"] is False
    assert route["gemini"]["production_executable"] is False
    assert route["hops"][3]["production_executable"] is False


@pytest.mark.parametrize("missing_field", ["pricing_policy", "cost_exceeded"])
def test_cost_guard_static_readback_requires_affirmative_pricing_and_cost_proof(
    missing_field: str,
) -> None:
    cost_guard = _usage(authenticated=True)
    cost_guard.pop(missing_field)

    route = production_route_readback(
        gemini_configured=True,
        gemini_enabled=True,
        gemini_authentication=cost_guard["authentication"],
        cost_guard=cost_guard,
    )

    assert route["gemini"]["cost_guard_ready"] is False
    assert route["gemini"]["production_executable"] is False


@pytest.mark.asyncio
async def test_status_endpoints_fail_closed_when_cost_readback_is_unavailable(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _configure(
        monkeypatch,
        limiter=_Limiter(error=RuntimeError("isolated readback failure")),
    )

    status = await get_ai_status()
    health = await get_ai_usage()

    assert status["gemini_production_executable"] is False
    assert status["production_route"]["gemini"]["cost_guard_ready"] is False
    assert health["gemini"]["readback_status"] == "unavailable"
    assert health["gemini_status"]["production_executable"] is False
    assert health["claude"] == {
        "provider": "claude",
        "production_executable": False,
        "mode": "non_executable_shadow_metadata_only",
    }