fix(provenance): populate entries alias in get_lineage and read lineage_chain in lineage()

- Add entries alias in get_lineage() return dictionary so CLI and programmatic callers can access lineage entries via either key

- Update lineage() wrapper method to fallback to lineage_chain when entries is missing

- Add assertions in test_cli_lineage confirming lineage and entries lists are non-empty
This commit is contained in:
Sameer6305
2026-07-26 16:10:26 +05:30
parent 2dd06aadcd
commit 6cc2e67b92
2 changed files with 7 additions and 2 deletions
+4 -2
View File
@@ -557,9 +557,11 @@ class ProvenanceManager:
if isinstance(meta, dict):
aggregated_metadata.update(meta)
chain_dicts = [entry.to_dict() for entry in lineage_entries]
return {
"entity_id": entity_id,
"lineage_chain": [entry.to_dict() for entry in lineage_entries],
"lineage_chain": chain_dicts,
"entries": chain_dicts,
"source_documents": list(set(
e.source_document for e in lineage_entries
if e.source_document
@@ -675,7 +677,7 @@ class ProvenanceManager:
Dict containing entity_id, depth, count, lineage entries, and sources
"""
base_lineage = self.get_lineage(entity_id)
entries = base_lineage.get("entries", [])
entries = base_lineage.get("lineage_chain") or base_lineage.get("entries", [])
if len(entries) > depth:
entries = entries[:depth]
sources = self.get_all_sources(entity_id)
+3
View File
@@ -498,6 +498,9 @@ class TestProvenanceManager:
assert res["entity_id"] == "e_cli"
assert res["depth"] == 2
assert isinstance(res["lineage"], list)
assert len(res["lineage"]) > 0
assert len(res["entries"]) > 0
assert res["lineage"][0]["entity_id"] == "e_cli"
assert isinstance(res["sources"], list)
def test_cli_audit_log(self):