DataOps-platform/tests/data_research/test_device_semantic_api.py

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"