DataOps-platform/tests/integration/test_device_asset_postgres.py

from __future__ import annotations

import os
import uuid
from datetime import datetime

import pytest


pytestmark = pytest.mark.integration


def test_device_assets_keep_source_identity_versions_and_search(
    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_asset_repository import (
        SqlAlchemyDeviceAssetRepository,
    )
    from app.core.data_research.device_assets import DeviceAssetService
    from app.models.data_research import (
        DeviceAsset,
        DeviceAssetSourceMapping,
        DeviceAssetVersion,
        IngestionSource,
    )

    app = create_app()
    app.config.update(TESTING=True)
    source_uid = str(uuid.uuid4())
    asset_uids = []
    try:
        with app.app_context():
            db.session.add(
                IngestionSource(
                    uid=source_uid,
                    source_type="database",
                    name="设备资产集成验收源",
                    config={
                        "database_type": "postgresql",
                        "database": "acceptance",
                        "schema": "asset",
                    },
                    permission_scope={},
                    status="active",
                    created_by="integration-test",
                )
            )
            db.session.commit()
            repository = SqlAlchemyDeviceAssetRepository(db.session)
            service = DeviceAssetService(
                repository,
                commit=db.session.commit,
                rollback=db.session.rollback,
            )

            created = service.import_records(
                {
                    "source_uid": source_uid,
                    "source_entity": "asset.equipment",
                    "records": [
                        {
                            "asset_type": "device",
                            "source_code": "EQ-PG-001",
                            "name": "循环水泵 A",
                            "location": "动力车间",
                            "organization": "设备动力部",
                            "responsible_person": "张工",
                            "source_updated_at": (
                                "2026-07-29T08:00:00+08:00"
                            ),
                            "attributes": {"model": "P-100"},
                        },
                        {
                            "asset_type": "component",
                            "source_code": "PART-PG-001",
                            "name": "循环水泵 A 轴承",
                            "location": "动力车间",
                            "organization": "设备动力部",
                            "responsible_person": "李工",
                            "attributes": {"parent_source_code": "EQ-PG-001"},
                        },
                    ],
                },
                actor_uid="integration-test",
            )
            asset_uids = [item.asset.uid for item in created.items]
            stable_uid = created.items[0].asset.uid

            unchanged = service.import_records(
                {
                    "source_uid": source_uid,
                    "source_entity": "asset.equipment",
                    "records": [
                        {
                            "asset_type": "device",
                            "source_code": "EQ-PG-001",
                            "name": "循环水泵 A",
                            "location": "动力车间",
                            "organization": "设备动力部",
                            "responsible_person": "张工",
                            "source_updated_at": (
                                "2026-07-29T09:00:00+08:00"
                            ),
                            "attributes": {"model": "P-100"},
                        }
                    ],
                },
                actor_uid="integration-test",
            )
            updated = service.import_records(
                {
                    "source_uid": source_uid,
                    "source_entity": "asset.equipment",
                    "records": [
                        {
                            "asset_type": "device",
                            "source_code": "EQ-PG-001",
                            "name": "循环水泵 A",
                            "location": "二号动力车间",
                            "organization": "设备动力部",
                            "responsible_person": "张工",
                            "source_updated_at": (
                                "2026-07-29T10:00:00+08:00"
                            ),
                            "attributes": {"model": "P-100"},
                        }
                    ],
                },
                actor_uid="integration-test-2",
            )

            assert unchanged.unchanged_count == 1
            assert updated.updated_count == 1
            assert updated.items[0].asset.uid == stable_uid
            assert updated.items[0].asset.current_version == 2

            for keyword in (
                "循环水泵",
                "EQ-PG-001",
                "二号动力车间",
                "张工",
            ):
                rows, total = repository.search(
                    {"keyword": keyword},
                    page=1,
                    page_size=20,
                )
                assert total >= 1
                assert stable_uid in {row.uid for row in rows}

            filtered, total = repository.search(
                {
                    "asset_type": "device",
                    "status": "active",
                    "source_uid": source_uid,
                },
                page=1,
                page_size=20,
            )
            assert total == 1
            assert filtered[0].uid == stable_uid

            detail = service.get(stable_uid)
            versions = service.versions(stable_uid)
            assert detail.asset.current_version == 2
            assert len(detail.mappings) == 1
            assert detail.mappings[0].source_code == "EQ-PG-001"
            assert detail.mappings[0].source_updated_at == (
                datetime.fromisoformat("2026-07-29T10:00:00+08:00")
            )
            assert [item.version for item in versions] == [2, 1]
            assert versions[0].snapshot["location"] == "二号动力车间"
            assert versions[1].snapshot["location"] == "动力车间"
    finally:
        with app.app_context():
            if asset_uids:
                db.session.query(DeviceAssetVersion).filter(
                    DeviceAssetVersion.asset_uid.in_(asset_uids)
                ).delete(synchronize_session=False)
                db.session.query(DeviceAssetSourceMapping).filter(
                    DeviceAssetSourceMapping.asset_uid.in_(asset_uids)
                ).delete(synchronize_session=False)
                db.session.query(DeviceAsset).filter(
                    DeviceAsset.uid.in_(asset_uids)
                ).delete(synchronize_session=False)
            db.session.query(IngestionSource).filter_by(
                uid=source_uid
            ).delete(synchronize_session=False)
            db.session.commit()