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