from __future__ import annotations from dataclasses import replace from datetime import datetime, timedelta import pytest SOURCE_UID = "00000000-0000-0000-0000-000000000101" class MemoryDeviceAssetRepository: def __init__(self, *, active_sources=(SOURCE_UID,)): self.active_sources = set(active_sources) self.assets = {} self.mappings = {} self.versions = {} def source_is_active(self, source_uid): return source_uid in self.active_sources @staticmethod def _mapping_key(source_uid, source_entity, asset_type, source_code): return source_uid, source_entity, asset_type, source_code def find_mapping( self, source_uid, source_entity, asset_type, source_code, *, for_update=False, ): del for_update return self.mappings.get( self._mapping_key( source_uid, source_entity, asset_type, source_code, ) ) def get(self, asset_uid): return self.assets.get(asset_uid) def create_asset(self, asset, mapping, version): self.assets[asset.uid] = asset self.mappings[ self._mapping_key( mapping.source_uid, mapping.source_entity, mapping.asset_type, mapping.source_code, ) ] = mapping self.versions[asset.uid] = [version] return asset, mapping def update_asset(self, asset, mapping, version): self.assets[asset.uid] = asset self.mappings[ self._mapping_key( mapping.source_uid, mapping.source_entity, mapping.asset_type, mapping.source_code, ) ] = mapping self.versions[asset.uid].append(version) return asset, mapping def touch_mapping(self, mapping, *, source_updated_at, last_seen_at): updated = replace( mapping, source_updated_at=source_updated_at, last_seen_at=last_seen_at, ) self.mappings[ self._mapping_key( mapping.source_uid, mapping.source_entity, mapping.asset_type, mapping.source_code, ) ] = updated return updated def search(self, filters, *, page, page_size): del filters records = list(self.assets.values()) start = (page - 1) * page_size return records[start : start + page_size], len(records) def list_mappings(self, asset_uid): return [ mapping for mapping in self.mappings.values() if mapping.asset_uid == asset_uid ] def list_versions(self, asset_uid): return list(self.versions.get(asset_uid, ())) def payload(**record_overrides): record = { "asset_type": "device", "source_code": "EQ-001", "name": "一号循环泵", "location": "一号车间", "organization": "动力部", "responsible_person": "设备管理员", "status": "active", "source_updated_at": "2026-07-29T08:00:00+08:00", "attributes": {"model": "P-100", "criticality": "A"}, } record.update(record_overrides) return { "source_uid": SOURCE_UID, "source_entity": "asset.equipment", "records": [record], } def service(repository=None): from app.core.data_research.device_assets import DeviceAssetService ids = iter( ( "00000000-0000-7000-8000-000000000201", "00000000-0000-7000-8000-000000000202", "00000000-0000-7000-8000-000000000203", "00000000-0000-7000-8000-000000000204", "00000000-0000-7000-8000-000000000205", "00000000-0000-7000-8000-000000000206", ) ) clock = iter( ( datetime(2026, 7, 29, 8, 1), datetime(2026, 7, 29, 8, 2), datetime(2026, 7, 29, 8, 3), ) ) return DeviceAssetService( repository or MemoryDeviceAssetRepository(), uid_factory=ids.__next__, now_factory=clock.__next__, ) def test_reimport_preserves_stable_uid_and_does_not_version_unchanged_data(): repository = MemoryDeviceAssetRepository() assets = service(repository) first = assets.import_records(payload(), actor_uid="editor-1") second = assets.import_records( payload(source_updated_at="2026-07-29T09:00:00+08:00"), actor_uid="editor-1", ) assert first.created_count == 1 assert first.updated_count == 0 assert second.unchanged_count == 1 assert second.items[0].asset.uid == first.items[0].asset.uid assert second.items[0].asset.current_version == 1 assert len(repository.versions[first.items[0].asset.uid]) == 1 assert second.items[0].mapping.source_updated_at.isoformat() == ( "2026-07-29T09:00:00+08:00" ) def test_changed_normalized_record_creates_one_next_immutable_version(): repository = MemoryDeviceAssetRepository() assets = service(repository) first = assets.import_records(payload(), actor_uid="editor-1") changed = assets.import_records( payload(location="二号车间"), actor_uid="editor-2", ) asset_uid = first.items[0].asset.uid assert changed.updated_count == 1 assert changed.items[0].asset.uid == asset_uid assert changed.items[0].asset.current_version == 2 assert [item.version for item in repository.versions[asset_uid]] == [1, 2] assert repository.versions[asset_uid][0].snapshot["location"] == "一号车间" assert repository.versions[asset_uid][1].snapshot["location"] == "二号车间" assert repository.versions[asset_uid][1].actor_uid == "editor-2" @pytest.mark.parametrize( "asset_type", ( "device", "component", "measurement_point", "alarm", "maintenance_record", ), ) def test_import_accepts_each_wp04_asset_type(asset_type): result = service().import_records( payload(asset_type=asset_type), actor_uid="editor-1", ) assert result.items[0].asset.asset_type == asset_type @pytest.mark.parametrize( ("bad_payload", "message"), ( ({"source_uid": "missing", "source_entity": "asset.equipment", "records": []}, "source"), (payload(asset_type="report"), "asset_type"), (payload(source_code=""), "source_code"), (payload(name=""), "name"), (payload(uid="caller-selected"), "uid"), (payload(attributes={"password": "secret"}), "secret"), (payload(attributes={"nested": {"api_key": "secret"}}), "secret"), ), ) def test_import_rejects_invalid_or_secret_bearing_records( bad_payload, message, ): from app.core.data_research.errors import DeviceAssetInvalid with pytest.raises(DeviceAssetInvalid, match=message): service().import_records(bad_payload, actor_uid="editor-1") def test_import_rejects_inactive_source_duplicate_identity_and_oversized_batch(): from app.core.data_research.errors import DeviceAssetInvalid inactive = service(MemoryDeviceAssetRepository(active_sources=())) with pytest.raises(DeviceAssetInvalid, match="active"): inactive.import_records(payload(), actor_uid="editor-1") duplicate = payload() duplicate["records"].append(dict(duplicate["records"][0])) with pytest.raises(DeviceAssetInvalid, match="duplicate"): service().import_records(duplicate, actor_uid="editor-1") oversized = payload() oversized["records"] = [ { **oversized["records"][0], "source_code": f"EQ-{index:04d}", } for index in range(501) ] with pytest.raises(DeviceAssetInvalid, match="500"): service().import_records(oversized, actor_uid="editor-1") def test_import_rejects_attributes_larger_than_the_catalog_boundary(): from app.core.data_research.errors import DeviceAssetInvalid with pytest.raises(DeviceAssetInvalid, match="attributes"): service().import_records( payload(attributes={"notes": "x" * 65536}), actor_uid="editor-1", ) def test_search_returns_asset_details_with_source_mappings(): repository = MemoryDeviceAssetRepository() assets = service(repository) imported = assets.import_records(payload(), actor_uid="editor-1") records, total = assets.search( {"keyword": "循环泵"}, page=1, page_size=20, ) assert total == 1 assert records[0].asset.uid == imported.items[0].asset.uid assert records[0].mappings == (imported.items[0].mapping,) def test_default_catalog_timestamps_keep_the_china_timezone(): from app.core.data_research.device_assets import DeviceAssetService ids = iter( ( "00000000-0000-7000-8000-000000000401", "00000000-0000-7000-8000-000000000402", "00000000-0000-7000-8000-000000000403", ) ) result = DeviceAssetService( MemoryDeviceAssetRepository(), uid_factory=ids.__next__, ).import_records(payload(), actor_uid="editor-1") created_at = result.items[0].asset.created_at assert created_at.tzinfo is not None assert created_at.utcoffset() == timedelta(hours=8) @pytest.mark.parametrize( "filters", ( {"source_uid": "not-a-uuid"}, {"asset_type": "report"}, {"status": "deleted"}, {"keyword": "x" * 301}, ), ) def test_search_rejects_invalid_catalog_filters(filters): from app.core.data_research.errors import DeviceAssetInvalid with pytest.raises(DeviceAssetInvalid): service().search(filters, page=1, page_size=20)