76 lines
3.4 KiB
Python
76 lines
3.4 KiB
Python
"""
|
|
retention_service.run_weekly_cleanup() — CLAUDE.md §22 DATA_RETENTION.
|
|
Uses explicit ISO date strings (not the mocks' internal sequence
|
|
counters) since these tests are specifically about real date-based
|
|
filtering, matching the pattern test_api_extended.py already uses for
|
|
scrape_runs' started_at.
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
from services import retention_service
|
|
from repositories.scrape_repository import scrape_repository
|
|
from repositories.alert_repository import alert_repository
|
|
from repositories.price_history_repository import price_history_repository
|
|
|
|
OLD = (datetime.now(timezone.utc) - timedelta(days=200)).isoformat()
|
|
RECENT = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
|
|
|
|
|
|
def test_deletes_scrape_runs_older_than_90_days():
|
|
old_run = scrape_repository.create({'tenant_id': 't1', 'triggered_by': 'cron', 'status': 'complete', 'started_at': OLD})
|
|
recent_run = scrape_repository.create({'tenant_id': 't1', 'triggered_by': 'cron', 'status': 'complete', 'started_at': RECENT})
|
|
|
|
result = retention_service.run_weekly_cleanup()
|
|
|
|
assert result['scrape_runs_deleted'] == 1
|
|
assert scrape_repository.get_by_id(old_run['id']) is None
|
|
assert scrape_repository.get_by_id(recent_run['id']) is not None
|
|
|
|
|
|
def test_deletes_only_read_alerts_older_than_90_days():
|
|
old_read = alert_repository.create({
|
|
'tenant_id': 't1', 'competitor_id': 'c1', 'alert_type': 'new_product',
|
|
'message': 'x', 'read': True, 'created': OLD,
|
|
})
|
|
old_unread = alert_repository.create({
|
|
'tenant_id': 't1', 'competitor_id': 'c1', 'alert_type': 'new_product',
|
|
'message': 'x', 'read': False, 'created': OLD,
|
|
})
|
|
recent_read = alert_repository.create({
|
|
'tenant_id': 't1', 'competitor_id': 'c1', 'alert_type': 'new_product',
|
|
'message': 'x', 'read': True, 'created': RECENT,
|
|
})
|
|
|
|
result = retention_service.run_weekly_cleanup()
|
|
|
|
assert result['alerts_deleted'] == 1
|
|
assert alert_repository.get_by_id(old_read['id']) is None
|
|
assert alert_repository.get_by_id(old_unread['id']) is not None # unread never auto-deleted, regardless of age
|
|
assert alert_repository.get_by_id(recent_read['id']) is not None
|
|
|
|
|
|
def test_deletes_price_history_older_than_365_days():
|
|
very_old = (datetime.now(timezone.utc) - timedelta(days=400)).isoformat()
|
|
old_record = price_history_repository.create({'tenant_id': 't1', 'product_id': 'p1', 'scraped_at': very_old})
|
|
recent_record = price_history_repository.create({'tenant_id': 't1', 'product_id': 'p1', 'scraped_at': RECENT})
|
|
|
|
result = retention_service.run_weekly_cleanup()
|
|
|
|
assert result['price_history_deleted'] == 1
|
|
assert price_history_repository.list_for_tenant('t1') == [recent_record]
|
|
assert old_record not in price_history_repository.list_for_tenant('t1')
|
|
|
|
|
|
def test_price_history_at_200_days_survives_365_day_retention():
|
|
"""200-day-old price_history is past the 90-day scrape_runs/alerts window but under the 365-day price_history one."""
|
|
record = price_history_repository.create({'tenant_id': 't1', 'product_id': 'p1', 'scraped_at': OLD})
|
|
|
|
result = retention_service.run_weekly_cleanup()
|
|
|
|
assert result['price_history_deleted'] == 0
|
|
assert price_history_repository.list_for_tenant('t1') == [record]
|
|
|
|
|
|
def test_returns_zero_counts_when_nothing_to_clean():
|
|
result = retention_service.run_weekly_cleanup()
|
|
assert result == {'scrape_runs_deleted': 0, 'alerts_deleted': 0, 'price_history_deleted': 0}
|