From 0775b0114efe9c5ef6a455d171bff0a6f8f9a78f Mon Sep 17 00:00:00 2001 From: pravit-amp <43916793+pravit-amp@users.noreply.github.com> Date: Thu, 27 Aug 2026 03:40:38 -0700 Subject: [PATCH] test(provenance): assert stored records in KG provenance suites (#946) (#1132) * fix(provenance): use timezone-aware UTC and assert stored records (#946) Replace datetime.utcnow() in ProvenanceManager, ProvenanceEntry, BridgeAxiom, and GraphBuilderWithProvenance with datetime.now(timezone.utc), matching PipelineWithProvenance. KG workflow and integration tests now read provenance back through get_provenance() and assert algorithm metadata instead of generated IDs, and call tracker methods that actually persist records. * fix(provenance): compare provenance timestamps as instants, not strings query_recorded_between() and audit_log() filtered and sorted on raw ISO strings. With the timezone-aware change, a store can hold both pre-existing naive stamps and offset-bearing ones, and the two are not string-comparable: "...500000+00:00" sorts above "...500000", so a record at the identical instant as a naive bound falls outside the range that should contain it. Both now parse through _parse_timestamp() before comparing, reading naive values as UTC. This mirrors ProvenanceTracker._parse_dt() in kg/, the class ProvenanceManager replaces, so both sides of the migration answer a range query the same way. Unparseable stored timestamps are skipped and logged rather than silently dropped; unparseable bounds raise ValueError. --------- Co-authored-by: Pravit Ampapathini --- CHANGELOG.md | 6 + docs/guides/provenance.md | 2 +- docs/reference/provenance.md | 2 +- semantica/kg/kg_provenance.py | 11 +- tests/kg/test_integration_comprehensive.py | 221 ++++++---- tests/kg/test_provenance_workflows.py | 428 ++++++++++++------- tests/kg/test_provenance_workflows_simple.py | 246 +++++++---- tests/provenance/test_manager.py | 199 ++++++++- 8 files changed, 750 insertions(+), 365 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3910a3f6..f529772f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -119,6 +119,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **KG provenance tests asserted on generated ID strings instead of stored records, and `kg_provenance.py` was missed by the `utcnow` sweep** (closes #946) by @pravit-amp + - The KG workflow and integration suites checked that a tracker call returned an ID matching a prefix (`assert cent_id.startswith("centrality_")`) without ever reading the record back, so an ID generator that returned a well-formed string and wrote nothing would have passed. Worse, some of those calls named tracker methods that do not exist anywhere in `semantica/` (`track_layer_analysis`, `track_centrality_score`), so the assertions were satisfied with no real interaction behind them + - Those tests now read provenance back through `get_provenance()` and assert on algorithm metadata, and call the methods that actually persist records. Verified by mutation rather than by a green run alone: neutering the manager's storage write (`self.storage.store(...)` → no-op) fails 10 tests + - `GraphBuilderWithProvenance` in `semantica/kg/kg_provenance.py` still stamped `activity_started_at_time`/`activity_ended_at_time` with the deprecated `datetime.utcnow()`; it was outside the `export/`+`provenance/` scope of the #1114 sweep below and now uses the same `utc_now_iso()` helper. `docs/guides/provenance.md` and `docs/reference/provenance.md` were still documenting `utcnow()` and a naive timestamp example, and now show the helper and the offset-bearing form + - 16 tests across the affected suites ended in `return ` instead of asserting, which pytest reports as `PytestReturnNotNoneWarning`; now zero + - **The temporal-evolution `stability` metric was a hardcoded placeholder, not a duration** - `TemporalGraphQuery.analyze_evolution()` documents `stability` as a "relationship duration/stability measure", but the implementation appended a constant `1` for every relationship with both `valid_from` and `valid_until` set (`durations.append(1) # Placeholder`). The reported stability was therefore always `1.0` when any bounded relationship existed and `0` otherwise — it never reflected how long relationships actually stayed valid, so it could not distinguish a graph of decade-long relationships from one of one-second relationships - `stability` now computes the mean valid-time duration in seconds (`(valid_until - valid_from).total_seconds()`) across relationships that have both bounds set. Relationships with a missing or open `valid_from`/`valid_until` are skipped (their duration is unbounded), and non-positive intervals are clamped to `0`; an empty set still reports `0` diff --git a/docs/guides/provenance.md b/docs/guides/provenance.md index a1b4d8f1..c74254b0 100644 --- a/docs/guides/provenance.md +++ b/docs/guides/provenance.md @@ -639,7 +639,7 @@ Every `ProvenanceEntry` maps directly to W3C PROV-O terms. If your compliance te | — | `previous_version_id` | This entry corrects/replaces a prior version of the *same* fact | | `prov:wasDerivedFrom` | `derived_from_id` | This entry was derived from a *different* source entity | | `prov:used` | `used_entities` | Entity IDs consumed to produce this one | -| `prov:generatedAtTime` | `timestamp` | ISO datetime, auto-set to `datetime.utcnow()` at write time | +| `prov:generatedAtTime` | `timestamp` | ISO datetime, auto-set to `utc_now_iso()` at write time | | `prov:qualifiedInvalidation` | `invalidated`, `invalidated_at_time`, `invalidated_by`, `invalidation_reason` | A retraction/correction recorded as a tombstone via `ProvenanceManager.invalidate()`, never a hard delete | | `prov:startedAtTime` / `prov:endedAtTime` | `activity_started_at_time`, `activity_ended_at_time` | Typed Activity timing — pass an `ActivityRecord` via the `activity=` kwarg to set these together with `activity_id` | | `prov:qualifiedGeneration`/`Generation`, `qualifiedUsage`/`Usage`, `qualifiedDerivation`/`Derivation` | (derived from the fields above) | Additive qualified forms of `wasGeneratedBy`/`used`/`wasDerivedFrom`, emitted automatically alongside the plain triples | diff --git a/docs/reference/provenance.md b/docs/reference/provenance.md index 6a7e4ced..fe94eb16 100644 --- a/docs/reference/provenance.md +++ b/docs/reference/provenance.md @@ -250,7 +250,7 @@ entry = ProvenanceEntry( source_document="report.pdf", # str: default "" source_location="Page 4", # Optional[str]: default None source_quote="Relevant text...", # Optional[str]: default None - timestamp="2024-01-01T12:00:00", # str: auto-set to utcnow() + timestamp="2024-01-01T12:00:00+00:00", # str: auto-set to utc_now_iso() first_seen=None, # Optional[str]: ISO timestamp last_updated=None, # Optional[str]: ISO timestamp confidence=0.9, # float: default 1.0 diff --git a/semantica/kg/kg_provenance.py b/semantica/kg/kg_provenance.py index 784386d0..4c50f2f0 100644 --- a/semantica/kg/kg_provenance.py +++ b/semantica/kg/kg_provenance.py @@ -54,10 +54,11 @@ Version: 1.0.0 """ from typing import Any, Dict, List, Optional -from datetime import datetime import uuid import time +from ..utils.helpers import utc_now_iso + class GraphBuilderWithProvenance: """ @@ -103,7 +104,7 @@ class GraphBuilderWithProvenance: def build(self, sources, **kwargs): """Build graph with provenance tracking.""" - activity_started_at_time = datetime.utcnow().isoformat() + activity_started_at_time = utc_now_iso() # Track the build operation (recorded before the build runs, so it # has no end time yet — this is the "in progress" marker). if self.provenance and self._prov_manager: @@ -124,7 +125,7 @@ class GraphBuilderWithProvenance: ) result = self._builder.build(sources, **kwargs) - activity_ended_at_time = datetime.utcnow().isoformat() + activity_ended_at_time = utc_now_iso() # Track individual entities and relationships if available if self.provenance and self._prov_manager and hasattr(result, 'get'): @@ -180,7 +181,7 @@ class GraphBuilderWithProvenance: def build_single_source(self, kg_data, **kwargs): """Build graph from single source with provenance tracking.""" - activity_started_at_time = datetime.utcnow().isoformat() + activity_started_at_time = utc_now_iso() # Track the build operation (recorded before the build runs, so it # has no end time yet — this is the "in progress" marker). if self.provenance and self._prov_manager: @@ -200,7 +201,7 @@ class GraphBuilderWithProvenance: ) result = self._builder.build_single_source(kg_data, **kwargs) - activity_ended_at_time = datetime.utcnow().isoformat() + activity_ended_at_time = utc_now_iso() # Track entities and relationships if available if self.provenance and self._prov_manager and isinstance(result, dict): diff --git a/tests/kg/test_integration_comprehensive.py b/tests/kg/test_integration_comprehensive.py index 58316079..0c96b82a 100644 --- a/tests/kg/test_integration_comprehensive.py +++ b/tests/kg/test_integration_comprehensive.py @@ -7,6 +7,7 @@ Tests integration between all KG components and algorithms. import pytest import networkx as nx import numpy as np +from datetime import datetime, timedelta from typing import Dict, List, Any, Tuple import time import json @@ -14,7 +15,6 @@ import json from semantica.kg import ( GraphBuilderWithProvenance, AlgorithmTrackerWithProvenance, - NodeEmbedder, SimilarityCalculator, PathFinder, LinkPredictor, @@ -24,6 +24,40 @@ from semantica.kg import ( ) +def _stored(owner, entity_id): + """Read a provenance record back. Fails if tracking only minted an ID.""" + record = owner._prov_manager.get_provenance(entity_id) + assert record is not None, ( + f"no stored provenance for {entity_id!r} — an ID was generated without a write" + ) + assert record.get("entity_id") == entity_id + return record + + +def _assert_utc_iso(value, field="timestamp"): + assert value, f"missing {field}" + parsed = datetime.fromisoformat(value) + assert parsed.tzinfo is not None, ( + f"{field}={value!r} is naive (datetime.utcnow leftover)" + ) + assert parsed.utcoffset() == timedelta(0), f"{field}={value!r} is not UTC" + return parsed + + +def _assert_tracked(owner, entity_id, source, **metadata): + record = _stored(owner, entity_id) + assert record["source_document"] == source + actual = record.get("metadata") or {} + for key, expected in metadata.items(): + assert actual.get(key) == expected, ( + f"{entity_id} metadata[{key!r}]={actual.get(key)!r}, expected {expected!r}" + ) + _assert_utc_iso(record["timestamp"], "timestamp") + if record.get("last_updated"): + _assert_utc_iso(record["last_updated"], "last_updated") + return record + + class TestComprehensiveIntegration: """Comprehensive integration tests for KG module.""" @@ -157,8 +191,6 @@ class TestComprehensiveIntegration: # Initialize all components builder = GraphBuilderWithProvenance(provenance=True) tracker = AlgorithmTrackerWithProvenance(provenance=True) - embedder = NodeEmbedder() - embedder.enable_provenance = True sim_calc = SimilarityCalculator() path_finder = PathFinder() link_predictor = LinkPredictor() @@ -351,14 +383,34 @@ class TestComprehensiveIntegration: source='comprehensive_integration_test' ) - # Verify all phases completed successfully + expected = { + 'construction': 'graph_construction', + 'centrality': 'centrality_calculation', + 'connectivity': 'connectivity_analysis', + 'community': 'community_detection', + 'similarity': 'similarity_calculation', + 'link_prediction': 'link_prediction', + 'path_analysis': 'path_analysis', + 'cross_layer': 'cross_layer_analysis', + } assert len(execution_ids) == 8 for phase, exec_id in execution_ids.items(): - assert exec_id is not None - assert len(exec_id) > 10 - - print(f"Full pipeline integration completed: {pipeline_id}") - return pipeline_id + _assert_tracked( + tracker, + exec_id, + source=pipeline_id, + entity_type=expected[phase], + ) + summary_record = _assert_tracked( + tracker, + summary_id, + source='comprehensive_integration_test', + entity_type='pipeline_summary', + pipeline_id=pipeline_id, + phases_count=8, + ) + assert summary_id.startswith('pipeline_summary_') + assert summary_record['metadata']['input_data_size'] == len(complex_graph_data) def test_multi_layer_network_analysis(self, multi_layer_network, realistic_embeddings): """Test multi-layer network analysis.""" @@ -382,39 +434,39 @@ class TestComprehensiveIntegration: # Centrality analysis if graph.number_of_nodes() > 0: - try: - degree_cent = centrality_calc.calculate_degree_centrality(graph_dict) - layer_results[f"{layer_name}_centrality"] = degree_cent - - # Track with provenance - cent_id = tracker.track_layer_analysis( - layer_name=layer_name, - graph=graph, - analysis_type='centrality', - results=degree_cent, - source=multi_layer_id - ) - - except Exception as e: - print(f"Centrality analysis failed for {layer_name}: {e}") + degree_cent = centrality_calc.calculate_degree_centrality(graph_dict) + layer_results[f"{layer_name}_centrality"] = degree_cent + cent_id = tracker.track_centrality_calculation( + graph=graph, + centrality_scores=degree_cent['centrality'], + method='degree', + source=multi_layer_id + ) + _assert_tracked( + tracker, + cent_id, + source=multi_layer_id, + method='degree', + scores_count=len(degree_cent['centrality']), + ) # Community detection if graph.number_of_edges() > 0: - try: - communities = community_detector.detect_communities(graph_dict, method='label_propagation') - layer_results[f"{layer_name}_communities"] = communities - - # Track with provenance - comm_id = tracker.track_layer_analysis( - layer_name=layer_name, - graph=graph, - analysis_type='communities', - results=communities, - source=multi_layer_id - ) - - except Exception as e: - print(f"Community detection failed for {layer_name}: {e}") + communities = community_detector.detect_communities(graph_dict, method='label_propagation') + layer_results[f"{layer_name}_communities"] = communities + comm_id = tracker.track_community_detection( + graph=graph, + communities=communities['communities'], + method='label_propagation', + source=multi_layer_id + ) + _assert_tracked( + tracker, + comm_id, + source=multi_layer_id, + method='label_propagation', + communities_count=len(communities['communities']), + ) # Cross-layer similarity analysis print("Cross-layer similarity analysis") @@ -433,39 +485,43 @@ class TestComprehensiveIntegration: similarity_score = len(common_nodes) / max(len(graph1.nodes()), len(graph2.nodes())) layer_similarities[f"{layer1_name}_{layer2_name}"] = similarity_score - # Track cross-layer analysis cross_layer_id = tracker.track_cross_layer_analysis( - multi_layer_network=multi_layer_network, - layer_similarities=layer_similarities, + graph_data=multi_layer_network, + cross_layer_results=layer_similarities, source='multi_layer_test' ) + _assert_tracked( + tracker, + cross_layer_id, + source='multi_layer_test', + entity_type='cross_layer_analysis', + layers_count=len(layer_similarities), + ) # Embedding-based entity similarity - print("Embedding-based entity similarity") - entity_similarities = sim_calc.pairwise_similarity(realistic_embeddings) - # Track embedding analysis embed_id = tracker.track_embedding_analysis( embeddings=realistic_embeddings, - similarities=entity_similarities, + analysis_results=entity_similarities, source='multi_layer_test' ) + _assert_tracked( + tracker, + embed_id, + source='multi_layer_test', + entity_type='embedding_analysis', + embeddings_count=len(realistic_embeddings), + ) - # Verify results assert len(layer_results) > 0 assert len(layer_similarities) > 0 assert len(entity_similarities) > 0 - - print(f"Multi-layer analysis completed") - print(f"Layers analyzed: {list(multi_layer_network.keys())}") - print(f"Layer similarities: {list(layer_similarities.keys())}") - - return multi_layer_id def test_error_handling_and_recovery(self): """Test error handling and recovery mechanisms.""" tracker = AlgorithmTrackerWithProvenance(provenance=True) + centrality_calc = CentralityCalculator() # Test with invalid graph data invalid_graph = { @@ -473,46 +529,24 @@ class TestComprehensiveIntegration: 'edges': [] } - # Should handle gracefully - try: - centrality_calc = CentralityCalculator() - result = centrality_calc.calculate_degree_centrality(invalid_graph) - # Should return empty result or handle gracefully - assert isinstance(result, dict) - except Exception as e: - # Should be a controlled exception - assert isinstance(e, (ValueError, RuntimeError)) + result = centrality_calc.calculate_degree_centrality(invalid_graph) + assert isinstance(result, dict) - # Test with invalid embeddings - invalid_embeddings = { - 'node1': [1, 2], # Different dimensions - 'node2': [1, 2, 3, 4] # Different dimensions - } - - try: - sim_calc = SimilarityCalculator() - result = sim_calc.batch_similarity( - embeddings=invalid_embeddings, - query_embedding=[1, 2, 3, 4], - method='cosine' - ) - # Should handle dimension mismatch - except Exception as e: - # Should handle gracefully - assert isinstance(e, ValueError) - - # Test provenance tracking with invalid data - try: - result = tracker.track_embedding_computation( - graph=None, # Invalid graph - algorithm='test', - embeddings={}, - parameters={} - ) - # Should either return None or handle gracefully - except Exception as e: - # Should be a controlled exception - assert isinstance(e, (ValueError, TypeError)) + # Provenance tracking with a None graph still writes a record. + result = tracker.track_embedding_computation( + graph=None, + algorithm='test', + embeddings={}, + parameters={}, + source='integration_error_recovery' + ) + _assert_tracked( + tracker, + result, + source='integration_error_recovery', + algorithm='test', + input_data_type='NoneType', + ) # Test graceful degradation when provenance is disabled tracker_no_prov = AlgorithmTrackerWithProvenance(provenance=False) @@ -526,8 +560,7 @@ class TestComprehensiveIntegration: # Should return None when provenance is disabled assert result is None - - print("Error handling and recovery test completed") + assert tracker_no_prov._prov_manager is None def test_performance_benchmarks(self, realistic_embeddings): """Test performance benchmarks with realistic data.""" diff --git a/tests/kg/test_provenance_workflows.py b/tests/kg/test_provenance_workflows.py index 493fd0da..1cfe9fcf 100644 --- a/tests/kg/test_provenance_workflows.py +++ b/tests/kg/test_provenance_workflows.py @@ -7,13 +7,13 @@ Tests complete provenance tracking workflows across multiple algorithms. import pytest import networkx as nx import time +from datetime import datetime, timedelta from typing import Dict, List, Any import uuid from semantica.kg import ( GraphBuilderWithProvenance, AlgorithmTrackerWithProvenance, - NodeEmbedder, SimilarityCalculator, LinkPredictor, CentralityCalculator, @@ -21,6 +21,40 @@ from semantica.kg import ( ) +def _stored(owner, entity_id): + """Read a provenance record back. Fails if tracking only minted an ID.""" + record = owner._prov_manager.get_provenance(entity_id) + assert record is not None, ( + f"no stored provenance for {entity_id!r} — an ID was generated without a write" + ) + assert record.get("entity_id") == entity_id + return record + + +def _assert_utc_iso(value, field="timestamp"): + assert value, f"missing {field}" + parsed = datetime.fromisoformat(value) + assert parsed.tzinfo is not None, ( + f"{field}={value!r} is naive (datetime.utcnow leftover)" + ) + assert parsed.utcoffset() == timedelta(0), f"{field}={value!r} is not UTC" + return parsed + + +def _assert_tracked(owner, entity_id, source, **metadata): + record = _stored(owner, entity_id) + assert record["source_document"] == source + actual = record.get("metadata") or {} + for key, expected in metadata.items(): + assert actual.get(key) == expected, ( + f"{entity_id} metadata[{key!r}]={actual.get(key)!r}, expected {expected!r}" + ) + _assert_utc_iso(record["timestamp"], "timestamp") + if record.get("last_updated"): + _assert_utc_iso(record["last_updated"], "last_updated") + return record + + class TestProvenanceWorkflows: """Test provenance tracking workflows.""" @@ -116,10 +150,25 @@ class TestProvenanceWorkflows: source=workflow_id ) - assert construction_id is not None + _assert_tracked( + tracker, + construction_id, + source=workflow_id, + entity_type="graph_construction", + entities_count=8, + relationships_count=12, + ) assert construction_id.startswith('graph_construction_') + + for entity in graph_result['entities']: + built = _stored(builder, entity['id']) + assert built['metadata']['operation'] == 'build_entity' + assert built['metadata']['entity_type'] == entity['type'] + _assert_utc_iso(built['activity_started_at_time'], 'activity_started_at_time') + _assert_utc_iso(built['activity_ended_at_time'], 'activity_ended_at_time') # Step 3: Track entity processing + processed = [] for entity in graph_result['entities']: entity_id = tracker.track_entity_processing( entity_id=entity['id'], @@ -127,9 +176,19 @@ class TestProvenanceWorkflows: entity_data=entity, source=workflow_id ) - assert entity_id is not None + _assert_tracked( + tracker, + entity_id, + source=workflow_id, + processed_entity_id=entity['id'], + processed_entity_type=entity['type'], + ) + processed.append(entity_id) + assert len(processed) == 8 + assert len(set(processed)) == 8 # Step 4: Track relationship processing + rel_ids = [] for relationship in graph_result['relationships']: rel_id = tracker.track_relationship_processing( relationship_id=f"{relationship['source']}-{relationship['target']}", @@ -137,10 +196,15 @@ class TestProvenanceWorkflows: relationship_data=relationship, source=workflow_id ) - assert rel_id is not None - - print(f"Graph construction workflow completed: {workflow_id}") - return workflow_id + _assert_tracked( + tracker, + rel_id, + source=workflow_id, + processed_relationship_type=relationship['type'], + ) + rel_ids.append(rel_id) + assert len(rel_ids) == 12 + assert len(set(rel_ids)) == 12 def test_embedding_workflow(self, workflow_graph, workflow_embeddings): """Test complete embedding workflow with provenance.""" @@ -176,8 +240,20 @@ class TestProvenanceWorkflows: source=workflow_id ) - assert embed_id is not None + _assert_tracked( + tracker, + embed_id, + source=workflow_id, + algorithm='node2vec', + node_count=6, + embedding_dimension=4, + ) assert embed_id.startswith('embedding_') + for node_id, vector in computed_embeddings.items(): + node_record = _stored(tracker, f"embedding_{node_id}") + assert node_record['metadata']['node_id'] == node_id + assert node_record['metadata']['execution_id'] == embed_id + assert node_record['metadata']['embedding_dimension'] == len(vector) # Step 2: Track embedding quality metrics quality_metrics = { @@ -196,10 +272,13 @@ class TestProvenanceWorkflows: source=workflow_id ) - assert quality_id is not None - - print(f"Embedding workflow completed: {workflow_id}") - return workflow_id + _assert_tracked( + tracker, + quality_id, + source=workflow_id, + algorithm='node2vec_quality_check', + ) + assert quality_id != embed_id def test_similarity_analysis_workflow(self, workflow_embeddings): """Test complete similarity analysis workflow with provenance.""" @@ -229,10 +308,18 @@ class TestProvenanceWorkflows: source=workflow_id ) - assert sim_id is not None + _assert_tracked( + tracker, + sim_id, + source=workflow_id, + method='cosine', + similarities_count=len(similarities), + ) assert sim_id.startswith('similarity_') + assert len(similarities) == 3 # Step 2: Track individual similarity results + result_ids = [] for node_id, similarity_score in similarities.items(): result_id = tracker.track_similarity_result( node_id=node_id, @@ -241,7 +328,17 @@ class TestProvenanceWorkflows: execution_id=sim_id, source=workflow_id ) - assert result_id is not None + record = _assert_tracked( + tracker, + result_id, + source=workflow_id, + node_id=node_id, + method='cosine', + execution_id=sim_id, + ) + assert record['metadata']['similarity_score'] == similarity_score + result_ids.append(result_id) + assert len(result_ids) == len(similarities) # Step 3: Track similarity threshold analysis threshold = 0.7 @@ -254,10 +351,14 @@ class TestProvenanceWorkflows: source=workflow_id ) - assert threshold_id is not None - - print(f"Similarity analysis workflow completed: {workflow_id}") - return workflow_id + _assert_tracked( + tracker, + threshold_id, + source=workflow_id, + threshold=threshold, + execution_id=sim_id, + ) + assert _stored(tracker, threshold_id)['metadata']['high_similarity_count'] == len(high_similarity) def test_link_prediction_workflow(self, workflow_graph): """Test complete link prediction workflow with provenance.""" @@ -287,7 +388,13 @@ class TestProvenanceWorkflows: source=workflow_id ) - assert pred_id is not None + _assert_tracked( + tracker, + pred_id, + source=workflow_id, + method=method, + predictions_count=len(predictions), + ) assert pred_id.startswith('link_prediction_') # Step 2: Track individual predictions @@ -300,10 +407,17 @@ class TestProvenanceWorkflows: execution_id=pred_id, source=workflow_id ) - assert result_id is not None - - print(f"Link prediction workflow completed: {workflow_id}") - return workflow_id + record = _assert_tracked( + tracker, + result_id, + source=workflow_id, + source_node=source, + target_node=target, + method=method, + execution_id=pred_id, + ) + assert record['metadata']['prediction_score'] == score + assert len(methods) == 3 def test_centrality_analysis_workflow(self, workflow_graph): """Test complete centrality analysis workflow with provenance.""" @@ -326,54 +440,42 @@ class TestProvenanceWorkflows: ('eigenvector', centrality_calc.calculate_eigenvector_centrality) ] + tracked_methods = [] for method_name, method_func in centrality_methods: try: - start_time = time.time() result = method_func(graph_dict) - calculation_time = time.time() - start_time - - cent_id = tracker.track_centrality_calculation( - graph=workflow_graph, - centrality_scores=result['centrality'], - method=method_name, - parameters={}, - calculation_time=calculation_time, - source=workflow_id - ) - - assert cent_id is not None - assert cent_id.startswith('centrality_') - - # Step 2: Track individual centrality scores - for node_id, score in result['centrality'].items(): - score_id = tracker.track_centrality_score( - node_id=node_id, - centrality_score=score, - method=method_name, - execution_id=cent_id, - source=workflow_id - ) - assert score_id is not None - - # Step 3: Track centrality ranking analysis - rankings = result['rankings'] - top_nodes = rankings[:3] # Top 3 nodes - - ranking_id = tracker.track_centrality_ranking( - execution_id=cent_id, - rankings=rankings, - top_nodes=top_nodes, - method=method_name, - source=workflow_id - ) - - assert ranking_id is not None - - except Exception as e: - print(f"Warning: {method_name} centrality failed: {e}") - - print(f"Centrality analysis workflow completed: {workflow_id}") - return workflow_id + except Exception: + # Algorithm failure is allowed for optional methods, not for degree. + if method_name == 'degree': + raise + continue + + calculation_time = 0.0 + cent_id = tracker.track_centrality_calculation( + graph=workflow_graph, + centrality_scores=result['centrality'], + method=method_name, + parameters={}, + calculation_time=calculation_time, + source=workflow_id + ) + _assert_tracked( + tracker, + cent_id, + source=workflow_id, + method=method_name, + scores_count=len(result['centrality']), + ) + assert cent_id.startswith('centrality_') + for node_id, score in result['centrality'].items(): + score_record = _stored(tracker, f"centrality_{node_id}_{cent_id}") + assert score_record['metadata']['node_id'] == node_id + assert score_record['metadata']['method'] == method_name + assert score_record['metadata']['centrality_score'] == score + tracked_methods.append(method_name) + + assert 'degree' in tracked_methods + assert len(tracked_methods) >= 1 def test_community_detection_workflow(self, workflow_graph): """Test complete community detection workflow with provenance.""" @@ -391,56 +493,39 @@ class TestProvenanceWorkflows: # Step 1: Track community detection methods = ['label_propagation', 'louvain'] + tracked_methods = [] for method in methods: try: - start_time = time.time() result = community_detector.detect_communities(graph_dict, method=method) - detection_time = time.time() - start_time - - comm_id = tracker.track_community_detection( - graph=workflow_graph, - communities=result['communities'], - method=method, - parameters={}, - detection_time=detection_time, - source=workflow_id - ) - - assert comm_id is not None - assert comm_id.startswith('community_') - - # Step 2: Track individual communities - for i, community in enumerate(result['communities']): - comm_result_id = tracker.track_community_result( - community_id=i, - nodes=community, - method=method, - execution_id=comm_id, - source=workflow_id - ) - assert comm_result_id is not None - - # Step 3: Track community quality metrics - quality_metrics = { - 'modularity': 0.3, - 'num_communities': len(result['communities']), - 'avg_community_size': len(result['communities']) / len(result['communities']) if result['communities'] else 0 - } - - quality_id = tracker.track_community_quality( - execution_id=comm_id, - metrics=quality_metrics, - method=method, - source=workflow_id - ) - - assert quality_id is not None - - except Exception as e: - print(f"Warning: {method} community detection failed: {e}") - - print(f"Community detection workflow completed: {workflow_id}") - return workflow_id + except Exception: + if method == 'label_propagation': + raise + continue + + comm_id = tracker.track_community_detection( + graph=workflow_graph, + communities=result['communities'], + method=method, + parameters={}, + detection_time=0.0, + source=workflow_id + ) + _assert_tracked( + tracker, + comm_id, + source=workflow_id, + method=method, + communities_count=len(result['communities']), + ) + assert comm_id.startswith('community_') + for i, community in enumerate(result['communities']): + comm_record = _stored(tracker, f"community_{comm_id}_{i}") + assert comm_record['metadata']['community_id'] == i + assert comm_record['metadata']['method'] == method + assert comm_record['metadata']['nodes'] == community + tracked_methods.append(method) + + assert 'label_propagation' in tracked_methods def test_comprehensive_provenance_workflow(self, workflow_data, workflow_graph, workflow_embeddings): """Test comprehensive provenance workflow combining all algorithms.""" @@ -555,29 +640,37 @@ class TestProvenanceWorkflows: source='comprehensive_test' ) - # Verify all execution IDs + # Verify all execution IDs were stored with the expected payload assert len(execution_ids) == 6 + expected_types = { + 'construction': ('graph_construction_', 'graph_construction'), + 'embedding': ('embedding_', 'embedding_computation'), + 'similarity': ('similarity_', 'similarity_calculation'), + 'link_prediction': ('link_prediction_', 'link_prediction'), + 'centrality': ('centrality_', 'centrality_calculation'), + 'community_detection': ('community_', 'community_detection'), + } for phase, exec_id in execution_ids.items(): - assert exec_id is not None - assert len(exec_id) > 10 - - # Verify all IDs are unique + prefix, entity_type = expected_types[phase] + assert exec_id.startswith(prefix) + _assert_tracked( + tracker, + exec_id, + source=master_workflow_id, + entity_type=entity_type, + ) + + summary_record = _assert_tracked( + tracker, + summary_id, + source='comprehensive_test', + entity_type='workflow_summary', + ) + assert summary_id.startswith('workflow_summary_') + assert summary_record['metadata']['master_workflow_id'] == master_workflow_id + all_ids = list(execution_ids.values()) + [summary_id] assert len(set(all_ids)) == len(all_ids) - - # Verify ID prefixes - assert execution_ids['construction'].startswith('graph_construction_') - assert execution_ids['embedding'].startswith('embedding_') - assert execution_ids['similarity'].startswith('similarity_') - assert execution_ids['link_prediction'].startswith('link_prediction_') - assert execution_ids['centrality'].startswith('centrality_') - assert execution_ids['community_detection'].startswith('community_') - assert summary_id.startswith('workflow_summary_') - - print(f"Comprehensive provenance workflow completed: {master_workflow_id}") - print(f"Execution IDs: {list(execution_ids.keys())}") - - return master_workflow_id def test_provenance_data_integrity(self, workflow_graph, workflow_embeddings): """Test provenance data integrity and consistency.""" @@ -618,24 +711,24 @@ class TestProvenanceWorkflows: ) operations.append(('link_prediction', link_id)) - # Verify data integrity + expected = { + 'embedding': ('embedding_', 'embedding_computation', embed_id), + 'similarity': ('similarity_', 'similarity_calculation', sim_id), + 'link_prediction': ('link_prediction_', 'link_prediction', link_id), + } for op_type, op_id in operations: - assert op_id is not None - assert len(op_id) > 10 - - # Verify ID format consistency - if op_type == 'embedding': - assert op_id.startswith('embedding_') - elif op_type == 'similarity': - assert op_id.startswith('similarity_') - elif op_type == 'link_prediction': - assert op_id.startswith('link_prediction_') - - # Verify workflow consistency + prefix, entity_type, expected_id = expected[op_type] + assert op_id == expected_id + assert op_id.startswith(prefix) + _assert_tracked( + tracker, + op_id, + source=workflow_id, + entity_type=entity_type, + ) + workflow_ids = [op_id for _, op_id in operations] - assert len(set(workflow_ids)) == len(workflow_ids) # All unique - - print(f"Provenance data integrity test completed: {workflow_id}") + assert len(set(workflow_ids)) == len(workflow_ids) def test_provenance_error_recovery(self): """Test provenance system error recovery.""" @@ -650,24 +743,27 @@ class TestProvenanceWorkflows: ) assert result is None # Should return None when provenance is disabled + assert tracker_no_prov._prov_manager is None - # Test error handling during tracking + # Tracking still records an execution when the graph is None — that is + # current production behavior, so assert the write rather than + # "either returns or raises". tracker = AlgorithmTrackerWithProvenance(provenance=True) - - # This should not raise exceptions even with invalid data - try: - result = tracker.track_embedding_computation( - graph=None, # Invalid graph - algorithm='test', - embeddings={}, - parameters={} - ) - # Should either return None or handle gracefully - except Exception as e: - # If it raises, it should be a controlled exception - assert isinstance(e, (ValueError, TypeError)) - - print("Provenance error recovery test completed") + result = tracker.track_embedding_computation( + graph=None, + algorithm='test', + embeddings={}, + parameters={}, + source='error_recovery' + ) + _assert_tracked( + tracker, + result, + source='error_recovery', + algorithm='test', + input_data_type='NoneType', + node_count=0, + ) if __name__ == '__main__': diff --git a/tests/kg/test_provenance_workflows_simple.py b/tests/kg/test_provenance_workflows_simple.py index 97b8fdd7..74ec1159 100644 --- a/tests/kg/test_provenance_workflows_simple.py +++ b/tests/kg/test_provenance_workflows_simple.py @@ -7,13 +7,13 @@ Tests provenance tracking workflows using only available methods. import pytest import networkx as nx import time +from datetime import datetime, timedelta from typing import Dict, List, Any import uuid from semantica.kg import ( GraphBuilderWithProvenance, AlgorithmTrackerWithProvenance, - NodeEmbedder, SimilarityCalculator, LinkPredictor, CentralityCalculator, @@ -21,6 +21,40 @@ from semantica.kg import ( ) +def _stored(owner, entity_id): + """Read a provenance record back. Fails if tracking only minted an ID.""" + record = owner._prov_manager.get_provenance(entity_id) + assert record is not None, ( + f"no stored provenance for {entity_id!r} — an ID was generated without a write" + ) + assert record.get("entity_id") == entity_id + return record + + +def _assert_utc_iso(value, field="timestamp"): + assert value, f"missing {field}" + parsed = datetime.fromisoformat(value) + assert parsed.tzinfo is not None, ( + f"{field}={value!r} is naive (datetime.utcnow leftover)" + ) + assert parsed.utcoffset() == timedelta(0), f"{field}={value!r} is not UTC" + return parsed + + +def _assert_tracked(owner, entity_id, source, **metadata): + record = _stored(owner, entity_id) + assert record["source_document"] == source + actual = record.get("metadata") or {} + for key, expected in metadata.items(): + assert actual.get(key) == expected, ( + f"{entity_id} metadata[{key!r}]={actual.get(key)!r}, expected {expected!r}" + ) + _assert_utc_iso(record["timestamp"], "timestamp") + if record.get("last_updated"): + _assert_utc_iso(record["last_updated"], "last_updated") + return record + + class TestProvenanceWorkflowsSimple: """Test provenance tracking workflows with available methods.""" @@ -93,11 +127,19 @@ class TestProvenanceWorkflowsSimple: source=workflow_id ) - assert embed_id is not None + _assert_tracked( + tracker, + embed_id, + source=workflow_id, + algorithm='node2vec', + node_count=6, + embedding_dimension=4, + ) assert embed_id.startswith('embedding_') - - print(f"Simple embedding workflow completed: {workflow_id}") - return workflow_id + for node_id in workflow_embeddings: + node_record = _stored(tracker, f"embedding_{node_id}") + assert node_record['metadata']['execution_id'] == embed_id + assert node_record['metadata']['node_id'] == node_id def test_similarity_workflow_simple(self, workflow_embeddings): """Test simple similarity workflow with provenance.""" @@ -124,11 +166,15 @@ class TestProvenanceWorkflowsSimple: source=workflow_id ) - assert sim_id is not None + _assert_tracked( + tracker, + sim_id, + source=workflow_id, + method='cosine', + similarities_count=len(similarities), + ) assert sim_id.startswith('similarity_') - - print(f"Simple similarity workflow completed: {workflow_id}") - return workflow_id + assert len(similarities) == 3 def test_link_prediction_workflow_simple(self, workflow_graph): """Test simple link prediction workflow with provenance.""" @@ -153,11 +199,15 @@ class TestProvenanceWorkflowsSimple: source=workflow_id ) - assert link_id is not None + _assert_tracked( + tracker, + link_id, + source=workflow_id, + method='preferential_attachment', + predictions_count=len(predictions), + ) assert link_id.startswith('link_prediction_') - - print(f"Simple link prediction workflow completed: {workflow_id}") - return workflow_id + assert len(predictions) >= 1 def test_centrality_workflow_simple(self, workflow_graph): """Test simple centrality workflow with provenance.""" @@ -184,11 +234,15 @@ class TestProvenanceWorkflowsSimple: source=workflow_id ) - assert cent_id is not None + _assert_tracked( + tracker, + cent_id, + source=workflow_id, + method='degree', + scores_count=len(degree_cent['centrality']), + ) assert cent_id.startswith('centrality_') - - print(f"Simple centrality workflow completed: {workflow_id}") - return workflow_id + assert set(degree_cent['centrality']) == set(workflow_graph.nodes()) def test_community_detection_workflow_simple(self, workflow_graph): """Test simple community detection workflow with provenance.""" @@ -215,11 +269,15 @@ class TestProvenanceWorkflowsSimple: source=workflow_id ) - assert comm_id is not None + _assert_tracked( + tracker, + comm_id, + source=workflow_id, + method='label_propagation', + communities_count=len(communities['communities']), + ) assert comm_id.startswith('community_') - - print(f"Simple community detection workflow completed: {workflow_id}") - return workflow_id + assert len(communities['communities']) >= 1 def test_graph_construction_workflow_simple(self, workflow_data): """Test simple graph construction workflow with provenance.""" @@ -237,23 +295,27 @@ class TestProvenanceWorkflowsSimple: assert len(graph_result['entities']) == 4 assert len(graph_result['relationships']) == 3 - # Track graph construction using embedding computation method as proxy - construction_id = tracker.track_embedding_computation( - graph={'nodes': list(graph_result['entities']), 'edges': list(graph_result['relationships'])}, - algorithm='graph_construction', - embeddings={'graph_size': len(graph_result['entities'])}, - parameters={ - 'entities_count': len(graph_result['entities']), - 'relationships_count': len(graph_result['relationships']) - }, + construction_id = tracker.track_graph_construction( + input_data=workflow_data, + output_graph=graph_result, + entities_count=len(graph_result['entities']), + relationships_count=len(graph_result['relationships']), + construction_time=0.0, source=workflow_id ) - - assert construction_id is not None - assert construction_id.startswith('embedding_') # Using embedding method as proxy - - print(f"Simple graph construction workflow completed: {workflow_id}") - return workflow_id + _assert_tracked( + tracker, + construction_id, + source=workflow_id, + entity_type='graph_construction', + entities_count=4, + relationships_count=3, + ) + assert construction_id.startswith('graph_construction_') + for entity in graph_result['entities']: + built = _stored(builder, entity['id']) + assert built['metadata']['operation'] == 'build_entity' + _assert_utc_iso(built['activity_started_at_time'], 'activity_started_at_time') def test_comprehensive_workflow_simple(self, workflow_data, workflow_graph, workflow_embeddings): """Test comprehensive workflow with all available methods.""" @@ -271,11 +333,12 @@ class TestProvenanceWorkflowsSimple: # Phase 1: Graph Construction graph_result = builder.build_single_source(workflow_data) - construction_id = tracker.track_embedding_computation( - graph={'nodes': list(graph_result['entities']), 'edges': list(graph_result['relationships'])}, - algorithm='graph_construction', - embeddings={'graph_size': len(graph_result['entities'])}, - parameters={'entities_count': len(graph_result['entities'])}, + construction_id = tracker.track_graph_construction( + input_data=workflow_data, + output_graph=graph_result, + entities_count=len(graph_result['entities']), + relationships_count=len(graph_result['relationships']), + construction_time=0.0, source=master_workflow_id ) execution_ids['construction'] = construction_id @@ -351,27 +414,25 @@ class TestProvenanceWorkflowsSimple: ) execution_ids['community_detection'] = comm_id - # Verify all execution IDs + expected = { + 'construction': ('graph_construction_', 'graph_construction'), + 'embedding': ('embedding_', 'embedding_computation'), + 'similarity': ('similarity_', 'similarity_calculation'), + 'link_prediction': ('link_prediction_', 'link_prediction'), + 'centrality': ('centrality_', 'centrality_calculation'), + 'community_detection': ('community_', 'community_detection'), + } assert len(execution_ids) == 6 for phase, exec_id in execution_ids.items(): - assert exec_id is not None - assert len(exec_id) > 10 - - # Verify all IDs are unique - all_ids = list(execution_ids.values()) - assert len(set(all_ids)) == len(all_ids) - - # Verify ID prefixes - assert execution_ids['embedding'].startswith('embedding_') - assert execution_ids['similarity'].startswith('similarity_') - assert execution_ids['link_prediction'].startswith('link_prediction_') - assert execution_ids['centrality'].startswith('centrality_') - assert execution_ids['community_detection'].startswith('community_') - - print(f"Comprehensive workflow completed: {master_workflow_id}") - print(f"Execution IDs: {list(execution_ids.keys())}") - - return master_workflow_id + prefix, entity_type = expected[phase] + assert exec_id.startswith(prefix) + _assert_tracked( + tracker, + exec_id, + source=master_workflow_id, + entity_type=entity_type, + ) + assert len(set(execution_ids.values())) == len(execution_ids) def test_provenance_data_integrity_simple(self, workflow_graph, workflow_embeddings): """Test provenance data integrity with available methods.""" @@ -412,24 +473,19 @@ class TestProvenanceWorkflowsSimple: ) operations.append(('link_prediction', link_id)) - # Verify data integrity + expected = { + 'embedding': 'embedding_computation', + 'similarity': 'similarity_calculation', + 'link_prediction': 'link_prediction', + } for op_type, op_id in operations: - assert op_id is not None - assert len(op_id) > 10 - - # Verify ID format consistency - if op_type == 'embedding': - assert op_id.startswith('embedding_') - elif op_type == 'similarity': - assert op_id.startswith('similarity_') - elif op_type == 'link_prediction': - assert op_id.startswith('link_prediction_') - - # Verify workflow consistency - workflow_ids = [op_id for _, op_id in operations] - assert len(set(workflow_ids)) == len(workflow_ids) # All unique - - print(f"Provenance data integrity test completed: {workflow_id}") + _assert_tracked( + tracker, + op_id, + source=workflow_id, + entity_type=expected[op_type], + ) + assert len(set(op_id for _, op_id in operations)) == 3 def test_provenance_error_recovery_simple(self): """Test provenance system error recovery.""" @@ -443,25 +499,25 @@ class TestProvenanceWorkflowsSimple: parameters={} ) - assert result is None # Should return None when provenance is disabled - - # Test error handling during tracking + assert result is None + assert tracker_no_prov._prov_manager is None + tracker = AlgorithmTrackerWithProvenance(provenance=True) - - # This should not raise exceptions even with invalid data - try: - result = tracker.track_embedding_computation( - graph=None, # Invalid graph - algorithm='test', - embeddings={}, - parameters={} - ) - # Should either return None or handle gracefully - except Exception as e: - # If it raises, it should be a controlled exception - assert isinstance(e, (ValueError, TypeError)) - - print("Provenance error recovery test completed") + result = tracker.track_embedding_computation( + graph=None, + algorithm='test', + embeddings={}, + parameters={}, + source='error_recovery' + ) + _assert_tracked( + tracker, + result, + source='error_recovery', + algorithm='test', + input_data_type='NoneType', + node_count=0, + ) if __name__ == '__main__': diff --git a/tests/provenance/test_manager.py b/tests/provenance/test_manager.py index d771b158..a505fe92 100644 --- a/tests/provenance/test_manager.py +++ b/tests/provenance/test_manager.py @@ -6,8 +6,9 @@ chunk tracking, source tracking, and lineage tracing. """ import pytest +import warnings from unittest.mock import patch -from datetime import datetime +from datetime import datetime, timedelta, timezone from semantica.provenance import ProvenanceManager, SourceReference, ProvenanceEntry from semantica.provenance.storage import InMemoryStorage, SQLiteStorage @@ -733,8 +734,8 @@ class TestProvenanceManager: entity_type="entity", activity_id="test", source_document="doc_1", - first_seen=datetime.utcnow().isoformat(), - last_updated=datetime.utcnow().isoformat(), + first_seen=datetime.now(timezone.utc).isoformat(), + last_updated=datetime.now(timezone.utc).isoformat(), ) with patch.object(prov_mgr.storage, "store", side_effect=RuntimeError("storage error")), \ patch.object(prov_mgr.logger, "error") as mock_log_error: @@ -1672,3 +1673,195 @@ class TestActivityTimingAcrossWrappers: assert entry["activity_ended_at_time"] is not None +class TestTimezoneAwareUtcTimestamps: + """Issue #946 — timezone-aware UTC stamps without datetime.utcnow().""" + + def test_track_entity_stamps_timezone_aware_utc(self): + before = datetime.now(timezone.utc) - timedelta(seconds=1) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always", DeprecationWarning) + prov_mgr = ProvenanceManager() + entry = prov_mgr.track_entity("utc_entity", source="doc_1") + after = datetime.now(timezone.utc) + timedelta(seconds=1) + + utcnow_warnings = [ + w + for w in caught + if issubclass(w.category, DeprecationWarning) + and "utcnow" in str(w.message).lower() + ] + assert utcnow_warnings == [] + + for field in ("timestamp", "first_seen", "last_updated"): + value = getattr(entry, field) + parsed = datetime.fromisoformat(value) + assert parsed.tzinfo is not None, f"{field}={value!r} is naive" + assert parsed.utcoffset() == timedelta(0), f"{field}={value!r} is not UTC" + assert before <= parsed <= after + + stored = prov_mgr.get_provenance("utc_entity") + assert stored is not None + assert stored["source_document"] == "doc_1" + assert stored["last_updated"] == entry.last_updated + + def test_invalidate_stamps_timezone_aware_utc(self): + prov_mgr = ProvenanceManager() + prov_mgr.track_entity("utc_invalidate", source="doc_1") + result = prov_mgr.invalidate( + "utc_invalidate", agent_id="reviewer", reason="test" + ) + parsed = datetime.fromisoformat(result.invalidated_at_time) + assert parsed.tzinfo is not None + assert parsed.utcoffset() == timedelta(0) + + def test_graph_builder_activity_times_are_timezone_aware_utc(self): + from semantica.kg.kg_provenance import GraphBuilderWithProvenance + + builder = GraphBuilderWithProvenance(provenance=True, agent_id="builder_svc") + result = builder.build_single_source({ + "entities": [{"id": "person1", "type": "Person", "name": "Ada"}], + "relationships": [], + }) + assert result["entities"][0]["id"] == "person1" + + person = builder._prov_manager.get_provenance("person1") + assert person is not None + assert person["metadata"]["operation"] == "build_entity" + for field in ("activity_started_at_time", "activity_ended_at_time"): + parsed = datetime.fromisoformat(person[field]) + assert parsed.tzinfo is not None, ( + f"{field}={person[field]!r} is naive" + ) + assert parsed.utcoffset() == timedelta(0) + started = datetime.fromisoformat(person["activity_started_at_time"]) + ended = datetime.fromisoformat(person["activity_ended_at_time"]) + assert started <= ended + +class TestMixedFormatTimestampComparisons: + """Issue #946 review — naive and offset-bearing stamps must compare as instants. + + Records written before the timezone-aware change carry a naive UTC stamp + (``2026-08-19T22:35:38.501697``); records written after carry ``+00:00``. + Comparing the two as raw strings puts the offset-bearing form above a naive + bound at the identical instant, so the record falls outside a range that + should contain it. These tests pin the boundary in both directions. + """ + + NAIVE = "2026-08-19T12:00:00.500000" + AWARE = "2026-08-19T12:00:00.500000+00:00" + + def _manager_with(self, *timestamps): + prov_mgr = ProvenanceManager() + for index, stamp in enumerate(timestamps): + prov_mgr.storage.store( + ProvenanceEntry( + entity_id=f"entity_{index}", + entity_type="entity", + activity_id="test", + source_document="doc_1", + timestamp=stamp, + ) + ) + return prov_mgr + + def test_raw_string_compare_would_exclude_the_boundary_record(self): + """Guards the premise: the two forms are not string-comparable.""" + assert not (self.AWARE <= self.NAIVE) + assert datetime.fromisoformat(self.AWARE) == datetime.fromisoformat( + self.NAIVE + ).replace(tzinfo=timezone.utc) + + def test_query_range_with_naive_bounds_includes_aware_record(self): + prov_mgr = self._manager_with(self.AWARE) + results = prov_mgr.query_recorded_between( + "2026-08-19T12:00:00.500000", "2026-08-19T12:00:00.500000" + ) + assert [r["entity_id"] for r in results] == ["entity_0"] + + def test_query_range_with_aware_bounds_includes_naive_record(self): + prov_mgr = self._manager_with(self.NAIVE) + results = prov_mgr.query_recorded_between( + "2026-08-19T12:00:00.500000+00:00", "2026-08-19T12:00:00.500000+00:00" + ) + assert [r["entity_id"] for r in results] == ["entity_0"] + + def test_query_range_returns_both_formats_from_a_mixed_store(self): + prov_mgr = self._manager_with(self.NAIVE, self.AWARE) + results = prov_mgr.query_recorded_between( + "2026-08-19T11:00:00", "2026-08-19T13:00:00+00:00" + ) + assert {r["entity_id"] for r in results} == {"entity_0", "entity_1"} + + def test_query_range_sorts_mixed_formats_chronologically(self): + prov_mgr = self._manager_with( + "2026-08-19T12:00:02+00:00", # entity_0, latest + "2026-08-19T12:00:00", # entity_1, earliest + "2026-08-19T12:00:01+00:00", # entity_2, middle + ) + results = prov_mgr.query_recorded_between( + "2026-08-19T11:00:00", "2026-08-19T13:00:00" + ) + assert [r["entity_id"] for r in results] == [ + "entity_1", + "entity_2", + "entity_0", + ] + + def test_query_range_accepts_trailing_z(self): + prov_mgr = self._manager_with(self.NAIVE) + results = prov_mgr.query_recorded_between( + "2026-08-19T11:00:00Z", "2026-08-19T13:00:00Z" + ) + assert len(results) == 1 + + def test_query_range_skips_unparseable_timestamp(self): + prov_mgr = self._manager_with("not-a-timestamp", self.AWARE) + results = prov_mgr.query_recorded_between( + "2026-08-19T11:00:00", "2026-08-19T13:00:00" + ) + assert [r["entity_id"] for r in results] == ["entity_1"] + + def test_audit_log_since_naive_bound_includes_aware_record(self): + prov_mgr = self._manager_with(self.AWARE) + entries = prov_mgr.audit_log( + since="2026-08-19T12:00:00.500000", format="json" + ) + assert [e["entity_id"] for e in entries] == ["entity_0"] + + def test_audit_log_since_aware_bound_includes_naive_record(self): + prov_mgr = self._manager_with(self.NAIVE) + entries = prov_mgr.audit_log( + since="2026-08-19T12:00:00.500000+00:00", format="json" + ) + assert [e["entity_id"] for e in entries] == ["entity_0"] + + def test_audit_log_excludes_records_before_since_across_formats(self): + prov_mgr = self._manager_with( + "2026-08-19T11:59:59", # entity_0, before + "2026-08-19T12:00:01+00:00", # entity_1, after + ) + entries = prov_mgr.audit_log(since="2026-08-19T12:00:00+00:00", format="json") + assert [e["entity_id"] for e in entries] == ["entity_1"] + + def test_audit_log_sorts_mixed_formats_chronologically(self): + prov_mgr = self._manager_with( + "2026-08-19T12:00:02+00:00", + "2026-08-19T12:00:00", + "2026-08-19T12:00:01+00:00", + ) + entries = prov_mgr.audit_log(format="json") + assert [e["entity_id"] for e in entries] == [ + "entity_1", + "entity_2", + "entity_0", + ] + + def test_audit_log_unparseable_timestamp_sorts_first(self): + prov_mgr = self._manager_with("not-a-timestamp", "2026-08-19T12:00:00+00:00") + entries = prov_mgr.audit_log(format="json") + assert [e["entity_id"] for e in entries] == ["entity_0", "entity_1"] + + def test_audit_log_since_excludes_unparseable_timestamp(self): + prov_mgr = self._manager_with("not-a-timestamp", "2026-08-19T12:00:00+00:00") + entries = prov_mgr.audit_log(since="2026-08-19T11:00:00", format="json") + assert [e["entity_id"] for e in entries] == ["entity_1"]