541a32a9cb
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m38s
CD Pipeline / build-and-deploy (push) Failing after 24m19s
CD Pipeline / post-deploy-checks (push) Has been skipped
276 lines
8.9 KiB
Python
276 lines
8.9 KiB
Python
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",
|
|
"claude",
|
|
"gemini",
|
|
]
|
|
|
|
|
|
def _usage(
|
|
provider: str = "gemini",
|
|
*,
|
|
authenticated: bool | None = None,
|
|
) -> dict:
|
|
model = "claude-sonnet-5" if provider == "claude" else "gemini-2.5-flash-lite"
|
|
return {
|
|
"provider": provider,
|
|
"limited": True,
|
|
"readback_status": "verified",
|
|
"cost_exceeded": False,
|
|
"accounting": {
|
|
"complete": True,
|
|
"status": "receipt_backed",
|
|
},
|
|
"pricing_policy": {
|
|
"model": model,
|
|
"supported": True,
|
|
},
|
|
"total_cost_usd": {
|
|
"current": 0.0,
|
|
"reserved": 0.0,
|
|
"limit": 5.0,
|
|
"alert_threshold": 4.0,
|
|
},
|
|
"authentication": {
|
|
"provider": provider,
|
|
"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:
|
|
if self.error:
|
|
raise self.error
|
|
usage = self.usage or _usage(provider)
|
|
payload = dict(usage) if usage.get("provider") == provider else _usage(
|
|
provider,
|
|
authenticated=(usage.get("authentication") or {}).get("authenticated"),
|
|
)
|
|
return payload
|
|
|
|
|
|
def _configure(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
*,
|
|
limiter: _Limiter,
|
|
disabled: bool = False,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
config_module.settings,
|
|
"GEMINI_API_KEY",
|
|
"configured-test-value",
|
|
)
|
|
monkeypatch.setattr(
|
|
config_module.settings,
|
|
"CLAUDE_API_KEY",
|
|
"configured-test-value",
|
|
)
|
|
monkeypatch.setattr(config_module.settings, "GEMINI_EXECUTION_MODE", "production")
|
|
monkeypatch.setattr(config_module.settings, "GEMINI_CANARY_PERCENT", 0)
|
|
monkeypatch.setattr(config_module.settings, "CLAUDE_EXECUTION_MODE", "shadow")
|
|
monkeypatch.setattr(config_module.settings, "CLAUDE_CANARY_PERCENT", 0)
|
|
monkeypatch.setenv("ENABLE_GEMINI", "true")
|
|
monkeypatch.setenv("ENABLE_CLAUDE", "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",
|
|
"Anthropic Claude API",
|
|
"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 result["claude_configured"] is True
|
|
assert result["claude_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"][4]["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.asyncio
|
|
async def test_verified_gemini_canary_is_not_reported_as_production(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_configure(monkeypatch, limiter=_Limiter(_usage(authenticated=True)))
|
|
monkeypatch.setattr(config_module.settings, "GEMINI_EXECUTION_MODE", "canary")
|
|
monkeypatch.setattr(config_module.settings, "GEMINI_CANARY_PERCENT", 5)
|
|
|
|
result = await get_ai_status()
|
|
|
|
assert result["gemini_production_executable"] is False
|
|
assert result["gemini_canary_executable"] is True
|
|
assert result["gemini_execution_mode"] == "canary"
|
|
assert result["gemini_canary_percent"] == 5
|
|
|
|
|
|
@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"][4]["production_executable"] is False
|
|
|
|
|
|
def test_route_readback_exposes_cloud_boundary_and_mesh_gap() -> None:
|
|
route = production_route_readback(
|
|
gemini_configured=False,
|
|
gemini_enabled=False,
|
|
)
|
|
|
|
assert route["hops"][0]["data_boundary"] == "cloud"
|
|
assert route["hops"][1]["data_boundary"] == "cloud"
|
|
assert route["hops"][2]["data_boundary"] == "local"
|
|
assert route["transport_security"] == {
|
|
"status": "degraded_mesh_pending",
|
|
"production_closure": False,
|
|
"reason": "gcp_ollama_public_http_stopgap",
|
|
"replacement": "wireguard_mesh_10.77.114.x",
|
|
"sanitized_generation_gate": (
|
|
"durable_same_run_endpoint_bound_transport_receipt_required"
|
|
),
|
|
"receipt_ttl_max_seconds": 120,
|
|
}
|
|
|
|
|
|
@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"]["readback_status"] == "unavailable"
|
|
assert health["claude_status"]["production_executable"] is False
|