DataOps-platform/tests/data_research/test_device_semantics.py

from __future__ import annotations

from dataclasses import replace
from datetime import datetime, timedelta, timezone

import pytest

DOMAIN_UID = "01900000-0000-7000-8000-000000000501"
MANAGER_UID = "01900000-0000-7000-8000-000000000502"
OTHER_UID = "01900000-0000-7000-8000-000000000503"
SOURCE_UID = "01900000-0000-7000-8000-000000000504"


def memory_repository():
    from app.core.data_research.ontology.repository import (
        MemoryOntologyRepository,
    )

    ids = iter(
        (
            "01900000-0000-7000-8000-000000000511",
            "01900000-0000-7000-8000-000000000512",
            "01900000-0000-7000-8000-000000000513",
            "01900000-0000-7000-8000-000000000514",
        )
    )
    return MemoryOntologyRepository(uid_factory=ids.__next__)


def responsibility_lookup(manager_uid=MANAGER_UID):
    def lookup(resource_type, resource_uid):
        assert resource_type == "device_ontology"
        assert resource_uid
        return {
            "resource_type": resource_type,
            "resource_uid": resource_uid,
            "revision": 1,
            "assignments": [
                {
                    "user_id": manager_uid,
                    "responsibility_role": "asset_manager",
                    "raci_role": "accountable",
                }
            ],
        }

    return lookup


def test_device_graph_is_deterministic_complete_and_generically_valid():
    from app.core.data_research.device_semantics import (
        assess_device_graph,
        build_device_graph,
    )
    from app.core.data_research.ontology.validation import validate_graph

    first = build_device_graph(DOMAIN_UID)
    second = build_device_graph(DOMAIN_UID)
    profile = assess_device_graph(first)

    assert first == second
    assert {item["uid"] for item in first["classes"]} == {
        "device",
        "component",
        "measurement_point",
        "alarm",
        "maintenance_record",
        "location",
        "organization",
        "person",
        "fault",
        "cause",
        "action",
    }
    assert {
        item["uid"] for item in first["relations"]
    } >= {
        "component.part_of",
        "measurement_point.monitors",
        "alarm.raised_by",
        "maintenance_record.maintains",
        "device.located_at",
        "device.owned_by",
        "device.responsible_to",
        "alarm.classified_as",
        "fault.caused_by",
        "maintenance_record.uses_action",
    }
    assert {
        item["property_uid"]
        for item in first["element_mappings"]
    } >= {"device.platform_uid", "device.source_code"}
    assert profile.ready_to_publish is True
    assert profile.missing_classes == ()
    assert profile.missing_relations == ()
    assert validate_graph(first) == []


def test_device_profile_reports_missing_semantic_requirements():
    from app.core.data_research.device_semantics import (
        assess_device_graph,
        build_device_graph,
    )

    graph = build_device_graph(DOMAIN_UID)
    graph["classes"] = [
        item for item in graph["classes"] if item["uid"] != "fault"
    ]
    graph["relations"] = [
        item
        for item in graph["relations"]
        if item["uid"] != "alarm.classified_as"
    ]
    graph["element_mappings"] = [
        item
        for item in graph["element_mappings"]
        if item["property_uid"] != "device.platform_uid"
    ]

    profile = assess_device_graph(graph)

    assert profile.ready_to_publish is False
    assert profile.missing_classes == ("fault",)
    assert profile.missing_relations == ("alarm.classified_as",)
    assert profile.missing_mappings == ("device.platform_uid",)


def test_device_semantic_bootstrap_is_idempotent_and_keeps_one_draft():
    from app.core.data_research.device_semantics import DeviceSemanticService

    repository = memory_repository()
    commits = []
    service = DeviceSemanticService(
        repository,
        commit=lambda: commits.append(True),
    )

    first = service.bootstrap(
        {"owner_domain_uid": DOMAIN_UID},
        actor_uid=MANAGER_UID,
    )
    second = service.bootstrap(
        {"owner_domain_uid": DOMAIN_UID},
        actor_uid=MANAGER_UID,
    )

    assert first.created is True
    assert second.created is False
    assert second.ontology.uid == first.ontology.uid
    assert second.version.uid == first.version.uid
    assert second.profile.ready_to_publish is True
    assert len(repository.list()) == 1
    assert len(repository.list_versions(first.ontology.uid)) == 1
    assert len(commits) == 1


def test_device_publication_requires_complete_profile_and_accountable_manager():
    from app.core.data_research.device_semantics import (
        DeviceOntologyPublicationAuthorizer,
        DeviceSemanticForbidden,
        DeviceSemanticInvalid,
        DeviceSemanticService,
    )

    repository = memory_repository()
    result = DeviceSemanticService(repository).bootstrap(
        {"owner_domain_uid": DOMAIN_UID},
        actor_uid=MANAGER_UID,
    )
    authorizer = DeviceOntologyPublicationAuthorizer(
        repository,
        responsibility_lookup=responsibility_lookup(),
    )

    authorizer(result.version, MANAGER_UID)
    with pytest.raises(DeviceSemanticForbidden, match="accountable"):
        authorizer(result.version, OTHER_UID)

    incomplete = replace(
        result.version,
        graph_document=result.version.graph_document.from_dict(
            {
                **result.version.graph_document.to_dict(),
                "classes": [],
            }
        ),
    )
    with pytest.raises(DeviceSemanticInvalid, match="profile"):
        authorizer(incomplete, MANAGER_UID)


def test_non_device_ontology_does_not_require_device_responsibility():
    from app.core.data_research.device_semantics import (
        DeviceOntologyPublicationAuthorizer,
    )
    from tests.data_research.test_ontology_validation import base_graph

    repository = memory_repository()
    ontology = repository.create(
        code="CUSTOMER",
        name="客户本体",
        owner_uid=MANAGER_UID,
        domain_links=base_graph()["domain_links"],
    )
    version = repository.save_draft(
        ontology.uid,
        base_graph(),
        expected_revision=0,
        actor_uid=MANAGER_UID,
    )

    DeviceOntologyPublicationAuthorizer(
        repository,
        responsibility_lookup=lambda *_args: pytest.fail(
            "generic ontology must not query device responsibility"
        ),
    )(version, OTHER_UID)


class MemorySemanticCodeRepository:
    def __init__(self, ontology_uid):
        self.ontologies = {ontology_uid: "DEVICE_SEMANTIC"}
        self.codes = {}
        self.versions_by_code = {}
        self.reviews_by_code = {}

    def ontology_exists(self, ontology_uid):
        return ontology_uid in self.ontologies

    def is_device_ontology(self, ontology_uid):
        return self.ontologies.get(ontology_uid) == "DEVICE_SEMANTIC"

    def find_code(self, ontology_uid, code_type, canonical_code):
        return next(
            (
                item
                for item in self.codes.values()
                if (
                    item.ontology_uid,
                    item.code_type,
                    item.canonical_code,
                )
                == (ontology_uid, code_type, canonical_code)
            ),
            None,
        )

    def get_code(self, code_uid, *, for_update=False):
        del for_update
        return self.codes.get(code_uid)

    def create_code(self, record, version):
        if self.find_code(
            record.ontology_uid,
            record.code_type,
            record.canonical_code,
        ):
            from app.core.data_research.errors import (
                DeviceSemanticConflict,
            )

            raise DeviceSemanticConflict("semantic code already exists")
        self.codes[record.uid] = record
        self.versions_by_code[record.uid] = [version]
        self.reviews_by_code[record.uid] = []
        return record

    def revise_code(self, record, version):
        self.codes[record.uid] = record
        self.versions_by_code[record.uid].append(version)
        return record

    def set_status(self, record, status, *, updated_at, updated_by):
        updated = replace(
            record,
            status=status,
            updated_at=updated_at,
            updated_by=updated_by,
        )
        self.codes[record.uid] = updated
        return updated

    def review_code(self, record, review, status):
        updated = replace(
            record,
            status=status,
            updated_at=review.created_at,
            updated_by=review.actor_uid,
        )
        self.codes[record.uid] = updated
        self.reviews_by_code[record.uid].append(review)
        return updated, review

    def search(self, filters, *, page, page_size):
        records = list(self.codes.values())
        for name in ("ontology_uid", "code_type", "status"):
            if filters.get(name):
                records = [
                    item
                    for item in records
                    if getattr(item, name) == filters[name]
                ]
        keyword = filters.get("keyword")
        if keyword:
            folded = keyword.casefold()
            records = [
                item
                for item in records
                if folded in item.canonical_code.casefold()
                or folded in item.canonical_name.casefold()
            ]
        records.sort(key=lambda item: (item.code_type, item.canonical_code))
        start = (page - 1) * page_size
        return records[start : start + page_size], len(records)

    def list_versions(self, code_uid):
        return list(reversed(self.versions_by_code.get(code_uid, ())))

    def list_reviews(self, code_uid):
        return list(reversed(self.reviews_by_code.get(code_uid, ())))


def code_service(repository=None, *, authorizer=None):
    from app.core.data_research.device_semantics import (
        DeviceSemanticCodeService,
    )

    ontology_uid = "01900000-0000-7000-8000-000000000521"
    ids = iter(
        (
            "01900000-0000-7000-8000-000000000531",
            "01900000-0000-7000-8000-000000000532",
            "01900000-0000-7000-8000-000000000533",
            "01900000-0000-7000-8000-000000000534",
            "01900000-0000-7000-8000-000000000535",
        )
    )
    clock = iter(
        (
            datetime(2026, 7, 29, 15, 0, tzinfo=timezone(timedelta(hours=8))),
            datetime(2026, 7, 29, 15, 1, tzinfo=timezone(timedelta(hours=8))),
            datetime(2026, 7, 29, 15, 2, tzinfo=timezone(timedelta(hours=8))),
            datetime(2026, 7, 29, 15, 3, tzinfo=timezone(timedelta(hours=8))),
        )
    )
    code_repository = repository or MemorySemanticCodeRepository(
        ontology_uid
    )
    return (
        DeviceSemanticCodeService(
            code_repository,
            uid_factory=ids.__next__,
            now_factory=clock.__next__,
            review_authorizer=authorizer or (
                lambda _ontology_uid, _actor_uid: None
            ),
        ),
        code_repository,
        ontology_uid,
    )


@pytest.mark.parametrize("code_type", ("fault", "cause", "action"))
def test_semantic_code_supports_three_governed_types_and_immutable_revision(
    code_type,
):
    service, repository, ontology_uid = code_service()
    created = service.create(
        {
            "ontology_uid": ontology_uid,
            "code_type": code_type,
            "canonical_code": f"{code_type}-001",
            "canonical_name": f"{code_type}示例",
            "definition": "首版定义",
            "source_mappings": [
                {
                    "source_uid": SOURCE_UID,
                    "source_code": "SRC-001",
                }
            ],
        },
        actor_uid=MANAGER_UID,
    )
    revised = service.revise(
        created.uid,
        {
            "canonical_name": f"{code_type}统一名称",
            "definition": "修订定义",
            "source_mappings": list(created.source_mappings),
        },
        expected_version=1,
        actor_uid=MANAGER_UID,
    )

    assert created.canonical_code == f"{code_type.upper()}-001"
    assert revised.uid == created.uid
    assert revised.current_version == 2
    versions = repository.list_versions(created.uid)
    assert [item.version for item in versions] == [2, 1]
    assert versions[0].snapshot["canonical_name"] == f"{code_type}统一名称"
    assert versions[1].snapshot["canonical_name"] == f"{code_type}示例"


def test_semantic_code_rejects_secrets_and_unproven_suggestions():
    from app.core.data_research.errors import DeviceSemanticInvalid

    service, _repository, ontology_uid = code_service()
    base = {
        "ontology_uid": ontology_uid,
        "code_type": "fault",
        "canonical_code": "F-001",
        "canonical_name": "轴承过热",
    }

    with pytest.raises(DeviceSemanticInvalid, match="secret"):
        service.create(
            {
                **base,
                "source_mappings": [
                    {
                        "source_uid": SOURCE_UID,
                        "source_code": "SRC-001",
                        "token": "secret",
                    }
                ],
            },
            actor_uid=MANAGER_UID,
        )
    with pytest.raises(DeviceSemanticInvalid, match="secret"):
        service.create(
            {
                **base,
                "contact_phone": "13800000000",
            },
            actor_uid=MANAGER_UID,
        )
    with pytest.raises(DeviceSemanticInvalid, match="evidence"):
        service.create(
            {**base, "suggestion_source": "rule"},
            actor_uid=MANAGER_UID,
        )
    with pytest.raises(DeviceSemanticInvalid, match="confidence"):
        service.create(
            {
                **base,
                "suggestion_source": "ai",
                "evidence_uids": ["evidence-1"],
                "confidence": 1.2,
            },
            actor_uid=MANAGER_UID,
        )


def test_semantic_codes_cannot_be_attached_to_a_generic_ontology():
    from app.core.data_research.errors import DeviceSemanticInvalid

    service, repository, _ontology_uid = code_service()
    generic_uid = "01900000-0000-7000-8000-000000000522"
    repository.ontologies[generic_uid] = "CUSTOMER"

    with pytest.raises(DeviceSemanticInvalid, match="device ontology"):
        service.create(
            {
                "ontology_uid": generic_uid,
                "code_type": "fault",
                "canonical_code": "F-OTHER-001",
                "canonical_name": "非设备故障",
            },
            actor_uid=MANAGER_UID,
        )
def test_semantic_code_submit_and_review_are_stateful_and_audited():
    approvals = []
    service, repository, ontology_uid = code_service(
        authorizer=lambda ontology, actor: approvals.append(
            (ontology, actor)
        )
    )
    record = service.create(
        {
            "ontology_uid": ontology_uid,
            "code_type": "fault",
            "canonical_code": "F-001",
            "canonical_name": "轴承过热",
        },
        actor_uid=OTHER_UID,
    )
    submitted = service.submit(
        record.uid,
        expected_version=1,
        actor_uid=OTHER_UID,
    )
    reviewed, review = service.review(
        record.uid,
        {
            "decision": "approve",
            "reason": "与企业故障字典一致",
            "expected_version": 1,
        },
        actor_uid=MANAGER_UID,
    )

    assert submitted.status == "in_review"
    assert reviewed.status == "published"
    assert review.version == 1
    assert review.decision == "approve"
    assert review.actor_uid == MANAGER_UID
    assert approvals == [(ontology_uid, MANAGER_UID)]
    assert repository.list_reviews(record.uid) == [review]


def test_semantic_code_rejects_stale_revision_duplicate_and_bad_transition():
    from app.core.data_research.errors import (
        DeviceSemanticConflict,
        DeviceSemanticInvalid,
    )

    service, _repository, ontology_uid = code_service()
    payload = {
        "ontology_uid": ontology_uid,
        "code_type": "cause",
        "canonical_code": "C-001",
        "canonical_name": "润滑不足",
    }
    record = service.create(payload, actor_uid=MANAGER_UID)

    with pytest.raises(DeviceSemanticConflict, match="exists"):
        service.create(payload, actor_uid=MANAGER_UID)
    with pytest.raises(DeviceSemanticConflict, match="version"):
        service.revise(
            record.uid,
            {"canonical_name": "润滑油不足"},
            expected_version=0,
            actor_uid=MANAGER_UID,
        )
    with pytest.raises(DeviceSemanticInvalid, match="in_review"):
        service.review(
            record.uid,
            {
                "decision": "approve",
                "reason": "尚未提交",
                "expected_version": 1,
            },
            actor_uid=MANAGER_UID,
        )


def test_semantic_code_search_and_traceability_are_bounded():
    service, _repository, ontology_uid = code_service()
    fault = service.create(
        {
            "ontology_uid": ontology_uid,
            "code_type": "fault",
            "canonical_code": "F-001",
            "canonical_name": "轴承过热",
        },
        actor_uid=MANAGER_UID,
    )
    service.create(
        {
            "ontology_uid": ontology_uid,
            "code_type": "cause",
            "canonical_code": "C-001",
            "canonical_name": "润滑不足",
        },
        actor_uid=MANAGER_UID,
    )

    records, total = service.search(
        {
            "ontology_uid": ontology_uid,
            "code_type": "fault",
            "keyword": "过热",
        },
        page=1,
        page_size=20,
    )

    assert total == 1
    assert records == [fault]
    assert service.get(fault.uid) == fault
    assert service.versions(fault.uid)[0].snapshot["canonical_code"] == "F-001"