From 9faa5661f8df6bc34fc3da59a180809a923527d7 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Mon, 23 Mar 2026 15:53:13 +0530 Subject: [PATCH] Core temporal data model overhaul (#396) --- semantica/kg/__init__.py | 3 + semantica/kg/temporal_model.py | 176 ++++++++++ semantica/kg/temporal_query.py | 320 ++++++++++++++---- semantica/utils/__init__.py | 2 + semantica/utils/exceptions.py | 15 + .../test_temporal_versioning.py | 46 +++ tests/kg/test_kg.py | 74 ++++ 7 files changed, 565 insertions(+), 71 deletions(-) create mode 100644 semantica/kg/temporal_model.py diff --git a/semantica/kg/__init__.py b/semantica/kg/__init__.py index e844a1e8..54f1f373 100644 --- a/semantica/kg/__init__.py +++ b/semantica/kg/__init__.py @@ -126,6 +126,7 @@ from .temporal_query import ( TemporalPatternDetector, TemporalVersionManager, ) +from .temporal_model import BiTemporalFact, TemporalBound __all__ = [ # Core Classes @@ -137,6 +138,8 @@ __all__ = [ "TemporalGraphQuery", "TemporalPatternDetector", "TemporalVersionManager", + "TemporalBound", + "BiTemporalFact", "AlgorithmTrackerWithProvenance", "ProvenanceTracker", # Enhanced Graph Algorithms diff --git a/semantica/kg/temporal_model.py b/semantica/kg/temporal_model.py new file mode 100644 index 00000000..13bfa0dd --- /dev/null +++ b/semantica/kg/temporal_model.py @@ -0,0 +1,176 @@ +""" +Temporal data model helpers for knowledge graph relationships. +""" + +from __future__ import annotations + +import json +import re +import warnings +from dataclasses import dataclass +from datetime import datetime, timezone +from enum import Enum +from typing import Any, Dict, Optional + +from ..utils.exceptions import TemporalValidationError + + +class TemporalBound(Enum): + """Sentinel bounds for open-ended temporal intervals.""" + + OPEN = "OPEN" + + +@dataclass +class BiTemporalFact: + """ + Backward-compatible wrapper around existing relationship dictionaries. + + Design note: + Facts continue to live as plain relationship dicts in the graph. This wrapper + is only used internally for normalization so existing callers can keep + reading and writing `valid_from` / `valid_until` directly. + """ + + valid_from: Optional[datetime] + valid_until: Optional[datetime | TemporalBound] + recorded_at: datetime + superseded_at: datetime | TemporalBound = TemporalBound.OPEN + + @classmethod + def from_relationship(cls, relationship: Dict[str, Any]) -> "BiTemporalFact": + valid_until_raw = relationship.get("valid_until", TemporalBound.OPEN) + if "valid_until" in relationship and relationship.get("valid_until") is None: + warnings.warn( + "`valid_until=None` is deprecated; use TemporalBound.OPEN or an explicit date.", + DeprecationWarning, + stacklevel=2, + ) + if valid_until_raw is None: + valid_until_raw = TemporalBound.OPEN + + recorded_at_raw = relationship.get("recorded_at") + superseded_at_raw = relationship.get("superseded_at", TemporalBound.OPEN) + + return cls( + valid_from=parse_temporal_value(relationship.get("valid_from")), + valid_until=parse_temporal_bound(valid_until_raw), + recorded_at=parse_temporal_value(recorded_at_raw) if recorded_at_raw is not None else datetime.now(timezone.utc), + superseded_at=parse_temporal_bound(superseded_at_raw, default=TemporalBound.OPEN), + ) + + def to_relationship_fields(self) -> Dict[str, Any]: + return { + "valid_from": serialize_temporal_value(self.valid_from), + "valid_until": serialize_temporal_bound(self.valid_until), + "recorded_at": serialize_temporal_value(self.recorded_at), + "superseded_at": serialize_temporal_bound(self.superseded_at), + } + + +def _coerce_iso_like_string(value: str) -> str: + match = re.match( + r"^(?P\d{4})-(?P\d{1,2})-(?P\d{1,2})(?P.*)$", + value.strip(), + ) + if not match: + return value.strip() + + month = int(match.group("month")) + day = int(match.group("day")) + rest = match.group("rest") + return f"{match.group('year')}-{month:02d}-{day:02d}{rest}" + + +def parse_temporal_value(value: Any) -> Optional[datetime]: + if value is None: + return None + if isinstance(value, datetime): + dt = value + elif isinstance(value, (int, float)): + dt = datetime.fromtimestamp(value, timezone.utc) + elif isinstance(value, str): + normalized = _coerce_iso_like_string(value) + if normalized.endswith("Z"): + normalized = normalized[:-1] + "+00:00" + try: + dt = datetime.fromisoformat(normalized) + except ValueError as exc: + raise TemporalValidationError( + "Invalid temporal value", + temporal_context={"value": value}, + ) from exc + else: + raise TemporalValidationError( + "Unsupported temporal value type", + temporal_context={"value": value, "type": type(value).__name__}, + ) + + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc) + + +def parse_temporal_bound( + value: Any, + *, + default: Optional[datetime | TemporalBound] = None, +) -> Optional[datetime | TemporalBound]: + if value is None: + return default + if value == TemporalBound.OPEN or value == TemporalBound.OPEN.value: + return TemporalBound.OPEN + return parse_temporal_value(value) + + +def serialize_temporal_value(value: Optional[datetime]) -> Optional[str]: + if value is None: + return None + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def serialize_temporal_bound(value: Optional[datetime | TemporalBound]) -> Optional[str]: + if value in (None, TemporalBound.OPEN): + return None + return serialize_temporal_value(value) + + +def deserialize_relationship_temporal_fields(relationship: Dict[str, Any]) -> Dict[str, Any]: + normalized = dict(relationship) + fact = BiTemporalFact.from_relationship(normalized) + normalized.update(fact.to_relationship_fields()) + if fact.valid_until is TemporalBound.OPEN: + normalized["valid_until"] = TemporalBound.OPEN + if fact.superseded_at is TemporalBound.OPEN: + normalized["superseded_at"] = TemporalBound.OPEN + return normalized + + +def relationship_to_json_ready(relationship: Dict[str, Any]) -> Dict[str, Any]: + json_ready = dict(relationship) + for field in ("valid_from", "recorded_at"): + if field in json_ready: + json_ready[field] = serialize_temporal_value(parse_temporal_value(json_ready[field])) + for field in ("valid_until", "superseded_at"): + if field in json_ready: + json_ready[field] = serialize_temporal_bound(parse_temporal_bound(json_ready[field])) + return json_ready + + +def temporal_structure_to_json_ready(value: Any) -> Any: + """Recursively convert temporal values into JSON-safe primitives.""" + if isinstance(value, dict): + return {key: temporal_structure_to_json_ready(item) for key, item in value.items()} + if isinstance(value, list): + return [temporal_structure_to_json_ready(item) for item in value] + if isinstance(value, tuple): + return [temporal_structure_to_json_ready(item) for item in value] + if value is TemporalBound.OPEN: + return None + if isinstance(value, datetime): + return serialize_temporal_value(value) + return value + + +def dumps_relationship_json(relationship: Dict[str, Any]) -> str: + return json.dumps(relationship_to_json_ready(relationship)) diff --git a/semantica/kg/temporal_query.py b/semantica/kg/temporal_query.py index 076d9adb..f3b3c31e 100644 --- a/semantica/kg/temporal_query.py +++ b/semantica/kg/temporal_query.py @@ -1,36 +1,24 @@ """ Temporal Query Module - -This module provides comprehensive time-aware querying capabilities for the -Semantica framework, enabling temporal queries and analysis on knowledge -graphs with temporal information. - -Key Features: - - Time-point queries (query graph at specific time) - - Time-range queries (query within time intervals) - - Temporal pattern detection (sequences, cycles, trends) - - Graph evolution analysis - - Temporal path finding - - Temporal version management - -Main Classes: - - TemporalGraphQuery: Main temporal query engine - - TemporalPatternDetector: Temporal pattern detection engine - - TemporalVersionManager: Temporal version/snapshot management - -Example Usage: - >>> from semantica.kg import TemporalGraphQuery - >>> query_engine = TemporalGraphQuery() - >>> result = query_engine.query_at_time(graph, query, at_time="2024-01-01") - >>> evolution = query_engine.analyze_evolution(graph, start_time="2024-01-01") - -Author: Semantica Contributors -License: MIT """ +import copy +import warnings +from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional from ..utils.progress_tracker import get_progress_tracker +from .temporal_model import ( + BiTemporalFact, + TemporalBound, + deserialize_relationship_temporal_fields, + parse_temporal_bound, + parse_temporal_value, + relationship_to_json_ready, + serialize_temporal_value, + temporal_structure_to_json_ready, +) +from ..utils.exceptions import ProcessingError, TemporalValidationError class TemporalGraphQuery: @@ -105,6 +93,7 @@ class TemporalGraphQuery: at_time: Any, include_history: bool = False, temporal_precision: Optional[str] = None, + time_axis: str = "valid", **options, ) -> Dict[str, Any]: """ @@ -141,16 +130,8 @@ class TemporalGraphQuery: relationships = [] if "relationships" in graph: for rel in graph.get("relationships", []): - valid_from = self._parse_time(rel.get("valid_from")) - valid_until = self._parse_time(rel.get("valid_until")) - - # Check if relationship is valid at query time - if valid_from and self._compare_times(query_time, valid_from) < 0: - continue - if valid_until and self._compare_times(query_time, valid_until) > 0: - continue - - relationships.append(rel) + if self._relationship_active_at_time(rel, query_time, time_axis=time_axis): + relationships.append(rel) # Get entities entities = graph.get("entities", []) @@ -177,6 +158,7 @@ class TemporalGraphQuery: end_time: Any, temporal_aggregation: str = "union", include_intervals: bool = True, + time_axis: str = "valid", **options, ) -> Dict[str, Any]: """ @@ -216,16 +198,13 @@ class TemporalGraphQuery: relationships = [] if "relationships" in graph: for rel in graph.get("relationships", []): - valid_from = self._parse_time(rel.get("valid_from")) - valid_until = self._parse_time(rel.get("valid_until")) - - # Check if relationship overlaps with time range - if valid_from and self._compare_times(end, valid_from) < 0: - continue - if valid_until and self._compare_times(start, valid_until) > 0: - continue - - relationships.append(rel) + if self._relationship_overlaps_range( + rel, + start, + end, + time_axis=time_axis, + ): + relationships.append(rel) # Aggregate based on strategy if temporal_aggregation == "intersection": @@ -233,11 +212,7 @@ class TemporalGraphQuery: relationships = [ rel for rel in relationships - if self._parse_time(rel.get("valid_from")) <= start - and ( - not rel.get("valid_until") - or self._parse_time(rel.get("valid_until")) >= end - ) + if self._relationship_covers_range(rel, start, end, time_axis=time_axis) ] elif temporal_aggregation == "evolution": # Group by time periods @@ -448,6 +423,8 @@ class TemporalGraphQuery: # Build adjacency with temporal constraints adjacency = {} relationships = graph.get("relationships", []) + parsed_start_time = self._parse_time(start_time) if start_time else None + parsed_end_time = self._parse_time(end_time) if end_time else None for rel in relationships: s = rel.get("source") @@ -459,15 +436,15 @@ class TemporalGraphQuery: valid_until = self._parse_time(rel.get("valid_until")) if ( - start_time + parsed_start_time and valid_until - and self._compare_times(valid_until, start_time) < 0 + and self._compare_times(valid_until, parsed_start_time) < 0 ): continue if ( - end_time + parsed_end_time and valid_from - and self._compare_times(valid_from, end_time) > 0 + and self._compare_times(valid_from, parsed_end_time) > 0 ): continue @@ -509,26 +486,95 @@ class TemporalGraphQuery: } def _parse_time(self, time_value): - """Parse time value.""" - from datetime import datetime - - if time_value is None: - return None - - if isinstance(time_value, str): - return time_value - - if isinstance(time_value, datetime): - return time_value.isoformat() - - return str(time_value) + """Parse time value into a UTC-normalized datetime.""" + try: + return parse_temporal_value(time_value) + except TemporalValidationError: + raise + except Exception as exc: + raise TemporalValidationError( + "Invalid temporal value", + temporal_context={"value": time_value}, + ) from exc def _compare_times(self, time1, time2): - """Compare two time strings.""" + """Compare two UTC datetimes after granularity truncation.""" if time1 is None or time2 is None: return 0 + time1 = self._truncate_to_granularity(time1) + time2 = self._truncate_to_granularity(time2) return (time1 > time2) - (time1 < time2) + def _truncate_to_granularity(self, value: datetime) -> datetime: + granularity = getattr(self, "temporal_granularity", "second") + if granularity == "second": + return value.replace(microsecond=0) + if granularity == "minute": + return value.replace(second=0, microsecond=0) + if granularity == "hour": + return value.replace(minute=0, second=0, microsecond=0) + if granularity == "day": + return value.replace(hour=0, minute=0, second=0, microsecond=0) + if granularity == "week": + start_of_week = value - timedelta(days=value.weekday()) + return start_of_week.replace(hour=0, minute=0, second=0, microsecond=0) + if granularity == "month": + return value.replace(day=1, hour=0, minute=0, second=0, microsecond=0) + if granularity == "year": + return value.replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0) + return value + + def _get_axis_bounds(self, relationship: Dict[str, Any], axis: str): + normalized = deserialize_relationship_temporal_fields(relationship) + fact = BiTemporalFact.from_relationship(normalized) + if axis == "valid": + return fact.valid_from, fact.valid_until + if axis == "transaction": + return fact.recorded_at, fact.superseded_at + raise ValueError(f"Unsupported time axis: {axis}") + + def _is_point_in_bounds(self, point: datetime, start: Optional[datetime], end: Optional[datetime | TemporalBound]) -> bool: + if start and self._compare_times(point, start) < 0: + return False + if isinstance(end, datetime) and self._compare_times(point, end) > 0: + return False + return True + + def _range_overlaps_bounds(self, query_start: datetime, query_end: datetime, start: Optional[datetime], end: Optional[datetime | TemporalBound]) -> bool: + if start and self._compare_times(query_end, start) < 0: + return False + if isinstance(end, datetime) and self._compare_times(query_start, end) > 0: + return False + return True + + def _range_covered_by_bounds(self, query_start: datetime, query_end: datetime, start: Optional[datetime], end: Optional[datetime | TemporalBound]) -> bool: + if start and self._compare_times(start, query_start) > 0: + return False + if isinstance(end, datetime) and self._compare_times(end, query_end) < 0: + return False + return True + + def _relationship_active_at_time(self, relationship: Dict[str, Any], query_time: datetime, *, time_axis: str) -> bool: + axes = ["valid", "transaction"] if time_axis == "both" else [time_axis] + return all( + self._is_point_in_bounds(query_time, *self._get_axis_bounds(relationship, axis)) + for axis in axes + ) + + def _relationship_overlaps_range(self, relationship: Dict[str, Any], start: datetime, end: datetime, *, time_axis: str) -> bool: + axes = ["valid", "transaction"] if time_axis == "both" else [time_axis] + return all( + self._range_overlaps_bounds(start, end, *self._get_axis_bounds(relationship, axis)) + for axis in axes + ) + + def _relationship_covers_range(self, relationship: Dict[str, Any], start: datetime, end: datetime, *, time_axis: str) -> bool: + axes = ["valid", "transaction"] if time_axis == "both" else [time_axis] + return all( + self._range_covered_by_bounds(start, end, *self._get_axis_bounds(relationship, axis)) + for axis in axes + ) + def _group_by_time_periods(self, relationships, start, end): """Group relationships by time periods.""" # Simplified grouping @@ -848,7 +894,10 @@ class TemporalVersionManager: "author": change_entry.author, "description": change_entry.description, "entities": graph.get("entities", []).copy(), - "relationships": graph.get("relationships", []).copy(), + "relationships": [ + relationship_to_json_ready(rel) + for rel in graph.get("relationships", []).copy() + ], "metadata": options.get("metadata", {}) } @@ -856,10 +905,107 @@ class TemporalVersionManager: snapshot["checksum"] = compute_checksum(snapshot) # Store snapshot - self.storage.save(snapshot) + try: + self.storage.save(snapshot) + except Exception as exc: + raise ProcessingError( + "Failed to persist snapshot", + processing_context={"label": version_label, "author": author}, + ) from exc self.logger.info(f"Created snapshot '{version_label}' by {author}") return snapshot + + def apply_revision(self, snapshot: Dict[str, Any], revision: Dict[str, Any]) -> Dict[str, Any]: + """ + Apply a temporal revision without deleting the original facts. + + Design note: + Overlapping retroactive revisions against the same fact are handled by + superseding the latest matching version and emitting a warning when the + newly requested valid window overlaps a sibling fact on the same edge + and relationship type. This preserves all prior versions instead of + trying to collapse them into a single mutable record. + """ + revision_time = datetime.now(timezone.utc) + relationships = copy.deepcopy(snapshot.get("relationships", [])) + fact_ids = set(revision.get("fact_ids", [])) + new_valid_from = parse_temporal_value(revision.get("new_valid_from")) + new_valid_until = parse_temporal_bound(revision.get("new_valid_until"), default=TemporalBound.OPEN) + revision_type = revision.get("revision_type", "correction") + + revised_relationships = [] + provenance_event = { + "type": "temporal_revision", + "revision_type": revision_type, + "author": revision.get("author"), + "reason": revision.get("reason"), + "recorded_at": serialize_temporal_value(revision_time), + "fact_ids": list(fact_ids), + } + + for rel in relationships: + rel_id = rel.get("id") or self._relationship_key(rel) + if rel_id not in fact_ids: + revised_relationships.append(rel) + continue + + original = deserialize_relationship_temporal_fields(rel) + original["id"] = rel_id + original["superseded_at"] = serialize_temporal_value(revision_time) + original.setdefault("provenance", []).append( + { + **provenance_event, + "role": "superseded", + } + ) + revised_relationships.append(original) + + replacement = copy.deepcopy(rel) + replacement["id"] = f"{rel_id}__rev__{int(revision_time.timestamp())}" + replacement["valid_from"] = serialize_temporal_value(new_valid_from) + replacement["valid_until"] = ( + TemporalBound.OPEN if new_valid_until is TemporalBound.OPEN else serialize_temporal_value(new_valid_until) + ) + replacement["recorded_at"] = serialize_temporal_value(revision_time) + replacement["superseded_at"] = TemporalBound.OPEN + replacement.setdefault("provenance", []).append( + { + **provenance_event, + "role": "replacement", + "replaces": rel_id, + } + ) + self._warn_on_retroactive_overlap(replacement, relationships, revision_type) + revised_relationships.append(replacement) + + revised_snapshot = copy.deepcopy(snapshot) + revised_snapshot["relationships"] = [ + relationship_to_json_ready(rel) for rel in revised_relationships + ] + base_label = snapshot.get("label", "snapshot") + original_label = base_label + revised_label = f"{base_label}__revision__{int(revision_time.timestamp())}" + + original_snapshot = copy.deepcopy(snapshot) + original_snapshot["label"] = original_label + original_snapshot_inserted = False + if self.storage.get(original_label) is None: + self.storage.save(original_snapshot) + original_snapshot_inserted = True + + revised_snapshot["label"] = revised_label + revised_snapshot.setdefault("metadata", {})["revision_event"] = temporal_structure_to_json_ready(provenance_event) + try: + self.storage.save(revised_snapshot) + except Exception as exc: + if original_snapshot_inserted: + self.storage.delete(original_label) + raise ProcessingError( + "Failed to persist revised snapshot", + processing_context={"original_label": original_label, "revised_label": revised_label}, + ) from exc + return revised_snapshot def list_versions(self) -> List[Dict[str, Any]]: """ @@ -971,6 +1117,38 @@ class TemporalVersionManager: target = relationship.get("target", "") rel_type = relationship.get("type", relationship.get("relationship", "")) return f"{source}|{rel_type}|{target}" + + def _warn_on_retroactive_overlap( + self, + replacement: Dict[str, Any], + relationships: List[Dict[str, Any]], + revision_type: str, + ) -> None: + if revision_type != "retroactive": + return + + query = TemporalGraphQuery(temporal_granularity="second") + candidate_start, candidate_end = query._get_axis_bounds(replacement, "valid") + for sibling in relationships: + if sibling.get("source") != replacement.get("source"): + continue + if sibling.get("target") != replacement.get("target"): + continue + if sibling.get("type") != replacement.get("type"): + continue + sibling_start, sibling_end = query._get_axis_bounds(sibling, "valid") + if query._range_overlaps_bounds( + candidate_start or datetime.min.replace(tzinfo=timezone.utc), + candidate_end if isinstance(candidate_end, datetime) else datetime.max.replace(tzinfo=timezone.utc), + sibling_start, + sibling_end, + ): + warnings.warn( + "Retroactive revision overlaps an existing fact on the same edge.", + UserWarning, + stacklevel=2, + ) + return def _compute_entity_changes(self, entity1: Dict[str, Any], entity2: Dict[str, Any]) -> Dict[str, Any]: """ diff --git a/semantica/utils/__init__.py b/semantica/utils/__init__.py index 0c674dae..bd0e7e2d 100644 --- a/semantica/utils/__init__.py +++ b/semantica/utils/__init__.py @@ -63,6 +63,7 @@ from .exceptions import ( ProcessingError, QualityError, SemanticaError, + TemporalValidationError, ValidationError, format_exception, handle_exception, @@ -158,6 +159,7 @@ __all__ = [ # Exceptions "SemanticaError", "ValidationError", + "TemporalValidationError", "ProcessingError", "ConfigurationError", "QualityError", diff --git a/semantica/utils/exceptions.py b/semantica/utils/exceptions.py index 8c6c586c..1645441d 100644 --- a/semantica/utils/exceptions.py +++ b/semantica/utils/exceptions.py @@ -152,6 +152,21 @@ class ValidationError(SemanticaError): self.constraint = details.get("constraint") +class TemporalValidationError(ValidationError): + """ + Exception raised for invalid temporal values or inconsistent temporal state. + """ + + def __init__( + self, + message: str, + temporal_context: Optional[Dict[str, Any]] = None, + **details: Any, + ): + super().__init__(message, validation_context=temporal_context, **details) + self.error_code = "SEM001T" + + class ProcessingError(SemanticaError): """ Exception raised for data processing errors. diff --git a/tests/change_management/test_temporal_versioning.py b/tests/change_management/test_temporal_versioning.py index 1102e8c9..0f0000fd 100644 --- a/tests/change_management/test_temporal_versioning.py +++ b/tests/change_management/test_temporal_versioning.py @@ -8,7 +8,9 @@ graphs, including persistent storage, detailed change tracking, and audit trails import os import tempfile import pytest +from datetime import datetime, timezone from semantica.kg.temporal_query import TemporalVersionManager +from semantica.kg.temporal_model import TemporalBound from semantica.change_management import ChangeLogEntry, InMemoryVersionStorage, SQLiteVersionStorage from semantica.utils.exceptions import ValidationError, ProcessingError @@ -234,6 +236,50 @@ class TestTemporalVersionManager: with pytest.raises(ValidationError, match="Version not found: nonexistent"): manager.compare_versions("v1.0", "nonexistent") + + def test_apply_revision_preserves_history_and_revises_valid_time(self): + manager = TemporalVersionManager() + snapshot = manager.create_snapshot( + graph={ + "entities": [], + "relationships": [ + { + "id": "fact-1", + "source": "drug_a", + "target": "drug_b", + "type": "interacts_with", + "valid_from": "2021-01-01", + "valid_until": TemporalBound.OPEN, + } + ], + }, + version_label="v1.0", + author="alice@company.com", + description="Initial version", + ) + + revised = manager.apply_revision( + snapshot, + { + "fact_ids": ["fact-1"], + "new_valid_from": "2019-01-01", + "new_valid_until": None, + "revision_type": "retroactive", + "author": "alice@company.com", + "reason": "Backfilled evidence", + }, + ) + + query_engine = __import__("semantica.kg.temporal_query", fromlist=["TemporalGraphQuery"]).TemporalGraphQuery() + result = query_engine.query_at_time(revised, "", "2020-06-01") + + assert len(result["relationships"]) == 1 + assert result["relationships"][0]["id"].startswith("fact-1__rev__") + + original = manager.get_version("v1.0") + assert original is not None + assert manager.get_version(revised["label"]) is not None + assert len(manager.list_versions()) == 2 def test_detailed_entity_diff(self): """Test detailed entity-level differences.""" diff --git a/tests/kg/test_kg.py b/tests/kg/test_kg.py index 1588e38b..33544462 100644 --- a/tests/kg/test_kg.py +++ b/tests/kg/test_kg.py @@ -1,13 +1,17 @@ import unittest +import warnings from unittest.mock import MagicMock, patch import sys import os +from datetime import datetime, timezone # Add project root to path sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) from semantica.kg.graph_builder import GraphBuilder from semantica.kg.graph_analyzer import GraphAnalyzer +from semantica.kg.temporal_model import TemporalBound, deserialize_relationship_temporal_fields +from semantica.utils.exceptions import TemporalValidationError class TestGraphBuilder(unittest.TestCase): def setUp(self): @@ -317,5 +321,75 @@ class TestTemporalGraphQuery(unittest.TestCase): self.assertEqual(result["num_paths"], 1) self.assertEqual(len(result["paths"][0]["path"]), 3) # A, B, C + def test_parse_time_normalizes_equivalent_dates_and_utc(self): + parsed_a = self.query_engine._parse_time("2024-1-1") + parsed_b = self.query_engine._parse_time("2024-01-01") + parsed_c = self.query_engine._parse_time("2024-06-15T10:00:00+05:30") + + self.assertEqual(parsed_a, parsed_b) + self.assertEqual(parsed_c, datetime(2024, 6, 15, 4, 30, tzinfo=timezone.utc)) + + def test_parse_time_invalid_raises_temporal_validation_error(self): + with self.assertRaises(TemporalValidationError): + self.query_engine._parse_time("not-a-date") + + def test_query_at_time_supports_open_bound_and_none_warning(self): + graph = { + "relationships": [ + { + "source": "1", + "target": "2", + "type": "current", + "valid_from": "2024-01-01", + "valid_until": TemporalBound.OPEN, + }, + { + "source": "2", + "target": "3", + "type": "deprecated-none", + "valid_from": "2024-01-01", + "valid_until": None, + }, + ] + } + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + result = self.query_engine.query_at_time(graph, "", "2026-01-01") + + self.assertEqual(len(result["relationships"]), 2) + self.assertTrue(any(item.category is DeprecationWarning for item in caught)) + + def test_query_at_time_supports_transaction_axis(self): + graph = { + "relationships": [ + { + "source": "1", + "target": "2", + "type": "known-later", + "valid_from": "2019-01-01", + "valid_until": TemporalBound.OPEN, + "recorded_at": "2021-01-01", + "superseded_at": TemporalBound.OPEN, + } + ] + } + + result_2020 = self.query_engine.query_at_time( + graph, "", "2020-06-01", time_axis="transaction" + ) + result_2021 = self.query_engine.query_at_time( + graph, "", "2021-06-01", time_axis="transaction" + ) + + self.assertEqual(len(result_2020["relationships"]), 0) + self.assertEqual(len(result_2021["relationships"]), 1) + + def test_null_valid_until_deserializes_to_open(self): + relationship = deserialize_relationship_temporal_fields( + {"source": "1", "target": "2", "type": "rel", "valid_from": "2024-01-01", "valid_until": None} + ) + self.assertIs(relationship["valid_until"], TemporalBound.OPEN) + if __name__ == "__main__": unittest.main()