govoplan-dashboard/tests/test_dashboard_layouts.py

from __future__ import annotations

import unittest

from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from govoplan_core.auth import ApiPrincipal, get_api_principal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.db.session import get_session
from govoplan_dashboard.backend.db.models import DashboardLayout
from govoplan_dashboard.backend.router import router


def principal(
    *,
    tenant_id: str = "tenant-1",
    account_id: str = "account-1",
) -> ApiPrincipal:
    return ApiPrincipal(
        principal=PrincipalRef(
            account_id=account_id,
            membership_id=f"membership:{tenant_id}:{account_id}",
            tenant_id=tenant_id,
            scopes=frozenset(),
            group_ids=frozenset(),
        ),
        account=object(),
        user=object(),
    )


class DashboardLayoutApiTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine(
            "sqlite+pysqlite:///:memory:",
            connect_args={"check_same_thread": False},
            poolclass=StaticPool,
        )
        DashboardLayout.__table__.create(self.engine)
        self.session_factory = sessionmaker(
            bind=self.engine,
            autoflush=False,
            expire_on_commit=False,
        )
        self.active_principal = principal()
        app = FastAPI()
        app.include_router(router, prefix="/api/v1")

        def session_dependency():
            session = self.session_factory()
            try:
                yield session
            finally:
                session.close()

        app.dependency_overrides[get_session] = session_dependency
        app.dependency_overrides[get_api_principal] = (
            lambda: self.active_principal
        )
        self.client = TestClient(app)

    def tearDown(self) -> None:
        self.client.close()
        self.engine.dispose()

    def test_layouts_are_isolated_by_account_and_view(self) -> None:
        initial = self.client.get("/api/v1/dashboard/layout")
        self.assertEqual(200, initial.status_code)
        self.assertFalse(initial.json()["exists"])
        self.assertEqual(0, initial.json()["revision"])

        saved = self.client.put(
            "/api/v1/dashboard/layout",
            json={
                "expected_revision": 0,
                "layout_version": 1,
                "placements": [
                    {
                        "instance_id": "instance-1",
                        "widget_id": "dashboard.installed-modules",
                        "size": "wide",
                        "configuration": {
                            "maxItems": 5,
                            "showVersions": False,
                        },
                    }
                ],
                "known_widget_ids": ["dashboard.installed-modules"],
            },
        )
        self.assertEqual(200, saved.status_code)
        self.assertEqual(1, saved.json()["revision"])

        view_layout = self.client.get(
            "/api/v1/dashboard/layout",
            params={"view_id": "view-files"},
        )
        self.assertEqual(200, view_layout.status_code)
        self.assertFalse(view_layout.json()["exists"])

        self.active_principal = principal(account_id="account-2")
        other_account = self.client.get("/api/v1/dashboard/layout")
        self.assertEqual(200, other_account.status_code)
        self.assertFalse(other_account.json()["exists"])

        self.active_principal = principal()
        original = self.client.get("/api/v1/dashboard/layout")
        self.assertTrue(original.json()["exists"])
        self.assertEqual(
            "dashboard.installed-modules",
            original.json()["placements"][0]["widget_id"],
        )
        self.assertIs(
            False,
            original.json()["placements"][0]["configuration"]["showVersions"],
        )

    def test_stale_revision_is_rejected(self) -> None:
        payload = {
            "expected_revision": 0,
            "layout_version": 1,
            "placements": [],
            "known_widget_ids": [],
        }
        first = self.client.put("/api/v1/dashboard/layout", json=payload)
        self.assertEqual(200, first.status_code)

        stale = self.client.put("/api/v1/dashboard/layout", json=payload)
        self.assertEqual(409, stale.status_code)
        self.assertIn("another session", stale.json()["detail"])

    def test_duplicate_instance_ids_are_rejected(self) -> None:
        placement = {
            "instance_id": "duplicate",
            "widget_id": "example.widget",
            "size": "medium",
            "configuration": {},
        }
        response = self.client.put(
            "/api/v1/dashboard/layout",
            json={
                "expected_revision": 0,
                "layout_version": 1,
                "placements": [placement, placement],
                "known_widget_ids": ["example.widget"],
            },
        )

        self.assertEqual(422, response.status_code)
        with Session(self.engine) as session:
            self.assertEqual(0, session.query(DashboardLayout).count())

    def test_oversized_widget_configuration_is_rejected(self) -> None:
        response = self.client.put(
            "/api/v1/dashboard/layout",
            json={
                "expected_revision": 0,
                "layout_version": 1,
                "placements": [
                    {
                        "instance_id": "instance-1",
                        "widget_id": "example.widget",
                        "size": "medium",
                        "configuration": {"query": "x" * 4_001},
                    }
                ],
                "known_widget_ids": ["example.widget"],
            },
        )

        self.assertEqual(422, response.status_code)

    def test_account_layout_context_limit_is_enforced(self) -> None:
        with self.session_factory() as session:
            session.add_all(
                DashboardLayout(
                    tenant_id="tenant-1",
                    account_id="account-1",
                    context_key=f"view:view-{index}",
                    view_id=f"view-{index}",
                    placements=[],
                    known_widget_ids=[],
                )
                for index in range(100)
            )
            session.commit()

        response = self.client.put(
            "/api/v1/dashboard/layout",
            params={"view_id": "one-too-many"},
            json={
                "expected_revision": 0,
                "layout_version": 1,
                "placements": [],
                "known_widget_ids": [],
            },
        )

        self.assertEqual(422, response.status_code)
        self.assertIn("At most 100", response.json()["detail"])


if __name__ == "__main__":
    unittest.main()