diff --git a/CHANGELOG.md b/CHANGELOG.md index 7620f95f..cd3b7472 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Global default persistent storage for `ProvenanceManager`, plus a working `provenance` CLI** (#795, #802) by @Sameer6305 and @KaifAhmad1 + - Every ingestion/processing module (`kg_provenance.py`, `pipeline_provenance.py`, and 20+ other call sites) instantiated its own `ProvenanceManager()` with no `storage_path`, so all of them silently fell back to `InMemoryStorage` and the SQLite audit trail was never actually written. `ProvenanceManager.set_default_storage_path(path)` now sets a class-level default that every no-arg instantiation picks up, and `Semantica.__init__` wires `config.provenance.storage_path` into it automatically during orchestrator init + - Added the thread-safe `default_storage_path(path)` context manager (`semantica.provenance.default_storage_path`) for test isolation — it stacks nested overrides and guarantees restoration of the previous default on exit, even on exception, so tests can't leak global state into each other + - Fixed `ProvenanceManager.__init__` raising `TypeError` on the CLI's `config=` kwarg, and implemented the four methods the CLI already called but that didn't exist on the class: `lineage()`, `audit_log()`, `export_prov()` (W3C PROV-O turtle/ntriples/jsonld via `rdflib`), and `check()` — unblocking `semantica provenance lineage|audit|export|check` end-to-end + - Follow-up review fixes: `track_entity` no longer aliases a caller-supplied `used_entities` list (it copied the reference and later mutated it in place via `.append()`, which could corrupt a list the caller still held); removed dead fallback branches in `orchestrator.py`/`manager.py` left over from not realizing `Config.get()` already resolves dotted paths; added a `--dry-run` option to `provenance audit` to match `provenance export` (previously only the global `--dry-run` flag worked, not a local one); and `provenance check --strict` no longer prints a green "✓" success line immediately before failing — a failing check now renders as a warning before the `ClickException` is raised + - **Markdown round-trip export/import for `AgentMemory`** (#765, #786) by @SaurabhScripts and @Sameer6305 - `AgentMemory.export(format="markdown")` and `import_data(format="markdown")` add a human-editable, diff-friendly alternative to the existing JSON/dict serialization: one Markdown file per memory item, with `id`, `created_at`, `updated_at`, and `type`/`kind` in required YAML frontmatter and the memory content as the Markdown body - Exporting without a `destination` returns a single memory as a Markdown string; exporting a set requires a destination directory and writes one stable, content-hashed filename per memory ID, so re-exporting an unchanged set is byte-for-byte idempotent diff --git a/semantica/cli.py b/semantica/cli.py index c9c362e6..56d8e9a0 100644 --- a/semantica/cli.py +++ b/semantica/cli.py @@ -2523,9 +2523,11 @@ def provenance_lineage(cli_ctx: CLIContext, entity_id: str, depth: int, local_js default="table", show_default=True) @click.option("--output", default=None, type=click.Path()) @click.option("--json", "local_json", is_flag=True, default=False) +@click.option("--dry-run", "local_dry", is_flag=True, default=False) @click.pass_obj def provenance_audit(cli_ctx: CLIContext, since: Optional[str], fmt: str, - output: Optional[str], local_json: bool) -> None: + output: Optional[str], local_json: bool, + local_dry: bool) -> None: """Export the audit log. \b @@ -2535,6 +2537,9 @@ def provenance_audit(cli_ctx: CLIContext, since: Optional[str], fmt: str, cli_ctx = _require_ctx(cli_ctx) def _action() -> None: + if _is_dry(cli_ctx, local_dry): + _dry(cli_ctx, "export audit log", since=since, format=fmt, output=output) + return try: from .provenance import ProvenanceManager pm = ProvenanceManager(config=cli_ctx.config.to_dict()) @@ -2601,10 +2606,17 @@ def provenance_check(cli_ctx: CLIContext, strict: bool, local_json: bool) -> Non result = pm.check(strict=strict) except ImportError as exc: raise click.ClickException(f"Provenance module not available: {exc}") from exc + is_valid = not isinstance(result, dict) or result.get("valid", True) if _is_json(cli_ctx, local_json): _jecho(result if isinstance(result, dict) else {"valid": bool(result)}) - else: + elif is_valid: _ok(cli_ctx, f"Provenance check: {result}") + else: + _warn(cli_ctx, f"Provenance check: {result}") + if strict and not is_valid: + raise click.ClickException( + f"Provenance integrity check failed: {result.get('errors')} error(s)" + ) _run_with_error_handling(_action) diff --git a/semantica/core/config_manager.py b/semantica/core/config_manager.py index 219bd157..1446c0bf 100644 --- a/semantica/core/config_manager.py +++ b/semantica/core/config_manager.py @@ -51,7 +51,6 @@ Author: Semantica Contributors License: MIT """ -import json import os from pathlib import Path from typing import Any, Dict, List, Optional, Union @@ -67,7 +66,6 @@ from ..utils.helpers import ( set_nested_value, ) from ..utils.progress_tracker import get_progress_tracker -from ..utils.validators import validate_config class Config: @@ -157,6 +155,7 @@ class Config: self.security = config_data.get( "security", DEFAULT_CONFIG.get("security", {}) ) + self.provenance = config_data.get("provenance", {}) self.custom = config_data.get("custom", {}) def _load_from_env(self, config_dict: Dict[str, Any]) -> None: @@ -346,6 +345,7 @@ class Config: "logging": self.logging, "quality": self.quality, "security": self.security, + "provenance": self.provenance, "custom": self.custom, } diff --git a/semantica/core/orchestrator.py b/semantica/core/orchestrator.py index e9d9a138..2d1ead68 100644 --- a/semantica/core/orchestrator.py +++ b/semantica/core/orchestrator.py @@ -27,11 +27,11 @@ License: MIT from pathlib import Path from typing import Any, Dict, List, Optional, Union -from ..utils.exceptions import ConfigurationError, ProcessingError +from ..utils.exceptions import ProcessingError from ..utils.logging import get_logger, log_execution_time from ..utils.progress_tracker import get_progress_tracker from .config_manager import Config, ConfigManager -from .lifecycle import LifecycleManager, SystemState +from .lifecycle import LifecycleManager from .plugin_registry import PluginRegistry @@ -92,7 +92,10 @@ class Semantica: # Configure global provenance storage path if specified try: - prov_storage_path = self.config.get("provenance", {}).get("storage_path") + # Config.get() already resolves dotted paths against nested dicts + # (see get_nested_value), so this single lookup covers both + # top-level and config.provenance={"storage_path": ...} shapes. + prov_storage_path = self.config.get("provenance.storage_path") if prov_storage_path: from ..provenance import ProvenanceManager ProvenanceManager.set_default_storage_path(prov_storage_path) @@ -640,10 +643,10 @@ class Semantica: try: # Import key modules to verify they're available # These imports don't create instances, just verify module availability - from ..ingest import FileIngestor - from ..kg import GraphBuilder - from ..parse import DocumentParser - from ..pipeline import PipelineBuilder + from ..ingest import FileIngestor # noqa: F401 + from ..kg import GraphBuilder # noqa: F401 + from ..parse import DocumentParser # noqa: F401 + from ..pipeline import PipelineBuilder # noqa: F401 self.logger.debug("Framework modules verified and available") except (ImportError, OSError) as e: diff --git a/semantica/provenance/__init__.py b/semantica/provenance/__init__.py index 706f804a..c6faaea4 100644 --- a/semantica/provenance/__init__.py +++ b/semantica/provenance/__init__.py @@ -43,7 +43,7 @@ License: MIT from .schemas import ProvenanceEntry, SourceReference from .storage import ProvenanceStorage, InMemoryStorage, SQLiteStorage -from .manager import ProvenanceManager +from .manager import ProvenanceManager, default_storage_path from .integrity import compute_checksum, verify_checksum __all__ = [ @@ -58,6 +58,7 @@ __all__ = [ # Manager "ProvenanceManager", + "default_storage_path", # Utilities "compute_checksum", diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index 85b516ce..ab0abb26 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -24,15 +24,28 @@ Author: Semantica Contributors License: MIT """ -from typing import Optional, List, Dict, Any +from typing import Optional, List, Dict, Any, Union from collections.abc import Mapping from datetime import datetime +from contextlib import contextmanager +import copy +import threading -from .schemas import ProvenanceEntry, SourceReference, PropertySource +from .schemas import ProvenanceEntry, SourceReference from .storage import ProvenanceStorage, InMemoryStorage, SQLiteStorage from .integrity import compute_checksum +@contextmanager +def default_storage_path(path: Optional[str]): + """ + Context manager for temporarily setting ProvenanceManager._default_storage_path. + Guarantees restoration to the previous value on exit (safe for tests). + """ + with ProvenanceManager.default_storage_path(path): + yield + + class ProvenanceManager: """ Unified provenance tracking manager. @@ -55,17 +68,38 @@ class ProvenanceManager: """ _default_storage_path: Optional[str] = None + _lock = threading.RLock() + _path_stack: List[Optional[str]] = [] @classmethod - def set_default_storage_path(cls, path: str) -> None: + def set_default_storage_path(cls, path: Optional[str]) -> None: """Set a global default storage path for all new instances.""" - cls._default_storage_path = path + with cls._lock: + cls._default_storage_path = path + + @classmethod + @contextmanager + def default_storage_path(cls, path: Optional[str]): + """ + Context manager for temporarily setting the global default storage path. + Guarantees restoration to the previous value on exit (safe for tests). + """ + cls._lock.acquire() + try: + cls._path_stack.append(cls._default_storage_path) + cls._default_storage_path = path + yield + finally: + cls._default_storage_path = ( + cls._path_stack.pop() if cls._path_stack else None + ) + cls._lock.release() def __init__( self, storage: Optional[ProvenanceStorage] = None, storage_path: Optional[str] = None, - config: Optional[Dict[str, Any]] = None, + config: Optional[Any] = None, **kwargs ): """ @@ -74,17 +108,27 @@ class ProvenanceManager: Args: storage: Custom storage backend (optional) storage_path: Path to SQLite database (optional, uses in-memory if None) - config: Configuration dictionary (optional) + config: Configuration dictionary or mapping (optional) """ if storage: self.storage = storage return if not storage_path and config: - storage_path = config.get("provenance", {}).get("storage_path") + if isinstance(config, Mapping): + prov_config = config.get("provenance", {}) + prov_has_path = ( + isinstance(prov_config, Mapping) + and "storage_path" in prov_config + ) + if prov_has_path: + storage_path = prov_config.get("storage_path") + elif "storage_path" in config: + storage_path = config.get("storage_path") - if not storage_path and self._default_storage_path: - storage_path = self._default_storage_path + if not storage_path: + with self._lock: + storage_path = self._default_storage_path if storage_path: self.storage = SQLiteStorage(storage_path) @@ -125,12 +169,6 @@ class ProvenanceManager: if not isinstance(entity_id, str): raise TypeError(f"entity_id must be a string, got {type(entity_id).__name__}") - if not isinstance(entity_id, str): - raise TypeError(f"entity_id must be a string, got {type(entity_id).__name__}") - - if not isinstance(entity_id, str): - raise TypeError(f"entity_id must be a string, got {type(entity_id).__name__}") - # Check if entity already exists existing = self.storage.retrieve(entity_id) parent_id = kwargs.get("parent_entity_id") @@ -166,7 +204,6 @@ class ProvenanceManager: if existing: # Create a history entry for the previous state # Use timestamp or counter for uniqueness - import copy history_entry = copy.deepcopy(existing) history_id = f"{entity_id}:v:{existing.last_updated}" @@ -201,7 +238,8 @@ class ProvenanceManager: metadata=metadata or {}, first_seen=existing.first_seen if existing else datetime.utcnow().isoformat(), last_updated=datetime.utcnow().isoformat(), - parent_entity_id=parent_id # Link to history or explicit parent + parent_entity_id=parent_id, # Link to history or explicit parent + used_entities=list(kwargs.get("used_entities", [])), ) # Make the archived history entry discoverable via trace_lineage()'s @@ -523,9 +561,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 @@ -626,3 +666,170 @@ class ProvenanceManager: if e.source_document )) } + + # === CLI Integration Methods === + + def lineage(self, entity_id: str, depth: int = 3) -> Dict[str, Any]: + """ + Get lineage information formatted for CLI display. + + Args: + entity_id: ID of the entity to trace lineage for + depth: Maximum traversal depth (default: 3) + + Returns: + Dict containing entity_id, depth, count, lineage entries, and sources + """ + base_lineage = self.get_lineage(entity_id) + entries = base_lineage.get("lineage_chain", []) + if len(entries) > depth: + entries = entries[:depth] + sources = self.get_all_sources(entity_id) + return { + "entity_id": entity_id, + "depth": depth, + "chain_length": len(entries), + "source_documents": base_lineage.get("source_documents", []), + "lineage": entries, + "entries": entries, + "sources": sources, + "metadata": base_lineage.get("metadata", {}), + } + + def audit_log( + self, since: Optional[str] = None, format: str = "table" + ) -> Union[str, List[Dict[str, Any]]]: + """ + Export audit log of provenance entries. + + Args: + since: Optional ISO 8601 date string filter + format: Output format ('table', 'csv', 'json') + + Returns: + Formatted audit log as string or list of dicts + """ + entries = self.storage.retrieve_all() + if since: + entries = [e for e in entries if getattr(e, "timestamp", "") >= since] + entries.sort(key=lambda e: getattr(e, "timestamp", "")) + + if format == "json": + return [ + e.to_dict() if hasattr(e, "to_dict") else getattr(e, "__dict__", {}) + for e in entries + ] + elif format == "csv": + lines = ["entity_id,entity_type,activity_id,agent_id,timestamp"] + for e in entries: + ts = getattr(e, "timestamp", "") + lines.append( + f"{e.entity_id},{e.entity_type},{e.activity_id},{e.agent_id},{ts}" + ) + return "\n".join(lines) + else: + lines = [ + f"{'ENTITY_ID':<20} {'TYPE':<15} {'ACTIVITY':<15} {'TIMESTAMP':<25}" + ] + lines.append("-" * 75) + for e in entries: + ts = str(getattr(e, "timestamp", "")) + lines.append( + f"{str(e.entity_id):<20} {str(e.entity_type):<15}" + f" {str(e.activity_id):<15} {ts:<25}" + ) + return "\n".join(lines) + + def export_prov(self, format: str = "turtle") -> str: + """ + Export provenance as W3C PROV-O RDF. + + Args: + format: RDF format ('turtle', 'ntriples', 'jsonld') + + Returns: + Serialized RDF string + """ + from rdflib import Graph, Literal, Namespace, URIRef + from rdflib.namespace import RDF, XSD + + PROV = Namespace("http://www.w3.org/ns/prov#") + EX = Namespace("http://example.org/ns/") + + g = Graph() + g.bind("prov", PROV) + g.bind("ex", EX) + + for e in self.storage.retrieve_all(): + ent_uri = URIRef(EX[str(e.entity_id)]) + g.add((ent_uri, RDF.type, PROV.Entity)) + + if getattr(e, "timestamp", None): + g.add( + ( + ent_uri, + PROV.generatedAtTime, + Literal(e.timestamp, datatype=XSD.dateTime), + ) + ) + + if getattr(e, "agent_id", None) and e.agent_id != "unknown": + ag_uri = URIRef(EX[str(e.agent_id)]) + g.add((ag_uri, RDF.type, PROV.Agent)) + g.add((ent_uri, PROV.wasAttributedTo, ag_uri)) + + if getattr(e, "activity_id", None) and e.activity_id != "unknown": + act_uri = URIRef(EX[str(e.activity_id)]) + g.add((act_uri, RDF.type, PROV.Activity)) + g.add((ent_uri, PROV.wasGeneratedBy, act_uri)) + + if getattr(e, "parent_entity_id", None): + p_uri = URIRef(EX[str(e.parent_entity_id)]) + g.add((ent_uri, PROV.wasDerivedFrom, p_uri)) + + for u_id in getattr(e, "used_entities", []): + u_uri = URIRef(EX[str(u_id)]) + # Emit wasDerivedFrom only when this used entity is not the same + # as parent_entity_id — which already carries that triple above. + if u_id != getattr(e, "parent_entity_id", None): + g.add((ent_uri, PROV.wasDerivedFrom, u_uri)) + if getattr(e, "activity_id", None) and e.activity_id != "unknown": + act_uri = URIRef(EX[str(e.activity_id)]) + g.add((act_uri, PROV.used, u_uri)) + + rdf_format = "json-ld" if format == "jsonld" else format + return g.serialize(format=rdf_format) + + def check(self, strict: bool = False) -> Dict[str, Any]: + """ + Validate provenance integrity. + + Args: + strict: Whether to perform strict validation + + Returns: + Dictionary with validation results + """ + entries = self.storage.retrieve_all() + all_ids = {e.entity_id for e in entries} + + missing_refs = [] + for e in entries: + if getattr(e, "parent_entity_id", None): + if e.parent_entity_id not in all_ids: + missing_refs.append(f"{e.entity_id} -> {e.parent_entity_id}") + for u_id in getattr(e, "used_entities", []): + if u_id not in all_ids: + missing_refs.append(f"{e.entity_id} -> {u_id}") + + valid = len(missing_refs) == 0 + errors = len(missing_refs) + + return { + "valid": valid, + "total_entries": len(entries), + "missing_references": missing_refs, + "strict": strict, + "errors": errors, + } + diff --git a/tests/provenance/test_manager.py b/tests/provenance/test_manager.py index 6d704d8f..157e8300 100644 --- a/tests/provenance/test_manager.py +++ b/tests/provenance/test_manager.py @@ -11,7 +11,130 @@ from semantica.provenance import ProvenanceManager, SourceReference class TestProvenanceManager: """Test ProvenanceManager functionality.""" - + + @pytest.fixture(autouse=True) + def reset_default_storage_path(self): + original = ProvenanceManager._default_storage_path + try: + yield + finally: + ProvenanceManager._default_storage_path = original + + def test_default_storage_path_pattern(self, tmp_path): + """Test global default storage pattern, config kwarg, and test-isolation context manager.""" + from semantica.provenance import default_storage_path, InMemoryStorage, SQLiteStorage + + # 1. Default should fallback to InMemoryStorage when no path/config + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, InMemoryStorage) + + # 2. Config kwarg should extract storage_path gracefully (CLI bug fix) + cfg_path = str(tmp_path / "cfg_test.db") + cfg = {"provenance": {"storage_path": cfg_path}} + prov_mgr = ProvenanceManager(config=cfg) + assert isinstance(prov_mgr.storage, SQLiteStorage) + + # 3. Explicit storage_path arg overrides config and default + explicit_path = str(tmp_path / "explicit.db") + prov_mgr = ProvenanceManager(storage_path=explicit_path, config=cfg) + assert isinstance(prov_mgr.storage, SQLiteStorage) + + # 4. default_storage_path context manager should temporarily set default and restore on exit + ctx_path = str(tmp_path / "ctx_test.db") + with default_storage_path(ctx_path): + assert ProvenanceManager._default_storage_path == ctx_path + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, SQLiteStorage) + + # Should be restored after context exit + assert ProvenanceManager._default_storage_path is None + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, InMemoryStorage) + + # 5. Guaranteed restoration even on exception + exc_path = str(tmp_path / "exception_test.db") + try: + with ProvenanceManager.default_storage_path(exc_path): + assert ProvenanceManager._default_storage_path == exc_path + raise RuntimeError("Test exception inside context") + except RuntimeError: + pass + assert ProvenanceManager._default_storage_path is None + + # 6. Nested contexts should restore correctly from stack + outer_path = str(tmp_path / "outer.db") + inner_path = str(tmp_path / "inner.db") + with default_storage_path(outer_path): + assert ProvenanceManager._default_storage_path == outer_path + with default_storage_path(inner_path): + assert ProvenanceManager._default_storage_path == inner_path + assert ProvenanceManager._default_storage_path == outer_path + assert ProvenanceManager._default_storage_path is None + + def test_default_storage_path_concurrency_safety(self, tmp_path): + """Test that default_storage_path context manager is thread-safe under concurrent use.""" + import threading + import time + from semantica.provenance import default_storage_path, SQLiteStorage + + errors = [] + + def _worker(path, delay): + try: + with default_storage_path(path): + time.sleep(delay) + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, SQLiteStorage) + assert prov_mgr.storage.db_path == path + except Exception as e: + errors.append(e) + + path_a = str(tmp_path / "thread_a.db") + path_b = str(tmp_path / "thread_b.db") + + t1 = threading.Thread(target=_worker, args=(path_a, 0.05)) + t2 = threading.Thread(target=_worker, args=(path_b, 0.01)) + t1.start() + t2.start() + t1.join() + t2.join() + + assert not errors, f"Errors in concurrent context worker: {errors}" + assert ProvenanceManager._default_storage_path is None + + def test_isolation_part_1_with_context_manager(self, tmp_path): + """Part 1: Prove context manager sets default storage path for ProvenanceManager created inside context.""" + from semantica.provenance import default_storage_path, SQLiteStorage + iso_db = str(tmp_path / "iso_part1.db") + with default_storage_path(iso_db): + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, SQLiteStorage) + assert prov_mgr.storage.db_path == iso_db + + def test_isolation_part_2_no_leakage_after_context(self): + """Part 2: Prove that in a SEPARATE test function after context exit, ProvenanceManager falls back to InMemoryStorage without leakage.""" + from semantica.provenance import InMemoryStorage + assert ProvenanceManager._default_storage_path is None + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, InMemoryStorage) + + def test_orchestrator_config_wiring(self, tmp_path): + """Test Semantica orchestrator configures ProvenanceManager._default_storage_path.""" + from semantica.core import Semantica + from semantica.provenance import ProvenanceManager, InMemoryStorage, SQLiteStorage + + # 1. No path set in config means InMemoryStorage fallback + _ = Semantica() + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, InMemoryStorage) + + # 2. Config with provenance.storage_path sets default storage path + test_db = str(tmp_path / "orch_test.db") + _ = Semantica(config={"provenance": {"storage_path": test_db}}) + assert ProvenanceManager._default_storage_path == test_db + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, SQLiteStorage) + def test_initialization(self): """Test manager initialization.""" prov_mgr = ProvenanceManager() @@ -407,3 +530,71 @@ class TestProvenanceManager: assert e2.used_entities[0] in entity_ids, ( "Archived history entry should be reachable via used_entities in lineage" ) + + def test_cli_lineage(self): + """Test CLI lineage wrapper method on ProvenanceManager.""" + prov_mgr = ProvenanceManager() + prov_mgr.track_entity("e_cli", source="doc_cli") + res = prov_mgr.lineage("e_cli", depth=2) + 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): + """Test CLI audit_log wrapper method on ProvenanceManager.""" + prov_mgr = ProvenanceManager() + prov_mgr.track_entity("e_audit", source="doc_audit") + res_table = prov_mgr.audit_log(format="table") + assert isinstance(res_table, str) + assert "ENTITY_ID" in res_table + res_csv = prov_mgr.audit_log(format="csv") + assert "entity_id,entity_type,activity_id,agent_id,timestamp" in res_csv + res_json = prov_mgr.audit_log(format="json") + assert isinstance(res_json, list) + + def test_cli_export_prov(self): + """Test CLI export_prov method on ProvenanceManager.""" + prov_mgr = ProvenanceManager() + prov_mgr.track_entity("e_parent", source="doc_rdf") + prov_mgr.track_entity( + "e_child", + source="doc_rdf", + parent_entity_id="e_parent", + used_entities=["e_parent"], + activity_id="act_transform", + ) + ttl = prov_mgr.export_prov(format="turtle") + assert "prov:Entity" in ttl or "http://www.w3.org/ns/prov#Entity" in ttl + assert "wasDerivedFrom" in ttl + assert "used" in ttl + + def test_cli_check(self): + """Test CLI check integrity method on ProvenanceManager.""" + prov_mgr = ProvenanceManager() + prov_mgr.track_entity("e_valid_parent", source="doc_1") + prov_mgr.track_entity( + "e_valid_child", + source="doc_1", + parent_entity_id="e_valid_parent", + used_entities=["e_valid_parent"], + ) + check_res = prov_mgr.check(strict=True) + assert check_res["valid"] is True + assert check_res["total_entries"] >= 2 + assert check_res["errors"] == 0 + + # Confirm non-strict mode still reports valid=False and errors > 0 when references are missing + prov_mgr.track_entity( + "e_broken_child", + source="doc_1", + parent_entity_id="non_existent_parent", + ) + check_broken = prov_mgr.check(strict=False) + assert check_broken["valid"] is False + assert check_broken["errors"] >= 1 + assert "e_broken_child -> non_existent_parent" in check_broken["missing_references"] +