Coverage for apps / core / management / commands / cleanup_device_codes.py: 25%

1"""Management command to clean up expired device codes.""" 

2 

3from django.core.cache import cache 

4from django.core.management.base import BaseCommand 

5from django.utils import timezone 

6 

7from apps.core.models import DeviceCode 

8 

9CLEANUP_CACHE_KEY = "device_code_cleanup_last_run" 

10 

11 

12class Command(BaseCommand): 

13 help = "Delete expired and invalidated device codes" 

14 

15 def add_arguments(self, parser): 

16 parser.add_argument("--dry-run", action="store_true") 

17 

18 def handle(self, *args, **options): 

19 from django.db import connection 

20 

21 # Guard against running before migrations have created the table 

22 table_names = connection.introspection.table_names() 

23 if "core_devicecode" not in table_names: 

24 self.stdout.write("DeviceCode table does not exist yet — skipping cleanup.") 

25 return 

26 

27 now = timezone.now() 

28 expired = DeviceCode.objects.filter(expires_at__lt=now, status__in=["pending", "expired"]) 

29 invalidated = DeviceCode.objects.filter(status="invalidated") 

30 # Also clean authorized codes older than 1 hour (already consumed) 

31 one_hour_ago = now - timezone.timedelta(hours=1) 

32 consumed = DeviceCode.objects.filter(status="authorized", created_at__lt=one_hour_ago) 

33 

34 counts = { 

35 "expired": expired.count(), 

36 "invalidated": invalidated.count(), 

37 "consumed": consumed.count(), 

38 } 

39 total = sum(counts.values()) 

40 

41 if options.get("dry_run"): 

42 if total == 0: 

43 self.stdout.write("No device codes to clean up.") 

44 else: 

45 self.stdout.write( 

46 f"Would delete {total} device codes " 

47 f"({counts['expired']} expired, {counts['invalidated']} invalidated, " 

48 f"{counts['consumed']} consumed)." 

49 ) 

50 return 

51 

52 if total > 0: 

53 expired.delete() 

54 invalidated.delete() 

55 consumed.delete() 

56 

57 remaining = DeviceCode.objects.count() 

58 

59 # Record run stats in cache (no expiry — persists until next run) 

60 cache.set( 

61 CLEANUP_CACHE_KEY, 

62 { 

63 "time": now.isoformat(), 

64 "deleted": total, 

65 "remaining": remaining, 

66 **counts, 

67 }, 

68 timeout=None, 

69 ) 

70 

71 if total == 0: 

72 self.stdout.write("No device codes to clean up.") 

73 else: 

74 self.stdout.write(f"Deleted {total} device codes.")