DataOps-platform/tests/integration/test_unified_responsibility_postgres.py

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()