feat(sre): enforce typed controlled automation
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

This commit is contained in:
ogt
2026-07-16 17:32:01 +08:00
parent dfec0c9f7f
commit 541a32a9cb
130 changed files with 14849 additions and 1738 deletions

View File

@@ -45,15 +45,25 @@ class _FakeRoute:
class _FakeFailoverManager:
def __init__(self) -> None:
self.task_types: list[str] = []
self.calls: list[tuple[str, bool]] = []
async def select_provider(self, task_type: str = "general") -> _FakeRoute:
self.task_types.append(task_type)
async def select_provider(
self,
task_type: str = "general",
*,
emit_side_effects: bool = False,
) -> _FakeRoute:
self.calls.append((task_type, emit_side_effects))
return _FakeRoute()
class _UnorderedFailoverManager:
async def select_provider(self, task_type: str = "general") -> SimpleNamespace:
async def select_provider(
self,
task_type: str = "general",
*,
emit_side_effects: bool = False,
) -> SimpleNamespace:
return SimpleNamespace(
all_endpoints_in_order=lambda: [
_FakeEndpoint("ollama_local"),
@@ -170,7 +180,7 @@ async def test_alert_context_uses_exact_ollama_lane_then_gemini(
"ollama_local",
"gemini",
]
assert fake_failover.task_types == ["alert_fast"]
assert fake_failover.calls == [("alert_fast", True)]
assert fake_router.contexts[0]["intent_hint"] == "alert_triage"
assert fake_executor.context is not None
assert fake_executor.context["task_type"] == "alert_fast"
@@ -277,7 +287,7 @@ async def test_explicit_ai_governance_context_uses_ollama_first(
"ollama_local",
"gemini",
]
assert fake_failover.task_types == ["diagnose"]
assert fake_failover.calls == [("diagnose", True)]
@pytest.mark.asyncio
@@ -297,7 +307,7 @@ async def test_alert_context_uses_gcp_a_gcp_b_then_111_order(
)
assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"]
assert fake_failover.task_types == ["alert_fast"]
assert fake_failover.calls == [("alert_fast", True)]
@pytest.mark.asyncio