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