Coverage for app/venv/lib/python3.14/site-packages/weblate/trans/views/git.py: 33%

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

2# 

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

4from __future__ import annotations 

5 

6from typing import TYPE_CHECKING 

7 

8from django.contrib.auth.decorators import login_required 

9from django.core.exceptions import PermissionDenied 

10from django.shortcuts import redirect 

11from django.urls import reverse 

12from django.utils.translation import gettext 

13from django.views.decorators.http import require_POST 

14 

15from weblate.trans.models import Component, Project, Translation 

16from weblate.trans.util import redirect_param 

17from weblate.utils import messages 

18from weblate.utils.errors import report_error 

19from weblate.utils.lock import WeblateLockTimeoutError 

20from weblate.utils.views import parse_path 

21 

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

23 from weblate.auth.models import AuthenticatedHttpRequest 

24 

25 

26def execute_locked( 

27 request: AuthenticatedHttpRequest, obj, message, call, *args, **kwargs 

28): 

29 """Wrap function call and gracefully handle possible lock exception.""" 

30 try: 

31 result = call(*args, **kwargs) 

32 # With False the call is supposed to show errors on its own 

33 if result is None or result: 

34 messages.success(request, message) 

35 except WeblateLockTimeoutError: 

36 messages.error( 

37 request, 

38 gettext("Could not lock the repository, another operation is in progress."), 

39 ) 

40 if isinstance(obj, Project): 

41 report_error("Repository lock timeout", project=obj) 

42 elif isinstance(obj, Component): 

43 report_error("Repository lock timeout", project=obj.project) 

44 else: 

45 report_error("Repository lock timeout", project=obj.component.project) 

46 

47 return redirect_param(obj, "#repository") 

48 

49 

50@login_required 

51@require_POST 

52def update(request: AuthenticatedHttpRequest, path): 

53 obj = parse_path(request, path, (Project, Component, Translation)) 

54 if not request.user.has_perm("vcs.update", obj): 

55 raise PermissionDenied 

56 

57 result = execute_locked( 

58 request, 

59 obj, 

60 gettext( 

61 "All repositories have been updated, updates of translations are in progress." 

62 ), 

63 obj.do_update, 

64 request, 

65 method=request.GET.get("method"), 

66 ) 

67 if result: 

68 return redirect( 

69 "{}?info=1".format( 

70 reverse("show_progress", kwargs={"path": obj.get_url_path()}) 

71 ) 

72 ) 

73 return result 

74 

75 

76@login_required 

77@require_POST 

78def push(request: AuthenticatedHttpRequest, path): 

79 obj = parse_path(request, path, (Project, Component, Translation)) 

80 if not request.user.has_perm("vcs.push", obj): 

81 raise PermissionDenied 

82 

83 return execute_locked( 

84 request, obj, gettext("All repositories were pushed."), obj.do_push, request 

85 ) 

86 

87 

88@login_required 

89@require_POST 

90def reset(request: AuthenticatedHttpRequest, path): 

91 obj = parse_path(request, path, (Project, Component, Translation)) 

92 if not request.user.has_perm("vcs.reset", obj): 

93 raise PermissionDenied 

94 

95 result = execute_locked( 

96 request, 

97 obj, 

98 gettext( 

99 "All repositories have been reset, updates of translations are in progress." 

100 ), 

101 obj.do_reset, 

102 request, 

103 keep_changes="keep_changes" in request.POST, 

104 ) 

105 if result: 

106 return redirect( 

107 "{}?info=1".format( 

108 reverse("show_progress", kwargs={"path": obj.get_url_path()}) 

109 ) 

110 ) 

111 return result 

112 

113 

114@login_required 

115@require_POST 

116def cleanup(request: AuthenticatedHttpRequest, path): 

117 obj = parse_path(request, path, (Project, Component, Translation)) 

118 if not request.user.has_perm("vcs.reset", obj): 

119 raise PermissionDenied 

120 

121 return execute_locked( 

122 request, 

123 obj, 

124 gettext("All repositories have been cleaned up."), 

125 obj.do_cleanup, 

126 request, 

127 ) 

128 

129 

130@login_required 

131@require_POST 

132def file_sync(request: AuthenticatedHttpRequest, path): 

133 obj = parse_path(request, path, (Project, Component, Translation)) 

134 if not request.user.has_perm("vcs.reset", obj): 

135 raise PermissionDenied 

136 

137 return execute_locked( 

138 request, 

139 obj, 

140 gettext("Translation files have been synchronized."), 

141 obj.do_file_sync, 

142 request, 

143 ) 

144 

145 

146@login_required 

147@require_POST 

148def file_scan(request: AuthenticatedHttpRequest, path): 

149 obj = parse_path(request, path, (Project, Component, Translation)) 

150 if not request.user.has_perm("vcs.reset", obj): 

151 raise PermissionDenied 

152 

153 result = execute_locked( 

154 request, 

155 obj, 

156 gettext("Updates of translations are in progress."), 

157 obj.do_file_scan, 

158 request, 

159 ) 

160 if result: 

161 return redirect( 

162 "{}?info=1".format( 

163 reverse("show_progress", kwargs={"path": obj.get_url_path()}) 

164 ) 

165 ) 

166 return result 

167 

168 

169@login_required 

170@require_POST 

171def commit(request: AuthenticatedHttpRequest, path): 

172 obj = parse_path(request, path, (Project, Component, Translation)) 

173 if not request.user.has_perm("vcs.commit", obj): 

174 raise PermissionDenied 

175 

176 return execute_locked( 

177 request, 

178 obj, 

179 gettext("All pending translations were committed."), 

180 obj.commit_pending, 

181 "commit", 

182 request.user, 

183 )