DataOps-platform/tests/integration/test_unified_work_center_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

from app.core.governance.work_center import UnifiedWorkCenterService
from app.core.governance.work_center_repository import SqlAlchemyWorkCenterRepository

pytestmark = pytest.mark.integration


@pytest.fixture()
def postgres_center():
    database_url = os.environ.get("TEST_DATABASE_URL")
    if not database_url:
        pytest.skip("TEST_DATABASE_URL is required")
    engine = create_engine(database_url)
    session = Session(engine)
    prefix = uuid.uuid4().hex[:10]
    users = [str(uuid.uuid4()) for _ in range(4)]
    for index, user_uid in enumerate(users):
        session.execute(
            text(
                """
                INSERT INTO public.users (
                    id, username, display_name, password_hash, status
                ) VALUES (
                    CAST(:uid AS uuid), :username, :display_name, 'not-a-login', 'active'
                )
                """
            ),
            {
                "uid": user_uid,
                "username": f"wp07-{prefix}-{index}@example.test",
                "display_name": f"WP07 {index}",
            },
        )
    session.commit()
    service = UnifiedWorkCenterService(
        SqlAlchemyWorkCenterRepository(session),
        commit=session.commit,
        rollback=session.rollback,
    )
    yield service, session, users, prefix
    session.rollback()
    session.execute(
        text(
            "DELETE FROM public.outbox_events WHERE aggregate_type = 'governance_task' "
            "AND aggregate_id IN (SELECT uid::text FROM public.governance_tasks WHERE source_uid LIKE :prefix)"
        ),
        {"prefix": f"{prefix}-%"},
    )
    workflow_uids = [
        row[0]
        for row in session.execute(
            text("SELECT uid FROM public.governance_workflows WHERE code = :code"),
            {"code": f"WP07_{prefix.upper()}"},
        )
    ]
    session.execute(
        text(
            "DELETE FROM public.governance_notifications WHERE related_task_uid IN "
            "(SELECT uid FROM public.governance_tasks WHERE source_uid LIKE :prefix)"
        ),
        {"prefix": f"{prefix}-%"},
    )
    session.execute(
        text("DELETE FROM public.governance_tasks WHERE source_uid LIKE :prefix"),
        {"prefix": f"{prefix}-%"},
    )
    if workflow_uids:
        session.execute(
            text(
                "UPDATE public.governance_workflows SET active_version_uid = NULL "
                "WHERE uid = ANY(CAST(:uids AS uuid[]))"
            ),
            {"uids": [str(value) for value in workflow_uids]},
        )
        session.execute(
            text(
                "DELETE FROM public.governance_workflow_versions "
                "WHERE workflow_uid = ANY(CAST(:uids AS uuid[]))"
            ),
            {"uids": [str(value) for value in workflow_uids]},
        )
        session.execute(
            text(
                "DELETE FROM public.governance_workflows "
                "WHERE uid = ANY(CAST(:uids AS uuid[]))"
            ),
            {"uids": [str(value) for value in workflow_uids]},
        )
    session.execute(
        text(
            "DELETE FROM public.governance_notification_preferences "
            "WHERE user_uid = ANY(CAST(:uids AS uuid[]))"
        ),
        {"uids": users},
    )
    session.execute(
        text(
            "DELETE FROM public.governance_notification_templates "
            "WHERE created_by = ANY(CAST(:uids AS uuid[]))"
        ),
        {"uids": users},
    )
    session.execute(
        text("DELETE FROM public.users WHERE id = ANY(CAST(:uids AS uuid[]))"),
        {"uids": users},
    )
    session.commit()
    session.close()
    engine.dispose()


def test_four_sources_share_persisted_contract_audit_and_retry(postgres_center):
    center, session, users, prefix = postgres_center
    actor, reviewer_a, reviewer_b, _reviewer_c = users
    workflow = center.create_workflow(
        {
            "code": f"WP07_{prefix.upper()}",
            "name": "WP07 PostgreSQL contract",
            "subject_types": [
                "quality_issue",
                "semantic_governance",
                "data_product",
                "agent",
            ],
            "routes": [
                {
                    "priority": 1,
                    "conditions": [
                        {"field": "risk_level", "operator": "eq", "value": "high"}
                    ],
                    "approval_mode": "dual_control",
                    "reviewer_uids": [reviewer_a, reviewer_b],
                    "min_approvals": 2,
                    "due_hours": 1,
                    "timeout_action": "escalate",
                    "notification_channels": ["in_app", "email"],
                }
            ],
            "default_route": {
                "approval_mode": "any",
                "reviewer_uids": [reviewer_a, reviewer_b],
                "min_approvals": 1,
                "due_hours": 24,
                "timeout_action": "close",
                "notification_channels": ["in_app"],
            },
        },
        actor_uid=actor,
    )
    workflow = center.publish_workflow(
        workflow["uid"], expected_version=1, actor_uid=actor
    )
    template = center.create_notification_template(
        {
            "code": "task_created",
            "channel": "email",
            "subject_template": "[DataOps] {{task_code}} {{title}}",
            "body_template": "{{description}}",
        },
        actor_uid=actor,
    )
    center.revise_notification_template(
        template["uid"],
        {
            "subject_template": "[DataOps] {{task_code}} {{title}}",
            "body_template": "待办：{{description}}",
            "status": "active",
        },
        expected_version=1,
        actor_uid=actor,
    )
    preference = center.replace_notification_preferences(
        {
            "enabled_channels": ["in_app"],
            "subscribed_events": ["task_created"],
            "quiet_hours": {},
        },
        expected_revision=0,
        actor_uid=reviewer_b,
    )
    assert preference["revision"] == 1
    tasks = []
    for subject_type in (
        "quality_issue",
        "semantic_governance",
        "data_product",
        "agent",
    ):
        tasks.append(
            center.create_task(
                {
                    "workflow_uid": workflow["uid"],
                    "task_type": "approval",
                    "subject_type": subject_type,
                    "subject_uid": f"{prefix}-{subject_type}",
                    "source_type": subject_type,
                    "source_uid": f"{prefix}-{subject_type}",
                    "title": f"Review {subject_type}",
                    "description": "PostgreSQL integration evidence",
                    "priority": "high" if subject_type == "agent" else "medium",
                    "context": {"risk_level": "high" if subject_type == "agent" else "low"},
                },
                actor_uid=actor,
            )
        )

    approved = center.review_task(
        tasks[0]["uid"],
        {"decision": "approve", "reason": "verified"},
        expected_version=1,
        actor_uid=reviewer_a,
    )
    center.close_task(
        approved["uid"],
        {"resolution": "closed with evidence", "evidence_refs": ["evidence://wp07"]},
        expected_version=2,
        actor_uid=reviewer_a,
    )
    email_attempts = center.deliver_notifications(
        "email",
        lambda _notification: (_ for _ in ()).throw(RuntimeError("SMTP unavailable")),
        at=datetime.now(UTC) + timedelta(minutes=1),
    )
    dashboard = center.dashboard(at=datetime.now(UTC) + timedelta(days=2))

    assert {item["subject_type"] for item in tasks} == {
        "quality_issue",
        "semantic_governance",
        "data_product",
        "agent",
    }
    assert all(item["source_state_unchanged"] for item in tasks)
    assert email_attempts[0]["status"] == "pending"
    assert email_attempts[0]["attempts"] == 1
    assert email_attempts[0]["subject"].startswith("[DataOps] GWT-")
    suppressed = session.execute(
        text(
            "SELECT COUNT(*) FROM public.governance_notifications "
            "WHERE recipient_uid = CAST(:uid AS uuid) AND channel = 'email' "
            "AND status = 'suppressed'"
        ),
        {"uid": reviewer_b},
    ).scalar_one()
    assert suppressed == 1
    assert dashboard["by_type"]["quality_issue"] >= 1
    assert dashboard["closed_count"] >= 1
    actions = {
        row[0]
        for row in session.execute(
            text(
                "SELECT action FROM public.governance_task_events "
                "WHERE task_uid = CAST(:uid AS uuid)"
            ),
            {"uid": tasks[0]["uid"]},
        )
    }
    assert {"created", "reviewed", "closed"} <= actions
    outbox = session.execute(
        text(
            "SELECT COUNT(*) FROM public.outbox_events "
            "WHERE aggregate_type = 'governance_task' AND aggregate_id = :uid"
        ),
        {"uid": tasks[0]["uid"]},
    ).scalar_one()
    assert outbox == 2