DataOps-platform/tests/data_research/test_device_entity_resolution_api.py

from __future__ import annotations

from dataclasses import replace
from datetime import UTC, datetime

import pytest

CANDIDATE_UID = "01900000-0000-7000-8000-000000000701"
LEFT_UID = "01900000-0000-7000-8000-000000000702"
RIGHT_UID = "01900000-0000-7000-8000-000000000703"
MERGE_UID = "01900000-0000-7000-8000-000000000704"


class FakeDeviceEntityResolutionService:
    def __init__(self):
        from app.core.data_research.device_entity_resolution import (
            DeviceEntityCandidateRecord,
            DeviceEntityGenerationResult,
            DeviceEntityMergeRecord,
            DeviceEntityReviewRecord,
            DeviceEntityRollbackRecord,
        )

        now = datetime(2026, 7, 29, 16, 0, tzinfo=UTC)
        self.candidate = DeviceEntityCandidateRecord(
            uid=CANDIDATE_UID,
            left_asset_uid=LEFT_UID,
            right_asset_uid=RIGHT_UID,
            canonical_asset_uid=None,
            status="pending",
            suggestion_source="rule",
            confidence=0.99,
            explanation=(
                {
                    "signal": "name",
                    "matched": True,
                    "weight": 0.5,
                    "left": "一号循环泵",
                    "right": "一号循环泵",
                },
            ),
            evidence_uids=("evidence-1", "evidence-2"),
            model_provider=None,
            model_name=None,
            current_version=1,
            created_by="editor-1",
            reviewed_by=None,
            created_at=now,
            updated_at=now,
        )
        self.review_record = DeviceEntityReviewRecord(
            uid="01900000-0000-7000-8000-000000000705",
            candidate_uid=CANDIDATE_UID,
            version=1,
            decision="approve",
            reason="企业设备管理员确认",
            actor_uid="admin-1",
            created_at=now,
        )
        self.merge_record = DeviceEntityMergeRecord(
            uid=MERGE_UID,
            candidate_uid=CANDIDATE_UID,
            canonical_asset_uid=LEFT_UID,
            member_asset_uid=RIGHT_UID,
            review_uid=self.review_record.uid,
            snapshot={"canonical": {"uid": LEFT_UID}, "member": {"uid": RIGHT_UID}},
            actor_uid="admin-1",
            created_at=now,
        )
        self.rollback_record = DeviceEntityRollbackRecord(
            uid="01900000-0000-7000-8000-000000000706",
            merge_uid=MERGE_UID,
            candidate_uid=CANDIDATE_UID,
            reason="现场确认不是同一台设备",
            snapshot={"merge": {"uid": MERGE_UID}},
            actor_uid="admin-1",
            created_at=now,
        )
        self.generation = DeviceEntityGenerationResult(
            records=(self.candidate,),
            created_count=1,
            existing_count=0,
            evaluated_pair_count=1,
            auto_merged_count=0,
        )
        self.actions = []

    def search(self, filters, *, page, page_size):
        self.actions.append(("search", filters, page, page_size))
        return [self.candidate], 1

    def get(self, candidate_uid):
        self.actions.append(("get", candidate_uid))
        return self.candidate

    def generate(self, payload, actor_uid):
        self.actions.append(("generate", payload, actor_uid))
        return self.generation

    def submit_ai_candidate(self, payload, actor_uid):
        self.actions.append(("submit_ai", payload, actor_uid))
        return replace(
            self.candidate,
            suggestion_source="ai",
            model_provider=payload.get("model_provider"),
            model_name=payload.get("model_name"),
        )

    def review(self, candidate_uid, payload, actor_uid):
        self.actions.append(("review", candidate_uid, payload, actor_uid))
        return (
            replace(
                self.candidate,
                status="merged",
                canonical_asset_uid=LEFT_UID,
                current_version=2,
            ),
            self.review_record,
            self.merge_record,
        )

    def rollback(self, merge_uid, payload, actor_uid):
        self.actions.append(("rollback", merge_uid, payload, actor_uid))
        return (
            replace(
                self.candidate,
                status="rolled_back",
                canonical_asset_uid=LEFT_UID,
                current_version=3,
            ),
            self.rollback_record,
        )

    def reviews(self, candidate_uid):
        self.actions.append(("reviews", candidate_uid))
        return [self.review_record]

    def merges(self, candidate_uid):
        self.actions.append(("merges", candidate_uid))
        return [self.merge_record]

    def rollbacks(self, merge_uid):
        self.actions.append(("rollbacks", merge_uid))
        return [self.rollback_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

    service = FakeDeviceEntityResolutionService()

    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_entity_resolution_service",
        lambda: service,
        raising=False,
    )
    app = create_app()
    app.config.update(
        TESTING=True,
        DEVICE_ENTITY_AUTO_MERGE_ENABLED=False,
    )
    return app.test_client(), service


def test_viewer_reads_candidates_and_evidence_but_cannot_generate(client):
    http, service = client
    headers = {"Authorization": "Bearer viewer"}

    listed = http.get(
        "/api/development/v1/device-entities/candidates"
        "?status=pending&suggestion_source=rule&page=1&page_size=20",
        headers=headers,
    )
    detail = http.get(
        f"/api/development/v1/device-entities/candidates/{CANDIDATE_UID}",
        headers=headers,
    )
    reviews = http.get(
        f"/api/development/v1/device-entities/candidates/{CANDIDATE_UID}/reviews",
        headers=headers,
    )
    merges = http.get(
        f"/api/development/v1/device-entities/candidates/{CANDIDATE_UID}/merges",
        headers=headers,
    )
    rollbacks = http.get(
        f"/api/development/v1/device-entities/merges/{MERGE_UID}/rollbacks",
        headers=headers,
    )
    denied = http.post(
        "/api/development/v1/device-entities/candidates/generate",
        headers=headers,
        json={},
    )

    assert listed.status_code == 200
    assert listed.get_json()["data"]["auto_merge_enabled"] is False
    assert listed.get_json()["data"]["records"][0]["confidence"] == 0.99
    assert detail.get_json()["data"]["evidence_uids"] == [
        "evidence-1",
        "evidence-2",
    ]
    assert reviews.get_json()["data"]["records"][0]["decision"] == "approve"
    assert merges.get_json()["data"]["records"][0]["uid"] == MERGE_UID
    assert rollbacks.get_json()["data"]["records"][0]["reason"].startswith(
        "现场确认"
    )
    assert denied.status_code == 403
    assert service.actions[0] == (
        "search",
        {"status": "pending", "suggestion_source": "rule"},
        "1",
        "20",
    )


def test_editor_generates_and_submits_ai_candidate_but_cannot_review(client):
    http, service = client
    headers = {"Authorization": "Bearer editor"}
    generated = http.post(
        "/api/development/v1/device-entities/candidates/generate",
        headers=headers,
        json={"asset_type": "device", "threshold": 0.8},
    )
    submitted = http.post(
        "/api/development/v1/device-entities/candidates",
        headers=headers,
        json={
            "left_asset_uid": LEFT_UID,
            "right_asset_uid": RIGHT_UID,
            "confidence": 0.91,
            "model_provider": "governed-provider",
            "model_name": "entity-match-v1",
            "evidence_uids": ["evidence-1"],
            "explanation": "同一台设备",
        },
    )
    denied = http.post(
        f"/api/development/v1/device-entities/candidates/{CANDIDATE_UID}/review",
        headers=headers,
        json={
            "decision": "approve",
            "canonical_asset_uid": LEFT_UID,
            "expected_version": 1,
            "reason": "证据充分",
        },
    )

    assert generated.status_code == 201
    assert generated.get_json()["data"]["created_count"] == 1
    assert submitted.status_code == 201
    assert submitted.get_json()["data"]["suggestion_source"] == "ai"
    assert denied.status_code == 403
    assert service.actions[0][2] == "editor-1"
    assert service.actions[1][2] == "editor-1"


def test_admin_review_and_rollback_return_immutable_evidence(client):
    http, service = client
    headers = {"Authorization": "Bearer admin"}

    reviewed = http.post(
        f"/api/development/v1/device-entities/candidates/{CANDIDATE_UID}/review",
        headers=headers,
        json={
            "decision": "approve",
            "canonical_asset_uid": LEFT_UID,
            "expected_version": 1,
            "reason": "企业设备管理员确认",
        },
    )
    rolled_back = http.post(
        f"/api/development/v1/device-entities/merges/{MERGE_UID}/rollback",
        headers=headers,
        json={
            "expected_version": 2,
            "reason": "现场确认不是同一台设备",
        },
    )

    assert reviewed.status_code == 200
    assert reviewed.get_json()["data"]["candidate"]["status"] == "merged"
    assert reviewed.get_json()["data"]["review"]["actor_uid"] == "admin-1"
    assert reviewed.get_json()["data"]["merge"]["member_asset_uid"] == (
        RIGHT_UID
    )
    assert rolled_back.status_code == 200
    assert rolled_back.get_json()["data"]["candidate"]["status"] == (
        "rolled_back"
    )
    assert rolled_back.get_json()["data"]["rollback"]["merge_uid"] == (
        MERGE_UID
    )
    assert service.actions[-1][3] == "admin-1"