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", ]