Coverage for app/venv/lib/python3.14/site-packages/weblate/utils/apps.py: 72%
211 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 errno
8import os
9import sys
10import time
11from datetime import timedelta
12from itertools import chain
13from pathlib import Path
14from typing import TYPE_CHECKING, cast
16from celery.exceptions import TimeoutError as CeleryTimeoutError
17from django.apps import AppConfig
18from django.conf import settings
19from django.core.cache import cache
20from django.core.checks import Error, Info, register
21from django.core.exceptions import ImproperlyConfigured
22from django.core.mail import get_connection
23from django.db import DatabaseError
24from django.db.models import CharField, TextField
25from django.db.models.functions import MD5, Lower
26from django.db.models.lookups import Regex
27from django.utils import timezone
28from packaging.version import Version
30from weblate.utils.html import format_html_join_comma, list_to_tuples
32from .celery import is_celery_queue_long
33from .checks import weblate_check
34from .classloader import ClassLoader
35from .const import HEARTBEAT_FREQUENCY
36from .data import data_path
37from .db import (
38 MySQLSearchLookup,
39 PostgreSQLRegexLookup,
40 PostgreSQLSearchLookup,
41 PostgreSQLSubstringLookup,
42 measure_database_latency,
43 using_postgresql,
44)
45from .errors import init_error_collection
46from .site import check_domain, get_site_domain
47from .version import VERSION_BASE, get_latest_version
49if TYPE_CHECKING: 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true
50 from collections.abc import Iterable, Sequence
52 from django.core.checks import CheckMessage
53 from django.db.models.lookups import Lookup
56GOOD_CACHE = {"MemcachedCache", "PyLibMCCache", "DatabaseCache", "RedisCache"}
57DEFAULT_MAILS = {
58 "root@localhost",
59 "webmaster@localhost",
60 "noreply@example.com",
61 "weblate@example.com",
62}
63DEFAULT_SECRET_KEYS = {
64 "jm8fqjlg+5!#xu%e-oh#7!$aa7!6avf7ud*_v=chdrb9qdco6(",
65 "secret key used for tests only",
66}
69@register(deploy=True)
70def check_mail_connection(
71 *,
72 app_configs: Sequence[AppConfig] | None,
73 databases: Sequence[str] | None,
74 **kwargs,
75) -> Iterable[CheckMessage]:
76 errors = []
77 try:
78 connection = get_connection(timeout=5)
79 connection.open()
80 connection.close()
81 except Exception as error:
82 message = "Cannot send e-mail ({}), please check EMAIL_* settings."
83 errors.append(weblate_check("weblate.E003", message.format(error)))
85 return errors
88@register(deploy=True)
89def check_celery(
90 *,
91 app_configs: Sequence[AppConfig] | None,
92 databases: Sequence[str] | None,
93 **kwargs,
94) -> Iterable[CheckMessage]:
95 # Import this lazily to avoid evaluating settings too early
96 from weblate.utils.tasks import ping
98 errors = []
99 if settings.CELERY_TASK_ALWAYS_EAGER: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 errors.append(
101 weblate_check(
102 "weblate.E005", "Celery is configured in the eager mode", Error
103 )
104 )
105 elif settings.CELERY_BROKER_URL == "memory://": 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true
106 errors.append(
107 weblate_check(
108 "weblate.E026", "Celery is configured to store queue in local memory"
109 )
110 )
111 else:
112 if is_celery_queue_long(): 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true
113 errors.append(
114 weblate_check(
115 "weblate.E009",
116 "The Celery tasks queue is too long, either the worker "
117 "is not running, or is too slow.",
118 )
119 )
121 start = time.monotonic()
122 result = ping.delay()
123 try:
124 pong = result.get(timeout=10, disable_sync_subtasks=False)
125 cache.set("celery_latency", round(1000 * (time.monotonic() - start)))
126 current = ping()
127 # Check for outdated Celery running different version of configuration
128 if current != pong: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 if pong is None:
130 # Celery runs Weblate 4.0 or older
131 differing = ["version"]
132 else:
133 differing = [
134 key
135 for key, value in current.items()
136 if key not in pong or value != pong[key]
137 ]
138 errors.append(
139 weblate_check(
140 "weblate.E034",
141 "The Celery process is outdated or misconfigured."
142 " Following items differ: {}".format(
143 format_html_join_comma("{}", list_to_tuples(differing))
144 ),
145 )
146 )
147 except CeleryTimeoutError:
148 errors.append(
149 weblate_check(
150 "weblate.E019",
151 "The Celery does not process tasks, or is too slow "
152 "in processing them.",
153 )
154 )
155 except NotImplementedError:
156 errors.append(
157 weblate_check(
158 "weblate.E020",
159 "The Celery is not configured to store results, "
160 "CELERY_RESULT_BACKEND is probably not set.",
161 )
162 )
164 heartbeat = cache.get("celery_heartbeat")
165 loaded = cache.get("celery_loaded")
166 now = time.time()
167 if ( 167 ↛ 172line 167 didn't jump to line 172 because the condition on line 167 was never true
168 loaded
169 and now - loaded > HEARTBEAT_FREQUENCY
170 and (not heartbeat or now - heartbeat > HEARTBEAT_FREQUENCY * 10)
171 ):
172 errors.append(
173 weblate_check(
174 "weblate.C030",
175 "The Celery beat scheduler is not executing periodic tasks "
176 "in a timely manner.",
177 )
178 )
180 return errors
183@register(deploy=True)
184def check_database(
185 *,
186 app_configs: Sequence[AppConfig] | None,
187 databases: Sequence[str] | None,
188 **kwargs,
189) -> Iterable[CheckMessage]:
190 errors = []
191 if not using_postgresql(): 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 errors.append(
193 weblate_check(
194 "weblate.E006",
195 "Weblate performs best with PostgreSQL, consider migrating to it.",
196 Info,
197 )
198 )
200 try:
201 delta = measure_database_latency()
202 if delta > 120: 202 ↛ 203line 202 didn't jump to line 203 because the condition on line 202 was never true
203 errors.append(
204 weblate_check(
205 "weblate.C038",
206 f"The database seems slow, the query took {delta} milliseconds",
207 )
208 )
210 except DatabaseError as error:
211 errors.append(
212 weblate_check(
213 "weblate.C037",
214 f"Could not connect to the database: {error}",
215 )
216 )
218 return errors
221@register(deploy=True)
222def check_cache(
223 *,
224 app_configs: Sequence[AppConfig] | None,
225 databases: Sequence[str] | None,
226 **kwargs,
227) -> Iterable[CheckMessage]:
228 """Check for sane caching."""
229 errors = []
231 cache_backend = cast("str", settings.CACHES["default"]["BACKEND"]).split(".")[-1]
232 if cache_backend not in GOOD_CACHE: 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true
233 errors.append(
234 weblate_check(
235 "weblate.E007",
236 "The configured cache back-end will lead to serious "
237 "performance or consistency issues.",
238 )
239 )
241 if settings.ENABLE_AVATARS and "avatar" not in settings.CACHES: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 errors.append(
243 weblate_check(
244 "weblate.E008",
245 "Please set up separate avatar caching to reduce pressure "
246 "on the default cache.",
247 Error,
248 )
249 )
251 return errors
254@register(deploy=True)
255def check_settings(
256 *,
257 app_configs: Sequence[AppConfig] | None,
258 databases: Sequence[str] | None,
259 **kwargs,
260) -> Iterable[CheckMessage]:
261 """Check for sane settings."""
262 errors = []
264 if not settings.ADMINS or any(x[1] in DEFAULT_MAILS for x in settings.ADMINS): 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 errors.append(
266 weblate_check(
267 "weblate.E011",
268 "E-mail addresses for site admins is misconfigured",
269 Error,
270 )
271 )
273 if settings.EMAIL_BACKEND != "django.core.mail.backends.dummy.EmailBackend": 273 ↛ 289line 273 didn't jump to line 289 because the condition on line 273 was always true
274 if settings.SERVER_EMAIL in DEFAULT_MAILS: 274 ↛ 281line 274 didn't jump to line 281 because the condition on line 274 was always true
275 errors.append(
276 weblate_check(
277 "weblate.E012",
278 "The server e-mail address should be changed from its default value",
279 )
280 )
281 if settings.DEFAULT_FROM_EMAIL in DEFAULT_MAILS: 281 ↛ 289line 281 didn't jump to line 289 because the condition on line 281 was always true
282 errors.append(
283 weblate_check(
284 "weblate.E013",
285 'The "From" e-mail address should be changed from its default value',
286 )
287 )
289 if settings.SECRET_KEY in DEFAULT_SECRET_KEYS: 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true
290 errors.append(
291 weblate_check(
292 "weblate.E014",
293 "The cookie secret key should be changed from its default value",
294 )
295 )
297 if not settings.ALLOWED_HOSTS: 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true
298 errors.append(weblate_check("weblate.E015", "No allowed hosts are set up"))
299 return errors
302@register(deploy=True)
303def check_class_loader(
304 *,
305 app_configs: Sequence[AppConfig] | None,
306 databases: Sequence[str] | None,
307 **kwargs,
308) -> Iterable[CheckMessage]:
309 errors: list[CheckMessage] = []
310 for instance in ClassLoader.instances.values():
311 try:
312 instance.load_data()
313 except ImproperlyConfigured as error:
314 errors.append(weblate_check("weblate.E028", str(error)))
315 return errors
318@register
319def check_data_writable(
320 *,
321 app_configs: Sequence[AppConfig] | None,
322 databases: Sequence[str] | None,
323 **kwargs,
324) -> Iterable[CheckMessage]:
325 """Check we can write to data dir."""
326 errors = []
327 if not settings.DATA_DIR: 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true
328 return [
329 weblate_check(
330 "weblate.E002",
331 "DATA_DIR is not configured.",
332 )
333 ]
334 dirs: list[Path] = [
335 Path(settings.DATA_DIR),
336 data_path("home"),
337 data_path("ssh"),
338 data_path("vcs"),
339 data_path("backups"),
340 data_path("fonts"),
341 data_path("cache") / "fonts",
342 ]
343 message = "Path {} is not writable, check your DATA_DIR settings."
344 for path in dirs:
345 path.mkdir(parents=True, exist_ok=True)
346 if not os.access(path, os.W_OK): 346 ↛ 347line 346 didn't jump to line 347 because the condition on line 346 was never true
347 errors.append(weblate_check("weblate.E002", message.format(path)))
349 return errors
352@register
353def check_site(
354 *,
355 app_configs: Sequence[AppConfig] | None,
356 databases: Sequence[str] | None,
357 **kwargs,
358) -> Iterable[CheckMessage]:
359 errors = []
360 if not check_domain(get_site_domain()): 360 ↛ 361line 360 didn't jump to line 361 because the condition on line 360 was never true
361 errors.append(weblate_check("weblate.E017", "Correct the site domain"))
362 return errors
365@register(deploy=True)
366def check_perms(
367 *,
368 app_configs: Sequence[AppConfig] | None,
369 databases: Sequence[str] | None,
370 **kwargs,
371) -> Iterable[CheckMessage]:
372 """Check that the data dir can be written to."""
373 if not settings.DATA_DIR: 373 ↛ 374line 373 didn't jump to line 374 because the condition on line 373 was never true
374 return []
375 start = time.monotonic()
376 errors = []
377 uid = os.getuid()
378 message = "The path {} is owned by a different user, check your DATA_DIR settings."
379 for dirpath, dirnames, filenames in os.walk(settings.DATA_DIR):
380 for name in chain(dirnames, filenames):
381 # Skip toplevel lost+found dir, that one is typically owned by root
382 # on filesystem toplevel directory. Also skip settings-override.py
383 # used in the Docker container as that one is typically bind mouted
384 # with different permissions (and Weblate is not expected to write
385 # to it).
386 if dirpath == settings.DATA_DIR and name in { 386 ↛ 390line 386 didn't jump to line 390 because the condition on line 386 was never true
387 "lost+found",
388 "settings-override.py",
389 }:
390 continue
391 path = os.path.join(dirpath, name)
392 try:
393 stat = os.lstat(path)
394 except OSError as error:
395 # File was removed meanwhile
396 if error.errno == errno.ENOENT:
397 continue
398 raise
399 if stat.st_uid != uid: 399 ↛ 400line 399 didn't jump to line 400 because the condition on line 399 was never true
400 errors.append(weblate_check("weblate.E027", message.format(path)))
401 if time.monotonic() - start > 15: 401 ↛ 402line 401 didn't jump to line 402 because the condition on line 401 was never true
402 break
404 return errors
407@register(deploy=True)
408def check_errors(
409 *,
410 app_configs: Sequence[AppConfig] | None,
411 databases: Sequence[str] | None,
412 **kwargs,
413) -> Iterable[CheckMessage]:
414 """Check that error collection is configured."""
415 if hasattr(settings, "ROLLBAR") or settings.SENTRY_DSN: 415 ↛ 416line 415 didn't jump to line 416 because the condition on line 415 was never true
416 return []
417 return [
418 weblate_check(
419 "weblate.I021",
420 "Error collection is not set up, "
421 "it is highly recommended for production use",
422 Info,
423 )
424 ]
427@register
428def check_encoding(
429 *,
430 app_configs: Sequence[AppConfig] | None,
431 databases: Sequence[str] | None,
432 **kwargs,
433) -> Iterable[CheckMessage]:
434 """Check that the encoding is UTF-8."""
435 if sys.getfilesystemencoding() == "utf-8" and sys.getdefaultencoding() == "utf-8": 435 ↛ 437line 435 didn't jump to line 437 because the condition on line 435 was always true
436 return []
437 return [
438 weblate_check(
439 "weblate.C023",
440 "System encoding is not UTF-8, processing non-ASCII strings will break",
441 )
442 ]
445@register(deploy=True)
446def check_diskspace(
447 *,
448 app_configs: Sequence[AppConfig] | None,
449 databases: Sequence[str] | None,
450 **kwargs,
451) -> Iterable[CheckMessage]:
452 """Check free disk space."""
453 if settings.DATA_DIR: 453 ↛ 457line 453 didn't jump to line 457 because the condition on line 453 was always true
454 stat = os.statvfs(settings.DATA_DIR)
455 if stat.f_bavail * stat.f_bsize < 10000000: 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true
456 return [weblate_check("weblate.C032", "The disk is nearly full")]
457 return []
460@register(deploy=True)
461def check_version(
462 *,
463 app_configs: Sequence[AppConfig] | None,
464 databases: Sequence[str] | None,
465 **kwargs,
466) -> Iterable[CheckMessage]:
467 try:
468 latest = get_latest_version()
469 except (ValueError, OSError):
470 return []
471 if Version(latest.version) > Version(VERSION_BASE): 471 ↛ 487line 471 didn't jump to line 487 because the condition on line 471 was always true
472 # With release every two months, this gets triggered after three releases
473 if latest.timestamp + timedelta(days=180) < timezone.now(): 473 ↛ 474line 473 didn't jump to line 474 because the condition on line 473 was never true
474 return [
475 weblate_check(
476 "weblate.C031",
477 f"You Weblate version is outdated, please upgrade to {latest.version}.",
478 )
479 ]
480 return [
481 weblate_check(
482 "weblate.I031",
483 f"New Weblate version is available, please upgrade to {latest.version}.",
484 Info,
485 )
486 ]
487 return []
490class UtilsConfig(AppConfig):
491 name = "weblate.utils"
492 label = "utils"
493 verbose_name = "Utils"
495 def ready(self) -> None:
496 super().ready()
497 init_error_collection()
499 lookups: list[tuple[type[Lookup]] | tuple[type[Lookup], str]]
500 if using_postgresql(): 500 ↛ 507line 500 didn't jump to line 507 because the condition on line 500 was always true
501 lookups = [
502 (PostgreSQLSearchLookup,),
503 (PostgreSQLSubstringLookup,),
504 (PostgreSQLRegexLookup, "trgm_regex"),
505 ]
506 else:
507 lookups = [
508 (MySQLSearchLookup,),
509 (MySQLSearchLookup, "substring"),
510 (Regex, "trgm_regex"),
511 ]
513 lookups.append((cast("type[Lookup]", MD5),))
514 lookups.append((cast("type[Lookup]", Lower),))
516 for lookup in lookups:
517 CharField.register_lookup(*lookup)
518 TextField.register_lookup(*lookup)