Coverage for backend/django/core/auxiliary/services/csv_lifecycle.py: 89%

1from datetime import datetime, timedelta 

2 

3from django.conf import settings 

4from django.utils import timezone 

5 

6from core.auxiliary.models.UploadSession import UploadSessionPurpose 

7 

8 

9# Initial browser-managed uploads get a short retention window so abandoned 

10# multipart sessions stop consuming object-storage space quickly. 

11SHORT_CSV_LIFECYCLE_TTL = timedelta(hours=1) 

12# Non-ML CSV imports only need to survive long enough for the follow-up import job. 

13STANDARD_COMPLETED_CSV_LIFECYCLE_TTL = timedelta(days=1) 

14 

15 

16def completed_csv_lifecycle_ttl(purpose: str) -> timedelta: 

17 """Return the completed-object retention window for the given CSV purpose.""" 

18 if purpose == UploadSessionPurpose.ML_TRAINING_CSV: 

19 return timedelta(days=settings.SEAWEED_ML_UPLOAD_RETENTION_DAYS) 

20 if purpose in { 20 ↛ 25line 20 didn't jump to line 25 because the condition on line 20 was always true

21 UploadSessionPurpose.SCENARIO_CSV, 

22 UploadSessionPurpose.PINCH_UTILITY_CSV, 

23 }: 

24 return STANDARD_COMPLETED_CSV_LIFECYCLE_TTL 

25 raise ValueError(f"Unsupported CSV upload purpose: {purpose}") 

26 

27 

28def expires_at_from_ttl(ttl: timedelta, *, reference_time: datetime | None = None) -> datetime: 

29 """Convert a retention delta into an absolute expiry timestamp.""" 

30 return (reference_time or timezone.now()) + ttl