Coverage for backend/django/core/auxiliary/models/SolveCompletionEmail.py: 100%

1"""Persistence models for solve-completion email attempts.""" 

2 

3from django.db import models 

4 

5from authentication.user.models import User 

6from core.auxiliary.enums.generalEnums import TaskStatus 

7from core.managers import AccessControlManager 

8 

9 

10class SolveCompletionEmailDeliveryStatus(models.TextChoices): 

11 """Lifecycle states for an individual delivery attempt.""" 

12 

13 PENDING = "pending" 

14 SENT = "sent" 

15 FAILED = "failed" 

16 SKIPPED = "skipped" 

17 

18 

19class SolveCompletionEmailOutcome(models.TextChoices): 

20 """Template and summary classifications stored with each delivery record.""" 

21 

22 SINGLE_SUCCESS = "single_success" 

23 SINGLE_FAILURE = "single_failure" 

24 SINGLE_CANCELLED = "single_cancelled" 

25 MULTI_ALL_SUCCEEDED = "multi_all_succeeded" 

26 MULTI_MIXED = "multi_mixed" 

27 MULTI_ALL_FAILED = "multi_all_failed" 

28 MULTI_CANCELLED = "multi_cancelled" 

29 

30 

31class SolveCompletionEmail(models.Model): 

32 """Tracks solve completion email attempts and prevents duplicate delivery. 

33 

34 The record is created before the email is sent so the unique constraint can 

35 act as the dedupe boundary when duplicate completion events arrive. 

36 """ 

37 

38 task = models.ForeignKey( 

39 "Task", 

40 on_delete=models.CASCADE, 

41 related_name="solve_completion_emails", 

42 ) 

43 flowsheet = models.ForeignKey( 

44 "Flowsheet", 

45 on_delete=models.CASCADE, 

46 related_name="solve_completion_emails", 

47 ) 

48 scenario = models.ForeignKey( 

49 "Scenario", 

50 on_delete=models.CASCADE, 

51 related_name="solve_completion_emails", 

52 null=True, 

53 blank=True, 

54 ) 

55 recipient = models.ForeignKey( 

56 User, 

57 on_delete=models.CASCADE, 

58 related_name="solve_completion_emails", 

59 ) 

60 recipient_email = models.EmailField(max_length=255, null=True, blank=True) 

61 terminal_status = models.CharField(max_length=32, choices=TaskStatus.choices) 

62 outcome_key = models.CharField(max_length=64, choices=SolveCompletionEmailOutcome.choices) 

63 is_multi_solve = models.BooleanField(default=False) 

64 scheduled_count = models.PositiveIntegerField(default=0) 

65 successful_count = models.PositiveIntegerField(default=0) 

66 failed_count = models.PositiveIntegerField(default=0) 

67 cancelled_count = models.PositiveIntegerField(default=0) 

68 delivery_status = models.CharField( 

69 max_length=32, 

70 choices=SolveCompletionEmailDeliveryStatus.choices, 

71 default=SolveCompletionEmailDeliveryStatus.PENDING, 

72 ) 

73 sent_at = models.DateTimeField(null=True, blank=True) 

74 error = models.TextField(null=True, blank=True) 

75 created_at = models.DateTimeField(auto_now_add=True) 

76 updated_at = models.DateTimeField(auto_now=True) 

77 

78 objects = AccessControlManager() 

79 

80 class Meta: 

81 """Django model metadata for audit and dedupe behavior.""" 

82 

83 constraints = [ 

84 models.UniqueConstraint( 

85 fields=["task", "terminal_status"], 

86 name="unique_solve_completion_email_per_terminal_task", 

87 ) 

88 ]