DataOps-platform/tests/integration/test_governance_metrics_postgres.py

from __future__ import annotations

import os
import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import text

pytestmark = pytest.mark.integration


def test_governance_metrics_apply_scope_and_return_safe_details(monkeypatch):
    platform_url = os.environ.get("TEST_DATABASE_URL")
    if not platform_url:
        pytest.skip("TEST_DATABASE_URL is required")

    monkeypatch.setenv("DATABASE_URL", platform_url)
    from app import create_app, db
    from app.core.data_research.governance_metric_repository import (
        SqlAlchemyGovernanceMetricRepository,
    )
    from app.core.data_research.governance_metrics import GovernanceMetricAccess
    from app.models.data_research import (
        DeviceAsset,
        DeviceAssetSourceMapping,
        DeviceEntityMatchCandidate,
        DeviceEntityMatchReview,
        DeviceEntityMergeEvent,
        DeviceEntityMergeRollback,
        DeviceQualityIssue,
        DeviceQualityProfile,
        DeviceQualityProfileVersion,
        DeviceQualityRun,
        IngestionSource,
    )

    app = create_app()
    app.config.update(TESTING=True)
    suffix = uuid.uuid4().hex[:10]
    actor_uid = str(uuid.uuid4())
    now = datetime.now(UTC).replace(microsecond=0)
    domain_a = str(uuid.uuid4())
    domain_b = str(uuid.uuid4())
    source_uids = {
        "a": str(uuid.uuid4()),
        "b": str(uuid.uuid4()),
        "u": str(uuid.uuid4()),
    }
    asset_uids = {
        key: str(uuid.uuid4())
        for key in ("a1", "a2", "a3", "b", "u", "retired")
    }
    mapping_uids = {}
    candidate_uids = []
    merge_uids = []
    issue_uids = []
    profile_uid = str(uuid.uuid4())
    profile_version_uid = str(uuid.uuid4())
    run_uid = str(uuid.uuid4())
    baseline_admin = None

    try:
        with app.app_context():
            baseline_admin = SqlAlchemyGovernanceMetricRepository(
                db.session
            ).summary_counts(GovernanceMetricAccess(global_access=True))
            db.session.execute(
                text(
                    """
                    INSERT INTO public.users (
                        id, username, display_name, password_hash, status
                    ) VALUES (
                        CAST(:id AS uuid), :username, :username,
                        'integration-test', 'active'
                    )
                    """
                ),
                {"id": actor_uid, "username": f"wp11-actor-{suffix}"},
            )
            for key, scope in (
                ("a", {"business_domains": [domain_a]}),
                ("b", {"business_domains": [domain_b]}),
                ("u", {}),
            ):
                db.session.add(
                    IngestionSource(
                        uid=source_uids[key],
                        source_type="database",
                        name=f"WP11 source {key} {suffix}",
                        config={"password": f"source-secret-{key}"},
                        permission_scope=scope,
                        status="active",
                        created_by=actor_uid,
                    )
                )
            db.session.flush()
            for key, _source_key, complete, status in (
                ("a1", "a", True, "active"),
                ("a2", "a", False, "active"),
                ("a3", "a", True, "active"),
                ("b", "b", True, "active"),
                ("u", "u", True, "active"),
                ("retired", "a", True, "retired"),
            ):
                db.session.add(
                    DeviceAsset(
                        uid=asset_uids[key],
                        asset_type="device",
                        name=f"WP11设备-{suffix}-{key}",
                        status=status,
                        current_version=1,
                        content_hash=(key[0] * 64),
                        location=f"{key}车间",
                        organization="设备部" if complete else " ",
                        responsible_person="张工" if complete else None,
                        attributes={"password": f"asset-secret-{key}"},
                        created_by=actor_uid,
                        updated_by=actor_uid,
                        updated_at=now,
                    )
                )
            db.session.flush()
            for key, source_key, _complete, _status in (
                ("a1", "a", True, "active"),
                ("a2", "a", False, "active"),
                ("a3", "a", True, "active"),
                ("b", "b", True, "active"),
                ("u", "u", True, "active"),
                ("retired", "a", True, "retired"),
            ):
                mapping_uid = str(uuid.uuid4())
                mapping_uids[key] = mapping_uid
                db.session.add(
                    DeviceAssetSourceMapping(
                        uid=mapping_uid,
                        asset_uid=asset_uids[key],
                        source_uid=source_uids[source_key],
                        source_entity="asset.equipment",
                        asset_type="device",
                        source_code=f"EQ-WP11-{suffix}-{key}",
                        source_updated_at=now,
                        first_seen_at=now,
                        last_seen_at=now,
                    )
                )
            db.session.flush()

            def add_merge(left_key, right_key, *, rolled_back=False):
                candidate_uid = str(uuid.uuid4())
                review_uid = str(uuid.uuid4())
                merge_uid = str(uuid.uuid4())
                candidate_uids.append(candidate_uid)
                merge_uids.append(merge_uid)
                db.session.add(
                    DeviceEntityMatchCandidate(
                        uid=candidate_uid,
                        left_asset_uid=asset_uids[left_key],
                        right_asset_uid=asset_uids[right_key],
                        canonical_asset_uid=asset_uids[left_key],
                        status="rolled_back" if rolled_back else "merged",
                        suggestion_source="manual",
                        confidence=1,
                        explanation=[],
                        evidence_uids=[],
                        current_version=2 if rolled_back else 1,
                        created_by=actor_uid,
                    )
                )
                db.session.flush()
                db.session.add(
                    DeviceEntityMatchReview(
                        uid=review_uid,
                        candidate_uid=candidate_uid,
                        version=1,
                        decision="approve",
                        reason="WP11 integration",
                        actor_uid=actor_uid,
                    )
                )
                db.session.flush()
                db.session.add(
                    DeviceEntityMergeEvent(
                        uid=merge_uid,
                        candidate_uid=candidate_uid,
                        canonical_asset_uid=asset_uids[left_key],
                        member_asset_uid=asset_uids[right_key],
                        review_uid=review_uid,
                        snapshot={"password": "merge-secret"},
                        actor_uid=actor_uid,
                    )
                )
                db.session.flush()
                if rolled_back:
                    db.session.add(
                        DeviceEntityMergeRollback(
                            uid=str(uuid.uuid4()),
                            merge_uid=merge_uid,
                            candidate_uid=candidate_uid,
                            reason="WP11 rollback",
                            snapshot={"password": "rollback-secret"},
                            actor_uid=actor_uid,
                        )
                    )

            add_merge("a1", "a2")
            add_merge("a3", "b")
            add_merge("a1", "a3", rolled_back=True)

            db.session.add(
                DeviceQualityProfile(
                    uid=profile_uid,
                    code=f"wp11-{suffix}",
                    name="WP11 integration",
                    created_by=actor_uid,
                )
            )
            db.session.flush()
            db.session.add(
                DeviceQualityProfileVersion(
                    uid=profile_version_uid,
                    profile_uid=profile_uid,
                    version=1,
                    status="published",
                    rules=[],
                    content_hash=suffix.ljust(64, "0"),
                    created_by=actor_uid,
                    published_by=actor_uid,
                    published_at=now,
                )
            )
            db.session.flush()
            db.session.add(
                DeviceQualityRun(
                    uid=run_uid,
                    policy_version_uid=profile_version_uid,
                    policy_hash=suffix.ljust(64, "0"),
                    source_uid=source_uids["a"],
                    status="success",
                    total_assets=3,
                    total_violations=3,
                    score=50,
                    created_by=actor_uid,
                )
            )
            db.session.flush()
            for index, (asset_key, source_key, status, occurrence) in enumerate(
                (
                    ("a1", "a", "closed", 1),
                    ("a2", "a", "open", 2),
                    ("b", "b", "closed", 1),
                ),
                start=1,
            ):
                issue_uid = str(uuid.uuid4())
                issue_uids.append(issue_uid)
                db.session.add(
                    DeviceQualityIssue(
                        uid=issue_uid,
                        issue_code=f"W11{suffix[:6]}{index:02d}",
                        source_violation_uid=str(uuid.uuid4()),
                        source_run_uid=run_uid,
                        rule_code="asset_context_complete",
                        severity="error",
                        priority="high",
                        asset_uid=asset_uids[asset_key],
                        field_name="organization",
                        source_uid=source_uids[source_key],
                        source_mapping_uid=mapping_uids[asset_key],
                        message=f"issue-secret-{asset_key}",
                        evidence={"password": f"evidence-secret-{asset_key}"},
                        recurrence_key=(f"{suffix}-{asset_key}").ljust(64, "0"),
                        occurrence_number=occurrence,
                        status=status,
                        assignee_uid=actor_uid,
                        due_at=now - timedelta(days=1),
                        current_version=1,
                        created_by=actor_uid,
                        updated_by=actor_uid,
                        created_at=now,
                        updated_at=now,
                        closed_at=now if status == "closed" else None,
                    )
                )
            db.session.commit()

            repository = SqlAlchemyGovernanceMetricRepository(db.session)
            admin = GovernanceMetricAccess(global_access=True)
            domain_viewer = GovernanceMetricAccess(
                global_access=False,
                business_domain_uids=(domain_a,),
            )

            assert baseline_admin is not None
            assert repository.summary_counts(admin) == {
                "asset_completeness": (
                    baseline_admin["asset_completeness"][0] + 4,
                    baseline_admin["asset_completeness"][1] + 5,
                ),
                "responsibility_coverage": (
                    baseline_admin["responsibility_coverage"][0] + 4,
                    baseline_admin["responsibility_coverage"][1] + 5,
                ),
                "entity_mapping": (
                    baseline_admin["entity_mapping"][0] + 4,
                    baseline_admin["entity_mapping"][1] + 5,
                ),
                "issue_closure": (
                    baseline_admin["issue_closure"][0] + 2,
                    baseline_admin["issue_closure"][1] + 3,
                ),
                "issue_recurrence": (
                    baseline_admin["issue_recurrence"][0] + 1,
                    baseline_admin["issue_recurrence"][1] + 3,
                ),
            }
            assert repository.summary_counts(domain_viewer) == {
                "asset_completeness": (2, 3),
                "responsibility_coverage": (2, 3),
                "entity_mapping": (2, 3),
                "issue_closure": (1, 2),
                "issue_recurrence": (1, 2),
            }

            incomplete, incomplete_total = repository.list_details(
                domain_viewer,
                metric="asset_completeness",
                state="incomplete",
                page=1,
                page_size=20,
            )
            mapped, mapped_total = repository.list_details(
                domain_viewer,
                metric="entity_mapping",
                state="mapped",
                page=1,
                page_size=20,
            )
            recurrent, recurrent_total = repository.list_details(
                domain_viewer,
                metric="issue_recurrence",
                state="recurrent",
                page=1,
                page_size=20,
            )

            assert incomplete_total == 1
            assert incomplete[0]["asset_uid"] == asset_uids["a2"]
            assert incomplete[0]["missing_fields"] == [
                "organization",
                "responsible_person",
            ]
            assert mapped_total == 2
            assert {item["asset_uid"] for item in mapped} == {
                asset_uids["a1"],
                asset_uids["a2"],
            }
            assert recurrent_total == 1
            assert recurrent[0]["issue_uid"] == issue_uids[1]
            assert recurrent[0]["overdue"] is True
            serialized = repr((incomplete, mapped, recurrent))
            for secret in (
                "source-secret",
                "asset-secret",
                "merge-secret",
                "issue-secret",
                "evidence-secret",
                "permission_scope",
            ):
                assert secret not in serialized
    finally:
        with app.app_context():
            for model, column, values in (
                (DeviceQualityIssue, DeviceQualityIssue.uid, issue_uids),
                (
                    DeviceEntityMergeRollback,
                    DeviceEntityMergeRollback.merge_uid,
                    merge_uids,
                ),
                (
                    DeviceEntityMergeEvent,
                    DeviceEntityMergeEvent.uid,
                    merge_uids,
                ),
                (
                    DeviceEntityMatchReview,
                    DeviceEntityMatchReview.candidate_uid,
                    candidate_uids,
                ),
                (
                    DeviceEntityMatchCandidate,
                    DeviceEntityMatchCandidate.uid,
                    candidate_uids,
                ),
            ):
                if values:
                    db.session.query(model).filter(column.in_(values)).delete(
                        synchronize_session=False
                    )
            db.session.query(DeviceQualityRun).filter_by(uid=run_uid).delete()
            db.session.query(DeviceQualityProfileVersion).filter_by(
                uid=profile_version_uid
            ).delete()
            db.session.query(DeviceQualityProfile).filter_by(
                uid=profile_uid
            ).delete()
            if asset_uids:
                db.session.query(DeviceAssetSourceMapping).filter(
                    DeviceAssetSourceMapping.asset_uid.in_(asset_uids.values())
                ).delete(synchronize_session=False)
                db.session.query(DeviceAsset).filter(
                    DeviceAsset.uid.in_(asset_uids.values())
                ).delete(synchronize_session=False)
            db.session.query(IngestionSource).filter(
                IngestionSource.uid.in_(source_uids.values())
            ).delete(synchronize_session=False)
            db.session.execute(
                text(
                    "DELETE FROM public.users "
                    "WHERE id = CAST(:id AS uuid)"
                ),
                {"id": actor_uid},
            )
            db.session.commit()