DataOps-platform/tests/data_research/test_candidate_decisions.py

from __future__ import annotations

import pytest


class CandidateRepository:
    def __init__(self):
        self.candidates = {
            "candidate-1": {
                "uid": "candidate-1",
                "normalized_data": {
                    "code": "ORDER_ID",
                    "name_zh": "订单编号",
                    "name_en": "order_id",
                    "definition": "订单的唯一业务标识",
                    "logical_type": "string",
                },
                "evidence_uids": ["evidence-1"],
                "status": "candidate",
            },
            "candidate-2": {
                "uid": "candidate-2",
                "normalized_data": {"name_zh": "金额"},
                "evidence_uids": ["evidence-2"],
                "status": "candidate",
            },
        }
        self.decisions = []
        self.commits = 0
        self.rollbacks = 0

    def get(self, uid):
        return self.candidates.get(uid)

    def save_decisions(self, decisions):
        self.decisions.extend(decisions)

    def commit(self):
        self.commits += 1

    def rollback(self):
        self.rollbacks += 1


class ElementService:
    def __init__(self):
        self.created = []

    def create_draft(self, payload, actor_uid):
        self.created.append((payload, actor_uid))
        return type("Element", (), {"uid": "element-new", "status": "draft"})()


def test_batch_decisions_are_atomic_and_retain_evidence():
    from app.core.data_research.candidate_decisions import CandidateDecisionService

    repository = CandidateRepository()
    elements = ElementService()
    service = CandidateDecisionService(
        repository,
        data_elements=elements,
        uid_factory=iter(("decision-1", "decision-2")).__next__,
    )
    decisions = service.decide(
        [
            {"candidate_uid": "candidate-1", "action": "create"},
            {
                "candidate_uid": "candidate-2",
                "action": "map",
                "data_element_uid": "element-existing",
            },
        ],
        actor_uid="editor-1",
    )

    assert repository.commits == 1
    assert repository.rollbacks == 0
    assert decisions[0].data_element_uid == "element-new"
    assert decisions[0].evidence_uids == ("evidence-1",)
    assert elements.created[0][0]["evidence_uids"] == ["evidence-1"]
    assert elements.created[0][0].get("status") is None
    assert decisions[1].data_element_uid == "element-existing"


def test_invalid_batch_rolls_back_without_partial_decisions():
    from app.core.data_research.candidate_decisions import CandidateDecisionService
    from app.core.data_research.errors import CandidateDecisionInvalid

    repository = CandidateRepository()
    service = CandidateDecisionService(
        repository,
        data_elements=ElementService(),
    )

    with pytest.raises(CandidateDecisionInvalid, match="data_element_uid"):
        service.decide(
            [
                {"candidate_uid": "candidate-1", "action": "ignore"},
                {"candidate_uid": "candidate-2", "action": "reuse"},
            ],
            actor_uid="editor-1",
        )

    assert repository.decisions == []
    assert repository.commits == 0
    assert repository.rollbacks == 1


def test_candidate_decision_never_directly_publishes():
    from app.core.data_research.candidate_decisions import CandidateDecisionService
    from app.core.data_research.errors import CandidateDecisionInvalid

    repository = CandidateRepository()
    service = CandidateDecisionService(repository, data_elements=ElementService())

    with pytest.raises(CandidateDecisionInvalid, match="unsupported action"):
        service.decide(
            [{"candidate_uid": "candidate-1", "action": "publish"}],
            actor_uid="editor-1",
        )