Prepare GovOPlaN self-hosted release workflow
This commit is contained in:
403
tests/test_idm_api.py
Normal file
403
tests/test_idm_api.py
Normal file
@@ -0,0 +1,403 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user