124 lines
4.4 KiB
Python
124 lines
4.4 KiB
Python
from semantic_versioning import InMemoryObjectVersionStore, SemanticObjectVersion, classify_version_change, diff_versions
|
|
from sir import NodeKind, SemanticNode, SirSnapshot, SourceRef
|
|
|
|
|
|
def test_object_version_store_deduplicates_same_payload():
|
|
node = SemanticNode(
|
|
semantic_id="procedure.demo",
|
|
lineage_id="lineage.procedure.demo",
|
|
kind=NodeKind.PROCEDURE,
|
|
name="Demo",
|
|
qualified_name="Module.Demo",
|
|
source_ref=SourceRef(source_path="module.bsl"),
|
|
)
|
|
store = InMemoryObjectVersionStore()
|
|
|
|
first = store.append_node_version(node)
|
|
second = store.append_node_version(node)
|
|
|
|
assert first.version_id == second.version_id
|
|
assert len(store.history(node.lineage_id)) == 1
|
|
|
|
|
|
def test_object_version_store_versions_snapshot_nodes():
|
|
node = SemanticNode(
|
|
semantic_id="procedure.demo",
|
|
lineage_id="lineage.procedure.demo",
|
|
kind=NodeKind.PROCEDURE,
|
|
name="Demo",
|
|
qualified_name="Module.Demo",
|
|
source_ref=SourceRef(source_path="module.bsl"),
|
|
)
|
|
snapshot = SirSnapshot(snapshot_id="snapshot.demo", project_id="demo", nodes=[node])
|
|
store = InMemoryObjectVersionStore()
|
|
|
|
versions = store.append_snapshot_versions(snapshot, task_id="task.1")
|
|
|
|
assert versions[0].task_id == "task.1"
|
|
assert store.all_versions()[0].lineage_id == node.lineage_id
|
|
|
|
|
|
def test_object_version_diff_reports_payload_changes():
|
|
before = SemanticObjectVersion(
|
|
version_id="version.1",
|
|
lineage_id="lineage.procedure.demo",
|
|
semantic_id="procedure.demo",
|
|
object_hash="hash.1",
|
|
payload={"name": "Demo", "attributes": {"export": False, "source": "old"}},
|
|
)
|
|
after = SemanticObjectVersion(
|
|
version_id="version.2",
|
|
lineage_id="lineage.procedure.demo",
|
|
semantic_id="procedure.demo",
|
|
object_hash="hash.2",
|
|
payload={"name": "Demo", "attributes": {"export": True}, "kind": "PROCEDURE"},
|
|
)
|
|
|
|
diff = diff_versions(before, after)
|
|
|
|
assert diff.changed is True
|
|
assert [(entry.path, entry.kind) for entry in diff.entries] == [
|
|
("attributes.export", "CHANGE"),
|
|
("attributes.source", "REMOVE"),
|
|
("kind", "ADD"),
|
|
]
|
|
|
|
|
|
def test_object_version_store_finds_version_by_id():
|
|
node = SemanticNode(
|
|
semantic_id="procedure.demo",
|
|
lineage_id="lineage.procedure.demo",
|
|
kind=NodeKind.PROCEDURE,
|
|
name="Demo",
|
|
qualified_name="Module.Demo",
|
|
source_ref=SourceRef(source_path="module.bsl"),
|
|
)
|
|
store = InMemoryObjectVersionStore()
|
|
version = store.append_node_version(node)
|
|
|
|
assert store.find_version(version.version_id) == version
|
|
|
|
|
|
def test_classify_version_change_detects_object_rename():
|
|
before = SemanticObjectVersion(
|
|
version_id="version.1",
|
|
lineage_id="lineage.document.order",
|
|
semantic_id="document.order",
|
|
object_hash="hash.1",
|
|
payload={"name": "ЗаказПокупателя", "qualified_name": "Документ.ЗаказПокупателя"},
|
|
)
|
|
after = SemanticObjectVersion(
|
|
version_id="version.2",
|
|
lineage_id="lineage.document.order",
|
|
semantic_id="document.order",
|
|
object_hash="hash.2",
|
|
payload={"name": "ЗаказКлиента", "qualified_name": "Документ.ЗаказКлиента"},
|
|
)
|
|
|
|
change = classify_version_change(before, after)
|
|
|
|
assert change.change_kind == "RENAME"
|
|
assert {entry.path for entry in change.diff.entries} == {"name", "qualified_name"}
|
|
|
|
|
|
def test_classify_version_change_detects_object_move():
|
|
before = SemanticObjectVersion(
|
|
version_id="version.1",
|
|
lineage_id="lineage.form.order",
|
|
semantic_id="form.order",
|
|
object_hash="hash.1",
|
|
payload={"name": "ФормаДокумента", "qualified_name": "Документ.ЗаказПокупателя.ФормаДокумента"},
|
|
)
|
|
after = SemanticObjectVersion(
|
|
version_id="version.2",
|
|
lineage_id="lineage.form.order",
|
|
semantic_id="form.order",
|
|
object_hash="hash.2",
|
|
payload={"name": "ФормаДокумента", "qualified_name": "Документ.ЗаказКлиента.ФормаДокумента"},
|
|
)
|
|
|
|
change = classify_version_change(before, after)
|
|
|
|
assert change.change_kind == "MOVE"
|
|
assert [entry.path for entry in change.diff.entries] == ["qualified_name"]
|