151 lines
6.2 KiB
Python
151 lines
6.2 KiB
Python
"""'/admin/dashboard': real, never-fabricated per-section workspace counts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from ragcore.api.v1.control.dependencies import get_control_application, get_principal
|
|
from ragcore.application.authorization import AuthorizationService
|
|
from ragcore.application.control.api import ControlPlaneApplication
|
|
from ragcore.domain.authorization import Principal, PrincipalType
|
|
from ragcore.domain.control import Application, KnowledgeSpace, Policy, Role, Workspace
|
|
from ragcore.main import create_app
|
|
|
|
|
|
class FakeControlRepository:
|
|
def __init__(self, workspace: Workspace) -> None:
|
|
self.workspace = workspace
|
|
self.applications: list[Application] = []
|
|
self.spaces: list[KnowledgeSpace] = []
|
|
self.policies: list[Policy] = []
|
|
self.grants: list[object] = []
|
|
|
|
async def get_workspace(self, workspace_id: object) -> Workspace | None:
|
|
return self.workspace if workspace_id == self.workspace.id else None
|
|
|
|
async def list_applications(self, workspace_id: object) -> tuple[Application, ...]:
|
|
return tuple(self.applications)
|
|
|
|
async def list_spaces(self, workspace_id: object) -> tuple[KnowledgeSpace, ...]:
|
|
return tuple(self.spaces)
|
|
|
|
async def list_policies(self, workspace_id: object) -> tuple[Policy, ...]:
|
|
return tuple(self.policies)
|
|
|
|
async def list_grants(self, workspace_id: object) -> tuple[object, ...]:
|
|
return tuple(self.grants)
|
|
|
|
|
|
def _client(principal: Principal, repository: FakeControlRepository) -> TestClient:
|
|
service = ControlPlaneApplication(repository, AuthorizationService())
|
|
app = create_app()
|
|
app.dependency_overrides[get_principal] = lambda: principal
|
|
app.dependency_overrides[get_control_application] = lambda: service
|
|
return TestClient(app, raise_server_exceptions=False)
|
|
|
|
|
|
def test_dashboard_requires_authentication() -> None:
|
|
client = TestClient(create_app(), raise_server_exceptions=False)
|
|
response = client.get("/admin/dashboard")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_dashboard_shows_real_counts_for_a_workspace_admin() -> None:
|
|
workspace = Workspace(uuid4(), "Dash workspace", "dash-workspace")
|
|
principal = Principal(
|
|
uuid4(), workspace.id, PrincipalType.HUMAN, frozenset({Role.WORKSPACE_ADMIN})
|
|
)
|
|
repository = FakeControlRepository(workspace)
|
|
repository.applications.append(
|
|
workspace.create_application(id=uuid4(), name="Ingest", slug="ingest")
|
|
)
|
|
client = _client(principal, repository)
|
|
|
|
response = client.get("/admin/dashboard")
|
|
|
|
assert response.status_code == 200
|
|
assert 'href="#main-content"' in response.text
|
|
assert '<nav aria-label="Primary navigation">' in response.text
|
|
assert "registered application</p>" in response.text
|
|
assert "Not configured" in response.text
|
|
|
|
|
|
def test_dashboard_presents_truthful_source_to_answer_lifecycle() -> None:
|
|
workspace = Workspace(uuid4(), "Flow workspace", "flow-workspace")
|
|
principal = Principal(
|
|
uuid4(), workspace.id, PrincipalType.HUMAN, frozenset({Role.WORKSPACE_ADMIN})
|
|
)
|
|
repository = FakeControlRepository(workspace)
|
|
response = _client(principal, repository).get("/admin/dashboard")
|
|
|
|
assert response.status_code == 200
|
|
assert 'data-shell-title="Dashboard"' in response.text
|
|
assert 'data-rag-lifecycle="source-to-answer"' in response.text
|
|
assert 'href="/admin/sources"' in response.text
|
|
assert 'href="/admin/jobs"' in response.text
|
|
assert 'href="/admin/query-lab"' in response.text
|
|
assert 'href="/admin/control/applications"' in response.text
|
|
assert "Dense + sparse" in response.text
|
|
assert "RRF fusion" in response.text
|
|
assert "versioned citations" in response.text
|
|
assert "reranking" not in response.text.lower()
|
|
|
|
|
|
def test_product_styles_define_tokens_responsiveness_and_reduced_motion() -> None:
|
|
client = TestClient(create_app())
|
|
response = client.get("/static/admin.css")
|
|
|
|
assert response.status_code == 200
|
|
assert "--color-accent:" in response.text
|
|
assert ".lifecycle-grid" in response.text
|
|
assert "@media (max-width: 767px)" in response.text
|
|
assert "@media (prefers-reduced-motion: reduce)" in response.text
|
|
assert "animation-duration: 0.01ms" in response.text
|
|
|
|
script = client.get("/static/admin.js")
|
|
assert script.status_code == 200
|
|
assert 'matchMedia("(min-width: 1024px)")' in script.text
|
|
assert 'toggleAttribute("inert"' in script.text
|
|
assert 'addEventListener("change"' in script.text
|
|
|
|
|
|
def test_dashboard_marks_sections_restricted_for_an_unprivileged_principal() -> None:
|
|
workspace = Workspace(uuid4(), "Dash workspace", "dash-workspace")
|
|
principal = Principal(uuid4(), workspace.id, PrincipalType.HUMAN, frozenset({Role.VIEWER}))
|
|
repository = FakeControlRepository(workspace)
|
|
client = _client(principal, repository)
|
|
|
|
response = client.get("/admin/dashboard")
|
|
|
|
assert response.status_code == 200
|
|
assert "You are not authorized to view applications." in response.text
|
|
assert "You are not authorized to view knowledge spaces." in response.text
|
|
assert "You are not authorized to view policies." in response.text
|
|
assert "Restricted to platform administrators." in response.text
|
|
|
|
|
|
class _StubProfileActivationService:
|
|
async def list_recent(self, limit: int = 1) -> tuple[object, ...]:
|
|
return ()
|
|
|
|
|
|
def test_dashboard_shows_empty_state_for_a_workspace_with_nothing_yet() -> None:
|
|
workspace = Workspace(uuid4(), "Empty workspace", "empty-workspace")
|
|
principal = Principal(
|
|
uuid4(), workspace.id, PrincipalType.HUMAN, frozenset({Role.PLATFORM_ADMIN})
|
|
)
|
|
repository = FakeControlRepository(workspace)
|
|
service = ControlPlaneApplication(repository, AuthorizationService())
|
|
app = create_app()
|
|
app.dependency_overrides[get_principal] = lambda: principal
|
|
app.dependency_overrides[get_control_application] = lambda: service
|
|
app.state.profile_activation_service = _StubProfileActivationService()
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
response = client.get("/admin/dashboard")
|
|
|
|
assert response.status_code == 200
|
|
assert "No profile activation decisions have been recorded yet." in response.text
|