DataOps-platform/tests/data_research/test_device_assets.py

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)