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"