404 lines
20 KiB
Python
404 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from govoplan_access.backend.db.models import Account, User
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.configuration_control import configuration_control_snapshot, create_configuration_change_request
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
|
|
from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment, IdmTenantSettings
|
|
from govoplan_idm.backend.api.v1.routes import router as idm_router
|
|
from govoplan_idm.backend.manifest import get_manifest
|
|
from govoplan_organizations.backend.db.models import OrganizationFunction, OrganizationUnit
|
|
from tests.db_isolation import temporary_database
|
|
|
|
|
|
class IdmApiTests(unittest.TestCase):
|
|
def _principal(self, scopes: set[str]) -> ApiPrincipal:
|
|
account = Account(
|
|
id="account-idm-api",
|
|
email="idm-admin@example.test",
|
|
normalized_email="idm-admin@example.test",
|
|
display_name="IDM Admin",
|
|
)
|
|
user = User(
|
|
id="user-idm-api",
|
|
tenant_id="tenant-idm-api",
|
|
account_id=account.id,
|
|
email=account.email,
|
|
display_name=account.display_name,
|
|
settings={},
|
|
mail_profile_policy={},
|
|
)
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id=account.id,
|
|
membership_id=user.id,
|
|
tenant_id=user.tenant_id,
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=account,
|
|
user=user,
|
|
)
|
|
|
|
def _app(self, scopes: set[str]) -> FastAPI:
|
|
app = FastAPI()
|
|
app.include_router(idm_router, prefix="/api/v1")
|
|
app.dependency_overrides[get_api_principal] = lambda: self._principal(scopes)
|
|
return app
|
|
|
|
def test_idm_hard_depends_on_identity_and_organizations(self) -> None:
|
|
manifest = get_manifest()
|
|
self.assertEqual(("identity", "organizations"), manifest.dependencies)
|
|
|
|
def test_organization_identity_candidates_are_idm_bridge(self) -> None:
|
|
root = Path(tempfile.mkdtemp(prefix="govoplan-idm-api-"))
|
|
try:
|
|
with temporary_database(f"sqlite:///{root / 'idm.db'}") as database:
|
|
Base.metadata.create_all(bind=database.engine)
|
|
with database.session() as session:
|
|
identity = Identity(
|
|
id="identity-idm-alice",
|
|
display_name="Alice Registry",
|
|
external_subject="alice@example.test",
|
|
source="local",
|
|
)
|
|
session.add(identity)
|
|
session.add(
|
|
IdentityAccountLink(
|
|
identity_id=identity.id,
|
|
account_id="account-idm-alice-primary",
|
|
is_primary=True,
|
|
source="local",
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
app = self._app({"organizations:function:assign"})
|
|
with TestClient(app) as client:
|
|
response = client.get("/api/v1/idm/organization-identities", params={"query": "registry"})
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
payload = response.json()
|
|
self.assertEqual(1, len(payload["identities"]))
|
|
item = payload["identities"][0]
|
|
self.assertEqual("identity-idm-alice", item["id"])
|
|
self.assertEqual("Alice Registry", item["display_name"])
|
|
self.assertEqual("account-idm-alice-primary", item["primary_account_id"])
|
|
finally:
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
|
|
def test_organization_function_assignments_are_owned_by_idm(self) -> None:
|
|
root = Path(tempfile.mkdtemp(prefix="govoplan-idm-assignments-api-"))
|
|
try:
|
|
with temporary_database(f"sqlite:///{root / 'idm-assignments.db'}") as database:
|
|
Base.metadata.create_all(bind=database.engine)
|
|
with database.session() as session:
|
|
identity = Identity(
|
|
id="identity-idm-bob",
|
|
display_name="Bob Registry",
|
|
external_subject="bob@example.test",
|
|
source="local",
|
|
)
|
|
unit = OrganizationUnit(
|
|
id="unit-idm-registry",
|
|
tenant_id="tenant-idm-api",
|
|
slug="registry",
|
|
name="Registry",
|
|
settings={},
|
|
)
|
|
function = OrganizationFunction(
|
|
id="function-idm-clerk",
|
|
tenant_id="tenant-idm-api",
|
|
organization_unit_id=unit.id,
|
|
slug="registry-clerk",
|
|
name="Registry Clerk",
|
|
settings={},
|
|
)
|
|
session.add_all([
|
|
identity,
|
|
IdentityAccountLink(
|
|
identity_id=identity.id,
|
|
account_id="account-idm-bob-primary",
|
|
is_primary=True,
|
|
source="local",
|
|
),
|
|
unit,
|
|
function,
|
|
])
|
|
session.commit()
|
|
|
|
app = self._app({"idm:organization_assignment:read", "idm:organization_assignment:write"})
|
|
with TestClient(app) as client:
|
|
create_response = client.post(
|
|
"/api/v1/idm/organization-function-assignments",
|
|
json={
|
|
"identity_id": "identity-idm-bob",
|
|
"account_id": "account-idm-bob-primary",
|
|
"function_id": "function-idm-clerk",
|
|
"applies_to_subunits": True,
|
|
},
|
|
)
|
|
self.assertEqual(201, create_response.status_code, create_response.text)
|
|
created = create_response.json()
|
|
self.assertEqual("identity-idm-bob", created["identity_id"])
|
|
self.assertEqual("function-idm-clerk", created["function_id"])
|
|
self.assertEqual("unit-idm-registry", created["organization_unit_id"])
|
|
self.assertTrue(created["applies_to_subunits"])
|
|
|
|
list_response = client.get("/api/v1/idm/organization-function-assignments")
|
|
self.assertEqual(200, list_response.status_code, list_response.text)
|
|
self.assertEqual([created["id"]], [item["id"] for item in list_response.json()["assignments"]])
|
|
|
|
patch_response = client.patch(
|
|
f"/api/v1/idm/organization-function-assignments/{created['id']}",
|
|
json={"is_active": False, "applies_to_subunits": False},
|
|
)
|
|
self.assertEqual(200, patch_response.status_code, patch_response.text)
|
|
patched = patch_response.json()
|
|
self.assertFalse(patched["is_active"])
|
|
self.assertFalse(patched["applies_to_subunits"])
|
|
|
|
with database.session() as session:
|
|
self.assertEqual(1, session.query(IdmOrganizationFunctionAssignment).count())
|
|
finally:
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
|
|
def test_legacy_organization_assign_scope_remains_assignment_compatibility(self) -> None:
|
|
root = Path(tempfile.mkdtemp(prefix="govoplan-idm-legacy-scope-"))
|
|
try:
|
|
with temporary_database(f"sqlite:///{root / 'idm-legacy.db'}") as database:
|
|
Base.metadata.create_all(bind=database.engine)
|
|
with database.session() as session:
|
|
session.add(Identity(id="identity-idm-legacy", display_name="Legacy Person", source="local"))
|
|
session.add(IdentityAccountLink(identity_id="identity-idm-legacy", account_id="account-idm-legacy", is_primary=True, source="local"))
|
|
session.add(OrganizationUnit(id="unit-idm-legacy", tenant_id="tenant-idm-api", slug="legacy", name="Legacy", settings={}))
|
|
session.add(
|
|
OrganizationFunction(
|
|
id="function-idm-legacy",
|
|
tenant_id="tenant-idm-api",
|
|
organization_unit_id="unit-idm-legacy",
|
|
slug="legacy-clerk",
|
|
name="Legacy Clerk",
|
|
settings={},
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
app = self._app({"organizations:function:assign"})
|
|
with TestClient(app) as client:
|
|
create_response = client.post(
|
|
"/api/v1/idm/organization-function-assignments",
|
|
json={
|
|
"identity_id": "identity-idm-legacy",
|
|
"account_id": "account-idm-legacy",
|
|
"function_id": "function-idm-legacy",
|
|
},
|
|
)
|
|
self.assertEqual(201, create_response.status_code, create_response.text)
|
|
list_response = client.get("/api/v1/idm/organization-function-assignments")
|
|
self.assertEqual(200, list_response.status_code, list_response.text)
|
|
self.assertEqual([create_response.json()["id"]], [item["id"] for item in list_response.json()["assignments"]])
|
|
finally:
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
|
|
def test_idm_assignment_change_requests_can_be_required_per_tenant(self) -> None:
|
|
root = Path(tempfile.mkdtemp(prefix="govoplan-idm-change-control-"))
|
|
try:
|
|
with temporary_database(f"sqlite:///{root / 'idm-change-control.db'}") as database:
|
|
Base.metadata.create_all(bind=database.engine)
|
|
with database.session() as session:
|
|
session.add(Identity(id="identity-idm-change", display_name="Change Person", source="local"))
|
|
session.add(IdentityAccountLink(identity_id="identity-idm-change", account_id="account-idm-change", is_primary=True, source="local"))
|
|
session.add(OrganizationUnit(id="unit-idm-change", tenant_id="tenant-idm-api", slug="change", name="Change", settings={}))
|
|
session.add(
|
|
OrganizationFunction(
|
|
id="function-idm-change",
|
|
tenant_id="tenant-idm-api",
|
|
organization_unit_id="unit-idm-change",
|
|
slug="change-clerk",
|
|
name="Change Clerk",
|
|
settings={},
|
|
)
|
|
)
|
|
session.add(
|
|
IdmTenantSettings(
|
|
tenant_id="tenant-idm-api",
|
|
require_assignment_change_requests=True,
|
|
audit_detail_level="standard",
|
|
settings={},
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
app = self._app({"idm:organization_assignment:write", "idm:organization_assignment:read"})
|
|
payload = {
|
|
"identity_id": "identity-idm-change",
|
|
"account_id": "account-idm-change",
|
|
"function_id": "function-idm-change",
|
|
"applies_to_subunits": True,
|
|
}
|
|
with TestClient(app) as client:
|
|
blocked = client.post("/api/v1/idm/organization-function-assignments", json=payload)
|
|
self.assertEqual(409, blocked.status_code, blocked.text)
|
|
self.assertEqual("configuration_request_required", blocked.json()["detail"]["code"])
|
|
|
|
with database.session() as session:
|
|
request = create_configuration_change_request(
|
|
session,
|
|
key="idm.organization_assignments",
|
|
value={
|
|
"resource_type": "organization_function_assignment",
|
|
"operation": "created",
|
|
"payload": payload,
|
|
},
|
|
actor_user_id="user-idm-api",
|
|
actor_scopes=("idm:organization_assignment:write",),
|
|
dry_run=True,
|
|
target={
|
|
"tenant_id": "tenant-idm-api",
|
|
"resource_type": "organization_function_assignment",
|
|
"operation": "created",
|
|
},
|
|
)
|
|
session.commit()
|
|
|
|
with TestClient(app) as client:
|
|
applied = client.post(
|
|
"/api/v1/idm/organization-function-assignments",
|
|
json={**payload, "change_request_id": request["id"]},
|
|
)
|
|
self.assertEqual(201, applied.status_code, applied.text)
|
|
self.assertEqual("identity-idm-change", applied.json()["identity_id"])
|
|
|
|
with database.session() as session:
|
|
history = configuration_control_snapshot(session)["history"]
|
|
self.assertEqual("idm.organization_assignments", history[0]["key"])
|
|
self.assertEqual("idm.organization_assignment.updated", history[0]["audit_event"])
|
|
finally:
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
|
|
def test_idm_settings_can_be_read_and_updated(self) -> None:
|
|
root = Path(tempfile.mkdtemp(prefix="govoplan-idm-settings-"))
|
|
try:
|
|
with temporary_database(f"sqlite:///{root / 'idm-settings.db'}") as database:
|
|
Base.metadata.create_all(bind=database.engine)
|
|
app = self._app({"idm:settings:read", "idm:settings:write"})
|
|
with TestClient(app) as client:
|
|
initial = client.get("/api/v1/idm/settings")
|
|
self.assertEqual(200, initial.status_code, initial.text)
|
|
self.assertFalse(initial.json()["require_assignment_change_requests"])
|
|
|
|
updated = client.patch(
|
|
"/api/v1/idm/settings",
|
|
json={
|
|
"require_assignment_change_requests": True,
|
|
"audit_detail_level": "full",
|
|
"change_retention_days": 90,
|
|
},
|
|
)
|
|
self.assertEqual(200, updated.status_code, updated.text)
|
|
payload = updated.json()
|
|
self.assertTrue(payload["require_assignment_change_requests"])
|
|
self.assertEqual("full", payload["audit_detail_level"])
|
|
self.assertEqual(90, payload["change_retention_days"])
|
|
finally:
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
|
|
def test_delegated_assignments_require_delegable_function_and_source_assignment(self) -> None:
|
|
root = Path(tempfile.mkdtemp(prefix="govoplan-idm-delegation-rules-"))
|
|
try:
|
|
with temporary_database(f"sqlite:///{root / 'idm-delegation.db'}") as database:
|
|
Base.metadata.create_all(bind=database.engine)
|
|
with database.session() as session:
|
|
session.add_all([
|
|
Identity(id="identity-idm-source", display_name="Source Person", source="local"),
|
|
IdentityAccountLink(identity_id="identity-idm-source", account_id="account-idm-source", is_primary=True, source="local"),
|
|
Identity(id="identity-idm-delegate", display_name="Delegate Person", source="local"),
|
|
IdentityAccountLink(identity_id="identity-idm-delegate", account_id="account-idm-delegate", is_primary=True, source="local"),
|
|
OrganizationUnit(id="unit-idm-delegation", tenant_id="tenant-idm-api", slug="delegation", name="Delegation", settings={}),
|
|
OrganizationFunction(
|
|
id="function-idm-not-delegable",
|
|
tenant_id="tenant-idm-api",
|
|
organization_unit_id="unit-idm-delegation",
|
|
slug="not-delegable",
|
|
name="Not Delegable",
|
|
delegable=False,
|
|
settings={},
|
|
),
|
|
OrganizationFunction(
|
|
id="function-idm-delegable",
|
|
tenant_id="tenant-idm-api",
|
|
organization_unit_id="unit-idm-delegation",
|
|
slug="delegable",
|
|
name="Delegable",
|
|
delegable=True,
|
|
settings={},
|
|
),
|
|
])
|
|
session.add(
|
|
IdmOrganizationFunctionAssignment(
|
|
id="assignment-idm-source",
|
|
tenant_id="tenant-idm-api",
|
|
identity_id="identity-idm-source",
|
|
account_id="account-idm-source",
|
|
function_id="function-idm-delegable",
|
|
organization_unit_id="unit-idm-delegation",
|
|
applies_to_subunits=False,
|
|
source="direct",
|
|
settings={},
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
app = self._app({"idm:organization_assignment:read", "idm:organization_assignment:write"})
|
|
with TestClient(app) as client:
|
|
missing_source = client.post(
|
|
"/api/v1/idm/organization-function-assignments",
|
|
json={
|
|
"identity_id": "identity-idm-delegate",
|
|
"account_id": "account-idm-delegate",
|
|
"function_id": "function-idm-delegable",
|
|
"source": "delegated",
|
|
},
|
|
)
|
|
self.assertEqual(422, missing_source.status_code, missing_source.text)
|
|
|
|
not_delegable = client.post(
|
|
"/api/v1/idm/organization-function-assignments",
|
|
json={
|
|
"identity_id": "identity-idm-delegate",
|
|
"account_id": "account-idm-delegate",
|
|
"function_id": "function-idm-not-delegable",
|
|
"source": "delegated",
|
|
"delegated_from_assignment_id": "assignment-idm-source",
|
|
},
|
|
)
|
|
self.assertEqual(422, not_delegable.status_code, not_delegable.text)
|
|
|
|
delegated = client.post(
|
|
"/api/v1/idm/organization-function-assignments",
|
|
json={
|
|
"identity_id": "identity-idm-delegate",
|
|
"account_id": "account-idm-delegate",
|
|
"function_id": "function-idm-delegable",
|
|
"source": "delegated",
|
|
"delegated_from_assignment_id": "assignment-idm-source",
|
|
},
|
|
)
|
|
self.assertEqual(201, delegated.status_code, delegated.text)
|
|
self.assertEqual("delegated", delegated.json()["source"])
|
|
finally:
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|