"""Tests for Event utils.""" import re import pytest from lib import constants from lib.abacus_event import AbacusEvent from lib.utils.event import get_abacus_event, validate_event_for_handler def test_get_abacus_event_with_params(): """Test get_abacus_event with params in kwargs.""" params = { 'abacus_event_id': 1, 'event_name': 'test_event', 'user_id': 'user_123', 'user_type': 'test_type' } event = get_abacus_event(params=params) assert isinstance(event, AbacusEvent) def test_get_abacus_event_with_conf(): """Test get_abacus_event with conf in args.""" class MockArgs: conf = { 'abacus_event_id': 2, 'event_name': 'another_event', 'user_id': 'user_456', 'user_type': 'another_type' } event = get_abacus_event(MockArgs()) assert isinstance(event, AbacusEvent) def test_get_abacus_event_no_params_or_conf(): """Test get_abacus_event with no params or conf.""" with pytest.raises(IndexError): get_abacus_event() def test_validate_event_for_handler_valid_event(): """Test validate_event_for_handler with a valid event.""" event = AbacusEvent( abacus_event_id=1, rolled_back_at=None, target_type='test_type', event_name='test_event' ) validate_event_for_handler(event, target_type='test_type', event_name='test_event') def test_validate_event_for_handler_missing_event_id(): """Test validate_event_for_handler with missing event ID.""" event = AbacusEvent( abacus_event_id=None, rolled_back_at=None, target_type='test_type', event_name='test_event' ) expected_error = 'Event ID missing' with pytest.raises(AssertionError, match=re.escape(expected_error)): validate_event_for_handler(event) def test_validate_event_for_handler_rolled_back_event(): """Test validate_event_for_handler with a rolled back event.""" event = AbacusEvent( abacus_event_id=1, rolled_back_at='2023-10-10 10:00:00', target_type='test_type', event_name='test_event' ) expected_error = 'Event 1 has been rolled back.' with pytest.raises(AssertionError, match=re.escape(expected_error)): validate_event_for_handler(event) def test_validate_event_for_handler_unexpected_target_type(): """Test validate_event_for_handler with unexpected target type.""" event = AbacusEvent( abacus_event_id=1, rolled_back_at=None, target_type='unexpected_type', event_name='test_event' ) expected_error = constants.ERROR_UNEXPECTED_EVENT.format( field='target_type', expected='test_type', actual='unexpected_type' ) with pytest.raises(AssertionError, match=re.escape(expected_error)): validate_event_for_handler(event, target_type='test_type') def test_validate_event_for_handler_unexpected_event_name(): """Test validate_event_for_handler with unexpected event name.""" event = AbacusEvent( abacus_event_id=1, rolled_back_at=None, target_type='test_type', event_name='unexpected_event' ) expected_error = constants.ERROR_UNEXPECTED_EVENT.format( field='event_name', expected='test_event', actual='unexpected_event' ) with pytest.raises(AssertionError, match=re.escape(expected_error)): validate_event_for_handler(event, event_name='test_event') def test_validate_event_for_handler_event_name_not_in_list(): """Test validate_event_for_handler with event name not in list.""" event = AbacusEvent( abacus_event_id=1, rolled_back_at=None, target_type='test_type', event_name='unexpected_event' ) expected_error = constants.ERROR_UNEXPECTED_EVENT.format( field='event_name', expected=['test_event1', 'test_event2'], actual='unexpected_event' ) print(expected_error) with pytest.raises(AssertionError, match=re.escape(expected_error)): validate_event_for_handler(event, event_names=['test_event1', 'test_event2'])