671 lines
24 KiB
Python
671 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from structlog.testing import capture_logs
|
|
|
|
from src.services import ai_router as ai_router_module
|
|
from src.services import openclaw as openclaw_module
|
|
from src.services.ai_providers.interfaces import AIResult
|
|
from src.services.ai_router import AIProviderRegistry, AIRouterExecutor
|
|
from src.services.openclaw import OpenClawService, build_paid_cloud_alert_context
|
|
|
|
|
|
class _Redis:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
cached_provider: str | None = None,
|
|
paid_disabled: bool = False,
|
|
durable_receipts: dict[str, dict[str, Any]] | None = None,
|
|
) -> None:
|
|
self.cached_provider = cached_provider
|
|
self.paid_disabled = paid_disabled
|
|
self.durable_receipts = durable_receipts or {}
|
|
self.set_calls: list[tuple[str, str, int | None]] = []
|
|
|
|
async def get(self, key: str) -> str | None:
|
|
if key.startswith("ai:control:disabled:"):
|
|
return "1" if self.paid_disabled else "0"
|
|
if key.startswith("ai:ollama:unavailable_receipt:"):
|
|
receipt_id = key.removeprefix("ai:ollama:unavailable_receipt:")
|
|
receipt = self.durable_receipts.get(receipt_id)
|
|
return json.dumps(receipt) if receipt else None
|
|
if self.cached_provider:
|
|
return json.dumps(
|
|
{
|
|
"provider": self.cached_provider,
|
|
"response": '{"cached":"must-not-run"}',
|
|
}
|
|
)
|
|
return None
|
|
|
|
async def set(self, key: str, value: str, ex: int | None = None) -> None:
|
|
self.set_calls.append((key, value, ex))
|
|
|
|
|
|
class _Provider:
|
|
is_enabled = True
|
|
capabilities = {"rca", "chat"}
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
*,
|
|
success: bool,
|
|
privacy_level: str,
|
|
attempt_order: list[str] | None = None,
|
|
failure_response: str = "",
|
|
failure_error: str = "bounded failure",
|
|
) -> None:
|
|
self.name = name
|
|
self.success = success
|
|
self.privacy_level = privacy_level
|
|
self.attempt_order = attempt_order
|
|
self.failure_response = failure_response
|
|
self.failure_error = failure_error
|
|
self.calls: list[tuple[str, dict[str, Any] | None]] = []
|
|
|
|
async def analyze(
|
|
self,
|
|
prompt: str,
|
|
context: dict[str, Any] | None = None,
|
|
) -> AIResult:
|
|
self.calls.append((prompt, context))
|
|
if self.attempt_order is not None:
|
|
self.attempt_order.append(self.name)
|
|
return AIResult(
|
|
raw_response=(
|
|
'{"action_title":"safe","description":"safe",'
|
|
'"suggested_action":"NO_ACTION","kubectl_command":"",'
|
|
'"target_resource":"api","namespace":"awoooi-prod",'
|
|
'"risk_level":"low","blast_radius":{"affected_pods":0,'
|
|
'"estimated_downtime":"0","related_services":[],'
|
|
'"data_impact":"NONE"},"reasoning":"safe","confidence":0.9}'
|
|
if self.success
|
|
else self.failure_response
|
|
),
|
|
success=self.success,
|
|
provider=self.name,
|
|
error=None if self.success else self.failure_error,
|
|
tokens=7,
|
|
audit_metadata={"generation_receipt_id": "paid-gen-1"},
|
|
)
|
|
|
|
|
|
class _Trace:
|
|
def __init__(self) -> None:
|
|
self.langfuse_trace_id = "trace-test"
|
|
self.generations: list[dict[str, Any]] = []
|
|
self.root_metadata: dict[str, Any] = {}
|
|
|
|
def __enter__(self) -> _Trace:
|
|
return self
|
|
|
|
def __exit__(self, *_args: Any) -> None:
|
|
return None
|
|
|
|
def generation(self, **kwargs: Any) -> None:
|
|
self.generations.append(kwargs)
|
|
|
|
def score(self, **_kwargs: Any) -> None:
|
|
return None
|
|
|
|
|
|
def _trace_factory(trace: _Trace):
|
|
def _factory(*_args: Any, **kwargs: Any) -> _Trace:
|
|
trace.root_metadata = dict(kwargs.get("metadata") or {})
|
|
return trace
|
|
|
|
return _factory
|
|
|
|
|
|
def _raw_alert() -> dict[str, Any]:
|
|
return {
|
|
"alert_type": "DockerContainerUnhealthy",
|
|
"severity": "warning",
|
|
"source": "alertmanager",
|
|
"target_resource": "192.168.0.110",
|
|
"namespace": "awoooi-prod",
|
|
"fingerprint": "fp-1",
|
|
"message": "Authorization: Bearer raw-secret-must-never-reach-cloud",
|
|
"labels": {"token": "raw-label-secret"},
|
|
"annotations": {"description": "raw annotation"},
|
|
"raw_logs": "raw-log-secret",
|
|
"headers": {"authorization": "raw-header-secret"},
|
|
}
|
|
|
|
|
|
def _prepared_context() -> dict[str, Any]:
|
|
service = object.__new__(OpenClawService)
|
|
prepared = service._prepare_paid_cloud_route_context(_raw_alert())
|
|
assert prepared is not None
|
|
prepared.update(
|
|
trace_id="trace-1",
|
|
run_id="run-1",
|
|
work_item_id="work-1",
|
|
alert_requires_ollama_before_cloud=True,
|
|
)
|
|
return prepared
|
|
|
|
|
|
def test_cloud_sanitizer_attests_only_allowlisted_structured_fields() -> None:
|
|
cloud_context, reason = build_paid_cloud_alert_context(_raw_alert())
|
|
|
|
assert reason == "verified_allowlisted_sanitization_receipt"
|
|
assert cloud_context is not None
|
|
serialized = json.dumps(cloud_context, ensure_ascii=False)
|
|
assert "raw-secret-must-never-reach-cloud" not in serialized
|
|
assert "raw-label-secret" not in serialized
|
|
assert "raw annotation" not in serialized
|
|
assert "raw-log-secret" not in serialized
|
|
assert "raw-header-secret" not in serialized
|
|
assert "192.168.0.110" not in serialized
|
|
assert "[PRIVATE_IP_REDACTED]" in serialized
|
|
receipt = cloud_context["cloud_sanitization_receipt"]
|
|
assert receipt["status"] == "verified"
|
|
assert receipt["raw_payload_forwarded"] is False
|
|
assert receipt["raw_log_payload_forwarded"] is False
|
|
assert receipt["secret_value_exposed"] is False
|
|
assert receipt["excluded_field_count"] == 5
|
|
|
|
|
|
def test_cloud_sanitizer_redacts_bare_keys_and_all_private_network_ranges() -> None:
|
|
alert = {
|
|
"alert_type": "HostKeyLeak-sk-ant-api03-ABCDEFGHIJKLMNOPQRSTUV",
|
|
"source": "AIzaSyABCDEFGHIJKLMNOPQRSTUVWXY123456",
|
|
"target_resource": "10.21.4.8 and 172.20.1.4",
|
|
"namespace": "[fd00::1234] and fe80::1",
|
|
"fingerprint": "Authorization: Bearer naked-secret-value",
|
|
}
|
|
|
|
cloud_context, reason = build_paid_cloud_alert_context(alert)
|
|
|
|
assert reason == "verified_allowlisted_sanitization_receipt"
|
|
assert cloud_context is not None
|
|
serialized = json.dumps(cloud_context, ensure_ascii=False)
|
|
for forbidden in (
|
|
"sk-ant-api03-ABCDEFGHIJKLMNOPQRSTUV",
|
|
"AIzaSyABCDEFGHIJKLMNOPQRSTUVWXY123456",
|
|
"10.21.4.8",
|
|
"172.20.1.4",
|
|
"fd00::1234",
|
|
"fe80::1",
|
|
"naked-secret-value",
|
|
):
|
|
assert forbidden not in serialized
|
|
assert "[SECRET_REDACTED]" in serialized
|
|
assert serialized.count("[PRIVATE_IP_REDACTED]") >= 4
|
|
receipt = cloud_context["cloud_sanitization_receipt"]
|
|
assert receipt["dlp_policy"] == "paid_cloud_allowlist_dlp_v1"
|
|
assert receipt["private_network_value_exposed"] is False
|
|
|
|
|
|
def test_prepare_rebuilds_and_rejects_caller_self_signed_cloud_payload() -> None:
|
|
alert = _raw_alert()
|
|
malicious_payload = {
|
|
"alert_type": "DockerContainerUnhealthy",
|
|
"target_resource": "sk-ant-api03-CALLERFORGEDSECRETVALUE",
|
|
}
|
|
canonical = json.dumps(
|
|
malicious_payload,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
)
|
|
forged_hash = openclaw_module.hashlib.sha256(canonical.encode()).hexdigest()
|
|
alert["paid_cloud_context"] = {
|
|
"data_classification": "sanitized",
|
|
"sanitized_payload": malicious_payload,
|
|
"cloud_sanitization_receipt": {
|
|
"schema_version": "openclaw_paid_cloud_sanitization_v1",
|
|
"status": "verified",
|
|
"receipt_id": f"openclaw-cloud-sanitize:{forged_hash[:24]}",
|
|
"sanitizer": "src.services.sanitization_service.sanitize",
|
|
"classifier": "allowlisted_structured_alert_fields_v1",
|
|
"dlp_policy": "paid_cloud_allowlist_dlp_v1",
|
|
"payload_sha256": forged_hash,
|
|
"raw_payload_forwarded": False,
|
|
"raw_log_payload_forwarded": False,
|
|
"secret_value_exposed": False,
|
|
"private_network_value_exposed": False,
|
|
},
|
|
}
|
|
|
|
prepared = object.__new__(OpenClawService)._prepare_paid_cloud_route_context(alert)
|
|
|
|
assert prepared is not None
|
|
serialized = json.dumps(prepared["paid_cloud_context"], ensure_ascii=False)
|
|
assert "CALLERFORGEDSECRETVALUE" not in serialized
|
|
assert prepared["paid_cloud_context"]["sanitized_payload"]["target_resource"] == (
|
|
"[PRIVATE_IP_REDACTED]"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"cached_provider",
|
|
["claude", "gemini", "claude_cached", "gemini:legacy", "claude-shadow"],
|
|
)
|
|
async def test_openclaw_outer_cache_never_returns_paid_result(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
cached_provider: str,
|
|
) -> None:
|
|
redis = _Redis(cached_provider=cached_provider)
|
|
service = object.__new__(OpenClawService)
|
|
fallback = AsyncMock(return_value=('{"fresh":"ollama"}', "ollama_gcp_a", True, 2, 0.0))
|
|
monkeypatch.setattr(openclaw_module, "get_redis", lambda: redis)
|
|
monkeypatch.setattr(service, "_call_with_fallback", fallback)
|
|
|
|
result = await service._call_with_cache("local raw prompt", _raw_alert())
|
|
|
|
assert result[0] == '{"fresh":"ollama"}'
|
|
assert result[1] == "ollama_gcp_a"
|
|
assert result[3] is False
|
|
fallback.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"provider",
|
|
["claude", "gemini", "claude_cached", "gemini:legacy", "claude-shadow"],
|
|
)
|
|
async def test_openclaw_outer_cache_never_writes_paid_result(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provider: str,
|
|
) -> None:
|
|
redis = _Redis()
|
|
service = object.__new__(OpenClawService)
|
|
fallback = AsyncMock(return_value=('{"paid":"response"}', provider, True, 2, 0.01))
|
|
monkeypatch.setattr(openclaw_module, "get_redis", lambda: redis)
|
|
monkeypatch.setattr(service, "_call_with_fallback", fallback)
|
|
|
|
result = await service._call_with_cache("local raw prompt", _raw_alert())
|
|
|
|
assert result[1] == provider
|
|
assert result[3] is False
|
|
assert redis.set_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_actual_executor_uses_original_prompt_for_ollama_and_sanitized_prompt_for_paid(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
redis = _Redis()
|
|
trace = _Trace()
|
|
attempt_order: list[str] = []
|
|
gcp_a = _Provider(
|
|
"ollama_gcp_a",
|
|
success=False,
|
|
privacy_level="local",
|
|
attempt_order=attempt_order,
|
|
)
|
|
gcp_b = _Provider(
|
|
"ollama_gcp_b",
|
|
success=False,
|
|
privacy_level="local",
|
|
attempt_order=attempt_order,
|
|
)
|
|
local = _Provider(
|
|
"ollama_local",
|
|
success=False,
|
|
privacy_level="local",
|
|
attempt_order=attempt_order,
|
|
)
|
|
claude = _Provider(
|
|
"claude",
|
|
success=False,
|
|
privacy_level="cloud",
|
|
attempt_order=attempt_order,
|
|
failure_response="paid-claude-response-secret",
|
|
failure_error="paid-claude-error-secret",
|
|
)
|
|
gemini = _Provider(
|
|
"gemini",
|
|
success=True,
|
|
privacy_level="cloud",
|
|
attempt_order=attempt_order,
|
|
)
|
|
registry = AIProviderRegistry()
|
|
for provider in (gcp_a, gcp_b, local, claude, gemini):
|
|
registry.register(provider)
|
|
|
|
monkeypatch.setenv("ENABLE_GEMINI", "true")
|
|
monkeypatch.setenv("ENABLE_CLAUDE", "true")
|
|
monkeypatch.setattr(ai_router_module._settings, "MOCK_MODE", False)
|
|
monkeypatch.setattr("src.core.redis_client.get_redis", lambda: redis)
|
|
monkeypatch.setattr(
|
|
"src.services.langfuse_client.langfuse_trace",
|
|
_trace_factory(trace),
|
|
)
|
|
original_prompt = "LOCAL ONLY raw-secret-must-never-reach-cloud"
|
|
|
|
with capture_logs() as captured_logs:
|
|
result = await AIRouterExecutor(registry).execute(
|
|
prompt=original_prompt,
|
|
provider_order=[
|
|
"ollama_gcp_a",
|
|
"ollama_gcp_b",
|
|
"ollama_local",
|
|
"claude",
|
|
"gemini",
|
|
],
|
|
context=_prepared_context(),
|
|
)
|
|
|
|
assert result.success is True
|
|
assert result.provider == "gemini"
|
|
assert attempt_order == [
|
|
"ollama_gcp_a",
|
|
"ollama_gcp_b",
|
|
"ollama_local",
|
|
"claude",
|
|
"gemini",
|
|
]
|
|
assert [provider.calls[0][0] for provider in (gcp_a, gcp_b, local)] == [
|
|
original_prompt,
|
|
original_prompt,
|
|
original_prompt,
|
|
]
|
|
paid_prompt, paid_context = gemini.calls[0]
|
|
assert paid_prompt != original_prompt
|
|
assert "raw-secret-must-never-reach-cloud" not in paid_prompt
|
|
assert paid_context is not None
|
|
assert "message" not in paid_context
|
|
assert "labels" not in paid_context
|
|
assert "annotations" not in paid_context
|
|
assert "raw_logs" not in paid_context
|
|
assert "headers" not in paid_context
|
|
claude_prompt, claude_context = claude.calls[0]
|
|
assert claude_prompt == paid_prompt
|
|
assert claude_context == paid_context
|
|
paid_trace = next(item for item in trace.generations if item["name"] == "gemini_call")
|
|
assert paid_trace["input"] is None
|
|
assert paid_trace["output"] is None
|
|
assert paid_trace["metadata"]["content_redacted"] is True
|
|
assert paid_trace["metadata"]["sanitization_receipt_id"]
|
|
assert "raw-secret-must-never-reach-cloud" not in json.dumps(paid_trace)
|
|
assert "DockerContainerUnhealthy" not in json.dumps(trace.root_metadata)
|
|
serialized_logs = json.dumps(captured_logs, ensure_ascii=False)
|
|
assert "paid-claude-response-secret" not in serialized_logs
|
|
assert "paid-claude-error-secret" not in serialized_logs
|
|
assert not redis.set_calls
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cloud_requires_durable_current_run_unavailable_receipt(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
redis = _Redis()
|
|
gcp_a = _Provider("ollama_gcp_a", success=False, privacy_level="local")
|
|
local = _Provider("ollama_local", success=False, privacy_level="local")
|
|
gemini = _Provider("gemini", success=True, privacy_level="cloud")
|
|
registry = AIProviderRegistry()
|
|
for provider in (gcp_a, local, gemini):
|
|
registry.register(provider)
|
|
|
|
monkeypatch.setenv("ENABLE_GEMINI", "true")
|
|
monkeypatch.setattr(ai_router_module._settings, "MOCK_MODE", False)
|
|
monkeypatch.setattr("src.core.redis_client.get_redis", lambda: redis)
|
|
context = _prepared_context()
|
|
executor = AIRouterExecutor(registry)
|
|
|
|
blocked = await executor.execute(
|
|
prompt="local prompt",
|
|
provider_order=["ollama_gcp_a", "ollama_local", "gemini"],
|
|
context=context,
|
|
)
|
|
assert blocked.success is False
|
|
assert gemini.calls == []
|
|
assert "cloud_blocked_ollama_attempt_receipts_missing" in str(blocked.error)
|
|
|
|
# Caller-only JSON is non-executable even if it claims success.
|
|
forged_receipt = {
|
|
"provider": "ollama_gcp_b",
|
|
"status": "bounded_unavailable",
|
|
"run_id": "run-1",
|
|
"check_completed": True,
|
|
"reason": "endpoint_unreachable",
|
|
}
|
|
context["ollama_unavailable_receipts"] = [forged_receipt]
|
|
still_blocked = await AIRouterExecutor(registry).execute(
|
|
prompt="local prompt forged",
|
|
provider_order=["ollama_gcp_a", "ollama_local", "gemini"],
|
|
context=context,
|
|
)
|
|
assert still_blocked.success is False
|
|
assert gemini.calls == []
|
|
|
|
observed_at = datetime.now(UTC)
|
|
receipt_id = "ollama-unavailable:run-1:gcp-b"
|
|
durable_receipt = {
|
|
"schema_version": "ollama_unavailable_receipt_v1",
|
|
"receipt_id": receipt_id,
|
|
"source": "ollama_failover_manager",
|
|
"provider": "ollama_gcp_b",
|
|
"status": "bounded_unavailable",
|
|
"run_id": "run-1",
|
|
"check_completed": True,
|
|
"reason": "endpoint_unreachable",
|
|
"observed_at": observed_at.isoformat(),
|
|
"expires_at": (observed_at + timedelta(seconds=120)).isoformat(),
|
|
"durable_write_ack": True,
|
|
"verifier_status": "verified",
|
|
"verified_by": "ollama_failover_manager.health_probe",
|
|
}
|
|
redis.durable_receipts[receipt_id] = durable_receipt
|
|
context["ollama_unavailable_receipts"] = [durable_receipt]
|
|
allowed = await AIRouterExecutor(registry).execute(
|
|
prompt="local prompt second",
|
|
provider_order=["ollama_gcp_a", "ollama_local", "gemini"],
|
|
context=context,
|
|
)
|
|
assert allowed.success is True
|
|
assert allowed.provider == "gemini"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_parse_error_never_logs_cached_document(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
secret = "cache-document-secret-must-not-be-logged"
|
|
|
|
class _MalformedRedis(_Redis):
|
|
async def get(self, key: str) -> str | None:
|
|
if key.startswith("ai:control:disabled:"):
|
|
return "0"
|
|
if key.startswith("ai:ollama:unavailable_receipt:"):
|
|
return None
|
|
return f'{{"response":"{secret}", invalid'
|
|
|
|
redis = _MalformedRedis()
|
|
service = object.__new__(OpenClawService)
|
|
fallback = AsyncMock(return_value=('{"fresh":"ollama"}', "ollama_gcp_a", True, 2, 0.0))
|
|
monkeypatch.setattr(openclaw_module, "get_redis", lambda: redis)
|
|
monkeypatch.setattr(service, "_call_with_fallback", fallback)
|
|
|
|
with capture_logs() as captured_logs:
|
|
await service._call_with_cache("local prompt", _raw_alert())
|
|
|
|
serialized_logs = json.dumps(captured_logs, ensure_ascii=False)
|
|
assert secret not in serialized_logs
|
|
assert "JSONDecodeError" in serialized_logs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_actual_executor_cache_parse_error_is_content_bounded(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
secret = "airouter-cache-document-secret-must-not-be-logged"
|
|
|
|
class _MalformedRedis(_Redis):
|
|
async def get(self, key: str) -> str | None:
|
|
if key.startswith("ai:control:disabled:"):
|
|
return "0"
|
|
if key.startswith("ai:ollama:unavailable_receipt:"):
|
|
return None
|
|
return f'{{"response":"{secret}", invalid'
|
|
|
|
provider = _Provider("ollama_gcp_a", success=True, privacy_level="local")
|
|
registry = AIProviderRegistry()
|
|
registry.register(provider)
|
|
monkeypatch.setattr(ai_router_module._settings, "MOCK_MODE", False)
|
|
monkeypatch.setattr(
|
|
"src.core.redis_client.get_redis",
|
|
lambda: _MalformedRedis(),
|
|
)
|
|
|
|
with capture_logs() as captured_logs:
|
|
result = await AIRouterExecutor(registry).execute(
|
|
prompt="local prompt",
|
|
provider_order=["ollama_gcp_a"],
|
|
context={"intent_hint": "diagnose"},
|
|
)
|
|
|
|
assert result.success is True
|
|
serialized_logs = json.dumps(captured_logs, ensure_ascii=False)
|
|
assert secret not in serialized_logs
|
|
assert "JSONDecodeError" in serialized_logs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_actual_executor_rejects_self_signed_cloud_receipt(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
redis = _Redis()
|
|
providers = [
|
|
_Provider("ollama_gcp_a", success=False, privacy_level="local"),
|
|
_Provider("ollama_gcp_b", success=False, privacy_level="local"),
|
|
_Provider("ollama_local", success=False, privacy_level="local"),
|
|
_Provider("gemini", success=True, privacy_level="cloud"),
|
|
]
|
|
registry = AIProviderRegistry()
|
|
for provider in providers:
|
|
registry.register(provider)
|
|
context = _prepared_context()
|
|
cloud_context = context["paid_cloud_context"]
|
|
cloud_context["sanitized_payload"]["target_resource"] = (
|
|
"sk-ant-api03-CALLERFORGEDSECRETVALUE"
|
|
)
|
|
canonical = json.dumps(
|
|
cloud_context["sanitized_payload"],
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
)
|
|
forged_hash = openclaw_module.hashlib.sha256(canonical.encode()).hexdigest()
|
|
receipt = cloud_context["cloud_sanitization_receipt"]
|
|
receipt["payload_sha256"] = forged_hash
|
|
receipt["receipt_id"] = f"openclaw-cloud-sanitize:{forged_hash[:24]}"
|
|
|
|
monkeypatch.setenv("ENABLE_GEMINI", "true")
|
|
monkeypatch.setattr(ai_router_module._settings, "MOCK_MODE", False)
|
|
monkeypatch.setattr("src.core.redis_client.get_redis", lambda: redis)
|
|
result = await AIRouterExecutor(registry).execute(
|
|
prompt="local prompt",
|
|
provider_order=[
|
|
"ollama_gcp_a",
|
|
"ollama_gcp_b",
|
|
"ollama_local",
|
|
"gemini",
|
|
],
|
|
context=context,
|
|
)
|
|
|
|
assert result.success is False
|
|
assert providers[-1].calls == []
|
|
assert "alert_cloud_sanitization_receipt_untrusted" in str(result.error)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ollama_circuit_open_cannot_slide_to_paid_cloud(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
redis = _Redis()
|
|
providers = [
|
|
_Provider("ollama_gcp_a", success=False, privacy_level="local"),
|
|
_Provider("ollama_gcp_b", success=False, privacy_level="local"),
|
|
_Provider("ollama_local", success=False, privacy_level="local"),
|
|
_Provider("gemini", success=True, privacy_level="cloud"),
|
|
]
|
|
registry = AIProviderRegistry()
|
|
for provider in providers:
|
|
registry.register(provider)
|
|
executor = AIRouterExecutor(registry)
|
|
circuit = executor._get_circuit_breaker("ollama_gcp_a")
|
|
circuit._failure_count = circuit._failure_threshold
|
|
circuit._last_failure_time = 10**12
|
|
|
|
monkeypatch.setenv("ENABLE_GEMINI", "true")
|
|
monkeypatch.setattr(ai_router_module._settings, "MOCK_MODE", False)
|
|
monkeypatch.setattr("src.core.redis_client.get_redis", lambda: redis)
|
|
result = await executor.execute(
|
|
prompt="local prompt",
|
|
provider_order=["ollama_gcp_a", "ollama_gcp_b", "ollama_local", "gemini"],
|
|
context=_prepared_context(),
|
|
)
|
|
|
|
assert result.success is False
|
|
assert providers[-1].calls == []
|
|
assert "cloud_blocked_ollama_circuit_open" in str(result.error)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_paid_langfuse_generation_contains_only_hashes_and_receipt(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
trace = _Trace()
|
|
service = object.__new__(OpenClawService)
|
|
monkeypatch.setattr(openclaw_module.settings, "USE_AI_ROUTER", False)
|
|
monkeypatch.setattr(openclaw_module.settings, "MOCK_MODE", False)
|
|
monkeypatch.setattr(
|
|
openclaw_module,
|
|
"langfuse_trace",
|
|
_trace_factory(trace),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_resolve_alert_provider_order",
|
|
AsyncMock(
|
|
return_value=["ollama_gcp_a", "ollama_gcp_b", "ollama_local", "gemini"]
|
|
),
|
|
)
|
|
monkeypatch.setattr(service, "_call_ollama", AsyncMock(return_value=("local failed", False)))
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_call_gemini",
|
|
AsyncMock(return_value=("paid-response-secret", True, 3, 0.01)),
|
|
)
|
|
monkeypatch.setattr(service, "_get_model_name", lambda provider: provider)
|
|
|
|
result = await service._call_with_fallback(
|
|
"LOCAL raw-secret-must-never-reach-cloud",
|
|
alert_context=_raw_alert(),
|
|
)
|
|
|
|
assert result[1] == "gemini"
|
|
paid_trace = next(item for item in trace.generations if item["name"] == "gemini_call")
|
|
assert paid_trace["input"] is None
|
|
assert paid_trace["output"] is None
|
|
serialized = json.dumps(paid_trace)
|
|
assert "raw-secret-must-never-reach-cloud" not in serialized
|
|
assert "paid-response-secret" not in serialized
|
|
assert paid_trace["metadata"]["prompt_sha256"]
|
|
assert paid_trace["metadata"]["response_sha256"]
|
|
assert "fp-1" not in json.dumps(trace.root_metadata)
|
|
|
|
|
|
def test_parse_failure_logs_hash_and_length_not_raw_response(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
service = object.__new__(OpenClawService)
|
|
secret = "raw-parse-secret-must-not-be-logged"
|
|
|
|
assert service._parse_analysis_result(secret) is None
|
|
|
|
assert secret not in caplog.text
|