fix(provenance): reset in-memory chain state on clear

This commit is contained in:
Sameer6305
2026-08-04 00:01:50 +05:30
parent 0a8330cbb0
commit e9e05fedbd
3 changed files with 52 additions and 0 deletions
+3
View File
@@ -317,6 +317,9 @@ class InMemoryStorage(ProvenanceStorage):
"""
count = len(self._entries)
self._entries.clear()
with self._seq_lock:
self._seq_counter = 0
self._chain_head = None
return count
@contextmanager
+24
View File
@@ -466,6 +466,30 @@ class TestProvenanceManager:
lineage = prov_mgr.get_lineage("entity_1")
assert lineage == {}
def test_clear_resets_chain_state(self):
"""Regression: clear() must fully reset chain state so the first write
after clear starts a fresh chain and verify_chain() passes."""
prov_mgr = ProvenanceManager()
prov_mgr.track_entity("e1", source="doc1")
prov_mgr.track_entity("e2", source="doc1")
prov_mgr.clear()
# Chain head must be empty immediately after clear
assert prov_mgr.storage.get_chain_head() is None
# First write after clear starts a fresh chain (sequence_id=1, previous_checksum=None)
entry = prov_mgr.track_entity("e3", source="doc2")
assert entry.sequence_id == 1
assert entry.previous_checksum is None
# verify_chain() must pass on the fresh chain
prov_mgr.track_entity("e4", source="doc2")
result = prov_mgr.verify_chain()
assert result["valid"] is True
assert result["total_entries"] == 2
assert result["broken_links"] == []
def test_retrack_with_explicit_parent_overrides_history_link(self):
"""#742 — re-tracking an entity with an explicit parent_entity_id must
honor the new value, not silently replace it with an auto-generated
+25
View File
@@ -135,6 +135,31 @@ class TestInMemoryStorage:
assert count == 1
assert len(storage.retrieve_all()) == 0
def test_clear_resets_chain_state(self):
"""Issue #825 fix: clear() must reset chain state so get_chain_head()
returns None and the first write after clear starts a fresh chain."""
storage = InMemoryStorage()
entry1 = ProvenanceEntry(
entity_id="e1", entity_type="entity", activity_id="act",
sequence_id=1, checksum="checksum_1",
)
storage.store(entry1)
assert storage.get_chain_head() == (1, "checksum_1")
storage.clear()
# Chain head must be None after clear
assert storage.get_chain_head() is None
# First write after clear starts a fresh chain
entry2 = ProvenanceEntry(
entity_id="e2", entity_type="entity", activity_id="act",
sequence_id=1, checksum="checksum_fresh",
)
storage.store(entry2)
assert storage.get_chain_head() == (1, "checksum_fresh")
def test_get_chain_head(self):
"""Issue #825, Part A item 2 — chain head reporting."""
storage = InMemoryStorage()