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