DataOps-platform/tests/data_research/test_graph_projection.py

from __future__ import annotations

import pytest


class FakeGraph:
    def __init__(self):
        self.calls = []

    def run(self, statement, **parameters):
        self.calls.append((statement, parameters))


def test_published_element_projection_is_uid_based_and_idempotent():
    from app.core.data_research.graph_projection import DataElementGraphProjector

    graph = FakeGraph()
    projector = DataElementGraphProjector(graph)
    payload = {
        "uid": "element-1",
        "version": 3,
        "status": "published",
        "snapshot": {"code": "ORDER_ID", "name_zh": "订单编号"},
    }

    projector.project(payload)
    projector.project(payload)

    assert len(graph.calls) == 2
    assert "MERGE (element:DataElement {uid: $uid})" in graph.calls[0][0]
    assert graph.calls[0][1]["uid"] == "element-1"
    assert graph.calls[0][1]["version"] == 3


def test_unpublished_element_is_never_projected():
    from app.core.data_research.graph_projection import DataElementGraphProjector
    from app.core.data_research.errors import DataElementProjectionInvalid

    graph = FakeGraph()
    with pytest.raises(DataElementProjectionInvalid, match="published"):
        DataElementGraphProjector(graph).project(
            {"uid": "element-1", "version": 2, "status": "in_review", "snapshot": {}}
        )
    assert graph.calls == []