138 lines
5.1 KiB
Python
138 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_access.backend.admin.service import ensure_default_roles, set_user_roles
|
|
from govoplan_access.backend.db.models import Account, Role, User, UserRoleAssignment
|
|
from govoplan_access.backend.security.sessions import collect_user_authorization_context
|
|
from govoplan_core.core.modules import RoleTemplate
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.tenancy.scope import Tenant, create_scope_tables, scope_registry
|
|
|
|
|
|
DOCS_READER = RoleTemplate(
|
|
slug="docs_reader",
|
|
name="Documentation reader",
|
|
description="Authenticated documentation baseline.",
|
|
permissions=("docs:documentation:read",),
|
|
default_authenticated=True,
|
|
)
|
|
|
|
|
|
class DefaultAuthenticatedRoleTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
create_scope_tables(self.engine)
|
|
Base.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.session.add(self.tenant)
|
|
self.session.flush()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
Base.metadata.drop_all(bind=self.engine)
|
|
scope_registry.metadata.drop_all(bind=self.engine)
|
|
self.engine.dispose()
|
|
|
|
def _add_user(self, suffix: str) -> User:
|
|
account = Account(
|
|
id=f"account-{suffix}",
|
|
email=f"{suffix}@example.test",
|
|
normalized_email=f"{suffix}@example.test",
|
|
)
|
|
user = User(
|
|
id=f"user-{suffix}",
|
|
tenant_id=self.tenant.id,
|
|
account_id=account.id,
|
|
email=account.email,
|
|
)
|
|
self.session.add_all([account, user])
|
|
self.session.flush()
|
|
return user
|
|
|
|
def test_materialized_default_role_is_implicit_and_cannot_be_assigned_or_removed(self) -> None:
|
|
first = self._add_user("first")
|
|
self._add_user("second")
|
|
|
|
with (
|
|
patch(
|
|
"govoplan_access.backend.admin.service.role_templates_for_level",
|
|
return_value=(DOCS_READER,),
|
|
),
|
|
patch(
|
|
"govoplan_access.backend.security.sessions.role_templates_for_level",
|
|
return_value=(DOCS_READER,),
|
|
),
|
|
):
|
|
roles = ensure_default_roles(self.session, self.tenant)
|
|
default_role = roles["docs_reader"]
|
|
self.assertFalse(default_role.is_assignable)
|
|
self.assertEqual(self.session.query(UserRoleAssignment).count(), 0)
|
|
|
|
# The database row is only an administration projection. Even if
|
|
# stale or tampered, it must never broaden the manifest baseline.
|
|
default_role.permissions = ["mail:profile:write"]
|
|
self.session.flush()
|
|
|
|
set_user_roles(self.session, user=first, role_ids=[])
|
|
context = collect_user_authorization_context(
|
|
self.session,
|
|
first,
|
|
account=first.account,
|
|
include_system=False,
|
|
)
|
|
self.assertEqual(self.session.query(UserRoleAssignment).count(), 0)
|
|
self.assertIn("docs:documentation:read", context.scopes)
|
|
self.assertNotIn("mail:profile:write", context.scopes)
|
|
self.assertEqual([role.slug for role in context.tenant_roles], ["docs_reader"])
|
|
|
|
def test_first_authorized_request_grants_default_without_database_mutation(self) -> None:
|
|
user = self._add_user("new")
|
|
|
|
with patch(
|
|
"govoplan_access.backend.security.sessions.role_templates_for_level",
|
|
return_value=(DOCS_READER,),
|
|
):
|
|
context = collect_user_authorization_context(
|
|
self.session,
|
|
user,
|
|
account=user.account,
|
|
include_system=False,
|
|
)
|
|
|
|
self.assertIn("docs:documentation:read", context.scopes)
|
|
self.assertEqual(context.tenant_roles, [])
|
|
self.assertEqual(self.session.query(Role).count(), 0)
|
|
self.assertEqual(self.session.query(UserRoleAssignment).count(), 0)
|
|
self.assertFalse(self.session.new)
|
|
self.assertFalse(self.session.dirty)
|
|
|
|
user_id = user.id
|
|
self.session.commit()
|
|
self.session.close()
|
|
self.session = self.Session()
|
|
persisted_user = self.session.get(User, user_id)
|
|
with patch(
|
|
"govoplan_access.backend.security.sessions.role_templates_for_level",
|
|
return_value=(DOCS_READER,),
|
|
):
|
|
reopened_context = collect_user_authorization_context(
|
|
self.session,
|
|
persisted_user,
|
|
account=persisted_user.account,
|
|
include_system=False,
|
|
)
|
|
self.assertIn("docs:documentation:read", reopened_context.scopes)
|
|
self.assertEqual(self.session.query(Role).count(), 0)
|
|
self.assertEqual(self.session.query(UserRoleAssignment).count(), 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|