Coverage for website/payments/views.py: 100.00%

1from decimal import Decimal 

2 

3from django.apps import apps 

4from django.conf import settings 

5from django.contrib import messages 

6from django.contrib.auth.decorators import login_required 

7from django.contrib.messages.views import SuccessMessageMixin 

8from django.core.exceptions import ( 

9 DisallowedRedirect, 

10 PermissionDenied, 

11 SuspiciousOperation, 

12) 

13from django.db.models import QuerySet, Sum 

14from django.http import Http404, HttpResponse, HttpResponseRedirect 

15from django.shortcuts import redirect 

16from django.urls import reverse_lazy 

17from django.utils import timezone 

18from django.utils.decorators import method_decorator 

19from django.utils.http import url_has_allowed_host_and_scheme 

20from django.utils.translation import gettext_lazy as _ 

21from django.views.generic import ListView 

22from django.views.generic.edit import CreateView, FormView, UpdateView 

23 

24from dateutil.relativedelta import relativedelta 

25 

26from payments import services 

27from payments.exceptions import PaymentError 

28from payments.forms import BankAccountForm, BankAccountUserRevokeForm, PaymentCreateForm 

29from payments.models import BankAccount, Payment, PaymentUser 

30from payments.payables import payables 

31 

32 

33@method_decorator(login_required, name="dispatch") 

34class BankAccountCreateView(SuccessMessageMixin, CreateView): 

35 model = BankAccount 

36 form_class = BankAccountForm 

37 success_url = reverse_lazy("payments:bankaccount-list") 

38 success_message = _("Bank account saved successfully.") 

39 

40 def get_context_data(self, **kwargs) -> dict: 

41 context = super().get_context_data(**kwargs) 

42 context["mandate_no"] = services.derive_next_mandate_no(self.request.member) 

43 context["creditor_id"] = settings.SEPA_CREDITOR_ID 

44 return context 

45 

46 def post(self, request, *args, **kwargs) -> HttpResponse: 

47 request.POST = request.POST.dict() 

48 request.POST["owner"] = self.request.member.pk 

49 if "direct_debit" in request.POST: 

50 request.POST["valid_from"] = timezone.now() 

51 request.POST["mandate_no"] = services.derive_next_mandate_no( 

52 self.request.member 

53 ) 

54 else: 

55 request.POST["valid_from"] = None 

56 request.POST["mandate_no"] = None 

57 request.POST["signature"] = None 

58 return super().post(request, *args, **kwargs) 

59 

60 def form_valid(self, form: BankAccountForm) -> HttpResponse: 

61 BankAccount.objects.filter( 

62 owner=PaymentUser.objects.get(pk=self.request.member.pk), mandate_no=None 

63 ).delete() 

64 BankAccount.objects.filter( 

65 owner=PaymentUser.objects.get(pk=self.request.member.pk) 

66 ).exclude(mandate_no=None).update(valid_until=timezone.now()) 

67 return super().form_valid(form) 

68 

69 

70@method_decorator(login_required, name="dispatch") 

71class BankAccountRevokeView(SuccessMessageMixin, UpdateView): 

72 model = BankAccount 

73 form_class = BankAccountUserRevokeForm 

74 success_url = reverse_lazy("payments:bankaccount-list") 

75 success_message = _("Direct debit authorisation successfully revoked.") 

76 

77 def get_queryset(self) -> QuerySet: 

78 return ( 

79 super() 

80 .get_queryset() 

81 .filter( 

82 owner=PaymentUser.objects.get(pk=self.request.member.pk), 

83 valid_until=None, 

84 ) 

85 .exclude(mandate_no=None) 

86 ) 

87 

88 def form_invalid(self, form): 

89 messages.error( 

90 self.request, 

91 _( 

92 "The mandate for this bank account cannot be revoked right now, as it is used for payments that have not yet been processed. Contact treasurer@thalia.nu to revoke your mandate." 

93 ), 

94 ) 

95 super().form_invalid(form) 

96 return HttpResponseRedirect(self.get_success_url()) 

97 

98 def get(self, *args, **kwargs) -> HttpResponse: 

99 return redirect("payments:bankaccount-list") 

100 

101 def post(self, request, *args, **kwargs) -> HttpResponse: 

102 request.POST = request.POST.dict() 

103 request.POST["valid_until"] = timezone.now() 

104 return super().post(request, *args, **kwargs) 

105 

106 

107@method_decorator(login_required, name="dispatch") 

108class BankAccountListView(ListView): 

109 model = BankAccount 

110 

111 def get_context_data(self, *args, **kwargs): 

112 context = super().get_context_data(*args, **kwargs) 

113 context.update( 

114 { 

115 "payment_user": PaymentUser.objects.get(pk=self.request.member.pk), 

116 } 

117 ) 

118 return context 

119 

120 def get_queryset(self) -> QuerySet: 

121 return ( 

122 super() 

123 .get_queryset() 

124 .filter(owner=PaymentUser.objects.get(pk=self.request.member.pk)) 

125 ) 

126 

127 

128@method_decorator(login_required, name="dispatch") 

129class PaymentListView(ListView): 

130 model = Payment 

131 

132 def get_queryset(self) -> QuerySet: 

133 year = self.kwargs.get("year", timezone.now().year) 

134 month = self.kwargs.get("month", timezone.now().month) 

135 

136 return ( 

137 super() 

138 .get_queryset() 

139 .filter( 

140 paid_by=PaymentUser.objects.get(pk=self.request.member.pk), 

141 created_at__year=year, 

142 created_at__month=month, 

143 ) 

144 ) 

145 

146 def get_context_data(self, *args, **kwargs): 

147 filters = [] 

148 for i in range(13): 

149 new_now = timezone.now() - relativedelta(months=i) 

150 filters.append({"year": new_now.year, "month": new_now.month}) 

151 

152 context = super().get_context_data(*args, **kwargs) 

153 context.update( 

154 { 

155 "filters": filters, 

156 "total": context["object_list"] 

157 .aggregate(Sum("amount")) 

158 .get("amount__sum"), 

159 "tpay_balance": PaymentUser.objects.get( 

160 pk=self.request.member.pk 

161 ).tpay_balance, 

162 "year": self.kwargs.get("year", timezone.now().year), 

163 "month": self.kwargs.get("month", timezone.now().month), 

164 } 

165 ) 

166 return context 

167 

168 

169@method_decorator(login_required, name="dispatch") 

170class PaymentProcessView(SuccessMessageMixin, FormView): 

171 """Defines a view that allows the user to add a Thalia Pay payment to a Payable object using a POST request. 

172 

173 The user should be authenticated. 

174 """ 

175 

176 form_class = PaymentCreateForm 

177 success_message = _("Your payment has been processed successfully.") 

178 template_name = "payments/payment_form.html" 

179 

180 payable = None 

181 

182 def get_success_url(self): 

183 return self.request.POST["next"] 

184 

185 def dispatch(self, request, *args, **kwargs): 

186 if not PaymentUser.objects.get(pk=request.member.pk).tpay_enabled: 

187 raise PermissionDenied 

188 return super().dispatch(request, *args, **kwargs) 

189 

190 def get_context_data(self, **kwargs): 

191 if self.payable is None: 

192 raise Http404("Payable does not exist") 

193 context = super().get_context_data(**kwargs) 

194 context.update({"payable": self.payable}) 

195 context.update({"payable_hash": hash(self.payable)}) 

196 context.update( 

197 { 

198 "new_balance": PaymentUser.objects.get( 

199 pk=self.payable.payment_payer.pk 

200 ).tpay_balance 

201 - Decimal(self.payable.payment_amount) 

202 } 

203 ) 

204 return context 

205 

206 def _check_payment_allowed(self, request, payable_hash): 

207 if not self.payable.payment_payer or ( 

208 self.payable.payment_payer.pk 

209 != PaymentUser.objects.get(pk=self.request.member.pk).pk 

210 ): 

211 return _("You are not allowed to process this payment.") 

212 

213 if self.payable.payment_amount == 0: 

214 return _("No payment required for amount of €0.00") 

215 

216 if self.payable.payment: 

217 return _("This object has already been paid for.") 

218 

219 if not self.payable.tpay_allowed: 

220 return _("You are not allowed to use Thalia Pay for this payment.") 

221 

222 if str(hash(self.payable)) != str(payable_hash): 

223 return _( 

224 "This object has been changed in the mean time. You have not paid." 

225 ) 

226 

227 return None 

228 

229 def post(self, request, *args, **kwargs): 

230 if not ( 

231 request.POST.keys() 

232 >= {"app_label", "model_name", "payable", "payable_hash", "next"} 

233 ): 

234 raise SuspiciousOperation("Missing POST parameters") 

235 

236 if not url_has_allowed_host_and_scheme( 

237 request.POST["next"], allowed_hosts={request.get_host()} 

238 ): 

239 raise DisallowedRedirect 

240 

241 app_label = request.POST["app_label"] 

242 model_name = request.POST["model_name"] 

243 payable_pk = request.POST["payable"] 

244 payable_hash = request.POST["payable_hash"] 

245 

246 try: 

247 payable_model = apps.get_model(app_label=app_label, model_name=model_name) 

248 except LookupError as error: 

249 raise Http404("This app model does not exist.") from error 

250 

251 try: 

252 payable_obj = payable_model.objects.get(pk=payable_pk) 

253 except payable_model.DoesNotExist as error: 

254 raise Http404("This payable does not exist.") from error 

255 

256 self.payable = payables.get_payable(payable_obj) 

257 

258 error = self._check_payment_allowed(request, payable_hash) 

259 if error: 

260 messages.error(self.request, error) 

261 return redirect(request.POST["next"]) 

262 

263 if "_save" not in request.POST: 

264 context = self.get_context_data(**kwargs) 

265 return self.render_to_response(context) 

266 

267 return super().post(request, *args, **kwargs) 

268 

269 def form_valid(self, form): 

270 try: 

271 services.create_payment( 

272 self.payable, 

273 PaymentUser.objects.get(pk=self.request.member.pk), 

274 Payment.TPAY, 

275 ) 

276 except PaymentError as e: 

277 messages.error(self.request, str(e)) 

278 return super().form_valid(form)