DataOps-platform/tests/integration/test_device_quality_postgres.py

from __future__ import annotations

import os
import uuid

import pytest
from sqlalchemy import create_engine, text

pytestmark = pytest.mark.integration


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

    engine = create_engine(platform_url)
    try:
        with engine.connect() as connection:
            definition = connection.execute(
                text(
                    """
                    SELECT pg_get_constraintdef(oid)
                    FROM pg_constraint
                    WHERE conname =
                        'governance_responsibility_scopes_resource_type_check'
                    """
                )
            ).scalar_one()
        assert "device_quality" in definition
    finally:
        engine.dispose()


def test_device_quality_postgres_keeps_version_run_and_source_evidence(
    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.device_quality import DeviceQualityService
    from app.core.data_research.device_quality_repository import (
        SqlAlchemyDeviceQualityRepository,
    )
    from app.models.data_research import (
        DeviceAsset,
        DeviceAssetSourceMapping,
        DeviceQualityAssetScore,
        DeviceQualityProfile,
        DeviceQualityProfileVersion,
        DeviceQualityRuleResult,
        DeviceQualityRun,
        DeviceQualityViolationSample,
        IngestionSource,
    )

    app = create_app()
    app.config.update(TESTING=True)
    source_uid = str(uuid.uuid4())
    asset_uid = str(uuid.uuid4())
    mapping_uid = str(uuid.uuid4())
    profile_uids = []
    run_uids = []
    try:
        with app.app_context():
            db.session.add(
                IngestionSource(
                    uid=source_uid,
                    source_type="database",
                    name="WP07 PostgreSQL 质量验收源",
                    config={
                        "database_type": "postgresql",
                        "database": "acceptance",
                        "schema": "asset",
                    },
                    permission_scope={},
                    status="active",
                    created_by="integration-test",
                )
            )
            db.session.add(
                DeviceAsset(
                    uid=asset_uid,
                    asset_type="device",
                    name="WP07 验收设备",
                    status="active",
                    current_version=1,
                    content_hash="a" * 64,
                    location="动力车间",
                    organization="设备动力部",
                    responsible_person="张工",
                    attributes={"model": "P-100"},
                    created_by="integration-test",
                    updated_by="integration-test",
                )
            )
            db.session.flush()
            db.session.add(
                DeviceAssetSourceMapping(
                    uid=mapping_uid,
                    asset_uid=asset_uid,
                    source_uid=source_uid,
                    source_entity="asset.equipment",
                    asset_type="device",
                    source_code="EQ-WP07-001",
                )
            )
            db.session.commit()

            repository = SqlAlchemyDeviceQualityRepository(db.session)
            quality = DeviceQualityService(
                repository,
                publish_authorizer=lambda _actor: None,
                commit=db.session.commit,
                rollback=db.session.rollback,
            )
            had_profile = repository.latest_version() is not None
            draft = quality.bootstrap(actor_uid="integration-test")
            if not had_profile:
                profile_uids.append(draft.profile_uid)
            published = (
                quality.publish(
                    draft.uid,
                    actor_uid="integration-test",
                )
                if draft.status == "draft"
                else draft
            )
            run = quality.run(
                actor_uid="integration-test",
                source_uid=source_uid,
            )
            run_uids.append(run.uid)
            second_run = quality.run(
                actor_uid="integration-test",
                source_uid=source_uid,
            )
            run_uids.append(second_run.uid)

            loaded, results = quality.get_run(run.uid)
            violations, violation_total = quality.violations(
                run.uid,
                rule_code=None,
                page=1,
                page_size=100,
            )
            scores, score_total = quality.asset_scores(
                run.uid,
                page=1,
                page_size=100,
            )
            recent_runs, run_total = quality.runs(
                page=1,
                page_size=100,
            )

            assert published.status == "published"
            assert loaded.policy_version_uid == published.uid
            assert loaded.policy_hash == published.content_hash
            assert loaded.total_assets == 1
            assert loaded.score == 100
            assert len(results) == 7
            assert violation_total == 0
            assert violations == []
            assert score_total == 1
            assert scores[0].asset_uid == asset_uid
            assert scores[0].score == 100
            assert run_total >= 2
            assert {run.uid, second_run.uid} <= {
                item.uid for item in recent_runs
            }

            snapshots, total = repository.load_assets(
                source_uid=source_uid,
                limit=5_001,
            )
            assert total == 1
            assert snapshots[0].uid == asset_uid
            assert snapshots[0].current_version == 1
            assert snapshots[0].mappings[0].uid == mapping_uid
    finally:
        with app.app_context():
            if run_uids:
                db.session.query(DeviceQualityViolationSample).filter(
                    DeviceQualityViolationSample.run_uid.in_(run_uids)
                ).delete(synchronize_session=False)
                db.session.query(DeviceQualityAssetScore).filter(
                    DeviceQualityAssetScore.run_uid.in_(run_uids)
                ).delete(synchronize_session=False)
                db.session.query(DeviceQualityRuleResult).filter(
                    DeviceQualityRuleResult.run_uid.in_(run_uids)
                ).delete(synchronize_session=False)
                db.session.query(DeviceQualityRun).filter(
                    DeviceQualityRun.uid.in_(run_uids)
                ).delete(synchronize_session=False)
            if profile_uids:
                db.session.query(DeviceQualityProfileVersion).filter(
                    DeviceQualityProfileVersion.profile_uid.in_(profile_uids)
                ).delete(synchronize_session=False)
                db.session.query(DeviceQualityProfile).filter(
                    DeviceQualityProfile.uid.in_(profile_uids)
                ).delete(synchronize_session=False)
            db.session.query(DeviceAssetSourceMapping).filter_by(
                uid=mapping_uid
            ).delete(synchronize_session=False)
            db.session.query(DeviceAsset).filter_by(
                uid=asset_uid
            ).delete(synchronize_session=False)
            db.session.query(IngestionSource).filter_by(
                uid=source_uid
            ).delete(synchronize_session=False)
            db.session.commit()