govoplan-access/tests/test_governance_projection.py

from __future__ import annotations

import unittest

from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker

from govoplan_access.backend.db.models import Account, Group, Role, User, UserRoleAssignment
from govoplan_access.backend.governance_materializer import SqlAccessGovernanceMaterializer
from govoplan_core.core.access import (
    GovernanceProjectionBatch,
    GovernanceProjectionCommand,
    GovernanceTemplateMaterialization,
)
from govoplan_core.db.base import Base


def _command(index: int, *, kind: str = "role", operation: str = "upsert") -> GovernanceProjectionCommand:
    return GovernanceProjectionCommand(
        assignment_id=f"assignment-{kind}-{index}",
        operation=operation,  # type: ignore[arg-type]
        template=GovernanceTemplateMaterialization(
            template_id=f"template-{kind}",
            kind=kind,  # type: ignore[arg-type]
            tenant_id=f"tenant-{index}",
            slug=f"managed-{kind}",
            name=f"Managed {kind}",
            permissions=("access:role:read",) if kind == "role" else (),
            required=True,
        ),
        provenance={"source": "test", "assignment_mode": "required"},
    )


class GovernanceProjectionTests(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)
        self.session = self.Session()
        self.materializer = SqlAccessGovernanceMaterializer()

    def tearDown(self) -> None:
        self.session.close()
        Base.metadata.drop_all(bind=self.engine)
        self.engine.dispose()

    def test_bulk_projection_is_idempotent_and_returns_per_assignment_outcomes(self) -> None:
        commands = tuple(_command(index) for index in range(5))
        first = self.materializer.reconcile(
            self.session,
            GovernanceProjectionBatch(operation_id="first", commands=commands),
        )
        second = self.materializer.reconcile(
            self.session,
            GovernanceProjectionBatch(operation_id="second", commands=commands),
        )

        self.assertEqual(["created"] * 5, [item.status for item in first.outcomes])
        self.assertEqual(["unchanged"] * 5, [item.status for item in second.outcomes])
        self.assertEqual(5, self.session.query(Role).count())
        self.assertEqual(
            {item.assignment_id for item in commands},
            {item.assignment_id for item in second.outcomes},
        )
        self.assertTrue(all(item.provenance["source"] == "test" for item in second.outcomes))

    def test_removal_isolated_blocker_preserves_other_batch_outcomes(self) -> None:
        first, second = _command(1), _command(2)
        created = self.materializer.reconcile(
            self.session,
            GovernanceProjectionBatch(operation_id="create", commands=(first, second)),
        )
        roles = {item.tenant_id: item.resource_id for item in created.outcomes}
        account = Account(
            id="account-1",
            email="assigned@example.test",
            normalized_email="assigned@example.test",
        )
        user = User(
            id="user-1",
            tenant_id="tenant-1",
            account_id=account.id,
            email=account.email,
        )
        self.session.add_all([account, user])
        self.session.flush()
        self.session.add(
            UserRoleAssignment(
                tenant_id="tenant-1",
                user_id=user.id,
                role_id=roles["tenant-1"],
            )
        )
        self.session.flush()

        removals = tuple(
            GovernanceProjectionCommand(
                assignment_id=item.assignment_id,
                operation="remove",
                template=item.template,
                provenance=item.provenance,
            )
            for item in (first, second)
        )
        result = self.materializer.reconcile(
            self.session,
            GovernanceProjectionBatch(operation_id="remove", commands=removals),
        )

        self.assertEqual(["blocked", "removed"], [item.status for item in result.outcomes])
        self.assertEqual(("role_has_users",), result.outcomes[0].blocker_codes)
        self.assertIsNotNone(self.session.get(Role, roles["tenant-1"]))
        self.assertIsNone(self.session.get(Role, roles["tenant-2"]))

    def test_dry_run_does_not_mutate(self) -> None:
        result = self.materializer.reconcile(
            self.session,
            GovernanceProjectionBatch(
                operation_id="preview",
                commands=(_command(1, kind="group"),),
                dry_run=True,
            ),
        )
        self.assertEqual("created", result.outcomes[0].status)
        self.assertEqual(0, self.session.query(Group).count())

    def test_bulk_read_query_count_does_not_grow_per_assignment(self) -> None:
        def select_count(size: int) -> int:
            count = 0

            def record_select(_conn, _cursor, statement, _parameters, _context, _executemany):
                nonlocal count
                if statement.lstrip().upper().startswith("SELECT"):
                    count += 1

            event.listen(self.engine, "before_cursor_execute", record_select)
            try:
                self.materializer.reconcile(
                    self.session,
                    GovernanceProjectionBatch(
                        operation_id=f"preview-{size}",
                        commands=tuple(
                            _command(index, kind="group" if index % 2 else "role")
                            for index in range(size)
                        ),
                        dry_run=True,
                    ),
                )
            finally:
                event.remove(self.engine, "before_cursor_execute", record_select)
            return count

        small = select_count(2)
        large = select_count(200)
        self.assertEqual(small, large)
        self.assertLessEqual(large, 4)


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