from __future__ import annotations import os import uuid from datetime import UTC, datetime, timedelta import pytest from sqlalchemy import create_engine, text from sqlalchemy.orm import Session pytestmark = pytest.mark.integration OBJECT_TYPES = ( "data_asset", "semantic_term", "data_standard", "quality_policy", "data_product", "agent", ) def test_unified_responsibility_chain_delegation_policy_and_audit(): database_url = os.environ.get("TEST_DATABASE_URL") if not database_url: pytest.skip("TEST_DATABASE_URL is required") from app.core.governance.responsibilities import ( ResponsibilityService, SqlAlchemyResponsibilityRepository, ) from app.core.governance.unified_responsibilities import ( UnifiedResponsibilityService, ) from app.core.governance.unified_responsibility_repository import ( SqlAlchemyUnifiedResponsibilityRepository, ) engine = create_engine(database_url) marker = uuid.uuid4().hex[:10] users = [str(uuid.uuid4()) for _ in range(3)] actor, owner_a, owner_b = users organization_uid = f"wp06-org-{marker}" domain_a = f"wp06-domain-a-{marker}" domain_b = f"wp06-domain-b-{marker}" object_uids = { object_type: f"wp06-{object_type}-{marker}" for object_type in OBJECT_TYPES } session = Session(engine) try: for index, user_uid in enumerate(users): session.execute( text( """ INSERT INTO public.users ( id, username, display_name, password_hash, status ) VALUES ( CAST(:id AS uuid), :username, :display_name, 'not-a-login-secret', 'active' ) """ ), { "id": user_uid, "username": f"wp06_{marker}_{index}", "display_name": f"WP06 user {index}", }, ) session.commit() matrix = ResponsibilityService(SqlAlchemyResponsibilityRepository(session)) unified = UnifiedResponsibilityService( SqlAlchemyUnifiedResponsibilityRepository(session), commit=session.commit, rollback=session.rollback, ) matrix.replace( resource_type="organization", resource_uid=organization_uid, assignments=[ { "user_id": owner_a, "responsibility_role": "organization_owner", "raci_role": "accountable", } ], expected_revision=0, actor_uid=actor, ) matrix.replace( resource_type="business_domain", resource_uid=domain_b, assignments=[ { "user_id": owner_b, "responsibility_role": "domain_owner", "raci_role": "accountable", } ], expected_revision=0, actor_uid=actor, ) session.commit() unified.set_parent( { "resource_type": "business_domain", "resource_uid": domain_a, "parent_type": "organization", "parent_uid": organization_uid, }, expected_revision=0, actor_uid=actor, ) unified.set_parent( { "resource_type": "business_domain", "resource_uid": domain_b, "parent_type": "organization", "parent_uid": organization_uid, }, expected_revision=0, actor_uid=actor, ) for object_type, object_uid in object_uids.items(): unified.set_parent( { "resource_type": object_type, "resource_uid": object_uid, "parent_type": "business_domain", "parent_uid": domain_a, }, expected_revision=0, actor_uid=actor, ) resolved = { object_type: unified.resolve(object_type, object_uid) for object_type, object_uid in object_uids.items() } assert all(item["status"] == "resolved" for item in resolved.values()) assert { item["final_owners"][0]["effective_user_id"] for item in resolved.values() } == {owner_a} now = datetime.now(UTC) delegation = unified.create_delegation( { "source_user_uid": owner_a, "delegate_user_uid": owner_b, "scope_type": "business_domain", "scope_uid": domain_a, "responsibility_role": "organization_owner", "delegation_type": "temporary", "starts_at": (now - timedelta(minutes=1)).isoformat(), "ends_at": (now + timedelta(minutes=1)).isoformat(), "reason": "WP06 PostgreSQL acceptance", }, actor_uid=actor, ) delegated = unified.resolve("data_asset", object_uids["data_asset"]) assert delegated["final_owners"][0]["effective_user_id"] == owner_b unified.expire_delegations( at=now + timedelta(minutes=2), actor_uid=actor ) assert unified.resolve( "data_asset", object_uids["data_asset"], at=now + timedelta(minutes=2) )["final_owners"][0]["effective_user_id"] == owner_a central = unified.create_policy( { "code": f"WP06_CENTRAL_{marker.upper()}", "name": "WP06 central policy", "policy_type": "central_policy", "scope_type": "organization", "scope_uid": organization_uid, "definition": { "required_responsibility_roles": ["organization_owner"], "require_unique_accountable": True, "max_delegation_days": 30, }, }, actor_uid=actor, ) unified.publish_policy(central["uid"], expected_version=1, actor_uid=actor) joint = unified.create_policy( { "code": f"WP06_JOINT_{marker.upper()}", "name": "WP06 joint review", "policy_type": "joint_review", "scope_type": "data_product", "scope_uid": object_uids["data_product"], "definition": { "business_domain_uids": [domain_a, domain_b], "min_approvals": 2, "require_all_domains": True, }, }, actor_uid=actor, ) unified.publish_policy(joint["uid"], expected_version=1, actor_uid=actor) review = unified.evaluate_joint_review( "data_product", object_uids["data_product"], [ {"user_uid": owner_a, "domain_uid": domain_a, "decision": "approve"}, {"user_uid": owner_b, "domain_uid": domain_b, "decision": "approve"}, ], ) assert review["status"] == "approved" assert review["deterministic"] is True audit_count = session.execute( text( """ SELECT COUNT(*) FROM public.governance_responsibility_audit_events WHERE actor_uid = CAST(:actor_uid AS uuid) """ ), {"actor_uid": actor}, ).scalar_one() assert audit_count >= 14 assert delegation["current_version"] == 1 finally: session.rollback() session.execute( text( """ UPDATE public.governance_responsibility_policies SET active_version_uid = NULL WHERE created_by = CAST(:actor_uid AS uuid); DELETE FROM public.governance_responsibility_policy_versions WHERE created_by = CAST(:actor_uid AS uuid); DELETE FROM public.governance_responsibility_policies WHERE created_by = CAST(:actor_uid AS uuid); DELETE FROM public.governance_responsibility_delegations WHERE created_by = CAST(:actor_uid AS uuid); DELETE FROM public.governance_responsibility_hierarchy WHERE updated_by = CAST(:actor_uid AS uuid); DELETE FROM public.governance_responsibility_scopes WHERE updated_by = CAST(:actor_uid AS uuid); DELETE FROM public.governance_responsibility_audit_events WHERE actor_uid = CAST(:actor_uid AS uuid); DELETE FROM public.users WHERE id::text = ANY(:user_uids); """ ), {"actor_uid": actor, "user_uids": users}, ) session.commit() session.close() engine.dispose()