intermittent commit

tests/test_admin_batch_helpers.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_access.backend.api.v1.admin_common import (
+    _accounts_by_user_id,
+    _group_member_ids_by_group_id,
+    _groups_by_user_id,
+    _roles_by_group_id,
+    _roles_by_user_id,
+    _tenant_role_assignment_counts,
+)
+from govoplan_access.backend.db.models import Account, Group, GroupRoleAssignment, Role, User, UserGroupMembership, UserRoleAssignment
+from govoplan_core.db.base import Base
+
+
+class AdminBatchHelperTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(bind=self.engine)
+        self.Session = sessionmaker(bind=self.engine)
+
+    def tearDown(self) -> None:
+        Base.metadata.drop_all(bind=self.engine)
+
+    def test_access_admin_batch_maps_related_rows(self) -> None:
+        session = self.Session()
+        account = Account(id="account-1", email="ada@example.test", normalized_email="ada@example.test")
+        user = User(id="user-1", tenant_id="tenant-1", account_id=account.id, email=account.email)
+        group = Group(id="group-1", tenant_id="tenant-1", slug="clerks", name="Clerks")
+        role = Role(id="role-1", tenant_id="tenant-1", slug="reader", name="Reader", permissions=["admin:users:read"])
+        session.add_all([account, user, group, role])
+        session.flush()
+        session.add(UserGroupMembership(tenant_id="tenant-1", user_id=user.id, group_id=group.id))
+        session.add(UserRoleAssignment(tenant_id="tenant-1", user_id=user.id, role_id=role.id))
+        session.add(GroupRoleAssignment(tenant_id="tenant-1", group_id=group.id, role_id=role.id))
+        session.commit()
+
+        accounts_by_user = _accounts_by_user_id(session, [user.id])
+        group_member_ids = _group_member_ids_by_group_id(session, tenant_id="tenant-1", group_ids=[group.id])
+        groups_by_user = _groups_by_user_id(session, tenant_id="tenant-1", user_ids=[user.id])
+        roles_by_group = _roles_by_group_id(session, tenant_id="tenant-1", group_ids=[group.id])
+        roles_by_user = _roles_by_user_id(session, tenant_id="tenant-1", user_ids=[user.id])
+        role_counts = _tenant_role_assignment_counts(session, [role.id])
+
+        self.assertEqual(accounts_by_user[user.id].id, account.id)
+        self.assertEqual(group_member_ids, {group.id: [user.id]})
+        self.assertEqual([item.id for item in groups_by_user[user.id]], [group.id])
+        self.assertEqual([item.id for item in roles_by_group[group.id]], [role.id])
+        self.assertEqual([item.id for item in roles_by_user[user.id]], [role.id])
+        self.assertEqual(role_counts, {role.id: (1, 1)})
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_admin_pagination.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import unittest
+
+from fastapi import HTTPException
+
+from govoplan_access.backend.api.v1.routes import (
+    _decode_full_delta_cursor,
+    _encode_full_delta_cursor,
+    _full_delta_page,
+    _page_query,
+)
+
+
+class FakeQuery:
+    def __init__(self, rows: list[int]) -> None:
+        self._rows = rows
+        self._offset = 0
+        self._limit: int | None = None
+
+    def order_by(self, *_args: object) -> FakeQuery:
+        return self
+
+    def count(self) -> int:
+        return len(self._rows)
+
+    def offset(self, value: int) -> FakeQuery:
+        clone = FakeQuery(self._rows)
+        clone._offset = value
+        clone._limit = self._limit
+        return clone
+
+    def limit(self, value: int) -> FakeQuery:
+        clone = FakeQuery(self._rows)
+        clone._offset = self._offset
+        clone._limit = value
+        return clone
+
+    def all(self) -> list[int]:
+        end = None if self._limit is None else self._offset + self._limit
+        return self._rows[self._offset:end]
+
+
+class AdminPaginationTests(unittest.TestCase):
+    def test_page_query_returns_page_and_metadata(self) -> None:
+        rows, metadata = _page_query(FakeQuery(list(range(7))), page=2, page_size=3)
+
+        self.assertEqual(rows, [3, 4, 5])
+        self.assertEqual(metadata, {"total": 7, "page": 2, "page_size": 3, "pages": 3})
+
+    def test_full_delta_page_returns_cursor_until_last_page(self) -> None:
+        rows, metadata, watermark, has_more = _full_delta_page(
+            FakeQuery(list(range(7))),
+            page=2,
+            page_size=3,
+            scope="users",
+            snapshot_sequence=42,
+        )
+
+        self.assertEqual(rows, [3, 4, 5])
+        self.assertEqual(metadata, {"total": 7, "page": 2, "page_size": 3, "pages": 3})
+        self.assertEqual(watermark, "full:users:3:42")
+        self.assertTrue(has_more)
+
+    def test_full_delta_page_returns_sequence_watermark_on_last_page(self) -> None:
+        rows, metadata, watermark, has_more = _full_delta_page(
+            FakeQuery(list(range(7))),
+            page=3,
+            page_size=3,
+            scope="users",
+            snapshot_sequence=42,
+        )
+
+        self.assertEqual(rows, [6])
+        self.assertEqual(metadata, {"total": 7, "page": 3, "page_size": 3, "pages": 3})
+        self.assertEqual(watermark, "seq:42")
+        self.assertFalse(has_more)
+
+    def test_full_delta_cursor_round_trips_and_rejects_wrong_scope(self) -> None:
+        cursor = _encode_full_delta_cursor("users", page=3, snapshot_sequence=42)
+
+        self.assertEqual(_decode_full_delta_cursor(cursor, scope="users"), (3, 42))
+        self.assertIsNone(_decode_full_delta_cursor("seq:42", scope="users"))
+        with self.assertRaises(HTTPException):
+            _decode_full_delta_cursor(cursor, scope="groups")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_auth_dependencies.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import unittest
+from typing import Iterable
+
+from fastapi import HTTPException
+from starlette.requests import Request
+
+from govoplan_access.backend.auth.dependencies import _extract_token, _requires_csrf, _resolve_legacy_principal_ref
+from govoplan_core.settings import settings
+
+
+def request_for(*, method: str = "GET", headers: Iterable[tuple[str, str]] = ()) -> Request:
+    return Request(
+        {
+            "type": "http",
+            "method": method,
+            "path": "/",
+            "headers": [(key.lower().encode("latin-1"), value.encode("latin-1")) for key, value in headers],
+        }
+    )
+
+
+class AuthDependencyTests(unittest.TestCase):
+    def test_extract_token_prefers_explicit_api_key(self) -> None:
+        request = request_for(headers=[("authorization", "Bearer session-token")])
+
+        self.assertEqual(_extract_token(request, "Bearer session-token", "api-key-token"), ("api-key-token", "api_key"))
+
+    def test_extract_token_supports_bearer_and_cookie_sources(self) -> None:
+        self.assertEqual(_extract_token(request_for(), "Bearer session-token", None), ("session-token", "bearer"))
+
+        cookie_request = request_for(headers=[("cookie", f"{settings.auth_session_cookie_name}=cookie-token")])
+        self.assertEqual(_extract_token(cookie_request, None, None), ("cookie-token", "cookie"))
+
+    def test_requires_csrf_only_for_mutating_methods(self) -> None:
+        self.assertFalse(_requires_csrf(request_for(method="GET")))
+        self.assertTrue(_requires_csrf(request_for(method="POST")))
+
+    def test_legacy_principal_resolver_rejects_missing_token_before_db_lookup(self) -> None:
+        with self.assertRaises(HTTPException) as raised:
+            _resolve_legacy_principal_ref(request_for(), None, authorization=None, x_api_key=None)  # type: ignore[arg-type]
+
+        self.assertEqual(raised.exception.status_code, 401)
+        self.assertEqual(raised.exception.detail, "Missing API key or session token")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_optional_tenancy_contract.py

@@ -15,8 +15,8 @@ class OptionalTenancyContractTests(unittest.TestCase):
 
         dependencies = tuple(project["dependencies"])
 
-        self.assertIn("govoplan-core>=0.1.6", dependencies)
-        self.assertNotIn("govoplan-tenancy>=0.1.6", dependencies)
+        self.assertIn("govoplan-core>=0.1.8", dependencies)
+        self.assertNotIn("govoplan-tenancy>=0.1.8", dependencies)
         self.assertFalse(any(item.startswith("govoplan-tenancy") for item in dependencies))
 
     def test_tenancy_is_declared_as_optional_module_integration(self) -> None:

tests/test_permission_catalog_contract.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_access.backend.permissions import catalog as access_catalog
+from govoplan_core.security import permissions as core_permissions
+from govoplan_core.security.scope_aliases import LEGACY_SCOPE_ALIASES
+
+
+class PermissionCatalogContractTests(unittest.TestCase):
+    def test_access_reuses_core_legacy_scope_aliases(self) -> None:
+        self.assertIs(access_catalog.LEGACY_SCOPE_ALIASES, LEGACY_SCOPE_ALIASES)
+        self.assertIs(core_permissions.LEGACY_SCOPE_ALIASES, LEGACY_SCOPE_ALIASES)
+
+    def test_legacy_alias_grants_match_in_core_and_access(self) -> None:
+        for legacy_scope, aliases in LEGACY_SCOPE_ALIASES.items():
+            for alias in aliases:
+                self.assertTrue(core_permissions.scope_grants(legacy_scope, alias))
+                self.assertTrue(access_catalog.scope_grants(legacy_scope, alias))
+
+    def test_access_legacy_catalog_includes_non_access_platform_scopes(self) -> None:
+        scopes = {permission.scope for permission in access_catalog.permission_catalog(include_legacy=True)}
+        self.assertIn("campaign:queue", scopes)
+        self.assertIn("files:upload", scopes)
+        self.assertIn("mail_servers:test", scopes)
+
+
+if __name__ == "__main__":
+    unittest.main()