feat: add advisor rollout gate and rollback playbook

tests/unit/service/test_advisor_rollout_service.py

@@ -0,0 +1,85 @@
+from types import SimpleNamespace
+
+import pytest
+
+from app.core.contracts import RequestContext
+from app.core.errors import ForbiddenAgentError
+from app.service.advisor_rollout_service import AdvisorRolloutService
+
+
+def context(
+    *, roles: tuple[str, ...] = ("customer",), customer_ids: tuple[str, ...] = ()
+) -> RequestContext:
+    return RequestContext(
+        user_id="9001", trace_id="rollout-test", roles=roles,
+        customer_ids=customer_ids,
+    )
+
+
+def settings(*, enabled: bool, customer_ids: str) -> SimpleNamespace:
+    return SimpleNamespace(
+        advisor_rollout_enabled=enabled,
+        advisor_rollout_customer_ids=customer_ids,
+    )
+
+
+def test_disabled_rollout_allows_everyone(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "app.service.advisor_rollout_service.get_settings",
+        lambda: settings(enabled=False, customer_ids=""),
+    )
+    assert AdvisorRolloutService.is_allowed(context())
+
+
+@pytest.mark.parametrize(
+    ("roles", "customer_ids", "allowed"),
+    [
+        (("customer",), (), True),
+        (("customer",), (), False),
+        (("admin",), (), True),
+        (("super_admin",), (), True),
+    ],
+)
+def test_enabled_rollout_uses_customer_whitelist_and_admin_bypass(
+    monkeypatch: pytest.MonkeyPatch,
+    roles: tuple[str, ...],
+    customer_ids: tuple[str, ...],
+    allowed: bool,
+) -> None:
+    whitelist = "9001" if roles == ("customer",) and not customer_ids else " 9002 , 9003 "
+    if roles == ("customer",) and not customer_ids and allowed is False:
+        whitelist = "9002"
+    monkeypatch.setattr(
+        "app.service.advisor_rollout_service.get_settings",
+        lambda: settings(enabled=True, customer_ids=whitelist),
+    )
+    actual = AdvisorRolloutService.is_allowed(context(roles=roles, customer_ids=customer_ids))
+    assert actual is allowed
+
+
+class FakeSession:
+    def __init__(self) -> None:
+        self.added: list[object] = []
+        self.commits = 0
+
+    def add(self, value: object) -> None:
+        self.added.append(value)
+
+    async def commit(self) -> None:
+        self.commits += 1
+
+
+async def test_denial_is_audited_and_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "app.service.advisor_rollout_service.get_settings",
+        lambda: settings(enabled=True, customer_ids="9002"),
+    )
+    session = FakeSession()
+
+    with pytest.raises(ForbiddenAgentError, match="尚未对该账号开放"):
+        await AdvisorRolloutService(session).ensure_allowed(context())  # type: ignore[arg-type]
+
+    assert session.commits == 1
+    audit = session.added[0]
+    assert audit.action_type == "advisor.rollout_denied"  # type: ignore[attr-defined]
+    assert audit.detail["reason"] == "not_in_rollout"  # type: ignore[attr-defined]