DataOps-platform/tests/data_research/test_ontology_repository.py

from __future__ import annotations

import pytest


def graph(name="Customer"):
    return {
        "classes": [{"uid": "class-1", "name": name}],
        "properties": [],
        "relations": [],
        "constraints": [],
        "domain_links": [
            {"domain_uid": "domain-a", "role": "owner"},
            {"domain_uid": "domain-b", "role": "contributor"},
        ],
        "element_mappings": [],
    }


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

    repository = MemoryOntologyRepository(uid_factory=iter(("ontology-1", "version-1")).__next__)
    ontology = repository.create(
        code="CUSTOMER_360",
        name="客户360本体",
        owner_uid="owner-1",
        domain_links=graph()["domain_links"],
    )
    assert ontology.uid == "ontology-1"
    assert len(ontology.domain_links) == 2

    with pytest.raises(OntologyConflict, match="code"):
        repository.create(code="CUSTOMER_360", name="重复", owner_uid="owner-2", domain_links=[])

    draft = repository.save_draft(
        ontology.uid, graph(), expected_revision=0, actor_uid="editor-1"
    )
    assert draft.parent_version_uid is None
    assert repository.get(ontology.uid).draft_revision == 1
    with pytest.raises(OntologyConflict, match="revision"):
        repository.save_draft(
            ontology.uid, graph("Changed"), expected_revision=0, actor_uid="editor-1"
        )


def test_published_version_is_immutable_and_hash_is_canonical():
    from app.core.data_research.ontology.repository import (
        MemoryOntologyRepository,
        OntologyImmutable,
        canonical_graph_hash,
    )

    ids = iter(("ontology-1", "version-1", "version-2"))
    repository = MemoryOntologyRepository(uid_factory=ids.__next__)
    ontology = repository.create(
        code="CUSTOMER", name="客户", owner_uid="owner-1", domain_links=graph()["domain_links"]
    )
    version = repository.save_draft(ontology.uid, graph(), expected_revision=0, actor_uid="editor-1")
    published = repository.mark_published(version.uid)
    assert published.content_hash == canonical_graph_hash(graph())
    assert published.parent_version_uid is None

    with pytest.raises(OntologyImmutable, match="published"):
        repository.replace_version(version.uid, graph("Illegal"))

    next_version = repository.save_draft(
        ontology.uid, graph("CustomerAccount"), expected_revision=1, actor_uid="editor-1"
    )
    assert next_version.parent_version_uid == published.uid
    assert next_version.version == 2


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

    ids = iter(("ontology-1", "version-1", "version-2"))
    repository = MemoryOntologyRepository(uid_factory=ids.__next__)
    ontology = repository.create(
        code="CUSTOMER",
        name="客户",
        owner_uid="owner-1",
        domain_links=graph()["domain_links"],
    )
    repository.save_draft(
        ontology.uid, graph(), expected_revision=0, actor_uid="editor-1"
    )
    repository.save_draft(
        ontology.uid,
        graph("CustomerAccount"),
        expected_revision=1,
        actor_uid="editor-1",
    )

    assert [item.uid for item in repository.list_versions(ontology.uid)] == [
        "version-2",
        "version-1",
    ]