184 lines
6.6 KiB
Python
184 lines
6.6 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
import unittest
|
|
|
|
from govoplan_core.core.policy import FunctionAssignmentGovernanceDecision
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.db.session import configure_database, reset_database
|
|
from govoplan_identity.backend.db import models as identity_models # noqa: F401
|
|
from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment
|
|
from govoplan_idm.backend.delegation_routes import validate_delegation_chain
|
|
from govoplan_organizations.backend.db import models as organization_models # noqa: F401
|
|
|
|
|
|
class DelegationRouteTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.database = configure_database("sqlite:///:memory:")
|
|
Base.metadata.create_all(
|
|
self.database.engine,
|
|
tables=[IdmOrganizationFunctionAssignment.__table__],
|
|
)
|
|
self.now = datetime(2026, 8, 22, 12, tzinfo=timezone.utc)
|
|
|
|
def tearDown(self) -> None:
|
|
reset_database(dispose=True)
|
|
|
|
def assignment(
|
|
self,
|
|
assignment_id: str,
|
|
*,
|
|
source: str = "direct",
|
|
parent: str | None = None,
|
|
active: bool = True,
|
|
valid_from: datetime | None = None,
|
|
valid_until: datetime | None = None,
|
|
) -> IdmOrganizationFunctionAssignment:
|
|
return IdmOrganizationFunctionAssignment(
|
|
id=assignment_id,
|
|
tenant_id="tenant-1",
|
|
identity_id=f"identity-{assignment_id}",
|
|
account_id=f"account-{assignment_id}",
|
|
function_id="function-1",
|
|
organization_unit_id="unit-1",
|
|
source=source,
|
|
delegated_from_assignment_id=parent,
|
|
is_active=active,
|
|
valid_from=valid_from,
|
|
valid_until=valid_until,
|
|
settings={},
|
|
)
|
|
|
|
@staticmethod
|
|
def decision(
|
|
*,
|
|
allowed: bool = True,
|
|
depth: int = 3,
|
|
validity_days: int | None = None,
|
|
) -> FunctionAssignmentGovernanceDecision:
|
|
return FunctionAssignmentGovernanceDecision(
|
|
allowed=True,
|
|
delegation_allowed=allowed,
|
|
maximum_delegation_depth=depth if allowed else 0,
|
|
maximum_delegated_validity_days=validity_days,
|
|
)
|
|
|
|
def test_complete_effective_chain_is_accepted(self) -> None:
|
|
with self.database.session() as session:
|
|
root = self.assignment(
|
|
"root",
|
|
valid_from=self.now - timedelta(days=30),
|
|
valid_until=self.now + timedelta(days=30),
|
|
)
|
|
first = self.assignment(
|
|
"first",
|
|
source="delegated",
|
|
parent="root",
|
|
valid_from=self.now - timedelta(days=10),
|
|
valid_until=self.now + timedelta(days=20),
|
|
)
|
|
second = self.assignment(
|
|
"second",
|
|
source="delegated",
|
|
parent="first",
|
|
valid_from=self.now - timedelta(days=1),
|
|
valid_until=self.now + timedelta(days=5),
|
|
)
|
|
session.add_all((root, first, second))
|
|
session.flush()
|
|
|
|
route = validate_delegation_chain(
|
|
session,
|
|
assignment=second,
|
|
tenant_id="tenant-1",
|
|
function_id="function-1",
|
|
decision=self.decision(depth=2),
|
|
effective_at=self.now,
|
|
)
|
|
|
|
self.assertTrue(route.effective)
|
|
self.assertEqual(("second", "first", "root"), route.chain_assignment_ids)
|
|
self.assertEqual(2, route.delegation_depth)
|
|
|
|
def test_policy_tightening_and_over_depth_fail_closed(self) -> None:
|
|
with self.database.session() as session:
|
|
root = self.assignment("root")
|
|
first = self.assignment("first", source="delegated", parent="root")
|
|
second = self.assignment("second", source="delegated", parent="first")
|
|
session.add_all((root, first, second))
|
|
session.flush()
|
|
|
|
disabled = validate_delegation_chain(
|
|
session,
|
|
assignment=second,
|
|
tenant_id="tenant-1",
|
|
function_id="function-1",
|
|
decision=self.decision(allowed=False),
|
|
effective_at=self.now,
|
|
)
|
|
shallow = validate_delegation_chain(
|
|
session,
|
|
assignment=second,
|
|
tenant_id="tenant-1",
|
|
function_id="function-1",
|
|
decision=self.decision(depth=1),
|
|
effective_at=self.now,
|
|
)
|
|
|
|
self.assertEqual("policy_tightened", disabled.code)
|
|
self.assertEqual("over_depth", shallow.code)
|
|
|
|
def test_expired_cyclic_and_overlong_routes_are_explained(self) -> None:
|
|
with self.database.session() as session:
|
|
expired = self.assignment(
|
|
"expired",
|
|
valid_until=self.now - timedelta(seconds=1),
|
|
)
|
|
cycle_a = self.assignment("cycle-a", source="delegated", parent="cycle-b")
|
|
cycle_b = self.assignment("cycle-b", source="delegated", parent="cycle-a")
|
|
overlong = self.assignment(
|
|
"overlong",
|
|
source="delegated",
|
|
parent="root",
|
|
valid_from=self.now,
|
|
valid_until=self.now + timedelta(days=31),
|
|
)
|
|
root = self.assignment("root")
|
|
session.add_all(
|
|
(expired, cycle_a, cycle_b, root, overlong)
|
|
)
|
|
session.flush()
|
|
|
|
expired_route = validate_delegation_chain(
|
|
session,
|
|
assignment=expired,
|
|
tenant_id="tenant-1",
|
|
function_id="function-1",
|
|
decision=self.decision(),
|
|
effective_at=self.now,
|
|
)
|
|
cyclic_route = validate_delegation_chain(
|
|
session,
|
|
assignment=cycle_a,
|
|
tenant_id="tenant-1",
|
|
function_id="function-1",
|
|
decision=self.decision(),
|
|
effective_at=self.now,
|
|
)
|
|
overlong_route = validate_delegation_chain(
|
|
session,
|
|
assignment=overlong,
|
|
tenant_id="tenant-1",
|
|
function_id="function-1",
|
|
decision=self.decision(validity_days=30),
|
|
effective_at=self.now,
|
|
)
|
|
|
|
self.assertEqual("expired", expired_route.code)
|
|
self.assertEqual("cyclic", cyclic_route.code)
|
|
self.assertEqual("policy_tightened", overlong_route.code)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|