import os os.environ['DB_ENV'] = 'test' import pytest from models.tracking_event import TrackingEvent from db.tracking import ( insert_tracking_event, get_tracking_events_by_child, get_tracking_events_by_user, anonymize_tracking_events_for_user ) from db.db import tracking_events_db def test_tracking_event_creation(): """Test creating a tracking event with factory method.""" event = TrackingEvent.create_event( user_id='user123', child_id='child456', entity_type='task', entity_id='task789', action='activated', points_before=10, points_after=20, metadata={'task_name': 'Homework'} ) assert event.user_id == 'user123' assert event.child_id == 'child456' assert event.entity_type == 'task' assert event.action == 'activated' assert event.points_before == 10 assert event.points_after == 20 assert event.delta == 10 assert event.metadata == {'task_name': 'Homework'} assert event.occurred_at # Should have ISO timestamp def test_tracking_event_delta_invariant(): """Test that delta invariant is enforced.""" with pytest.raises(ValueError, match="Delta invariant violated"): TrackingEvent( user_id='user123', child_id='child456', entity_type='task', entity_id='task789', action='activated', points_before=10, points_after=20, delta=5, # Wrong! Should be 10 occurred_at='2026-02-09T12:00:00Z' ) def test_insert_and_query_tracking_event(): """Test inserting and querying tracking events.""" tracking_events_db.truncate() event1 = TrackingEvent.create_event( user_id='user1', child_id='child1', entity_type='task', entity_id='task1', action='activated', points_before=0, points_after=10 ) event2 = TrackingEvent.create_event( user_id='user1', child_id='child1', entity_type='reward', entity_id='reward1', action='requested', points_before=10, points_after=10 ) insert_tracking_event(event1) insert_tracking_event(event2) # Query by child events, total = get_tracking_events_by_child('child1', limit=10, offset=0) assert total == 2 assert len(events) == 2 def test_query_with_filters(): """Test querying with entity_type and action filters.""" tracking_events_db.truncate() # Insert task activation task_event = TrackingEvent.create_event( user_id='user1', child_id='child1', entity_type='task', entity_id='task1', action='activated', points_before=0, points_after=10 ) insert_tracking_event(task_event) # Insert reward request reward_event = TrackingEvent.create_event( user_id='user1', child_id='child1', entity_type='reward', entity_id='reward1', action='requested', points_before=10, points_after=10 ) insert_tracking_event(reward_event) # Filter by entity_type events, total = get_tracking_events_by_child('child1', entity_type='task') assert total == 1 assert events[0].entity_type == 'task' # Filter by action events, total = get_tracking_events_by_child('child1', action='requested') assert total == 1 assert events[0].action == 'requested' def test_pagination(): """Test offset-based pagination.""" tracking_events_db.truncate() # Insert 5 events for i in range(5): event = TrackingEvent.create_event( user_id='user1', child_id='child1', entity_type='task', entity_id=f'task{i}', action='activated', points_before=i * 10, points_after=(i + 1) * 10 ) insert_tracking_event(event) # First page events, total = get_tracking_events_by_child('child1', limit=2, offset=0) assert total == 5 assert len(events) == 2 # Second page events, total = get_tracking_events_by_child('child1', limit=2, offset=2) assert total == 5 assert len(events) == 2 # Last page events, total = get_tracking_events_by_child('child1', limit=2, offset=4) assert total == 5 assert len(events) == 1 def test_anonymize_tracking_events(): """Test anonymizing tracking events on user deletion.""" tracking_events_db.truncate() event = TrackingEvent.create_event( user_id='user_to_delete', child_id='child1', entity_type='task', entity_id='task1', action='activated', points_before=0, points_after=10 ) insert_tracking_event(event) # Anonymize count = anonymize_tracking_events_for_user('user_to_delete') assert count == 1 # Verify user_id is None events, total = get_tracking_events_by_child('child1') assert total == 1 assert events[0].user_id is None assert events[0].child_id == 'child1' # Child data preserved def test_points_change_correctness(): """Test that points before/after/delta are tracked correctly.""" tracking_events_db.truncate() # Task activation (points increase) task_event = TrackingEvent.create_event( user_id='user1', child_id='child1', entity_type='task', entity_id='task1', action='activated', points_before=50, points_after=60 ) assert task_event.delta == 10 insert_tracking_event(task_event) # Reward redeem (points decrease) reward_event = TrackingEvent.create_event( user_id='user1', child_id='child1', entity_type='reward', entity_id='reward1', action='redeemed', points_before=60, points_after=40 ) assert reward_event.delta == -20 insert_tracking_event(reward_event) # Query and verify events, _ = get_tracking_events_by_child('child1') assert len(events) == 2 assert events[0].delta == -20 # Most recent (sorted desc) assert events[1].delta == 10 def test_no_points_change_for_request_and_cancel(): """Test that reward request and cancel have delta=0.""" tracking_events_db.truncate() # Request request_event = TrackingEvent.create_event( user_id='user1', child_id='child1', entity_type='reward', entity_id='reward1', action='requested', points_before=100, points_after=100 ) assert request_event.delta == 0 insert_tracking_event(request_event) # Cancel cancel_event = TrackingEvent.create_event( user_id='user1', child_id='child1', entity_type='reward', entity_id='reward1', action='cancelled', points_before=100, points_after=100 ) assert cancel_event.delta == 0 insert_tracking_event(cancel_event) events, _ = get_tracking_events_by_child('child1') assert all(e.delta == 0 for e in events)