Coverage for app/venv/lib/python3.14/site-packages/weblate/wladmin/models.py: 59%
215 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
1# Copyright © Michal Čihař <michal@weblate.org>
2#
3# SPDX-License-Identifier: GPL-3.0-or-later
5from __future__ import annotations
7import json
8from typing import TYPE_CHECKING, ClassVar, TypedDict
10import dateutil.parser
11from appconf import AppConf
12from django.conf import settings
13from django.contrib.admin import ModelAdmin
14from django.core.cache import cache
15from django.core.checks import run_checks
16from django.db import models
17from django.utils import timezone
18from django.utils.functional import cached_property
19from django.utils.translation import gettext, gettext_lazy
21from weblate.auth.models import User
22from weblate.trans.models import Component, Project
23from weblate.utils.backup import (
24 BackupError,
25 backup,
26 cleanup,
27 get_paper_key,
28 initialize,
29 make_password,
30 prune,
31)
32from weblate.utils.const import SUPPORT_STATUS_CACHE_KEY
33from weblate.utils.requests import request
34from weblate.utils.site import get_site_url
35from weblate.utils.stats import GlobalStats
36from weblate.utils.validators import validate_backup_path
37from weblate.vcs.ssh import ensure_ssh_key
39if TYPE_CHECKING: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true
40 from weblate.auth.models import AuthenticatedHttpRequest
43class WeblateConf(AppConf):
44 BACKGROUND_ADMIN_CHECKS = True
46 class Meta:
47 prefix = ""
50class WeblateModelAdmin(ModelAdmin):
51 """Customized Model Admin object."""
53 delete_confirmation_template = "wladmin/delete_confirmation.html"
54 delete_selected_confirmation_template = "wladmin/delete_selected_confirmation.html"
57class ConfigurationErrorManager(models.Manager["ConfigurationError"]):
58 def configuration_health_check(self, checks=None) -> None:
59 # Run deployment checks if needed
60 if checks is None: 60 ↛ 62line 60 didn't jump to line 62 because the condition on line 60 was always true
61 checks = run_checks(include_deployment_checks=True)
62 checks_dict = {check.id: check for check in checks}
63 criticals = {
64 "weblate.E002",
65 "weblate.E003",
66 "weblate.E007",
67 "weblate.E009",
68 "weblate.E012",
69 "weblate.E013",
70 "weblate.E014",
71 "weblate.E015",
72 "weblate.E017",
73 "weblate.E018",
74 "weblate.E019",
75 "weblate.C023",
76 "weblate.C029",
77 "weblate.C030",
78 "weblate.C031",
79 "weblate.C032",
80 "weblate.E034",
81 "weblate.C035",
82 "weblate.C036",
83 "weblate.C037",
84 "weblate.C038",
85 "weblate.C040",
86 "weblate.C041",
87 "weblate.E006",
88 }
89 removals = []
90 existing = {error.name: error for error in self.all()}
92 for check_id in criticals:
93 if check_id in checks_dict:
94 check = checks_dict[check_id]
95 if check_id in existing:
96 error = existing[check_id]
97 if error.message != check.msg: 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true
98 error.message = check.msg
99 error.save(update_fields=["message"])
100 else:
101 self.create(name=check_id, message=check.msg)
102 elif check_id in existing: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 removals.append(check_id)
105 if removals: 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true
106 self.filter(name__in=removals).delete()
109class ConfigurationError(models.Model):
110 name = models.CharField(unique=True, max_length=150)
111 message = models.TextField()
112 timestamp = models.DateTimeField(default=timezone.now)
113 ignored = models.BooleanField(default=False, db_index=True)
115 objects = ConfigurationErrorManager()
117 class Meta:
118 indexes: ClassVar = [
119 models.Index(fields=["ignored", "timestamp"]),
120 ]
121 verbose_name = "Configuration error"
122 verbose_name_plural = "Configuration errors"
124 def __str__(self) -> str:
125 return self.name
128SUPPORT_NAMES = {
129 # Translators: The Weblate server has no paid support by the Weblate team
130 "community": gettext_lazy("Community support"),
131 # Translators: The Weblate server is hosted by the Weblate team
132 "hosted": gettext_lazy("Hosted service"),
133 # Translators: The Weblate server has paid support from the Weblate team
134 "basic": gettext_lazy("Basic self-hosted support"),
135 # Translators: The Weblate server has extended support from the Weblate team
136 "extended": gettext_lazy("Extended self-hosted support"),
137 # Translators: The Weblate server has premium support from the Weblate team
138 "premium": gettext_lazy("Premium self-hosted support"),
139}
142class SupportStatusManager(models.Manager["SupportStatus"]):
143 def get_current(self, *, for_update: bool = False) -> SupportStatus:
144 base = self.filter(enabled=True)
145 if for_update: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 base = base.select_for_update()
147 try:
148 return base.latest("expiry")
149 except SupportStatus.DoesNotExist:
150 return SupportStatus(name="community", enabled=False)
153class SupportStatus(models.Model):
154 name = models.CharField(max_length=150)
155 secret = models.CharField(max_length=400)
156 expiry = models.DateTimeField(db_index=True, null=True)
157 in_limits = models.BooleanField(default=True)
158 discoverable = models.BooleanField(default=False)
159 limits = models.JSONField(default=dict)
160 has_subscription = models.BooleanField(default=False)
161 backup_repository = models.CharField(max_length=500, default="", blank=True)
162 enabled = models.BooleanField(default=True, db_index=True, blank=True)
164 objects = SupportStatusManager()
166 class Meta:
167 verbose_name = "Support status"
168 verbose_name_plural = "Support statuses"
170 def __str__(self) -> str:
171 return f"{self.name}:{self.expiry}"
173 def get_verbose(self):
174 if self.has_expired_support:
175 return gettext("Unpaid subscription")
176 return SUPPORT_NAMES.get(self.name, self.name)
178 def refresh(self) -> None:
179 stats = GlobalStats()
180 data = {
181 "secret": self.secret,
182 "site_url": get_site_url(),
183 "site_title": settings.SITE_TITLE,
184 "users": User.objects.count(),
185 "projects": Project.objects.count(),
186 "components": Component.objects.count(),
187 "languages": stats.languages,
188 "source_strings": stats.source_strings,
189 "strings": stats.all,
190 "words": stats.all_words,
191 }
192 if self.discoverable:
193 data["discoverable"] = 1
194 data["public_projects"] = json.dumps(
195 [
196 {
197 "name": project.name,
198 "url": project.get_absolute_url(),
199 "web": project.web,
200 }
201 for project in Project.objects.filter(
202 access_control=Project.ACCESS_PUBLIC
203 ).iterator()
204 ]
205 )
206 ssh_key = ensure_ssh_key()
207 if ssh_key:
208 data["ssh_key"] = ssh_key["key"]
209 response = request("post", settings.SUPPORT_API_URL, data=data, timeout=360)
210 response.raise_for_status()
211 payload = response.json()
212 self.name = payload["name"]
213 self.expiry = dateutil.parser.parse(payload["expiry"])
214 self.in_limits = payload["in_limits"]
215 self.limits = payload["limits"]
216 self.has_subscription = payload["has_subscription"]
217 self.backup_repository = payload["backup_repository"]
218 if self.backup_repository:
219 BackupService.objects.get_or_create(
220 repository=self.backup_repository, defaults={"enabled": False}
221 )
222 # Invalidate support status cache
223 cache.delete(SUPPORT_STATUS_CACHE_KEY)
225 def get_limits_details(self):
226 stats = GlobalStats()
227 current_values = {
228 "hosted_words": stats.all_words,
229 "hosted_strings": stats.all,
230 "source_strings": stats.source_strings,
231 "projects": Project.objects.count(),
232 "languages": stats.languages,
233 }
234 names = {
235 "hosted_words": gettext_lazy("Hosted words"),
236 "hosted_strings": gettext_lazy("Hosted strings"),
237 "source_strings": gettext_lazy("Source strings"),
238 "projects": gettext_lazy("Projects"),
239 "languages": gettext_lazy("Languages"),
240 }
241 result = []
242 for limit, value in self.limits.items():
243 if not value or limit not in names:
244 continue
245 current = current_values[limit]
246 result.append(
247 {
248 "name": names[limit],
249 "limit": value,
250 "current": current,
251 "in_limit": current < value,
252 }
253 )
254 return result
256 @property
257 def has_support(self) -> bool:
258 return self.name != "community"
260 @property
261 def has_expired_support(self) -> bool:
262 return self.pk is not None and self.has_subscription and not self.has_support
265class BackupService(models.Model):
266 repository = models.CharField(
267 max_length=500,
268 default="",
269 verbose_name=gettext_lazy("Backup repository URL"),
270 help_text=gettext_lazy(
271 "Use /path/to/repo for local backups "
272 "or user@host:/path/to/repo "
273 "or ssh://user@host:port/path/to/backups for remote SSH backups."
274 ),
275 validators=[validate_backup_path],
276 )
277 enabled = models.BooleanField(default=True)
278 timestamp = models.DateTimeField(default=timezone.now)
279 passphrase = models.CharField(max_length=100, default=make_password)
280 paperkey = models.TextField()
282 class Meta:
283 verbose_name = "Support service"
284 verbose_name_plural = "Support services"
286 def __str__(self) -> str:
287 return self.repository
289 @cached_property
290 def last_logs(self) -> models.QuerySet[BackupLog]:
291 return self.backuplog_set.order_by("-timestamp")[:10]
293 @cached_property
294 def has_errors(self) -> bool:
295 for log in self.last_logs:
296 if log.event == "error":
297 return True
298 if log.event == "backup":
299 return False
300 return False
302 def ensure_init(self) -> None:
303 if not self.paperkey:
304 try:
305 self.paperkey = get_paper_key(self.repository)
306 self.save(update_fields=["paperkey"])
307 except BackupError:
308 try:
309 log = initialize(self.repository, self.passphrase)
310 self.backuplog_set.create(event="init", log=log)
311 self.paperkey = get_paper_key(self.repository)
312 self.save()
313 except BackupError as error:
314 self.backuplog_set.create(event="error", log=str(error))
316 def backup(self) -> None:
317 try:
318 log = backup(self.repository, self.passphrase)
319 self.backuplog_set.create(event="backup", log=log)
320 except BackupError as error:
321 self.backuplog_set.create(event="error", log=str(error))
323 def prune(self) -> None:
324 try:
325 log = prune(self.repository, self.passphrase)
326 self.backuplog_set.create(event="prune", log=log)
327 except BackupError as error:
328 self.backuplog_set.create(event="error", log=str(error))
330 def cleanup(self) -> None:
331 initial = self.backuplog_set.filter(event="cleanup").exists()
332 try:
333 log = cleanup(self.repository, self.passphrase, initial=initial)
334 self.backuplog_set.create(event="cleanup", log=log)
335 except BackupError as error:
336 self.backuplog_set.create(event="error", log=str(error))
339class BackupLog(models.Model):
340 service = models.ForeignKey(BackupService, on_delete=models.deletion.CASCADE)
341 timestamp = models.DateTimeField(default=timezone.now)
342 event = models.CharField(
343 max_length=100,
344 choices=(
345 # Translators: Backup repository operation
346 ("backup", gettext_lazy("Backup performed")),
347 # Translators: Backup repository operation
348 ("error", gettext_lazy("Backup failed")),
349 # Translators: Backup repository operation
350 ("prune", gettext_lazy("Deleted the oldest backups")),
351 # Translators: Backup repository operation
352 ("cleanup", gettext_lazy("Cleaned up backup storage")),
353 # Translators: Backup repository operation
354 ("init", gettext_lazy("Repository initialization")),
355 ),
356 db_index=True,
357 )
358 log = models.TextField()
360 class Meta:
361 verbose_name = "Backup log"
362 verbose_name_plural = "Backup logs"
364 def __str__(self) -> str:
365 return f"{self.service}:{self.event}"
368class SupportStatusDict(TypedDict):
369 has_support: bool
370 in_limits: bool
371 has_expired_support: bool
372 backup_repository: str
373 name: str
374 is_hosted_weblate: bool
375 is_dedicated: bool
378def get_support_status(request: AuthenticatedHttpRequest) -> SupportStatusDict:
379 support_status: SupportStatusDict
380 if hasattr(request, "weblate_support_status"): 380 ↛ 381line 380 didn't jump to line 381 because the condition on line 380 was never true
381 support_status = request.weblate_support_status
382 else:
383 support_status = cache.get(SUPPORT_STATUS_CACHE_KEY)
384 if support_status is None:
385 support_status_instance = SupportStatus.objects.get_current()
386 is_hosted = support_status_instance.name == "hosted"
387 support_status = {
388 "name": support_status_instance.name,
389 "is_hosted_weblate": is_hosted
390 and settings.SITE_DOMAIN == "hosted.weblate.org",
391 "is_dedicated": is_hosted,
392 "has_support": support_status_instance.has_support,
393 "has_expired_support": support_status_instance.has_expired_support,
394 "in_limits": support_status_instance.in_limits,
395 "backup_repository": support_status_instance.backup_repository,
396 }
397 cache.set(SUPPORT_STATUS_CACHE_KEY, support_status, 86400)
398 request.weblate_support_status = support_status
399 return support_status