Coverage for app/venv/lib/python3.14/site-packages/weblate/trans/models/pending.py: 70%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5from __future__ import annotations 

6 

7from datetime import timedelta 

8from typing import TYPE_CHECKING 

9 

10from django.conf import settings 

11from django.db import models 

12from django.db.models import BooleanField, Case, OuterRef, Q, Subquery, TextField, When 

13from django.db.models.fields.json import KT 

14from django.db.models.functions import Cast 

15from django.utils import timezone 

16 

17from weblate.utils.db import using_postgresql 

18from weblate.utils.state import STATE_APPROVED, STATE_FUZZY, StringState 

19from weblate.utils.version import GIT_VERSION 

20 

21if TYPE_CHECKING: 21 ↛ 22line 21 didn't jump to line 22 because the condition on line 21 was never true

22 from datetime import datetime 

23 

24 from weblate.auth.models import User 

25 from weblate.trans.models import Component, Translation, Unit 

26 

27 

28class PendingChangeQuerySet(models.QuerySet): 

29 def for_component(self, component: Component): 

30 """Return pending changes for a specific component.""" 

31 return self.filter(unit__translation__component=component) 

32 

33 def for_translation(self, translation: Translation): 

34 from weblate.trans.models.project import CommitPolicyChoices 

35 

36 one_week_ago = timezone.now() - timedelta(days=7) 

37 

38 # filter changes that are new or eligible for retry 

39 eligible_for_attempt_filter = ( 

40 Q(metadata__last_failed__isnull=True) 

41 | ~Q(failed_revision=translation.revision) 

42 | ~Q(weblate_version=GIT_VERSION) 

43 | Q(last_failed__lt=one_week_ago.isoformat()) 

44 ) 

45 

46 qs = ( 

47 self.filter(unit__translation=translation) 

48 .annotate( 

49 # use KT and explicitly cast key value to string to avoid 

50 # django from treating string comparison values for RHS as JSON 

51 # on mariadb and mysql. 

52 failed_revision=Cast(KT("metadata__failed_revision"), TextField()), 

53 weblate_version=Cast(KT("metadata__weblate_version"), TextField()), 

54 last_failed=Cast(KT("metadata__last_failed"), TextField()), 

55 eligible_for_attempt=Case( 

56 When(eligible_for_attempt_filter, then=True), 

57 default=False, 

58 output_field=BooleanField(), 

59 ), 

60 ) 

61 .filter(Q(eligible_for_attempt=True) | Q(metadata__blocking_unit=True)) 

62 .order_by("unit_id", "timestamp") 

63 .values_list( 

64 "pk", "unit_id", "metadata__blocking_unit", "eligible_for_attempt" 

65 ) 

66 ) 

67 

68 blocked_units = set() 

69 eligible_pks = [] 

70 

71 # failed changes that are not yet eligible for retry and have blocking_unit=True 

72 # should prevent all newer changes from being committed 

73 for pk, unit_id, blocking_unit, eligible_for_attempt in qs.iterator(): 

74 if unit_id in blocked_units: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 continue 

76 

77 if eligible_for_attempt: 77 ↛ 79line 77 didn't jump to line 79 because the condition on line 77 was always true

78 eligible_pks.append(pk) 

79 elif blocking_unit: 

80 blocked_units.add(unit_id) 

81 

82 qs = self.filter(pk__in=eligible_pks) 

83 

84 policy = translation.component.project.commit_policy 

85 if policy == CommitPolicyChoices.ALL: 85 ↛ 88line 85 didn't jump to line 88 because the condition on line 85 was always true

86 return qs 

87 

88 filters = [] 

89 if policy == CommitPolicyChoices.WITHOUT_NEEDS_EDITING: 

90 filters.append(~Q(state=STATE_FUZZY)) 

91 elif policy == CommitPolicyChoices.APPROVED_ONLY: 

92 filters.append(Q(state=STATE_APPROVED)) 

93 

94 # For each unit, finds the last change that makes it eligible for committing 

95 # based on the project's commit policy, and returns all changes up to and 

96 # including that change. 

97 latest_eligible_changes = ( 

98 PendingUnitChange.objects.filter(*filters, unit_id=OuterRef("unit_id")) 

99 .order_by("-timestamp") 

100 .values("timestamp")[:1] 

101 ) 

102 return qs.filter(timestamp__lte=Subquery(latest_eligible_changes)) 

103 

104 def older_than(self, timestamp: datetime): 

105 """Return pending changes older than given timestamp.""" 

106 return self.filter(timestamp__lt=timestamp) 

107 

108 def select_for_update(self) -> PendingChangeQuerySet: # type: ignore[override] 

109 if using_postgresql(): 109 ↛ 113line 109 didn't jump to line 113 because the condition on line 109 was always true

110 # Use weaker locking and limit locking to this table only 

111 return super().select_for_update(no_key=True, of=("self",)) 

112 # Discard any select_related to avoid locking additional tables 

113 return super().select_for_update().select_related(None) 

114 

115 

116class PendingUnitChange(models.Model): 

117 """Stores actual change data that needs to be committed to a repository.""" 

118 

119 unit = models.ForeignKey( 

120 "trans.Unit", 

121 on_delete=models.CASCADE, 

122 related_name="pending_changes", 

123 db_index=True, 

124 ) 

125 author = models.ForeignKey( 

126 settings.AUTH_USER_MODEL, 

127 on_delete=models.CASCADE, 

128 db_index=True, 

129 ) 

130 target = models.TextField(default="", blank=True) 

131 explanation = models.TextField(default="", blank=True) 

132 source_unit_explanation = models.TextField(default="", blank=True) 

133 state = models.IntegerField(default=0, choices=StringState.choices, db_index=True) 

134 timestamp = models.DateTimeField(default=timezone.now, db_index=True) 

135 add_unit = models.BooleanField(default=False) 

136 metadata = models.JSONField(default=dict, blank=True, null=False) 

137 

138 objects = PendingChangeQuerySet.as_manager() 

139 

140 class Meta: 

141 app_label = "trans" 

142 verbose_name = "pending change" 

143 verbose_name_plural = "pending changes" 

144 

145 def __str__(self) -> str: 

146 return f"Pending change for {self.unit} -> {self.target} by {self.author}" 

147 

148 @classmethod 

149 def store_unit_change( 

150 cls, 

151 unit: Unit, 

152 *, 

153 author: User | None = None, 

154 target: str | None = None, 

155 explanation: str | None = None, 

156 state: int | None = None, 

157 add_unit: bool = False, 

158 source_unit_explanation: str | None = None, 

159 timestamp: datetime | None = None, 

160 ) -> PendingUnitChange: 

161 """Store complete change data for a unit by a specific author.""" 

162 # update current fields in disk_state details for comparison of 

163 # uncommitted disk state with incoming changes during check_sync. 

164 unit.store_disk_state() 

165 

166 if target is None: 166 ↛ 168line 166 didn't jump to line 168 because the condition on line 166 was always true

167 target = unit.target 

168 if explanation is None: 168 ↛ 170line 168 didn't jump to line 170 because the condition on line 168 was always true

169 explanation = unit.explanation 

170 if state is None: 170 ↛ 172line 170 didn't jump to line 172 because the condition on line 170 was always true

171 state = unit.state 

172 if source_unit_explanation is None: 172 ↛ 174line 172 didn't jump to line 174 because the condition on line 172 was always true

173 source_unit_explanation = unit.source_unit.explanation 

174 if author is None: 174 ↛ 177line 174 didn't jump to line 177 because the condition on line 174 was always true

175 author = unit.get_last_content_change()[0] 

176 

177 kwargs = { 

178 "unit": unit, 

179 "author": author, 

180 "target": target, 

181 "explanation": explanation, 

182 "state": state, 

183 "add_unit": add_unit, 

184 "source_unit_explanation": source_unit_explanation, 

185 } 

186 if timestamp is not None: 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true

187 kwargs["timestamp"] = timestamp 

188 

189 pending_unit_change = PendingUnitChange(**kwargs) 

190 pending_unit_change.save() 

191 return pending_unit_change