377 lines
13 KiB
Python
377 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import timedelta
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.postbox import PostboxActorRef
|
|
from govoplan_core.core.tasks import WorkItemQuery
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.db.session import (
|
|
DatabaseHandle,
|
|
get_database,
|
|
reset_database,
|
|
set_database,
|
|
)
|
|
from govoplan_core.security.time import utc_now
|
|
from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
|
|
from govoplan_identity.backend.directory import SqlIdentityDirectory
|
|
from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment
|
|
from govoplan_idm.backend.directory import SqlIdmDirectory
|
|
from govoplan_organizations.backend.db.models import (
|
|
OrganizationFunction,
|
|
OrganizationUnit,
|
|
)
|
|
from govoplan_organizations.backend.directory import SqlOrganizationDirectory
|
|
from govoplan_postbox.backend.db.models import (
|
|
Postbox,
|
|
PostboxAccessEvent,
|
|
PostboxAddress,
|
|
PostboxAttachmentReference,
|
|
PostboxBinding,
|
|
PostboxDelivery,
|
|
PostboxGrouping,
|
|
PostboxGroupingSource,
|
|
PostboxMessage,
|
|
PostboxMessageReceipt,
|
|
PostboxParticipant,
|
|
PostboxRoute,
|
|
PostboxTemplate,
|
|
PostboxTemplateRevision,
|
|
)
|
|
from govoplan_postbox.backend.service import PostboxService
|
|
from govoplan_postbox.backend.principals import actor_from_principal
|
|
from govoplan_postbox.backend.work_items import PostboxWorkItemProvider
|
|
|
|
|
|
TABLES = (
|
|
Identity.__table__,
|
|
IdentityAccountLink.__table__,
|
|
OrganizationUnit.__table__,
|
|
OrganizationFunction.__table__,
|
|
IdmOrganizationFunctionAssignment.__table__,
|
|
PostboxTemplate.__table__,
|
|
PostboxTemplateRevision.__table__,
|
|
PostboxAddress.__table__,
|
|
Postbox.__table__,
|
|
PostboxBinding.__table__,
|
|
PostboxMessage.__table__,
|
|
PostboxParticipant.__table__,
|
|
PostboxAttachmentReference.__table__,
|
|
PostboxDelivery.__table__,
|
|
PostboxRoute.__table__,
|
|
PostboxMessageReceipt.__table__,
|
|
PostboxGrouping.__table__,
|
|
PostboxGroupingSource.__table__,
|
|
PostboxAccessEvent.__table__,
|
|
)
|
|
|
|
|
|
class PostboxRealDirectoryAccessTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
try:
|
|
self.previous_database = get_database()
|
|
except RuntimeError:
|
|
self.previous_database = None
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(self.engine, tables=TABLES)
|
|
self.database = DatabaseHandle("sqlite:///:memory:", engine=self.engine)
|
|
set_database(self.database)
|
|
self.organizations = SqlOrganizationDirectory(
|
|
session_factory=self.database.SessionLocal
|
|
)
|
|
self.identities = SqlIdentityDirectory()
|
|
self.idm = SqlIdmDirectory(
|
|
identities=self.identities,
|
|
organizations=self.organizations,
|
|
)
|
|
self.service = PostboxService(
|
|
identities=self.identities,
|
|
idm=self.idm,
|
|
incumbencies=self.idm,
|
|
organizations=self.organizations,
|
|
)
|
|
with self.database.SessionLocal() as session:
|
|
session.add_all(
|
|
(
|
|
Identity(
|
|
id="identity-owner",
|
|
display_name="Owner",
|
|
source="test",
|
|
is_active=True,
|
|
settings={},
|
|
),
|
|
IdentityAccountLink(
|
|
id="link-owner",
|
|
identity_id="identity-owner",
|
|
account_id="account-owner",
|
|
is_primary=True,
|
|
source="test",
|
|
),
|
|
Identity(
|
|
id="identity-delegate",
|
|
display_name="Delegate",
|
|
source="test",
|
|
is_active=True,
|
|
settings={},
|
|
),
|
|
IdentityAccountLink(
|
|
id="link-delegate",
|
|
identity_id="identity-delegate",
|
|
account_id="account-delegate",
|
|
is_primary=True,
|
|
source="test",
|
|
),
|
|
OrganizationUnit(
|
|
id="unit-one",
|
|
tenant_id="tenant-1",
|
|
slug="unit-one",
|
|
name="Unit One",
|
|
is_active=True,
|
|
settings={},
|
|
),
|
|
OrganizationUnit(
|
|
id="unit-two",
|
|
tenant_id="tenant-1",
|
|
slug="unit-two",
|
|
name="Unit Two",
|
|
is_active=True,
|
|
settings={},
|
|
),
|
|
OrganizationFunction(
|
|
id="function-one",
|
|
tenant_id="tenant-1",
|
|
organization_unit_id="unit-one",
|
|
slug="clerk",
|
|
name="Clerk",
|
|
delegable=True,
|
|
is_active=True,
|
|
settings={},
|
|
),
|
|
)
|
|
)
|
|
session.commit()
|
|
postbox = self.service.create_exact_postbox(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
name="Unit One / Clerk",
|
|
organization_unit_id="unit-one",
|
|
function_id="function-one",
|
|
address_key=None,
|
|
description=None,
|
|
classification="internal",
|
|
actor_id="admin-1",
|
|
)
|
|
session.commit()
|
|
self.postbox_id = postbox.id
|
|
|
|
def tearDown(self) -> None:
|
|
if self.previous_database is None:
|
|
reset_database()
|
|
else:
|
|
set_database(self.previous_database)
|
|
self.database.dispose()
|
|
|
|
def _actor(self, account_id: str) -> PostboxActorRef:
|
|
return PostboxActorRef(
|
|
account_id=account_id,
|
|
authorized_actions=frozenset({"discover", "read", "reply"}),
|
|
)
|
|
|
|
def _decision(self, account_id: str):
|
|
with Session(self.engine) as session:
|
|
return self.service.explain_access(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
postbox_id=self.postbox_id,
|
|
actor=self._actor(account_id),
|
|
action="read",
|
|
)
|
|
|
|
def _add_assignment(
|
|
self,
|
|
*,
|
|
assignment_id: str,
|
|
identity_id: str,
|
|
account_id: str,
|
|
source: str = "direct",
|
|
delegated_from_assignment_id: str | None = None,
|
|
valid_until=None,
|
|
) -> None:
|
|
with self.database.SessionLocal() as session:
|
|
session.add(
|
|
IdmOrganizationFunctionAssignment(
|
|
id=assignment_id,
|
|
tenant_id="tenant-1",
|
|
identity_id=identity_id,
|
|
account_id=account_id,
|
|
function_id="function-one",
|
|
organization_unit_id="unit-one",
|
|
source=source,
|
|
delegated_from_assignment_id=delegated_from_assignment_id,
|
|
valid_until=valid_until,
|
|
is_active=True,
|
|
settings={},
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
def test_real_directory_reassignment_changes_current_holder_only(self) -> None:
|
|
self._add_assignment(
|
|
assignment_id="owner-assignment",
|
|
identity_id="identity-owner",
|
|
account_id="account-owner",
|
|
)
|
|
self.assertTrue(self._decision("account-owner").allowed)
|
|
|
|
with self.database.SessionLocal() as session:
|
|
assignment = session.get(
|
|
IdmOrganizationFunctionAssignment,
|
|
"owner-assignment",
|
|
)
|
|
assignment.is_active = False
|
|
session.commit()
|
|
self._add_assignment(
|
|
assignment_id="delegate-assignment",
|
|
identity_id="identity-delegate",
|
|
account_id="account-delegate",
|
|
)
|
|
|
|
self.assertFalse(self._decision("account-owner").allowed)
|
|
replacement = self._decision("account-delegate")
|
|
self.assertTrue(replacement.allowed)
|
|
self.assertEqual(replacement.assignment_ids, ("delegate-assignment",))
|
|
|
|
def test_real_directory_delegation_expires_with_its_source(self) -> None:
|
|
self._add_assignment(
|
|
assignment_id="owner-assignment",
|
|
identity_id="identity-owner",
|
|
account_id="account-owner",
|
|
)
|
|
self._add_assignment(
|
|
assignment_id="delegated-assignment",
|
|
identity_id="identity-delegate",
|
|
account_id="account-delegate",
|
|
source="delegated",
|
|
delegated_from_assignment_id="owner-assignment",
|
|
valid_until=utc_now() + timedelta(hours=1),
|
|
)
|
|
self.assertTrue(self._decision("account-delegate").allowed)
|
|
|
|
with self.database.SessionLocal() as session:
|
|
delegated = session.get(
|
|
IdmOrganizationFunctionAssignment,
|
|
"delegated-assignment",
|
|
)
|
|
delegated.valid_until = utc_now() - timedelta(seconds=1)
|
|
session.commit()
|
|
|
|
expired = self._decision("account-delegate")
|
|
self.assertFalse(expired.allowed)
|
|
self.assertEqual(expired.reason_code, "effective_assignment_missing")
|
|
|
|
def test_unread_message_is_projected_as_current_work_until_read(self) -> None:
|
|
self._add_assignment(
|
|
assignment_id="owner-assignment",
|
|
identity_id="identity-owner",
|
|
account_id="account-owner",
|
|
)
|
|
principal = ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-owner",
|
|
membership_id="membership-owner",
|
|
tenant_id="tenant-1",
|
|
identity_id="identity-owner",
|
|
scopes=frozenset({"postbox:postbox:read"}),
|
|
function_assignment_ids=frozenset({"owner-assignment"}),
|
|
),
|
|
account=object(),
|
|
user=object(),
|
|
)
|
|
provider = PostboxWorkItemProvider(service=self.service)
|
|
with self.database.SessionLocal() as session:
|
|
message = PostboxMessage(
|
|
tenant_id="tenant-1",
|
|
postbox_id=self.postbox_id,
|
|
subject="Review the submitted evidence",
|
|
status="delivered",
|
|
classification="internal",
|
|
sender_label="Permit service",
|
|
delivered_at=utc_now(),
|
|
wrapped_keys=[],
|
|
external_recipient_tokens=[],
|
|
metadata_={},
|
|
)
|
|
session.add(message)
|
|
session.commit()
|
|
|
|
page = provider.list_items(
|
|
session,
|
|
principal,
|
|
query=WorkItemQuery(tenant_id="tenant-1"),
|
|
)
|
|
self.assertEqual(1, page.total)
|
|
self.assertEqual(message.id, page.items[0].id)
|
|
self.assertEqual("owner-assignment", page.items[0].assignments[0].id)
|
|
|
|
self.service.mark_message(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
message_id=message.id,
|
|
actor=actor_from_principal(principal),
|
|
state="read",
|
|
)
|
|
session.commit()
|
|
self.assertEqual(
|
|
0,
|
|
provider.list_items(
|
|
session,
|
|
principal,
|
|
query=WorkItemQuery(tenant_id="tenant-1"),
|
|
).total,
|
|
)
|
|
|
|
def test_real_organization_state_and_function_move_fail_closed(self) -> None:
|
|
self._add_assignment(
|
|
assignment_id="owner-assignment",
|
|
identity_id="identity-owner",
|
|
account_id="account-owner",
|
|
)
|
|
self.assertTrue(self._decision("account-owner").allowed)
|
|
|
|
with self.database.SessionLocal() as session:
|
|
function = session.get(OrganizationFunction, "function-one")
|
|
function.is_active = False
|
|
session.commit()
|
|
inactive_function = self._decision("account-owner")
|
|
self.assertEqual(
|
|
inactive_function.reason_code,
|
|
"organization_function_inactive",
|
|
)
|
|
|
|
with self.database.SessionLocal() as session:
|
|
function = session.get(OrganizationFunction, "function-one")
|
|
function.is_active = True
|
|
unit = session.get(OrganizationUnit, "unit-one")
|
|
unit.is_active = False
|
|
session.commit()
|
|
inactive_unit = self._decision("account-owner")
|
|
self.assertEqual(inactive_unit.reason_code, "organization_unit_inactive")
|
|
|
|
with self.database.SessionLocal() as session:
|
|
unit = session.get(OrganizationUnit, "unit-one")
|
|
unit.is_active = True
|
|
function = session.get(OrganizationFunction, "function-one")
|
|
function.organization_unit_id = "unit-two"
|
|
session.commit()
|
|
moved = self._decision("account-owner")
|
|
self.assertEqual(moved.reason_code, "organization_function_reassigned")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|