feat: Implement task and reward tracking feature
All checks were successful
Gitea Actions Demo / build-and-push (push) Successful in 24s

- Added tracking events for tasks, penalties, and rewards with timestamps.
- Created new TinyDB table for tracking records to maintain audit history.
- Developed backend API for querying tracking events with filters and pagination.
- Implemented logging for tracking events with per-user rotating log files.
- Added unit tests for tracking event creation, querying, and anonymization.
- Deferred frontend changes for future implementation.
- Established acceptance criteria and documentation for the tracking feature.

feat: Introduce account deletion scheduler

- Implemented a scheduler to delete accounts marked for deletion after a configurable threshold.
- Added new fields to the User model to manage deletion status and attempts.
- Created admin API endpoints for managing deletion thresholds and viewing the deletion queue.
- Integrated error handling and logging for the deletion process.
- Developed unit tests for the deletion scheduler and related API endpoints.
- Documented the deletion process and acceptance criteria.
This commit is contained in:
2026-02-09 15:39:43 -05:00
parent 27f02224ab
commit 3dee8b80a2
20 changed files with 1450 additions and 0 deletions

View File

@@ -0,0 +1,254 @@
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)