forked from HAL9000/cleveragents-core
hamza.khyari
d5f7f15215
Temporal metadata fields (validFrom, validUntil, isCurrent, isRevisionOf) on UKO InformationUnit nodes enable revision chain tracking: when code changes, old nodes are marked historical and new revision nodes are created with back-links. Three storage tiers (hot/warm/cold) filter nodes by temporal scope (current/recent/all) with configurable retention (warm_retention_hours default 24h, cold_retention_days default 90d). Includes TemporalMetadata, TemporalNode, RevisionChain, TierQueryResult, TierRetentionConfig frozen domain models, TemporalBackend protocol, InMemoryTemporalBackend stub, TemporalService with structlog and DI, BackendSet.temporal typing upgrade from object|None to TemporalBackend|None. 67 Behave scenarios, 8 Robot Framework tests, ASV benchmarks, and reference documentation. ISSUES CLOSED: #577
432 lines
13 KiB
Python
432 lines
13 KiB
Python
"""Step definitions for Temporal Data Model — metadata and node model steps."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
from behave import given, then
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.domain.models.acms.temporal import (
|
|
TemporalMetadata,
|
|
TemporalNode,
|
|
)
|
|
|
|
__all__: list[str] = []
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TemporalMetadata steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a TemporalMetadata with valid_from now")
|
|
def step_given_temporal_metadata_defaults(context: Any) -> None:
|
|
context.temporal_meta = TemporalMetadata(
|
|
valid_from=datetime.now(tz=UTC),
|
|
)
|
|
|
|
|
|
@given(
|
|
"a TemporalMetadata with valid_from now and valid_until tomorrow"
|
|
' and is_current false and is_revision_of "{revision_of}"'
|
|
)
|
|
def step_given_temporal_metadata_all_fields(
|
|
context: Any,
|
|
revision_of: str,
|
|
) -> None:
|
|
now = datetime.now(tz=UTC)
|
|
context.temporal_meta = TemporalMetadata(
|
|
valid_from=now,
|
|
valid_until=now + timedelta(days=1),
|
|
is_current=False,
|
|
is_revision_of=revision_of,
|
|
)
|
|
|
|
|
|
@given("a TemporalMetadata with naive valid_from")
|
|
def step_given_temporal_metadata_naive_valid_from(context: Any) -> None:
|
|
context.temporal_meta = TemporalMetadata(
|
|
valid_from=datetime(2026, 1, 15, 10, 30, 0),
|
|
)
|
|
|
|
|
|
@given("a TemporalMetadata with naive valid_until")
|
|
def step_given_temporal_metadata_naive_valid_until(context: Any) -> None:
|
|
context.temporal_meta = TemporalMetadata(
|
|
valid_from=datetime.now(tz=UTC),
|
|
valid_until=datetime(2026, 12, 31, 23, 59, 59),
|
|
is_current=False,
|
|
)
|
|
|
|
|
|
@then("the temporal metadata is_current should be true")
|
|
def step_then_temporal_metadata_is_current_true(context: Any) -> None:
|
|
assert context.temporal_meta.is_current is True
|
|
|
|
|
|
@then("the temporal metadata is_current should be false")
|
|
def step_then_temporal_metadata_is_current_false(context: Any) -> None:
|
|
assert context.temporal_meta.is_current is False
|
|
|
|
|
|
@then("the temporal metadata valid_until should be None")
|
|
def step_then_temporal_metadata_valid_until_none(context: Any) -> None:
|
|
assert context.temporal_meta.valid_until is None
|
|
|
|
|
|
@then("the temporal metadata valid_until should not be None")
|
|
def step_then_temporal_metadata_valid_until_not_none(context: Any) -> None:
|
|
assert context.temporal_meta.valid_until is not None
|
|
|
|
|
|
@then("the temporal metadata is_revision_of should be None")
|
|
def step_then_temporal_metadata_revision_none(context: Any) -> None:
|
|
assert context.temporal_meta.is_revision_of is None
|
|
|
|
|
|
@then('the temporal metadata is_revision_of should be "{expected}"')
|
|
def step_then_temporal_metadata_revision_of(
|
|
context: Any,
|
|
expected: str,
|
|
) -> None:
|
|
assert context.temporal_meta.is_revision_of == expected
|
|
|
|
|
|
@then("the temporal metadata valid_from should have UTC timezone")
|
|
def step_then_temporal_metadata_valid_from_utc(context: Any) -> None:
|
|
assert context.temporal_meta.valid_from.tzinfo is not None
|
|
|
|
|
|
@then("the temporal metadata valid_until should have UTC timezone")
|
|
def step_then_temporal_metadata_valid_until_utc(context: Any) -> None:
|
|
assert context.temporal_meta.valid_until is not None
|
|
assert context.temporal_meta.valid_until.tzinfo is not None
|
|
|
|
|
|
@then("modifying the temporal metadata is_current should raise an error")
|
|
def step_then_temporal_metadata_frozen(context: Any) -> None:
|
|
try:
|
|
context.temporal_meta.is_current = False # type: ignore[misc]
|
|
msg = "Expected frozen model to reject mutation"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then("creating a TemporalMetadata with empty is_revision_of should raise ValueError")
|
|
def step_then_temporal_metadata_empty_revision(context: Any) -> None:
|
|
try:
|
|
TemporalMetadata(
|
|
valid_from=datetime.now(tz=UTC),
|
|
is_revision_of="",
|
|
)
|
|
msg = "Expected ValueError for empty is_revision_of"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then(
|
|
"creating a TemporalMetadata with whitespace is_revision_of should raise ValueError"
|
|
)
|
|
def step_then_temporal_metadata_whitespace_revision(context: Any) -> None:
|
|
try:
|
|
TemporalMetadata(
|
|
valid_from=datetime.now(tz=UTC),
|
|
is_revision_of=" ",
|
|
)
|
|
msg = "Expected ValueError for whitespace is_revision_of"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@given("a TemporalMetadata with non-UTC timezone")
|
|
def step_given_temporal_metadata_non_utc(context: Any) -> None:
|
|
est = timezone(timedelta(hours=-5))
|
|
context.temporal_meta = TemporalMetadata(
|
|
valid_from=datetime(2026, 6, 15, 12, 0, tzinfo=est),
|
|
is_current=True,
|
|
)
|
|
|
|
|
|
@then(
|
|
"creating a TemporalMetadata with valid_until before valid_from"
|
|
" should raise ValueError",
|
|
)
|
|
def step_then_temporal_valid_until_before_valid_from(context: Any) -> None:
|
|
try:
|
|
now = datetime.now(tz=UTC)
|
|
TemporalMetadata(
|
|
valid_from=now + timedelta(hours=1),
|
|
valid_until=now,
|
|
is_current=False,
|
|
)
|
|
msg = "Expected ValueError for valid_until < valid_from"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then(
|
|
"creating a TemporalMetadata with is_current true and valid_until set"
|
|
" should raise ValueError",
|
|
)
|
|
def step_then_temporal_is_current_with_valid_until(context: Any) -> None:
|
|
try:
|
|
now = datetime.now(tz=UTC)
|
|
TemporalMetadata(
|
|
valid_from=now,
|
|
valid_until=now + timedelta(hours=1),
|
|
is_current=True,
|
|
)
|
|
msg = "Expected ValueError for is_current=True with valid_until"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TemporalNode steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
'a TemporalNode with uri "{uri}" and resource "{resource}"'
|
|
' and path "{path}" without range'
|
|
)
|
|
def step_given_temporal_node_no_range(
|
|
context: Any,
|
|
uri: str,
|
|
resource: str,
|
|
path: str,
|
|
) -> None:
|
|
context.temporal_node = TemporalNode(
|
|
node_uri=uri,
|
|
source_resource=resource,
|
|
source_path=path,
|
|
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
|
|
)
|
|
|
|
|
|
@given(
|
|
'a TemporalNode with uri "{uri}" and resource "{resource}"'
|
|
' and path "{path}" and range "{src_range}"'
|
|
)
|
|
def step_given_temporal_node_with_range(
|
|
context: Any,
|
|
uri: str,
|
|
resource: str,
|
|
path: str,
|
|
src_range: str,
|
|
) -> None:
|
|
context.temporal_node = TemporalNode(
|
|
node_uri=uri,
|
|
source_resource=resource,
|
|
source_path=path,
|
|
source_range=src_range,
|
|
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
|
|
)
|
|
|
|
|
|
@then('the temporal node uri should be "{expected}"')
|
|
def step_then_temporal_node_uri(context: Any, expected: str) -> None:
|
|
assert context.temporal_node.node_uri == expected
|
|
|
|
|
|
@then('the temporal node source_resource should be "{expected}"')
|
|
def step_then_temporal_node_resource(context: Any, expected: str) -> None:
|
|
assert context.temporal_node.source_resource == expected
|
|
|
|
|
|
@then('the temporal node source_path should be "{expected}"')
|
|
def step_then_temporal_node_path(context: Any, expected: str) -> None:
|
|
assert context.temporal_node.source_path == expected
|
|
|
|
|
|
@then("the temporal node source_range should be None")
|
|
def step_then_temporal_node_range_none(context: Any) -> None:
|
|
assert context.temporal_node.source_range is None
|
|
|
|
|
|
@then('the temporal node source_range should be "{expected}"')
|
|
def step_then_temporal_node_range(context: Any, expected: str) -> None:
|
|
assert context.temporal_node.source_range == expected
|
|
|
|
|
|
@then("the temporal node should have temporal metadata")
|
|
def step_then_temporal_node_has_metadata(context: Any) -> None:
|
|
assert isinstance(context.temporal_node.temporal, TemporalMetadata)
|
|
|
|
|
|
@then("modifying the temporal node uri should raise an error")
|
|
def step_then_temporal_node_frozen(context: Any) -> None:
|
|
try:
|
|
context.temporal_node.node_uri = "changed" # type: ignore[misc]
|
|
msg = "Expected frozen model to reject mutation"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then("creating a TemporalNode with empty node_uri should raise ValueError")
|
|
def step_then_temporal_node_empty_uri(context: Any) -> None:
|
|
try:
|
|
TemporalNode(
|
|
node_uri="",
|
|
source_resource="RES",
|
|
source_path="src/a.py",
|
|
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
|
|
)
|
|
msg = "Expected ValueError"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then("creating a TemporalNode with whitespace node_uri should raise ValueError")
|
|
def step_then_temporal_node_whitespace_uri(context: Any) -> None:
|
|
try:
|
|
TemporalNode(
|
|
node_uri=" ",
|
|
source_resource="RES",
|
|
source_path="src/a.py",
|
|
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
|
|
)
|
|
msg = "Expected ValueError"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then("creating a TemporalNode with empty source_resource should raise ValueError")
|
|
def step_then_temporal_node_empty_resource(context: Any) -> None:
|
|
try:
|
|
TemporalNode(
|
|
node_uri="uko:test",
|
|
source_resource="",
|
|
source_path="src/a.py",
|
|
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
|
|
)
|
|
msg = "Expected ValueError"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then("creating a TemporalNode with empty source_path should raise ValueError")
|
|
def step_then_temporal_node_empty_path(context: Any) -> None:
|
|
try:
|
|
TemporalNode(
|
|
node_uri="uko:test",
|
|
source_resource="RES",
|
|
source_path="",
|
|
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
|
|
)
|
|
msg = "Expected ValueError"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then("creating a TemporalNode with empty source_range should raise ValueError")
|
|
def step_then_temporal_node_empty_range(context: Any) -> None:
|
|
try:
|
|
TemporalNode(
|
|
node_uri="uko:test",
|
|
source_resource="RES",
|
|
source_path="src/a.py",
|
|
source_range="",
|
|
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
|
|
)
|
|
msg = "Expected ValueError"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then("creating a TemporalNode with whitespace source_range should raise ValueError")
|
|
def step_then_temporal_node_whitespace_range(context: Any) -> None:
|
|
try:
|
|
TemporalNode(
|
|
node_uri="uko:test",
|
|
source_resource="RES",
|
|
source_path="src/a.py",
|
|
source_range=" ",
|
|
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
|
|
)
|
|
msg = "Expected ValueError"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Round 4: string datetime coercion (NEW-3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a TemporalMetadata with valid_from as ISO-8601 string")
|
|
def step_given_metadata_string_datetime(context: Any) -> None:
|
|
context.temporal_meta = TemporalMetadata.model_validate(
|
|
{
|
|
"valid_from": "2025-06-15T12:00:00+03:00",
|
|
"is_current": True,
|
|
}
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Round 5: zero-width Unicode (NEW-14) + symmetric invariant (NEW-15)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then(
|
|
"creating a TemporalNode with zero-width Unicode node_uri should raise ValueError"
|
|
)
|
|
def step_then_zero_width_uri_raises(context: Any) -> None:
|
|
try:
|
|
TemporalNode(
|
|
node_uri="\u200b\u200c\u200d",
|
|
source_resource="RES001",
|
|
source_path="src/a.py",
|
|
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
|
|
)
|
|
msg = "Expected ValueError for zero-width Unicode URI"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then(
|
|
"creating a TemporalMetadata with is_current false"
|
|
" and no valid_until should raise ValueError"
|
|
)
|
|
def step_then_historical_without_valid_until_raises(context: Any) -> None:
|
|
try:
|
|
TemporalMetadata(
|
|
valid_from=datetime.now(tz=UTC),
|
|
is_current=False,
|
|
)
|
|
msg = "Expected ValueError for is_current=False without valid_until"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|
|
|
|
|
|
@then(
|
|
"creating a TemporalMetadata with zero-width Unicode is_revision_of"
|
|
" should raise ValueError"
|
|
)
|
|
def step_then_zero_width_revision_of_raises(context: Any) -> None:
|
|
try:
|
|
TemporalMetadata(
|
|
valid_from=datetime.now(tz=UTC),
|
|
is_revision_of="\u200b\u200c",
|
|
)
|
|
msg = "Expected ValueError for zero-width Unicode is_revision_of"
|
|
raise AssertionError(msg)
|
|
except ValidationError:
|
|
pass
|