ss-tools/backend/tests/api/test_scenario_live_bindings_api.py

# #region Test.Api.ScenarioLiveBindings [C:3] [TYPE Module] [SEMANTICS test,api,scenario,execution,live-binding,admin]
# @BRIEF Admin CRUD for deployment live bindings via an isolated FastAPI app + fake ConfigManager.
# @RELATION BINDS_TO -> [Api.ScenarioLiveBindings]
# @RELATION BINDS_TO -> [ScenarioExecution.Runner.ConfiguredBinding]
# @TEST_CONTRACT: GET/PUT/PATCH /api/scenario-live-bindings -> validated settings.scenario_live_execution_bindings
# @TEST_EDGE: invalid_snapshot -> 422 before persistence
# @TEST_EDGE: binding_ref_mismatch -> 422
# @TEST_EDGE: unknown_toggle -> 404
# @TEST_INVARIANT Api.ScenarioLiveBindings: a registered, enabled binding is resolved server-side by
#                 ScenarioExecution.Runner.ConfiguredBinding; disabled/unregistered is not.
#                 -> VERIFIED_BY: test_registered_binding_resolves_server_side
# @TEST_INVARIANT Api.ScenarioLiveBindings: non-admin is denied 403 by the real has_permission checker.
#                 -> VERIFIED_BY: test_non_admin_denied_403
from __future__ import annotations

from fastapi import FastAPI
from fastapi.testclient import TestClient

from src.api.routes.dashboard_testing.scenario_live_bindings import router
from src.core.config_models import GlobalSettings
from src.dependencies import get_config_manager, get_current_user
from src.models.auth import Role, User

_BINDING_REF = "ss-prod-d11-live-001"
_SNAPSHOT = {
    "binding_ref": _BINDING_REF,
    "environment_id": "env-prod-01",
    "dashboard_release_id": "release-1",
    "release_fingerprint": "f" * 64,
    "dashboard_id": 11,
    "query_model_fingerprint": "q" * 64,
    "execution_principal_fingerprint": "p" * 64,
    "rls_security_fingerprint": "r" * 64,
    "browser_safe_checkpoint_ref": None,
    "browser_action_binding_ref": None,
    "evidence_owner_type": "scenario_run",
    "evidence_ref_policy": "draft_storage_raw_response",
}


_QUERY_MODEL = {
    "schema_version": 1,
    "environment_id": "env-prod-01",
    "dashboard_id": 11,
    "title": "Sales Dashboard",
    "query_model_fingerprint": "q" * 64,
}


class _FakeConfig:
    def __init__(self) -> None:
        self.settings = GlobalSettings()


class _FakeConfigManager:
    def __init__(self) -> None:
        self._config = _FakeConfig()

    def get_config(self):
        return self._config

    def update_global_settings(self, settings):
        self._config.settings = settings
        return self._config


def _admin() -> User:
    role = Role(id="role-admin", name="admin", is_admin=True)
    role.permissions = []
    user = User(id="user-admin", username="admin", email="admin@test.com")
    user.roles = [role]
    return user


def _outsider() -> User:
    role = Role(id="role-viewer", name="viewer", is_admin=False)
    role.permissions = []
    user = User(id="user-viewer", username="viewer", email="viewer@test.com")
    user.roles = [role]
    return user


class _Env:
    def __init__(self) -> None:
        self.config_manager = _FakeConfigManager()
        self.app = FastAPI()
        self.app.include_router(router)
        self.app.dependency_overrides[get_config_manager] = lambda: self.config_manager
        self.client = TestClient(self.app)

    def as_user(self, user: User) -> None:
        self.app.dependency_overrides[get_current_user] = lambda: user


def _put_body(**overrides) -> dict:
    body = {"enabled": True, "binding_snapshot": dict(_SNAPSHOT), "query_model_snapshot": dict(_QUERY_MODEL)}
    body.update(overrides)
    return body


# #region Test.Api.ScenarioLiveBindings.Crud [C:2] [TYPE Function]
def test_list_empty_then_register_and_toggle():
    env = _Env()
    env.as_user(_admin())

    assert env.client.get("/api/scenario-live-bindings").json() == []

    created = env.client.put(f"/api/scenario-live-bindings/{_BINDING_REF}", json=_put_body())
    assert created.status_code == 200
    assert created.json()["binding_ref"] == _BINDING_REF
    assert created.json()["enabled"] is True

    listed = env.client.get("/api/scenario-live-bindings").json()
    assert [item["binding_ref"] for item in listed] == [_BINDING_REF]

    toggled = env.client.patch(f"/api/scenario-live-bindings/{_BINDING_REF}", json={"enabled": False})
    assert toggled.status_code == 200
    assert toggled.json()["enabled"] is False


def test_put_rejects_invalid_snapshot_without_persisting():
    env = _Env()
    env.as_user(_admin())
    bad = {**_SNAPSHOT, "evidence_owner_type": "draft"}
    response = env.client.put(f"/api/scenario-live-bindings/{_BINDING_REF}", json=_put_body(binding_snapshot=bad))
    assert response.status_code == 422
    assert env.config_manager.get_config().settings.scenario_live_execution_bindings == []


def test_put_rejects_binding_ref_mismatch():
    env = _Env()
    env.as_user(_admin())
    response = env.client.put("/api/scenario-live-bindings/other-ref", json=_put_body())
    assert response.status_code == 422
    assert env.config_manager.get_config().settings.scenario_live_execution_bindings == []


def test_put_rejects_invalid_and_mismatched_query_model():
    env = _Env()
    env.as_user(_admin())
    invalid = env.client.put(
        f"/api/scenario-live-bindings/{_BINDING_REF}",
        json=_put_body(query_model_snapshot={"totally": "bogus"}),
    )
    assert invalid.status_code == 422

    mismatched = env.client.put(
        f"/api/scenario-live-bindings/{_BINDING_REF}",
        json=_put_body(query_model_snapshot={**_QUERY_MODEL, "environment_id": "other-env"}),
    )
    assert mismatched.status_code == 422

    fingerprint_mismatch = env.client.put(
        f"/api/scenario-live-bindings/{_BINDING_REF}",
        json=_put_body(query_model_snapshot={**_QUERY_MODEL, "query_model_fingerprint": "z" * 64}),
    )
    assert fingerprint_mismatch.status_code == 422
    assert env.config_manager.get_config().settings.scenario_live_execution_bindings == []


def test_toggle_unknown_binding_is_404():
    env = _Env()
    env.as_user(_admin())
    response = env.client.patch("/api/scenario-live-bindings/missing-ref", json={"enabled": True})
    assert response.status_code == 404


def test_non_admin_denied_403():
    env = _Env()
    env.as_user(_outsider())
    assert env.client.get("/api/scenario-live-bindings").status_code == 403
    assert env.client.put(f"/api/scenario-live-bindings/{_BINDING_REF}", json=_put_body()).status_code == 403
    assert env.client.patch(f"/api/scenario-live-bindings/{_BINDING_REF}", json={"enabled": True}).status_code == 403
# #endregion Test.Api.ScenarioLiveBindings.Crud


# #region Test.Api.ScenarioLiveBindings.Resolve [C:2] [TYPE Function]
def test_registered_binding_resolves_server_side():
    from src.services.dashboard_testing.execution.start_run import _resolve_configured_live_binding

    env = _Env()
    env.as_user(_admin())
    env.client.put(f"/api/scenario-live-bindings/{_BINDING_REF}", json=_put_body())
    plan = {"steps": [{"dashboard_id": 11}]}
    principal = _SNAPSHOT["execution_principal_fingerprint"]

    resolved = _resolve_configured_live_binding(
        environment_id="env-prod-01", plan=plan, principal_fingerprint=principal,
        config_manager=env.config_manager,
    )
    assert resolved is not None
    assert resolved.binding_ref == _BINDING_REF

    # A different principal or environment never adopts the binding.
    assert _resolve_configured_live_binding(
        environment_id="env-prod-01", plan=plan, principal_fingerprint="z" * 64,
        config_manager=env.config_manager,
    ) is None

    env.client.patch(f"/api/scenario-live-bindings/{_BINDING_REF}", json={"enabled": False})
    assert _resolve_configured_live_binding(
        environment_id="env-prod-01", plan=plan, principal_fingerprint=principal,
        config_manager=env.config_manager,
    ) is None
# #endregion Test.Api.ScenarioLiveBindings.Resolve
# #endregion Test.Api.ScenarioLiveBindings