RAGcore-Public/tests/web/profiles/test_profiles_web.py

from __future__ import annotations

from datetime import UTC, datetime
from uuid import uuid4

from fastapi.testclient import TestClient

from ragcore.api.v1.control.dependencies import get_principal
from ragcore.application.profiles import (
    ActivationDecision,
    ActivationDecisionNotFoundError,
    AlreadyRolledBackError,
    CandidateAliasTarget,
    NotActivatedError,
)
from ragcore.domain.authorization import Principal, PrincipalType
from ragcore.domain.control import Role
from ragcore.evaluation import RetrievalRunProfile
from ragcore.evaluation.metrics.models import GateResult
from ragcore.main import create_app
from ragcore.web.profiles.dependencies import get_profile_activation_service

WORKSPACE_ID = uuid4()


def _principal(role: Role) -> Principal:
    return Principal(uuid4(), WORKSPACE_ID, PrincipalType.HUMAN, frozenset({role}))


def _run_profile(version: str) -> RetrievalRunProfile:
    return RetrievalRunProfile(
        embedding_profile_digest="a" * 64,
        sparse_profile_version="technical-v1",
        index_alias="rag_dense_hybrid_active",
        index_collection_version=version,
    )


def _decision(**overrides: object) -> ActivationDecision:
    values: dict[str, object] = {
        "id": uuid4(),
        "created_at": datetime.now(UTC),
        "dataset_id": "sample",
        "dataset_version": "v1",
        "baseline_run_id": uuid4(),
        "candidate_run_id": uuid4(),
        "baseline_profile": _run_profile("v1"),
        "candidate_profile": _run_profile("v2"),
        "gate": GateResult(thresholds_version="v1", passed=True, regressions=()),
        "activated": True,
        "candidate_target": CandidateAliasTarget(
            alias_name="rag_dense_hybrid_active",
            collection_name="rag_dense_hybrid_v2",
            version="v2",
            model_digest="b" * 64,
        ),
        "index_operation_id": uuid4(),
        "previous_collection": "rag_dense_hybrid_v1",
    }
    values.update(overrides)
    return ActivationDecision(**values)


class StubProfileActivationService:
    def __init__(self, decisions: tuple[ActivationDecision, ...] = ()) -> None:
        self._decisions = decisions
        self.rollback_calls: list[object] = []
        self.rollback_error: Exception | None = None

    async def list_recent(self, limit: int = 20) -> tuple[ActivationDecision, ...]:
        return self._decisions[:limit]

    async def rollback(self, decision_id: object) -> ActivationDecision:
        self.rollback_calls.append(decision_id)
        if self.rollback_error is not None:
            raise self.rollback_error
        return self._decisions[0].with_rollback(rolled_back_at=datetime.now(UTC))


def _app(service: StubProfileActivationService, principal: Principal) -> TestClient:
    app = create_app()
    app.dependency_overrides[get_principal] = lambda: principal
    app.dependency_overrides[get_profile_activation_service] = lambda: service
    return TestClient(app, raise_server_exceptions=False)


def test_page_loads_with_landmarks_and_empty_state_for_platform_admin() -> None:
    client = _app(StubProfileActivationService(), _principal(Role.PLATFORM_ADMIN))
    response = client.get("/admin/profiles")
    assert response.status_code == 200
    assert 'href="#main-content"' in response.text
    assert "No profile activation decisions have been recorded yet." in response.text


def test_non_platform_admin_is_forbidden() -> None:
    client = _app(StubProfileActivationService(), _principal(Role.WORKSPACE_ADMIN))
    response = client.get("/admin/profiles")
    assert response.status_code == 403
    assert "Access denied" in response.text


def test_unauthenticated_request_is_rejected() -> None:
    app = create_app()
    stub = StubProfileActivationService()
    app.dependency_overrides[get_profile_activation_service] = lambda: stub
    client = TestClient(app, raise_server_exceptions=False)
    response = client.get("/admin/profiles")
    assert response.status_code == 401


def test_decision_list_shows_gate_and_status_without_leaking_secrets() -> None:
    decision = _decision()
    service = StubProfileActivationService((decision,))
    client = _app(service, _principal(Role.PLATFORM_ADMIN))
    response = client.get("/admin/profiles")
    assert response.status_code == 200
    assert "v1" in response.text and "v2" in response.text
    assert "passed" in response.text
    assert "activated" in response.text
    assert "api_key" not in response.text.lower()
    assert "secret" not in response.text.lower()


def test_rollback_succeeds_for_platform_admin() -> None:
    decision = _decision()
    service = StubProfileActivationService((decision,))
    client = _app(service, _principal(Role.PLATFORM_ADMIN))
    response = client.post(f"/admin/profiles/{decision.id}/rollback")
    assert response.status_code == 200
    assert service.rollback_calls == [decision.id]


def test_rollback_is_forbidden_for_non_platform_admin() -> None:
    decision = _decision()
    service = StubProfileActivationService((decision,))
    client = _app(service, _principal(Role.WORKSPACE_ADMIN))
    response = client.post(f"/admin/profiles/{decision.id}/rollback")
    assert response.status_code == 403
    assert service.rollback_calls == []


def test_rollback_of_unknown_decision_is_a_stable_not_found() -> None:
    service = StubProfileActivationService((_decision(),))
    service.rollback_error = ActivationDecisionNotFoundError("not found")
    client = _app(service, _principal(Role.PLATFORM_ADMIN))
    response = client.post(f"/admin/profiles/{uuid4()}/rollback")
    assert response.status_code == 404


def test_rollback_of_a_rejected_decision_is_a_stable_conflict() -> None:
    service = StubProfileActivationService((_decision(),))
    service.rollback_error = NotActivatedError("never activated")
    client = _app(service, _principal(Role.PLATFORM_ADMIN))
    response = client.post(f"/admin/profiles/{uuid4()}/rollback")
    assert response.status_code == 409


def test_rollback_twice_is_a_stable_conflict() -> None:
    service = StubProfileActivationService((_decision(),))
    service.rollback_error = AlreadyRolledBackError("already rolled back")
    client = _app(service, _principal(Role.PLATFORM_ADMIN))
    response = client.post(f"/admin/profiles/{uuid4()}/rollback")
    assert response.status_code == 409