DataOps-platform/tests/data_research/test_ontology_change_sets.py

from __future__ import annotations

import pytest


def suggestions():
    from app.core.data_research.ontology.suggestions import Suggestion

    return [
        Suggestion("s-1", "class", {"uid": "customer", "name": "Customer"}, ("domain-1",), 0.95, "rule"),
        Suggestion("s-2", "property", {"uid": "name", "name": "name", "class_uid": "customer"}, ("element-1",), 0.9, "rule"),
    ]


def test_partial_decisions_apply_only_accepted_or_edited_changes():
    from app.core.data_research.ontology.change_sets import ChangeSetService

    service = ChangeSetService(uid_factory=lambda: "change-set-1")
    change_set = service.create("ontology-1", suggestions(), actor_uid="editor-1")
    decided = service.decide(
        change_set,
        [{"suggestion_uid": "s-1", "decision": "accept", "reason": "符合业务定义"}],
        actor_uid="reviewer-1",
    )
    graph = service.apply({"classes": [], "properties": []}, decided)

    assert graph["classes"][0]["uid"] == "customer"
    assert graph["properties"] == []  # undecided suggestions never apply
    assert decided.decisions[0].actor_uid == "reviewer-1"
    assert decided.decisions[0].reason == "符合业务定义"


def test_edit_reject_and_conflict_detection_are_audited():
    from app.core.data_research.ontology.change_sets import ChangeSetConflict, ChangeSetService

    service = ChangeSetService(uid_factory=lambda: "change-set-1")
    change_set = service.create("ontology-1", suggestions(), actor_uid="editor-1")
    decided = service.decide(
        change_set,
        [
            {"suggestion_uid": "s-1", "decision": "edit", "payload": {"uid": "customer", "name": "Customer360"}, "reason": "统一命名"},
            {"suggestion_uid": "s-2", "decision": "reject", "reason": "定义不充分"},
        ],
        actor_uid="reviewer-1",
    )
    assert service.apply({"classes": [], "properties": []}, decided)["classes"][0]["name"] == "Customer360"

    duplicate = suggestions() + [replace_suggestion(suggestions()[0], uid="s-3")]
    with pytest.raises(ChangeSetConflict, match="target"):
        service.create("ontology-1", duplicate, actor_uid="editor-1")


def replace_suggestion(value, *, uid):
    from dataclasses import replace

    return replace(value, uid=uid, payload={**value.payload, "name": "Conflicting"})