Allegro
bcf8c31270
- Add JSON schema at schemas/state.json with full validation rules - Implement State, StateType, StateMetadata dataclasses in models/state.py - Support immutable state objects with versioning and TTL - Include serialization/deserialization (JSON and dict) - Add next_version() for state history chaining - Comprehensive test suite with 19 tests (100% pass) - Full documentation at docs/state-schema.md Features: - UUID validation for all ID fields - Optimistic locking via version numbers - Flexible payload structure - Metadata support (source, provenance, tags) - Expiration checking via TTL - Complete type hints Closes Issue #3
264 lines
8.5 KiB
Python
264 lines
8.5 KiB
Python
"""
|
|
Tests for State Model and Schema Validation
|
|
"""
|
|
|
|
import json
|
|
import uuid
|
|
import pytest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
# Add parent to path for imports
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from models.state import State, StateType, StateMetadata, load_schema
|
|
|
|
|
|
class TestStateMetadata:
|
|
"""Tests for StateMetadata class."""
|
|
|
|
def test_default_creation(self):
|
|
"""Test creating metadata with defaults."""
|
|
meta = StateMetadata()
|
|
assert meta.source == "unknown"
|
|
assert meta.provenance is None
|
|
assert meta.created_by is None
|
|
assert meta.tags == []
|
|
|
|
def test_full_creation(self):
|
|
"""Test creating metadata with all fields."""
|
|
meta = StateMetadata(
|
|
source="test-source",
|
|
provenance="trace-123",
|
|
created_by="electra",
|
|
tags=["seed", "test"]
|
|
)
|
|
assert meta.source == "test-source"
|
|
assert meta.provenance == "trace-123"
|
|
assert meta.created_by == "electra"
|
|
assert meta.tags == ["seed", "test"]
|
|
|
|
def test_to_dict(self):
|
|
"""Test metadata serialization."""
|
|
meta = StateMetadata(source="test", tags=["a", "b"])
|
|
d = meta.to_dict()
|
|
assert d["source"] == "test"
|
|
assert d["tags"] == ["a", "b"]
|
|
assert d["provenance"] is None
|
|
|
|
def test_from_dict(self):
|
|
"""Test metadata deserialization."""
|
|
data = {"source": "src", "provenance": "prov", "created_by": "user", "tags": ["t"]}
|
|
meta = StateMetadata.from_dict(data)
|
|
assert meta.source == "src"
|
|
assert meta.provenance == "prov"
|
|
|
|
|
|
class TestState:
|
|
"""Tests for State class."""
|
|
|
|
def test_create_with_defaults(self):
|
|
"""Test State.create factory method."""
|
|
entity_id = str(uuid.uuid4())
|
|
state = State.create(
|
|
entity_id=entity_id,
|
|
state_type=StateType.ACTIVE,
|
|
payload={"key": "value"}
|
|
)
|
|
|
|
assert state.entity_id == entity_id
|
|
assert state.state_type == StateType.ACTIVE
|
|
assert state.payload == {"key": "value"}
|
|
assert state.version == 1
|
|
assert state.previous_state_id is None
|
|
assert state.ttl is None
|
|
# Verify UUID format
|
|
uuid.UUID(state.id) # Should not raise
|
|
|
|
def test_create_with_metadata(self):
|
|
"""Test State.create with metadata."""
|
|
meta = StateMetadata(source="test", tags=["important"])
|
|
state = State.create(
|
|
entity_id=str(uuid.uuid4()),
|
|
state_type=StateType.PENDING,
|
|
payload={},
|
|
metadata=meta
|
|
)
|
|
assert state.metadata.source == "test"
|
|
assert state.metadata.tags == ["important"]
|
|
|
|
def test_invalid_uuid_raises(self):
|
|
"""Test that invalid UUIDs raise ValueError."""
|
|
with pytest.raises(ValueError, match="Invalid UUID"):
|
|
State(
|
|
id="not-a-uuid",
|
|
entity_id=str(uuid.uuid4()),
|
|
state_type=StateType.ACTIVE,
|
|
payload={},
|
|
timestamp=datetime.now(timezone.utc),
|
|
version=1
|
|
)
|
|
|
|
def test_invalid_version_raises(self):
|
|
"""Test that version < 1 raises ValueError."""
|
|
with pytest.raises(ValueError, match="Version must be >= 1"):
|
|
State(
|
|
id=str(uuid.uuid4()),
|
|
entity_id=str(uuid.uuid4()),
|
|
state_type=StateType.ACTIVE,
|
|
payload={},
|
|
timestamp=datetime.now(timezone.utc),
|
|
version=0
|
|
)
|
|
|
|
def test_negative_ttl_raises(self):
|
|
"""Test that negative TTL raises ValueError."""
|
|
with pytest.raises(ValueError, match="TTL must be non-negative"):
|
|
State(
|
|
id=str(uuid.uuid4()),
|
|
entity_id=str(uuid.uuid4()),
|
|
state_type=StateType.ACTIVE,
|
|
payload={},
|
|
timestamp=datetime.now(timezone.utc),
|
|
version=1,
|
|
ttl=-1
|
|
)
|
|
|
|
def test_to_dict_serialization(self):
|
|
"""Test state to dictionary conversion."""
|
|
state = State.create(
|
|
entity_id=str(uuid.uuid4()),
|
|
state_type=StateType.ACTIVE,
|
|
payload={"count": 42}
|
|
)
|
|
d = state.to_dict()
|
|
|
|
assert d["id"] == state.id
|
|
assert d["entity_id"] == state.entity_id
|
|
assert d["state_type"] == "active"
|
|
assert d["payload"] == {"count": 42}
|
|
assert "timestamp" in d
|
|
assert d["version"] == 1
|
|
|
|
def test_to_json_serialization(self):
|
|
"""Test state to JSON conversion."""
|
|
state = State.create(
|
|
entity_id=str(uuid.uuid4()),
|
|
state_type=StateType.ARCHIVED,
|
|
payload={"archived": True}
|
|
)
|
|
json_str = state.to_json()
|
|
|
|
# Should be valid JSON
|
|
parsed = json.loads(json_str)
|
|
assert parsed["state_type"] == "archived"
|
|
assert parsed["payload"]["archived"] is True
|
|
|
|
def test_from_dict_deserialization(self):
|
|
"""Test state from dictionary conversion."""
|
|
entity_id = str(uuid.uuid4())
|
|
data = {
|
|
"id": str(uuid.uuid4()),
|
|
"entity_id": entity_id,
|
|
"state_type": "pending",
|
|
"payload": {"status": "waiting"},
|
|
"timestamp": "2026-04-02T12:00:00+00:00",
|
|
"version": 2,
|
|
"metadata": {"source": "test"},
|
|
"previous_state_id": str(uuid.uuid4()),
|
|
"ttl": 3600
|
|
}
|
|
|
|
state = State.from_dict(data)
|
|
assert state.entity_id == entity_id
|
|
assert state.state_type == StateType.PENDING
|
|
assert state.version == 2
|
|
assert state.ttl == 3600
|
|
|
|
def test_from_json_deserialization(self):
|
|
"""Test state from JSON conversion."""
|
|
state = State.create(
|
|
entity_id=str(uuid.uuid4()),
|
|
state_type=StateType.DELETED,
|
|
payload={"deleted_at": "2026-04-01"}
|
|
)
|
|
json_str = state.to_json()
|
|
|
|
restored = State.from_json(json_str)
|
|
assert restored.id == state.id
|
|
assert restored.state_type == StateType.DELETED
|
|
assert restored.payload == state.payload
|
|
|
|
def test_is_expired_no_ttl(self):
|
|
"""Test that state without TTL never expires."""
|
|
state = State.create(
|
|
entity_id=str(uuid.uuid4()),
|
|
state_type=StateType.ACTIVE,
|
|
payload={}
|
|
)
|
|
assert not state.is_expired()
|
|
|
|
def test_is_expired_with_ttl(self):
|
|
"""Test TTL expiration logic."""
|
|
state = State.create(
|
|
entity_id=str(uuid.uuid4()),
|
|
state_type=StateType.ACTIVE,
|
|
payload={},
|
|
ttl=1 # 1 second TTL
|
|
)
|
|
assert not state.is_expired()
|
|
# Note: We can't easily test actual expiration without sleeping
|
|
|
|
def test_next_version(self):
|
|
"""Test creating next version of state."""
|
|
state = State.create(
|
|
entity_id=str(uuid.uuid4()),
|
|
state_type=StateType.ACTIVE,
|
|
payload={"count": 1},
|
|
version=5
|
|
)
|
|
|
|
next_state = state.next_version({"count": 2})
|
|
|
|
assert next_state.version == 6
|
|
assert next_state.previous_state_id == state.id
|
|
assert next_state.entity_id == state.entity_id
|
|
assert next_state.payload == {"count": 2}
|
|
assert next_state.state_type == state.state_type
|
|
|
|
def test_all_state_types(self):
|
|
"""Test all state type enum values."""
|
|
entity_id = str(uuid.uuid4())
|
|
|
|
for state_type in StateType:
|
|
state = State.create(
|
|
entity_id=entity_id,
|
|
state_type=state_type,
|
|
payload={}
|
|
)
|
|
assert state.state_type == state_type
|
|
assert state.to_dict()["state_type"] == state_type.value
|
|
|
|
|
|
class TestSchema:
|
|
"""Tests for JSON Schema."""
|
|
|
|
def test_schema_loads(self):
|
|
"""Test that schema file loads correctly."""
|
|
schema = load_schema()
|
|
assert schema["title"] == "SEED State Schema"
|
|
assert "properties" in schema
|
|
assert "id" in schema["required"]
|
|
assert "entity_id" in schema["required"]
|
|
|
|
def test_schema_has_examples(self):
|
|
"""Test that schema includes examples."""
|
|
schema = load_schema()
|
|
assert "examples" in schema
|
|
assert len(schema["examples"]) > 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|