Coverage for src/meshadmin/server/networks/views/group.py: 64%

1from django.conf import settings 

2from django.contrib.auth.mixins import LoginRequiredMixin 

3from django.core.paginator import EmptyPage, PageNotAnInteger, Paginator 

4from django.db import models 

5from django.http import HttpResponse 

6from django.shortcuts import get_object_or_404, render 

7from django.urls import reverse_lazy 

8from django.views.generic import CreateView, DeleteView, DetailView, UpdateView, View 

9 

10from meshadmin.server.networks.forms import GroupConfigForm, GroupForm, RuleForm 

11from meshadmin.server.networks.mixins import NetworkPermissionMixin 

12from meshadmin.server.networks.models import Group, GroupConfig, Network, Rule 

13 

14 

15class GroupDetailView(LoginRequiredMixin, NetworkPermissionMixin, DetailView): 

16 model = Group 

17 template_name = "networks/group/detail.html" 

18 

19 def get_network(self): 

20 return self.get_object().network 

21 

22 def get_context_data(self, **kwargs): 

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

24 hosts = self.object.host_set.all() 

25 search = self.request.GET.get("search", "") 

26 if search: 

27 hosts = hosts.filter(name__icontains=search) 

28 

29 if self.request.GET.get("stale_config"): 

30 from datetime import timedelta 

31 

32 from django.utils import timezone 

33 

34 stale_threshold = timezone.now() - timedelta(hours=24) 

35 hosts = hosts.filter( 

36 models.Q(last_config_refresh__lt=stale_threshold) 

37 | models.Q(last_config_refresh__isnull=True) 

38 ) 

39 

40 per_page = int(self.request.GET.get("per_page", settings.PAGINATION_PER_PAGE)) 

41 paginator = Paginator(hosts, per_page) 

42 page = self.request.GET.get("hosts_page", 1) 

43 

44 try: 

45 hosts = paginator.page(page) 

46 except (PageNotAnInteger, EmptyPage): 

47 hosts = paginator.page(1) 

48 

49 context.update( 

50 { 

51 "hosts": hosts, 

52 "paginator": paginator, 

53 "page_obj": hosts, 

54 "per_page": per_page, 

55 "per_page_options": [25, 50, 100], 

56 "network": self.object.network, 

57 "fields": [ 

58 { 

59 "name": field.name, 

60 "verbose_name": field.verbose_name, 

61 "value": getattr(self.object, field.name), 

62 } 

63 for field in self.model._meta.fields 

64 ], 

65 "rules": self.object.rules.all(), 

66 "configs": self.object.config_overrides.all(), 

67 } 

68 ) 

69 return context 

70 

71 def get_template_names(self): 

72 if self.request.htmx: 

73 return ["networks/network/_hosts_table.html"] 

74 return [self.template_name] 

75 

76 

77class GroupCreateView(LoginRequiredMixin, NetworkPermissionMixin, CreateView): 

78 model = Group 

79 form_class = GroupForm 

80 template_name = "networks/group/form.html" 

81 

82 def get_network(self): 

83 return get_object_or_404(Network, id=self.kwargs.get("network_id")) 

84 

85 def form_valid(self, form): 

86 response = super().form_valid(form) 

87 if self.request.htmx: 

88 context = self.get_context_data() 

89 context["object"] = self.object 

90 context["action"] = "Update" 

91 return render(self.request, "networks/group/_form_content.html", context) 

92 return response 

93 

94 def get_success_url(self): 

95 return reverse_lazy("networks:group-detail", kwargs={"pk": self.object.pk}) 

96 

97 def get_form_kwargs(self): 

98 kwargs = super().get_form_kwargs() 

99 network_id = self.kwargs.get("network_id") 

100 if network_id: 

101 network = get_object_or_404(Network, id=network_id) 

102 kwargs["network"] = network 

103 return kwargs 

104 

105 def get_context_data(self, **kwargs): 

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

107 context.update( 

108 { 

109 "action": "Create", 

110 "model_name": self.model._meta.verbose_name, 

111 "network_id": self.kwargs.get("network_id"), 

112 } 

113 ) 

114 return context 

115 

116 

117class GroupUpdateView(LoginRequiredMixin, NetworkPermissionMixin, UpdateView): 

118 model = Group 

119 form_class = GroupForm 

120 template_name = "networks/group/form.html" 

121 

122 def get_network(self): 

123 return self.get_object().network 

124 

125 def get_context_data(self, **kwargs): 

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

127 context.update( 

128 { 

129 "action": "Update", 

130 "network_id": self.get_object().network.id, 

131 "rules": self.get_object().rules.all(), 

132 } 

133 ) 

134 return context 

135 

136 def form_valid(self, form): 

137 response = super().form_valid(form) 

138 if self.request.htmx: 

139 context = self.get_context_data() 

140 context["object"] = self.object 

141 context["action"] = "Update" 

142 return render(self.request, "networks/group/_form_content.html", context) 

143 return response 

144 

145 def get_success_url(self): 

146 return reverse_lazy("networks:group-detail", kwargs={"pk": self.object.pk}) 

147 

148 

149class GroupDeleteView(LoginRequiredMixin, NetworkPermissionMixin, DeleteView): 

150 model = Group 

151 template_name = "networks/shared/delete.html" 

152 

153 def get_network(self): 

154 return self.get_object().network 

155 

156 def get_success_url(self): 

157 return ( 

158 reverse_lazy( 

159 "networks:network-detail", kwargs={"pk": self.object.network.pk} 

160 ) 

161 + "#groups-section" 

162 ) 

163 

164 def get_context_data(self, **kwargs): 

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

166 context["model_name"] = self.model._meta.verbose_name 

167 return context 

168 

169 

170class RuleFormModalView(LoginRequiredMixin, NetworkPermissionMixin, View): 

171 def get_network(self): 

172 group = get_object_or_404(Group, id=self.request.GET.get("group")) 

173 return group.network 

174 

175 def get(self, request): 

176 group_id = request.GET.get("group") 

177 rule_id = request.GET.get("rule") 

178 security_group = get_object_or_404(Group, id=group_id) 

179 

180 if rule_id: 

181 rule = get_object_or_404(Rule, id=rule_id) 

182 form = RuleForm(instance=rule) 

183 else: 

184 form = RuleForm( 

185 initial={"security_group": security_group}, 

186 instance=Rule(security_group=security_group), 

187 ) 

188 

189 return render( 

190 request, 

191 "networks/rule/_form_modal.html", 

192 { 

193 "form": form, 

194 "security_group": security_group, 

195 "rule": rule if rule_id else None, 

196 }, 

197 ) 

198 

199 

200class GroupAddRuleView(LoginRequiredMixin, NetworkPermissionMixin, View): 

201 def get_network(self): 

202 group_id = self.request.POST.get("security_group") 

203 group = get_object_or_404(Group, id=group_id) 

204 return group.network 

205 

206 def post(self, request): 

207 group_id = request.POST.get("security_group") 

208 security_group = get_object_or_404(Group, id=group_id) 

209 

210 form = RuleForm(request.POST, initial={"security_group_id": group_id}) 

211 if form.is_valid(): 

212 rule = form.save(commit=False) 

213 rule.security_group = security_group 

214 rule.save() 

215 

216 if request.POST.getlist("groups"): 

217 rule.groups.set( 

218 Group.objects.filter(id__in=request.POST.getlist("groups")) 

219 ) 

220 

221 context = {"rules": Rule.objects.filter(security_group=security_group)} 

222 return render( 

223 request, 

224 "networks/group/_rules_list.html", 

225 context, 

226 ) 

227 else: 

228 response = render( 

229 request, 

230 "networks/rule/_form_modal.html", 

231 { 

232 "form": form, 

233 "security_group": security_group, 

234 }, 

235 ) 

236 response["HX-Retarget"] = "#modal-content" 

237 return response 

238 

239 

240class GroupConfigModalView(LoginRequiredMixin, NetworkPermissionMixin, View): 

241 def get_network(self): 

242 group = get_object_or_404(Group, id=self.request.GET.get("group")) 

243 return group.network 

244 

245 def get(self, request): 

246 group_id = request.GET.get("group") 

247 config_id = request.GET.get("config") 

248 group = get_object_or_404(Group, id=group_id) 

249 

250 if config_id: 

251 config = get_object_or_404(GroupConfig, id=config_id) 

252 form = GroupConfigForm(instance=config) 

253 else: 

254 form = GroupConfigForm( 

255 instance=GroupConfig(group=group), 

256 ) 

257 

258 return render( 

259 request, 

260 "networks/group/_config_modal.html", 

261 { 

262 "form": form, 

263 "group": group, 

264 "config": config if config_id else None, 

265 }, 

266 ) 

267 

268 

269class GroupAddUpdateConfigView(LoginRequiredMixin, NetworkPermissionMixin, View): 

270 def get_network(self): 

271 group = get_object_or_404(Group, id=self.request.POST.get("group")) 

272 return group.network 

273 

274 def post(self, request): 

275 group_id = request.POST.get("group") 

276 config_id = request.POST.get("config") 

277 group = get_object_or_404(Group, id=group_id) 

278 

279 if config_id: 

280 config = get_object_or_404(GroupConfig, id=config_id) 

281 form = GroupConfigForm(request.POST, instance=config) 

282 else: 

283 key = request.POST.get("key") 

284 existing_config = GroupConfig.objects.filter(group=group, key=key).first() 

285 

286 if existing_config: 

287 form = GroupConfigForm(request.POST, instance=existing_config) 

288 else: 

289 form = GroupConfigForm(request.POST) 

290 

291 if form.is_valid(): 

292 config = form.save(commit=False) 

293 config.group = group 

294 config.save() 

295 return render( 

296 request, 

297 "networks/group/_configs_list.html", 

298 { 

299 "configs": group.config_overrides.all(), 

300 }, 

301 ) 

302 else: 

303 response = render( 

304 request, 

305 "networks/group/_config_modal.html", 

306 { 

307 "form": form, 

308 "group": group, 

309 "config": config if config_id else None, 

310 }, 

311 ) 

312 response["HX-Retarget"] = "#config-modal-content" 

313 return response 

314 

315 

316class GroupConfigDeleteView(LoginRequiredMixin, NetworkPermissionMixin, DeleteView): 

317 model = GroupConfig 

318 template_name = "networks/shared/delete.html" 

319 

320 def get_network(self): 

321 return self.get_object().group.network 

322 

323 def delete(self, request, *args, **kwargs): 

324 self.object = self.get_object() 

325 self.object.delete() 

326 return HttpResponse("") 

327 

328 def get_success_url(self): 

329 return ( 

330 reverse_lazy("networks:group-detail", kwargs={"pk": self.object.group.pk}) 

331 + "#groups-section" 

332 ) 

333 

334 def get_context_data(self, **kwargs): 

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

336 context["model_name"] = self.model._meta.verbose_name 

337 return context