govoplan-access/tests/test_service_accounts.py

from __future__ import annotations

import unittest

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from govoplan_access.backend.db.base import AccessBase
from govoplan_access.backend.db.models import (
    Account,
    ApiKey,
    AuthSession,
    User,
)
from govoplan_access.backend.service_accounts import (
    ServiceAccountConflictError,
    create_service_account,
    retire_service_account,
    update_service_account,
)
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.tenancy.scope import (
    Tenant,
    create_scope_tables,
    scope_registry,
)


class ServiceAccountTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite:///:memory:")
        create_scope_tables(self.engine)
        AccessBase.metadata.create_all(bind=self.engine)
        self.Session = sessionmaker(bind=self.engine)
        self.session = self.Session()
        self.tenant = Tenant(
            id="tenant-1",
            slug="tenant-1",
            name="Tenant 1",
        )
        self.account = Account(
            id="account-1",
            email="admin@example.test",
            normalized_email="admin@example.test",
        )
        self.user = User(
            id="user-1",
            tenant_id=self.tenant.id,
            account_id=self.account.id,
            email=self.account.email,
        )
        self.session.add_all(
            (self.tenant, self.account, self.user)
        )
        self.session.commit()

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

    def _principal(
        self,
        scopes: frozenset[str] = frozenset({"tenant:*"}),
    ) -> ApiPrincipal:
        return ApiPrincipal(
            principal=PrincipalRef(
                account_id=self.account.id,
                membership_id=self.user.id,
                tenant_id=self.tenant.id,
                scopes=scopes,
            ),
            account=self.account,
            user=self.user,
        )

    def test_create_builds_a_non_login_identity_without_secrets(self) -> None:
        item = create_service_account(
            self.session,
            tenant=self.tenant,
            principal=self._principal(),
            name=" Monthly import ",
            description="Runs the governed monthly import.",
            scope_ceiling=(
                "dataflow:pipeline:run",
                "datasources:catalogue:read",
            ),
        )

        backing_account = self.session.get(
            Account,
            item.account_id,
        )
        membership = self.session.get(
            User,
            item.membership_id,
        )
        self.assertEqual("Monthly import", item.name)
        self.assertEqual(
            [
                "dataflow:pipeline:run",
                "datasources:catalogue:read",
            ],
            item.scope_ceiling,
        )
        self.assertEqual(
            "service_account",
            backing_account.auth_provider,
        )
        self.assertIsNone(backing_account.password_hash)
        self.assertEqual(
            "service_account",
            membership.auth_provider,
        )
        self.assertIsNone(membership.password_hash)
        self.assertEqual(
            0,
            self.session.query(ApiKey)
            .filter(ApiKey.user_id == membership.id)
            .count(),
        )
        self.assertEqual(
            0,
            self.session.query(AuthSession)
            .filter(AuthSession.user_id == membership.id)
            .count(),
        )

    def test_scope_escalation_and_stale_updates_fail_closed(self) -> None:
        principal = self._principal(
            frozenset({"dataflow:pipeline:run"})
        )
        with self.assertRaises(PermissionError):
            create_service_account(
                self.session,
                tenant=self.tenant,
                principal=principal,
                name="Escalating worker",
                description=None,
                scope_ceiling=("system:settings:write",),
            )

        item = create_service_account(
            self.session,
            tenant=self.tenant,
            principal=principal,
            name="Bounded worker",
            description=None,
            scope_ceiling=("dataflow:pipeline:run",),
        )
        updated = update_service_account(
            self.session,
            tenant_id=self.tenant.id,
            service_account_id=item.id,
            principal=principal,
            expected_revision=1,
            changes={"description": "Updated"},
        )
        self.assertEqual(2, updated.revision)
        with self.assertRaises(ServiceAccountConflictError):
            update_service_account(
                self.session,
                tenant_id=self.tenant.id,
                service_account_id=item.id,
                principal=principal,
                expected_revision=1,
                changes={"description": "Stale"},
            )

    def test_retirement_revokes_the_backing_principal(self) -> None:
        principal = self._principal()
        item = create_service_account(
            self.session,
            tenant=self.tenant,
            principal=principal,
            name="Retired worker",
            description=None,
            scope_ceiling=("dataflow:pipeline:run",),
        )

        retired = retire_service_account(
            self.session,
            tenant_id=self.tenant.id,
            service_account_id=item.id,
            principal=principal,
            expected_revision=1,
        )

        self.assertFalse(retired.is_active)
        self.assertIsNotNone(retired.retired_at)
        self.assertFalse(
            self.session.get(Account, retired.account_id).is_active
        )
        self.assertFalse(
            self.session.get(User, retired.membership_id).is_active
        )


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