"""Unit tests for EventBridge schemas.""" import pytest from pydantic import ValidationError from src.schemas import EventBridgeEvent, EventBridgeEventMetadata class TestEventBridgeEventMetadata: """Test EventBridgeEventMetadata Pydantic model.""" def test_valid_metadata(self): """Test creating valid metadata with all required fields.""" metadata = EventBridgeEventMetadata( correlation_id='corr-123', outbox_event_id=456, target_id=789, target_type='order', created_at='2023-01-01T10:00:00+00:00', processed_at='2023-01-01T10:05:00+00:00', ) assert metadata.correlation_id == 'corr-123' assert metadata.outbox_event_id == 456 assert metadata.target_id == 789 assert metadata.target_type == 'order' assert metadata.created_at == '2023-01-01T10:00:00+00:00' assert metadata.processed_at == '2023-01-01T10:05:00+00:00' def test_created_at_required(self): """Test that created_at is a required field.""" with pytest.raises(ValidationError) as exc_info: EventBridgeEventMetadata( correlation_id='corr-123', outbox_event_id=456, target_id=789, target_type='order', # Missing created_at processed_at='2023-01-01T10:05:00+00:00', ) errors = exc_info.value.errors() assert len(errors) == 1 assert errors[0]['loc'] == ('created_at',) assert errors[0]['type'] == 'missing' def test_processed_at_required(self): """Test that processed_at is a required field.""" with pytest.raises(ValidationError) as exc_info: EventBridgeEventMetadata( correlation_id='corr-123', outbox_event_id=456, target_id=789, target_type='order', created_at='2023-01-01T10:00:00+00:00', # Missing processed_at ) errors = exc_info.value.errors() assert len(errors) == 1 assert errors[0]['loc'] == ('processed_at',) assert errors[0]['type'] == 'missing' def test_timestamps_as_iso8601_strings(self): """Test that timestamps are stored as strings (ISO 8601 format).""" metadata = EventBridgeEventMetadata( correlation_id='corr-123', outbox_event_id=456, target_id=789, target_type='order', created_at='2023-06-15T14:30:00+00:00', processed_at='2023-06-15T14:35:00+00:00', ) assert isinstance(metadata.created_at, str) assert isinstance(metadata.processed_at, str) assert 'T' in metadata.created_at assert 'T' in metadata.processed_at def test_timestamps_with_timezone_offset(self): """Test that timestamps can include timezone offsets.""" metadata = EventBridgeEventMetadata( correlation_id='corr-123', outbox_event_id=456, target_id=789, target_type='order', created_at='2023-06-15T14:30:00-05:00', # EST processed_at='2023-06-15T14:35:00+00:00', # UTC ) assert metadata.created_at == '2023-06-15T14:30:00-05:00' assert metadata.processed_at == '2023-06-15T14:35:00+00:00' def test_metadata_serialization(self): """Test that metadata can be serialized to dict/JSON.""" metadata = EventBridgeEventMetadata( correlation_id='corr-123', outbox_event_id=456, target_id=789, target_type='order', created_at='2023-01-01T10:00:00+00:00', processed_at='2023-01-01T10:05:00+00:00', ) data = metadata.model_dump() assert data['correlation_id'] == 'corr-123' assert data['outbox_event_id'] == 456 assert data['created_at'] == '2023-01-01T10:00:00+00:00' assert data['processed_at'] == '2023-01-01T10:05:00+00:00' class TestEventBridgeEventDetail: """Test EventBridgeEventDetail Pydantic model.""" def test_valid_detail_with_data(self): """Test creating a valid detail with data and metadata.""" from src.schemas import EventBridgeEventDetail detail = EventBridgeEventDetail( metadata=EventBridgeEventMetadata( correlation_id='corr-456', outbox_event_id=123, target_id=789, target_type='file_upload', created_at='2023-01-01T10:00:00+00:00', processed_at='2023-01-01T10:05:00+00:00', ), data={'file_id': 789, 'status': 'completed'}, ) assert detail.metadata.correlation_id == 'corr-456' assert detail.metadata.outbox_event_id == 123 assert detail.data == {'file_id': 789, 'status': 'completed'} def test_valid_detail_with_null_data(self): """Test creating a valid detail with null data.""" from src.schemas import EventBridgeEventDetail detail = EventBridgeEventDetail( metadata=EventBridgeEventMetadata( correlation_id='corr-456', outbox_event_id=123, target_id=789, target_type='file_upload', created_at='2023-01-01T10:00:00+00:00', processed_at='2023-01-01T10:05:00+00:00', ), data=None, ) assert detail.metadata.correlation_id == 'corr-456' assert detail.data is None def test_detail_serialization_with_timestamps(self): """Test that detail serialization includes timestamps in metadata.""" from src.schemas import EventBridgeEventDetail detail = EventBridgeEventDetail( metadata=EventBridgeEventMetadata( correlation_id='corr-789', outbox_event_id=999, target_id=111, target_type='order', created_at='2023-03-15T08:30:00+00:00', processed_at='2023-03-15T08:35:00+00:00', ), data={'order_id': 111, 'total': 99.99}, ) data = detail.model_dump() assert 'metadata' in data assert data['metadata']['created_at'] == '2023-03-15T08:30:00+00:00' assert data['metadata']['processed_at'] == '2023-03-15T08:35:00+00:00' assert data['data']['order_id'] == 111 def test_detail_json_serialization(self): """Test that detail can be serialized to JSON string.""" from src.schemas import EventBridgeEventDetail detail = EventBridgeEventDetail( metadata=EventBridgeEventMetadata( correlation_id='corr-abc', outbox_event_id=555, target_id=666, target_type='payment', created_at='2023-05-20T12:00:00+00:00', processed_at='2023-05-20T12:01:00+00:00', ), data={'amount': 150.00, 'currency': 'USD'}, ) json_str = detail.model_dump_json() assert isinstance(json_str, str) assert 'corr-abc' in json_str assert '2023-05-20T12:00:00+00:00' in json_str assert '2023-05-20T12:01:00+00:00' in json_str assert '150.0' in json_str def test_detail_metadata_required(self): """Test that metadata is a required field.""" from src.schemas import EventBridgeEventDetail with pytest.raises(ValidationError) as exc_info: EventBridgeEventDetail( # Missing metadata data={'foo': 'bar'} ) errors = exc_info.value.errors() assert len(errors) == 1 assert errors[0]['loc'] == ('metadata',) assert errors[0]['type'] == 'missing' def test_detail_with_nested_data_structure(self): """Test that detail can handle complex nested data structures.""" from src.schemas import EventBridgeEventDetail detail = EventBridgeEventDetail( metadata=EventBridgeEventMetadata( correlation_id='corr-nested', outbox_event_id=777, target_id=888, target_type='order', created_at='2023-07-10T09:00:00+00:00', processed_at='2023-07-10T09:05:00+00:00', ), data={ 'order': { 'id': 888, 'items': [ {'sku': 'ABC-123', 'qty': 2}, {'sku': 'DEF-456', 'qty': 1}, ], 'shipping': {'address': '123 Main St', 'city': 'NYC'}, }, 'timestamps': { 'ordered': '2023-07-10T08:00:00+00:00', 'confirmed': '2023-07-10T08:30:00+00:00', }, }, ) assert detail.data['order']['id'] == 888 assert len(detail.data['order']['items']) == 2 assert detail.data['order']['shipping']['city'] == 'NYC' assert detail.metadata.created_at == '2023-07-10T09:00:00+00:00' class TestEventBridgeEvent: """Test EventBridgeEvent Pydantic model (full AWS event structure).""" def test_valid_event_structure(self): """Test creating a valid EventBridge event with all required fields.""" from src.schemas import EventBridgeEventDetail event = EventBridgeEvent( source='abacus.outbox', detail_type='order.created', detail=EventBridgeEventDetail( metadata=EventBridgeEventMetadata( correlation_id='corr-123', outbox_event_id=456, target_id=789, target_type='order', created_at='2023-01-01T10:00:00+00:00', processed_at='2023-01-01T10:05:00+00:00', ), data={'order_id': 789, 'amount': 100.00}, ), ) assert event.source == 'abacus.outbox' assert event.detail_type == 'order.created' assert event.detail.metadata.created_at == '2023-01-01T10:00:00+00:00' assert event.detail.data['order_id'] == 789 def test_event_with_optional_fields(self): """Test EventBridge event with optional fields set.""" from src.schemas import EventBridgeEventDetail event = EventBridgeEvent( source='abacus.outbox', detail_type='payment.completed', detail=EventBridgeEventDetail( metadata=EventBridgeEventMetadata( correlation_id='corr-456', outbox_event_id=789, target_id=999, target_type='payment', created_at='2023-02-01T11:00:00+00:00', processed_at='2023-02-01T11:02:00+00:00', ), data=None, ), event_bus_name='custom-bus', resources=['arn:aws:events:us-east-1:123456789012:rule/my-rule'], trace_header='Root=1-67891234-abcdef', ) assert event.event_bus_name == 'custom-bus' assert len(event.resources) == 1 assert event.trace_header == 'Root=1-67891234-abcdef' def test_event_serialization_aliases(self): """Test that EventBridge event uses correct serialization aliases.""" from src.schemas import EventBridgeEvent, EventBridgeEventDetail event = EventBridgeEvent( source='abacus.outbox', detail_type='test.event', detail=EventBridgeEventDetail( metadata=EventBridgeEventMetadata( correlation_id='corr-test', outbox_event_id=111, target_id=222, target_type='test', created_at='2023-03-01T12:00:00+00:00', processed_at='2023-03-01T12:01:00+00:00', ), data={'test': 'value'}, ), ) # Serialize with aliases serialized = event.model_dump(by_alias=True) # Check that aliases are used in serialization assert 'Source' in serialized assert 'DetailType' in serialized assert 'Detail' in serialized assert 'EventBusName' in serialized # Detail should be serialized as JSON string assert isinstance(serialized['Detail'], str)