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
274 lines
8.3 KiB
Python
274 lines
8.3 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,
|
|
*,
|
|
eval_results: list[list[object]] | None = None,
|
|
) -> None:
|
|
self.value = value
|
|
self.set = AsyncMock(return_value=True)
|
|
self.delete = AsyncMock(return_value=1)
|
|
self.eval_results = list(eval_results or [])
|
|
self.eval_calls: list[tuple[object, ...]] = []
|
|
|
|
async def get(self, _key: str) -> str | None:
|
|
return self.value
|
|
|
|
async def eval(self, *args):
|
|
self.eval_calls.append(args)
|
|
if not self.eval_results:
|
|
raise AssertionError("unexpected Redis eval")
|
|
return self.eval_results.pop(0)
|
|
|
|
|
|
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,
|
|
eval_results=[
|
|
[0, "paired_paid_state_not_enabled", ""],
|
|
[1, "paid_provider_state_set"],
|
|
],
|
|
)
|
|
|
|
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
|
|
assert await ai_control.set_provider_disabled("gemini", False) is False
|
|
assert len(redis.eval_calls) == 2
|
|
disable_call = redis.eval_calls[1]
|
|
assert "ai:control:disabled:claude" in disable_call
|
|
assert "ai:control:disabled:gemini" in disable_call
|
|
assert "ai:control:paid_canary:activation_lock" in disable_call
|
|
assert "ai:control:paid_canary:active_receipt" in disable_call
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_paid_canary_lease_is_run_owned_and_persistent_promotion_is_blocked(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
redis = _ControlRedis(
|
|
eval_results=[
|
|
[1, "activation_lease_acquired"],
|
|
[1, "paid_canary_disabled"],
|
|
]
|
|
)
|
|
|
|
async def _redis() -> _ControlRedis:
|
|
return redis
|
|
|
|
monkeypatch.setattr(ai_control, "_get_redis", _redis)
|
|
|
|
owner = "AIA-SRE-013-run.attempt-123456789abc"
|
|
assert await ai_control.acquire_paid_provider_canary_lease(owner) is True
|
|
assert await ai_control.promote_paid_provider_canary_activation(
|
|
owner,
|
|
completion_record_id="row-4",
|
|
) is False
|
|
assert await ai_control.rollback_paid_provider_canary_activation(owner) is True
|
|
assert len(redis.eval_calls) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_paid_canary_execution_claim_is_atomic_and_owner_released(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
redis = _ControlRedis(
|
|
eval_results=[
|
|
[1, "execution_claim_acquired"],
|
|
[0, "execution_claim_held"],
|
|
[0, "execution_claim_owner_mismatch"],
|
|
[1, "execution_claim_released"],
|
|
]
|
|
)
|
|
|
|
async def _redis() -> _ControlRedis:
|
|
return redis
|
|
|
|
monkeypatch.setattr(ai_control, "_get_redis", _redis)
|
|
run_id = "267a3d26-3c29-470f-8fe3-388a2f408f39"
|
|
owner = f"{run_id}.claim-123456789abc"
|
|
|
|
assert (
|
|
await ai_control.acquire_paid_provider_canary_execution_claim(
|
|
run_id,
|
|
owner,
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
await ai_control.acquire_paid_provider_canary_execution_claim(
|
|
run_id,
|
|
f"{run_id}.claim-deadbeef0000",
|
|
)
|
|
is False
|
|
)
|
|
assert (
|
|
await ai_control.release_paid_provider_canary_execution_claim(
|
|
run_id,
|
|
f"{run_id}.claim-deadbeef0000",
|
|
)
|
|
is False
|
|
)
|
|
assert (
|
|
await ai_control.release_paid_provider_canary_execution_claim(run_id, owner)
|
|
is True
|
|
)
|
|
claim_keys = [call[2] for call in redis.eval_calls]
|
|
assert len(set(claim_keys)) == 1
|
|
assert str(claim_keys[0]).startswith(
|
|
"ai:control:paid_canary:execution_claim:"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"admission_result",
|
|
[
|
|
[0, "paired_paid_state_not_enabled", ""],
|
|
[0, "paid_activation_receipt_missing", ""],
|
|
],
|
|
)
|
|
async def test_stale_single_or_paired_legacy_zero_keys_fail_closed(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
admission_result: list[object],
|
|
) -> None:
|
|
redis = _ControlRedis(eval_results=[admission_result])
|
|
|
|
async def _redis() -> _ControlRedis:
|
|
return redis
|
|
|
|
monkeypatch.setattr(ai_control, "_get_redis", _redis)
|
|
|
|
assert await ai_control.is_provider_disabled(
|
|
"claude",
|
|
run_id="AIA-SRE-013-run",
|
|
) is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_persistent_pair_is_never_admitted(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
owner = "AIA-SRE-013-run.attempt-123456789abc"
|
|
redis = _ControlRedis(
|
|
eval_results=[
|
|
[0, "legacy_persistent_activation_blocked", f"{owner}|row-4"]
|
|
]
|
|
)
|
|
|
|
async def _redis() -> _ControlRedis:
|
|
return redis
|
|
|
|
monkeypatch.setattr(ai_control, "_get_redis", _redis)
|
|
|
|
assert await ai_control.is_provider_disabled("gemini") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"receipt",
|
|
[
|
|
"AIA-SRE-013-run.attempt-123456789abc",
|
|
"AIA-SRE-013-run.attempt-123456789abc|",
|
|
"|row-4",
|
|
"AIA-SRE-013-run.attempt-123456789abc|row-4|extra",
|
|
],
|
|
)
|
|
async def test_malformed_persistent_activation_receipt_fails_closed(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
receipt: str,
|
|
) -> None:
|
|
redis = _ControlRedis(
|
|
eval_results=[[1, "persistent_paid_activation_admitted", receipt]]
|
|
)
|
|
|
|
async def _redis() -> _ControlRedis:
|
|
return redis
|
|
|
|
monkeypatch.setattr(ai_control, "_get_redis", _redis)
|
|
|
|
assert await ai_control.is_provider_disabled("claude") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_persistent_readback_requires_exact_completion_record(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
owner = "AIA-SRE-013-run.attempt-123456789abc"
|
|
atomic_state = ["", -2, "0", -1, "0", -1, f"{owner}|row-4"]
|
|
redis = _ControlRedis(eval_results=[atomic_state, atomic_state])
|
|
|
|
async def _redis() -> _ControlRedis:
|
|
return redis
|
|
|
|
monkeypatch.setattr(ai_control, "_get_redis", _redis)
|
|
|
|
matching = await ai_control.get_paid_provider_canary_control_readback(
|
|
expected_owner=owner,
|
|
expected_completion_record_id="row-4",
|
|
)
|
|
mismatching = await ai_control.get_paid_provider_canary_control_readback(
|
|
expected_owner=owner,
|
|
expected_completion_record_id="row-5",
|
|
)
|
|
|
|
assert matching["persistent_enabled"] is False
|
|
assert matching["legacy_persistent_state_detected"] is True
|
|
assert matching["active_completion_receipt_matches"] is True
|
|
assert mismatching["persistent_enabled"] is False
|
|
assert mismatching["active_completion_receipt_matches"] is False
|
|
|
|
|
|
@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
|