Coverage for sm/views.py: 87%
253 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-24 19:06 +0000
1import logging
3from django.views.generic import TemplateView
4from django.contrib.auth.mixins import LoginRequiredMixin
5from django.contrib import messages
6from server.models import Model as Server
7from cluster.models import Model as Cluster
8from clusterpackage.models import Model as ClusterPackage
9from clusterpackagetype.models import Model as ClusterPackageType
10from clustersoftware.models import Model as ClusterSoftware
11from domain.models import Model as Domain
12from location.models import Model as Location
13from operatingsystem.models import Model as OS
14from patchtime.models import Model as Patchtime
15from servermodel.models import Model as ServerModel
16from status.models import Model as Status
17from vendor.models import Model as Vendor
18from django.db.models import Count, Q
19from django.core.exceptions import ObjectDoesNotExist
20from django.apps import apps
21from django.http import Http404, JsonResponse
22from django.conf import settings
23from django.urls import reverse
24from django.core.exceptions import PermissionDenied
26from django.db.models import ProtectedError
27from django.utils.translation import gettext as _
28from django.shortcuts import redirect
29from django.views import View
30from typing import Any, List
31from .mixins import filter_queryset_by_tenant, filter_history_queryset_by_tenant
32from .utils_starterpack import import_starter_pack
34logger = logging.getLogger(__name__)
37class SafeDeleteMixin:
38 """
39 Mixin to catch ProtectedError during deletion and show a friendly
40 message instead of raising a 500 error.
41 """
43 def form_valid(self, form: Any) -> Any:
44 try:
45 obj_name = str(self.object) # type: ignore
46 # Chain to the rest of the MRO (e.g. MultiTenantMixin) so tenant
47 # guards such as read-only global fixtures are enforced.
48 response = super().form_valid(form) # type: ignore
49 if hasattr(self, "success_message") and self.success_message: # type: ignore # noqa: E501
50 messages.success(
51 # type: ignore
52 self.request,
53 self.success_message % self.object.__dict__,
54 )
55 else:
56 messages.success(self.request, _("Successfully deleted %s") % obj_name)
57 return response
58 except ProtectedError as e:
59 instances = ", ".join(str(obj) for obj in e.protected_objects)
60 messages.error(
61 self.request,
62 _(
63 "%(name)s cannot be deleted because it is referenced by "
64 "%(count)d other resource(s): %(instances)s"
65 )
66 % {
67 "name": self.object, # type: ignore
68 "count": len(e.protected_objects),
69 "instances": instances,
70 },
71 )
72 return self.render_to_response(
73 self.get_context_data(object=self.object) # type: ignore
74 )
77class GenericBulkDeleteView(LoginRequiredMixin, View):
78 """
79 Generic bulk-delete view for models that use the standard
80 ``MultiTenantMixin`` list pattern.
82 Subclasses must define ``model`` and the request must include
83 ``selected_ids`` (a list of primary keys). Deletion runs through the
84 tenant-filtered queryset so users can only ever delete their own items.
85 """
87 model = None
89 def get_queryset(self) -> Any:
90 queryset = self.model.objects.all()
91 if self.request.user.is_superuser:
92 return queryset
93 user_groups = self.request.user.groups.all()
94 selected_groups = self.request.session.get("selected_groups", [])
95 if hasattr(self.model, "group"):
96 if selected_groups:
97 group_ids = [int(g) for g in selected_groups if g.isdigit()]
98 if group_ids:
99 return queryset.filter(
100 Q(group__id__in=group_ids) | Q(group__isnull=True)
101 )
102 return queryset.filter(Q(group__in=user_groups) | Q(group__isnull=True))
103 return queryset
105 def get_success_url(self) -> str:
106 return reverse("%s:index" % self.model._meta.app_label)
108 def post(self, request: Any, *args: Any, **kwargs: Any) -> Any:
109 ids = request.POST.getlist("selected_ids")
110 if not ids:
111 messages.warning(request, _("No items selected."))
112 return redirect(self.get_success_url())
114 model_name = self.model._meta.object_name.lower()
115 app_label = self.model._meta.app_label
116 delete_perm = "delete_%s" % model_name
117 if not request.user.has_perm("{}.{}".format(app_label, delete_perm)):
118 raise PermissionDenied
120 queryset = self.get_queryset().filter(pk__in=ids)
121 count = queryset.count()
122 if count == 0:
123 messages.warning(request, _("No items selected."))
124 return redirect(self.get_success_url())
126 try:
127 queryset.delete()
128 messages.success(
129 request,
130 _("Successfully deleted %d item(s).") % count,
131 )
132 except ProtectedError as e:
133 instances = ", ".join(str(obj) for obj in list(e.protected_objects)[:5])
134 messages.error(
135 request,
136 _("Some items cannot be deleted because they are referenced by: %s")
137 % instances,
138 )
139 return redirect(self.get_success_url())
142class GenericCSVExportView(LoginRequiredMixin, View):
143 """
144 Exports the model's rows (tenant-filtered) as CSV.
146 Subclasses define ``model`` and ``export_fields`` — an ordered list of
147 ``(header, attribute_path)`` tuples. Attribute paths may traverse related
148 objects with ``__``.
149 """
151 model = None
152 export_fields: List = []
153 filename = "export.csv"
155 def get_queryset(self) -> Any:
156 queryset = self.model.objects.all()
157 if self.request.user.is_superuser:
158 return queryset
159 user_groups = self.request.user.groups.all()
160 selected_groups = self.request.session.get("selected_groups", [])
161 if hasattr(self.model, "group"):
162 if selected_groups:
163 group_ids = [int(g) for g in selected_groups if g.isdigit()]
164 if group_ids:
165 return queryset.filter(
166 Q(group__id__in=group_ids) | Q(group__isnull=True)
167 )
168 return queryset.filter(Q(group__in=user_groups) | Q(group__isnull=True))
169 return queryset
171 def get_filename(self) -> str:
172 return self.filename
174 def resolve_value(self, obj: Any, path: str) -> Any:
175 value: Any = obj
176 for part in path.split("__"):
177 if value is None:
178 return ""
179 try:
180 value = getattr(value, part)
181 except (AttributeError, ObjectDoesNotExist):
182 return ""
183 return value
185 def get(self, request: Any, *args: Any, **kwargs: Any) -> Any:
186 import csv
188 from django.http import HttpResponse
190 response = HttpResponse(content_type="text/csv; charset=utf-8")
191 response["Content-Disposition"] = (
192 'attachment; filename="%s"' % self.get_filename()
193 )
195 writer = csv.writer(response)
196 writer.writerow([header for header, _ in self.export_fields])
197 for obj in self.get_queryset():
198 writer.writerow(
199 [self.resolve_value(obj, path) for _, path in self.export_fields]
200 )
201 return response
204class DashboardView(LoginRequiredMixin, TemplateView):
205 template_name = "dashboard.html"
207 def get_queryset_filtered(self, model: Any) -> Any:
208 if not hasattr(model, "group"):
209 return model.objects.all()
210 return filter_queryset_by_tenant(model.objects.all(), self.request)
212 def get_context_data(self, **kwargs: Any) -> Any:
213 context = super().get_context_data(**kwargs)
215 # Basic Stats (Filtered)
216 context["server_count"] = self.get_queryset_filtered(Server).count()
217 context["cluster_count"] = self.get_queryset_filtered(Cluster).count()
218 context["vendor_count"] = self.get_queryset_filtered(Vendor).count()
219 context["os_count"] = self.get_queryset_filtered(OS).count()
221 # Data for Charts (Filtered)
222 # OS Distribution
223 os_dist = (
224 self.get_queryset_filtered(Server)
225 .values("operatingsystem__vendor__name", "operatingsystem__version")
226 .annotate(count=Count("id"))
227 .order_by("-count")[:5]
228 )
230 context["os_labels"] = [
231 f"{item['operatingsystem__vendor__name']} "
232 f"{item['operatingsystem__version']}"
233 for item in os_dist
234 ]
235 context["os_data"] = [item["count"] for item in os_dist]
237 # Status Distribution
238 status_dist = (
239 self.get_queryset_filtered(Server)
240 .values("status__name")
241 .annotate(count=Count("id"))
242 .order_by("-count")
243 )
245 context["status_labels"] = [item["status__name"] for item in status_dist]
246 context["status_data"] = [item["count"] for item in status_dist]
248 # Recent Activity (Filtered)
249 context["recent_servers"] = (
250 self.get_queryset_filtered(Server).all().order_by("-id")[:5]
251 )
253 return context
256class SearchView(LoginRequiredMixin, TemplateView):
257 template_name = "search.html"
259 def get_queryset_filtered(self, model: Any) -> Any:
260 if not hasattr(model, "group"):
261 return model.objects.all()
262 return filter_queryset_by_tenant(model.objects.all(), self.request)
264 def get_template_names(self) -> List[str]:
265 if self.request.GET.get("ajax"):
266 return ["search_results_ajax.html"]
267 return [self.template_name]
269 def get_context_data(self, **kwargs: Any) -> Any:
270 context = super().get_context_data(**kwargs)
271 query = self.request.GET.get("q", "").lower()
272 context["query"] = query
274 # Navigation Quick Jumps
275 nav_targets = [
276 {"name": _("Dashboard"), "url": "/", "icon": "fa-gauge-high"},
277 {
278 "name": _("Group Management"),
279 "url": "/group/members/",
280 "icon": "fa-users",
281 },
282 {"name": _("Servers"), "url": "/server/", "icon": "fa-server"},
283 {"name": _("Server Models"), "url": "/servermodel/", "icon": "fa-cubes"},
284 {"name": _("Vendors"), "url": "/vendor/", "icon": "fa-industry"},
285 {"name": _("Clusters"), "url": "/cluster/", "icon": "fa-th-large"},
286 {
287 "name": _("Operating Systems"),
288 "url": "/operatingsystem/",
289 "icon": "fa-laptop",
290 },
291 {"name": _("Statuses"), "url": "/status/", "icon": "fa-tag"},
292 {"name": _("Locations"), "url": "/location/", "icon": "fa-map-marker-alt"},
293 {"name": _("Domains"), "url": "/domain/", "icon": "fa-globe"},
294 {"name": _("Patch Times"), "url": "/patchtime/", "icon": "fa-calendar"},
295 {
296 "name": _("Cluster Software"),
297 "url": "/clustersoftware/",
298 "icon": "fa-shield-halved",
299 },
300 {
301 "name": _("Cluster Packages"),
302 "url": "/clusterpackage/",
303 "icon": "fa-archive",
304 },
305 {
306 "name": _("API Documentation"),
307 "url": "/api/schema/swagger-ui/",
308 "icon": "fa-book",
309 },
310 ]
312 if len(query) >= 2:
313 # Filter navigation targets
314 context["nav_results"] = [
315 item for item in nav_targets if query in (item["name"].lower())
316 ]
318 # Servers: match hostname, IPs, serial, description, application,
319 # rack, and related fields. Exact hostname matches rank first.
320 server_qs = (
321 self.get_queryset_filtered(Server)
322 .filter(
323 Q(hostname__icontains=query)
324 | Q(primary_ip__icontains=query)
325 | Q(management_ip__icontains=query)
326 | Q(serial_nr__icontains=query)
327 | Q(description__icontains=query)
328 | Q(application__icontains=query)
329 | Q(rack__icontains=query)
330 | Q(domain__name__icontains=query)
331 | Q(status__name__icontains=query)
332 | Q(location__name__icontains=query)
333 | Q(tags__name__icontains=query)
334 )
335 .distinct()
336 )
337 # Manual relevance: exact hostname match ranks above substring,
338 # then alphabetical.
339 server_list = list(
340 sorted(
341 server_qs,
342 key=lambda s: (0 if s.hostname.lower() == query else 1, s.hostname),
343 )[:10]
344 )
345 context["servers"] = server_list
347 context["vendors"] = self.get_queryset_filtered(Vendor).filter(
348 name__icontains=query
349 )[:10]
350 context["clusters"] = (
351 self.get_queryset_filtered(Cluster)
352 .filter(
353 Q(name__icontains=query)
354 | Q(clustersoftware__name__icontains=query)
355 | Q(clustersoftware__version__icontains=query)
356 )
357 .distinct()[:10]
358 )
359 context["domains"] = self.get_queryset_filtered(Domain).filter(
360 name__icontains=query
361 )[:10]
362 context["locations"] = self.get_queryset_filtered(Location).filter(
363 Q(name__icontains=query) | Q(country__icontains=query)
364 )[:10]
365 context["statuses"] = self.get_queryset_filtered(Status).filter(
366 name__icontains=query
367 )[:10]
368 context["patchtimes"] = self.get_queryset_filtered(Patchtime).filter(
369 name__icontains=query
370 )[:10]
371 context["servermodels"] = self.get_queryset_filtered(ServerModel).filter(
372 Q(name__icontains=query) | Q(vendor__name__icontains=query)
373 )[:10]
374 context["os"] = self.get_queryset_filtered(OS).filter(
375 Q(version__icontains=query) | Q(vendor__name__icontains=query)
376 )[:10]
377 context["clustersoftware"] = self.get_queryset_filtered(
378 ClusterSoftware
379 ).filter(
380 Q(name__icontains=query)
381 | Q(version__icontains=query)
382 | Q(vendor__name__icontains=query)
383 )[
384 :10
385 ]
386 context["clusterpackagetypes"] = self.get_queryset_filtered(
387 ClusterPackageType
388 ).filter(name__icontains=query)[:10]
389 context["clusterpackages"] = self.get_queryset_filtered(
390 ClusterPackage
391 ).filter(
392 Q(name__icontains=query)
393 | Q(cluster__name__icontains=query)
394 | Q(description__icontains=query)
395 | Q(host__icontains=query)
396 )[
397 :10
398 ]
400 # Simple check if anything was found
401 result_lists = [
402 context["servers"],
403 context["vendors"],
404 context["clusters"],
405 context["domains"],
406 context["locations"],
407 context["statuses"],
408 context["patchtimes"],
409 context["servermodels"],
410 context["os"],
411 context["clustersoftware"],
412 context["clusterpackagetypes"],
413 context["clusterpackages"],
414 ]
415 context["has_results"] = (
416 bool(context["nav_results"])
417 or any(hasattr(rl, "exists") and rl.exists() for rl in result_lists)
418 or any(isinstance(rl, list) and rl for rl in result_lists)
419 )
421 # Group filter for the results page
422 if self.request.user.is_superuser:
423 from django.contrib.auth.models import Group
425 context["search_groups"] = Group.objects.all()
426 else:
427 context["search_groups"] = self.request.user.groups.all()
428 else:
429 context["has_results"] = False
430 context["query_too_short"] = True
432 return context
435class HistoryDiffView(LoginRequiredMixin, TemplateView):
436 template_name = "history_diff.html"
438 def get_context_data(self, **kwargs: Any) -> Any:
439 context = super().get_context_data(**kwargs)
440 app_label = kwargs.get("app_label")
441 history_id = kwargs.get("history_id")
443 try:
444 model = apps.get_model(app_label, "Model")
445 history_qs = filter_history_queryset_by_tenant(
446 model.history.all(), self.request # type: ignore
447 )
448 record = history_qs.get(history_id=history_id)
449 except (LookupError, ObjectDoesNotExist):
450 raise Http404("History record not found")
452 context["record"] = record
453 context["instance"] = record.instance
454 context["app_label"] = app_label
456 if record.prev_record:
457 context["diff"] = record.diff_against(record.prev_record)
458 else:
459 context["diff"] = None
461 return context
464class PatchScheduleView(LoginRequiredMixin, TemplateView):
465 """
466 Overview of servers grouped by their patch-time window, so operators can
467 see at a glance what is patched when.
468 """
470 template_name = "patch_schedule.html"
472 def get_servers(self) -> Any:
473 if self.request.user.is_superuser:
474 return Server.objects.all()
475 user_groups = self.request.user.groups.all()
476 selected_groups = self.request.session.get("selected_groups", [])
477 if selected_groups:
478 group_ids = [int(g) for g in selected_groups if g.isdigit()]
479 if group_ids:
480 return Server.objects.filter(
481 Q(group__id__in=group_ids) | Q(group__isnull=True)
482 )
483 return Server.objects.filter(Q(group__in=user_groups) | Q(group__isnull=True))
485 def get_context_data(self, **kwargs: Any) -> Any:
486 context = super().get_context_data(**kwargs)
487 servers = self.get_servers().select_related("patchtime", "status")
488 by_window: dict = {}
489 for server in servers:
490 window = server.patchtime.name if server.patchtime_id else _("Unassigned")
491 by_window.setdefault(window, []).append(server)
493 ordered = sorted(
494 by_window.items(), key=lambda item: (item[0] == _("Unassigned"), item[0])
495 )
496 context["windows"] = ordered
497 context["server_count"] = servers.count()
498 return context
501class ImportStarterPackView(LoginRequiredMixin, View):
502 def post(self, request: Any, *args: Any, **kwargs: Any) -> Any:
503 user_groups = request.user.groups.all()
504 if not user_groups.exists():
505 messages.error(request, _("You are not assigned to any group."))
506 return redirect("dashboard")
508 group = user_groups.first()
509 results = import_starter_pack(group)
511 messages.success(
512 request,
513 _("Imported %d vendors and %d operating systems into group %s.")
514 % (results["vendors"], results["os"], group.name),
515 )
516 return redirect("vendor:index")
519class TermsView(TemplateView):
520 template_name = "legal/terms.html"
523class PrivacyView(TemplateView):
524 template_name = "legal/privacy.html"
527class ImpressumView(TemplateView):
528 template_name = "legal/impressum.html"
531class HealthView(View):
532 def get(self, request: Any, *args: Any, **kwargs: Any) -> JsonResponse:
533 health = {
534 "status": "healthy",
535 "version": getattr(settings, "APP_VERSION", "unknown"),
536 "last_modification": getattr(settings, "APP_MODIFICATION_DATE", "unknown"),
537 "checks": {},
538 }
540 # Check database connection
541 try:
542 from django.db import connection
544 connection.cursor()
545 health["checks"]["database"] = "ok"
546 except Exception:
547 # Log the real error server-side; never leak connection internals
548 # (host/port/user) to unauthenticated callers.
549 logger.exception("Health check: database connection failed")
550 health["status"] = "unhealthy"
551 health["checks"]["database"] = "unavailable"
553 status_code = 200 if health["status"] == "healthy" else 503
554 return JsonResponse(health, status=status_code)