75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from src.services import ai_control
|
|
from src.services.ai_providers.interfaces import is_provider_enabled_by_env
|
|
|
|
|
|
class _ControlRedis:
|
|
def __init__(self, value: str | None = None) -> None:
|
|
self.value = value
|
|
self.set = AsyncMock(return_value=True)
|
|
self.delete = AsyncMock(return_value=1)
|
|
|
|
async def get(self, _key: str) -> str | None:
|
|
return self.value
|
|
|
|
|
|
def test_gemini_environment_gate_defaults_disabled(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.delenv("ENABLE_GEMINI", raising=False)
|
|
|
|
assert is_provider_enabled_by_env("gemini") is False
|
|
assert is_provider_enabled_by_env("ollama_gcp_a") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_paid_disable_state_is_fail_closed_when_redis_is_unavailable(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def _unavailable() -> None:
|
|
raise RuntimeError("isolated Redis failure")
|
|
|
|
monkeypatch.setattr(ai_control, "_get_redis", _unavailable)
|
|
|
|
with pytest.raises(RuntimeError, match="isolated Redis failure"):
|
|
await ai_control.is_provider_disabled("gemini")
|
|
assert await ai_control.is_provider_disabled("ollama_gcp_a") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gemini_missing_disable_key_is_disabled_and_control_is_persistent(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
redis = _ControlRedis(value=None)
|
|
|
|
async def _redis() -> _ControlRedis:
|
|
return redis
|
|
|
|
monkeypatch.setattr(ai_control, "_get_redis", _redis)
|
|
|
|
assert await ai_control.is_provider_disabled("gemini") is True
|
|
assert await ai_control.set_provider_disabled("gemini", True) is True
|
|
redis.set.assert_awaited_with("ai:control:disabled:gemini", "1")
|
|
assert await ai_control.set_provider_disabled("gemini", False) is True
|
|
redis.set.assert_awaited_with("ai:control:disabled:gemini", "0")
|
|
redis.delete.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_primary_override_cannot_jump_to_paid_or_legacy_provider() -> None:
|
|
assert await ai_control.set_primary_provider("gemini") is False
|
|
assert await ai_control.set_primary_provider("claude") is False
|
|
assert await ai_control.set_primary_provider("openclaw_nemo") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shadow_provider_cannot_be_enabled_for_execution() -> None:
|
|
assert await ai_control.set_provider_disabled("claude", False) is False
|
|
assert await ai_control.set_provider_disabled("nemotron", False) is False
|
|
assert await ai_control.set_provider_disabled("openclaw_nemo", False) is False
|