Coverage for app/venv/lib/python3.14/site-packages/weblate/wladmin/middleware.py: 83%

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

2# 

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

4 

5from __future__ import annotations 

6 

7import multiprocessing 

8import time 

9from random import randint 

10from threading import Thread 

11from typing import TYPE_CHECKING 

12 

13from django.conf import settings 

14from django.core.cache import cache 

15from django.shortcuts import redirect 

16from django.urls import reverse 

17 

18from weblate.accounts.models import AuditLog 

19from weblate.runner import main 

20 

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

22 from weblate.auth.models import AuthenticatedHttpRequest 

23 

24CHECK_CACHE_KEY = "weblate-health-check" 

25 

26 

27class ManageMiddleware: 

28 """ 

29 Middleware to trigger periodic management tasks. 

30 

31 These have to be triggered from the UWSGI context to be able to detect differences 

32 between Celery and UWSGI environments. 

33 """ 

34 

35 def __init__(self, get_response=None) -> None: 

36 self.get_response = get_response 

37 

38 @staticmethod 

39 def trigger_check() -> None: 

40 if not settings.BACKGROUND_ADMIN_CHECKS: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true

41 return 

42 # Update last execution timestamp 

43 cache.set(CHECK_CACHE_KEY, time.time()) 

44 

45 # Use safer spawn method 

46 context = multiprocessing.get_context("spawn") 

47 

48 # Spawn a management process to do a configuration health check 

49 # - using threads causes problems with a database connections 

50 # (SSL initialization and keeping open persistent connections) 

51 # - spawning directly the method using multiprocessing is not easy 

52 # because of missing invocation of django.setup() 

53 process = context.Process( 

54 target=main, kwargs={"argv": ["weblate", "configuration_health_check"]} 

55 ) 

56 process.start() 

57 

58 # Make sure we catch SIGCHLD from the spawned process 

59 thread = Thread(target=process.join) 

60 thread.start() 

61 

62 def __call__(self, request: AuthenticatedHttpRequest): 

63 if request.session.pop("redirect_to_donate", False): 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true

64 AuditLog.objects.create(request.user, request, "donate") 

65 return redirect(reverse("donate")) 

66 response = self.get_response(request) 

67 if ( 67 ↛ 72line 67 didn't jump to line 72 because the condition on line 67 was never true

68 request.resolver_match 

69 and request.resolver_match.view_name == "manage-performance" 

70 ): 

71 # Always trigger on the performance page 

72 self.trigger_check() 

73 elif randint(0, 100) == 1: # noqa: S311 

74 # Trigger when last check is too old 

75 last_run = cache.get(CHECK_CACHE_KEY) 

76 now = time.time() 

77 if last_run is None or now - last_run > 900: 

78 self.trigger_check() 

79 

80 return response