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

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 IdentityAccessControlManager 

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 flowsheet_state = models.ForeignKey( 

49 "FlowsheetState", 

50 on_delete=models.SET_NULL, 

51 related_name="solve_completion_emails", 

52 null=True, 

53 blank=True, 

54 ) 

55 scenario = models.ForeignKey( 

56 "Scenario", 

57 on_delete=models.SET_NULL, 

58 related_name="solve_completion_emails", 

59 null=True, 

60 blank=True, 

61 ) 

62 recipient = models.ForeignKey( 

63 User, 

64 on_delete=models.CASCADE, 

65 related_name="solve_completion_emails", 

66 ) 

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

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

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

70 is_multi_solve = models.BooleanField(default=False) 

71 scheduled_count = models.PositiveIntegerField(default=0) 

72 successful_count = models.PositiveIntegerField(default=0) 

73 failed_count = models.PositiveIntegerField(default=0) 

74 cancelled_count = models.PositiveIntegerField(default=0) 

75 delivery_status = models.CharField( 

76 max_length=32, 

77 choices=SolveCompletionEmailDeliveryStatus.choices, 

78 default=SolveCompletionEmailDeliveryStatus.PENDING, 

79 ) 

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

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

82 created_at = models.DateTimeField(auto_now_add=True) 

83 updated_at = models.DateTimeField(auto_now=True) 

84 

85 objects = IdentityAccessControlManager() 

86 

87 def clean(self): 

88 """Validate stable identity and nullable state/scenario provenance.""" 

89 

90 super().clean() 

91 if self.flowsheet_state_id is not None: 

92 if self.flowsheet_state.flowsheet_id != self.flowsheet_id: 

93 from django.core.exceptions import ValidationError 

94 

95 raise ValidationError( 

96 {"flowsheet_state": "Email state must belong to the email flowsheet."} 

97 ) 

98 if ( 

99 self.scenario_id is not None 

100 and self.scenario.flowsheet_state_id != self.flowsheet_state_id 

101 ): 

102 from django.core.exceptions import ValidationError 

103 

104 raise ValidationError( 

105 {"scenario": "Email scenario must belong to the email state."} 

106 ) 

107 

108 class Meta: 

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

110 

111 constraints = [ 

112 models.UniqueConstraint( 

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

114 name="unique_solve_completion_email_per_terminal_task", 

115 ) 

116 ]