from __future__ import annotations from dataclasses import replace from datetime import UTC, datetime import pytest ONTOLOGY_UID = "01900000-0000-7000-8000-000000000601" CODE_UID = "01900000-0000-7000-8000-000000000602" class FakeDeviceSemanticService: def __init__(self): from app.core.data_research.device_semantics import ( DeviceSemanticBootstrapResult, assess_device_graph, build_device_graph, ) from app.core.data_research.ontology.models import ( DomainLink, GraphDocument, Ontology, OntologyVersion, ) graph = build_device_graph( "01900000-0000-7000-8000-000000000603" ) ontology = Ontology( uid=ONTOLOGY_UID, code="DEVICE_SEMANTIC", name="设备语义本体", owner_uid="editor-1", domain_links=( DomainLink( "01900000-0000-7000-8000-000000000603", "owner", ), ), ) version = OntologyVersion( uid="01900000-0000-7000-8000-000000000604", ontology_uid=ONTOLOGY_UID, version=1, graph_document=GraphDocument.from_dict(graph), content_hash="a" * 64, ) self.result = DeviceSemanticBootstrapResult( ontology=ontology, version=version, profile=assess_device_graph(graph), created=True, ) self.actions = [] def bootstrap(self, payload, actor_uid): self.actions.append(("bootstrap", payload, actor_uid)) return self.result def profile(self): self.actions.append(("profile",)) return replace(self.result, created=False) class FakeDeviceSemanticCodeService: def __init__(self): from app.core.data_research.device_semantics import ( DeviceSemanticCodeRecord, DeviceSemanticCodeReviewRecord, DeviceSemanticCodeVersionRecord, ) now = datetime(2026, 7, 29, 16, 0, tzinfo=UTC) self.record = DeviceSemanticCodeRecord( uid=CODE_UID, ontology_uid=ONTOLOGY_UID, code_type="fault", canonical_code="F-001", canonical_name="轴承过热", definition="轴承温度超过治理阈值", status="draft", current_version=1, source_mappings=(), evidence_uids=("evidence-1",), suggestion_source="rule", confidence=None, created_by="editor-1", updated_by="editor-1", created_at=now, updated_at=now, ) self.version = DeviceSemanticCodeVersionRecord( uid="01900000-0000-7000-8000-000000000605", code_uid=CODE_UID, version=1, snapshot={ "ontology_uid": ONTOLOGY_UID, "code_type": "fault", "canonical_code": "F-001", "canonical_name": "轴承过热", }, created_by="editor-1", created_at=now, ) self.review_record = DeviceSemanticCodeReviewRecord( uid="01900000-0000-7000-8000-000000000606", code_uid=CODE_UID, version=1, decision="approve", reason="企业故障字典已确认", actor_uid="admin-1", created_at=now, ) self.actions = [] def search(self, filters, *, page, page_size): self.actions.append(("search", filters, page, page_size)) return [self.record], 1 def create(self, payload, actor_uid): self.actions.append(("create", payload, actor_uid)) return self.record def get(self, code_uid): self.actions.append(("get", code_uid)) return self.record def revise( self, code_uid, payload, *, expected_version, actor_uid, ): self.actions.append( ("revise", code_uid, payload, expected_version, actor_uid) ) return replace( self.record, canonical_name=payload.get( "canonical_name", self.record.canonical_name, ), current_version=2, ) def submit(self, code_uid, *, expected_version, actor_uid): self.actions.append( ("submit", code_uid, expected_version, actor_uid) ) return replace(self.record, status="in_review") def review(self, code_uid, payload, *, actor_uid): self.actions.append(("review", code_uid, payload, actor_uid)) return replace(self.record, status="published"), self.review_record def versions(self, code_uid): self.actions.append(("versions", code_uid)) return [self.version] def reviews(self, code_uid): self.actions.append(("reviews", code_uid)) return [self.review_record] @pytest.fixture() def client(monkeypatch): from flask import request from app import create_app from app.api.data_development import routes from app.core.system import permissions semantic_service = FakeDeviceSemanticService() code_service = FakeDeviceSemanticCodeService() def identity(): role = request.headers.get( "Authorization", "", ).removeprefix("Bearer ") if role not in {"viewer", "editor", "admin"}: return None return {"id": f"{role}-1", "roles": [role]} monkeypatch.setattr(permissions, "authenticate_request", identity) monkeypatch.setattr( routes, "get_device_semantic_service", lambda: semantic_service, ) monkeypatch.setattr( routes, "get_device_semantic_code_service", lambda: code_service, ) app = create_app() app.config.update(TESTING=True) return app.test_client(), semantic_service, code_service def test_viewer_reads_profile_codes_and_traceability_but_cannot_edit(client): http, _semantic_service, code_service = client headers = {"Authorization": "Bearer viewer"} profile = http.get( "/api/development/v1/device-semantics/profile", headers=headers, ) codes = http.get( "/api/development/v1/device-semantics/codes" f"?ontology_uid={ONTOLOGY_UID}&code_type=fault&page=1&page_size=10", headers=headers, ) versions = http.get( f"/api/development/v1/device-semantics/codes/{CODE_UID}/versions", headers=headers, ) reviews = http.get( f"/api/development/v1/device-semantics/codes/{CODE_UID}/reviews", headers=headers, ) denied = http.post( "/api/development/v1/device-semantics/codes", headers=headers, json={}, ) assert profile.status_code == 200 assert profile.get_json()["data"]["profile"]["ready_to_publish"] is True assert codes.status_code == 200 assert codes.get_json()["data"]["records"][0][ "canonical_code" ] == "F-001" assert versions.get_json()["data"]["records"][0]["version"] == 1 assert reviews.get_json()["data"]["records"][0][ "decision" ] == "approve" assert denied.status_code == 403 assert code_service.actions[0] == ( "search", {"ontology_uid": ONTOLOGY_UID, "code_type": "fault"}, "1", "10", ) def test_editor_bootstraps_creates_and_submits_but_cannot_review(client): http, semantic_service, code_service = client headers = {"Authorization": "Bearer editor"} bootstrap = http.post( "/api/development/v1/device-semantics/bootstrap", headers=headers, json={ "owner_domain_uid": ( "01900000-0000-7000-8000-000000000603" ) }, ) created = http.post( "/api/development/v1/device-semantics/codes", headers=headers, json={ "ontology_uid": ONTOLOGY_UID, "code_type": "fault", "canonical_code": "F-001", "canonical_name": "轴承过热", }, ) submitted = http.post( f"/api/development/v1/device-semantics/codes/{CODE_UID}/submit", headers=headers, json={"expected_version": 1}, ) denied = http.post( f"/api/development/v1/device-semantics/codes/{CODE_UID}/review", headers=headers, json={ "decision": "approve", "expected_version": 1, }, ) assert bootstrap.status_code == 201 assert created.status_code == 201 assert submitted.get_json()["data"]["status"] == "in_review" assert denied.status_code == 403 assert semantic_service.actions[0][2] == "editor-1" assert code_service.actions[-1] == ( "submit", CODE_UID, 1, "editor-1", ) def test_admin_review_returns_audited_decision(client): http, _semantic_service, code_service = client response = http.post( f"/api/development/v1/device-semantics/codes/{CODE_UID}/review", headers={"Authorization": "Bearer admin"}, json={ "decision": "approve", "reason": "企业故障字典已确认", "expected_version": 1, }, ) assert response.status_code == 200 assert response.get_json()["data"]["record"]["status"] == "published" assert response.get_json()["data"]["review"]["actor_uid"] == "admin-1" assert code_service.actions[-1][3] == "admin-1"