DataOps-platform/tests/data_research/test_ontology_publication.py

from __future__ import annotations

import pytest

from tests.data_research.test_ontology_validation import base_graph


def repository():
    from app.core.data_research.ontology.repository import MemoryOntologyRepository

    ids = iter(("ontology-1", "version-1", "run-1", "version-2"))
    repo = MemoryOntologyRepository(uid_factory=ids.__next__)
    ontology = repo.create(
        code="CUSTOMER",
        name="客户本体",
        owner_uid="owner-1",
        domain_links=base_graph()["domain_links"],
    )
    version = repo.save_draft(ontology.uid, base_graph(), expected_revision=0, actor_uid="editor-1")
    return repo, ontology, version


def test_publish_is_idempotent_atomic_and_emits_outbox_event():
    from app.core.data_research.ontology.publication import OntologyPublicationService

    repo, ontology, version = repository()
    events = []
    service = OntologyPublicationService(
        repo,
        outbox_enqueue=lambda **event: events.append(event),
        commit=lambda: events.append({"committed": True}),
    )
    first = service.publish(version.uid, idempotency_key="publish-1", actor_uid="admin-1")
    second = service.publish(version.uid, idempotency_key="publish-1", actor_uid="admin-1")

    assert first == second
    assert repo.get(ontology.uid).active_version_uid == version.uid
    assert [event.get("event_type") for event in events].count("ontology.version_published") == 1
    assert first.status == "published"


def test_invalid_graph_rolls_back_without_activation():
    from app.core.data_research.ontology.publication import (
        OntologyPublicationService,
        OntologyValidationFailed,
    )

    repo, ontology, version = repository()
    repo.replace_version(version.uid, {**base_graph(), "domain_links": []})
    rollbacks = []
    with pytest.raises(OntologyValidationFailed):
        OntologyPublicationService(repo, rollback=lambda: rollbacks.append(True)).publish(
            version.uid, idempotency_key="invalid", actor_uid="admin-1"
        )
    assert repo.get(ontology.uid).active_version_uid is None
    assert rollbacks == [True]


class GraphSession:
    def __init__(self, fail=False):
        self.fail = fail
        self.calls = []

    def run(self, statement, **params):
        if self.fail:
            raise ConnectionError("neo4j unavailable")
        self.calls.append((statement, params))


def test_projection_uses_active_version_uid_and_failure_is_retryable():
    from app.core.data_research.ontology.projection import OntologyGraphProjector

    graph = GraphSession()
    payload = {
        "ontology_uid": "ontology-1",
        "version_uid": "version-1",
        "version": 1,
        "graph_document": base_graph(),
    }
    OntologyGraphProjector(graph).project(payload)
    assert "active_version_uid" in graph.calls[0][0]
    assert graph.calls[0][1]["version_uid"] == "version-1"

    with pytest.raises(ConnectionError, match="neo4j unavailable"):
        OntologyGraphProjector(GraphSession(fail=True)).project(payload)


def test_version_diff_and_rollback_create_new_draft():
    from app.core.data_research.ontology.publication import OntologyPublicationService, diff_graphs

    repo, ontology, version = repository()
    service = OntologyPublicationService(repo)
    service.publish(version.uid, idempotency_key="publish-1", actor_uid="admin-1")
    changed = base_graph()
    changed["classes"] = changed["classes"] + [{"uid": "account", "name": "Account"}]
    assert diff_graphs(base_graph(), changed)["classes"]["added"] == ["account"]

    rollback = service.create_rollback(
        ontology.uid,
        target_version_uid=version.uid,
        expected_revision=1,
        actor_uid="admin-1",
    )
    assert rollback.version == 2
    assert rollback.parent_version_uid == version.uid
    assert rollback.status == "draft"