-
- {% bs_icon 'exclamation-triangle'%} {% trans "Warning" %}: {% trans "You don't have any Logs" %}
-
+
+ {% bs_icon 'exclamation-triangle'%}
+ {% if q or failed %}{% trans "No log rows match." %}{% else %}{% trans "There are no log rows yet." %}{% endif %}
{% else %}
-
+
| # |
- {% trans "Date" %} |
+ {% blocktrans %}Date ({{ time_zone }}){% endblocktrans %} |
{% trans "User" %} |
{% trans "Host" %} |
{% trans "Instance" %} |
{% trans "Message" %} |
{% trans "Result" %} |
{% trans "IP" %} |
-
+
-
+
{% for log in logs %}
-
+
| {{ log.id }} |
- {{ log.date|date:"M d H:i:s" }} |
+ {{ log.date|date:"Y-m-d H:i:s" }} |
{{ log.user }} |
{{ log.host }} |
- {{ log.instance }} |
+ {% if log.vm_pk %}{{ log.instance }}{% else %}{{ log.instance }}{% endif %} |
{{ log.message }} |
{% if log.success is None %}{% trans "Unknown" %}{% elif log.success %}{% trans "Done" %}{% else %}{% trans "Failed" %}{% endif %} |
{{ log.ip|default:"" }} |
@@ -53,11 +61,8 @@
- {% bootstrap_pagination logs %}
+ {% bootstrap_pagination logs extra=query %}
{% endif %}
{% endblock %}
-{% block script %}
-
-{% endblock script %}
diff --git a/admin/tests/test_logs_page.py b/admin/tests/test_logs_page.py
new file mode 100644
index 00000000..f46a67c7
--- /dev/null
+++ b/admin/tests/test_logs_page.py
@@ -0,0 +1,87 @@
+"""The Logs page searches every row on the server, can show the failed ones
+only, says the year and time zone, links a VM's rows to the VM, and downloads
+what it shows as an Excel workbook."""
+
+from io import BytesIO
+
+from computes.models import Compute
+from django.contrib.auth import get_user_model
+from django.test import TestCase, override_settings
+from django.urls import reverse
+from instances.models import Instance
+from logs.models import Logs
+from lxml import html
+from openpyxl import load_workbook
+
+UUID = "11111111-2222-3333-4444-555555555555"
+
+
+@override_settings(TIME_ZONE="UTC")
+class LogsPageTests(TestCase):
+ def setUp(self):
+ self.admin = get_user_model().objects.create_superuser("lg-admin", "lg@example.com", "pw")
+ self.client.force_login(self.admin)
+ Logs.objects.all().delete()
+ # the row to find is the oldest: past the first page
+ Logs.objects.create(user="alice", host="host1", instance="web", instance_uuid=UUID,
+ message="VM web: delete", success=False, ip="192.0.2.7")
+ Logs.objects.bulk_create(Logs(user="bob", instance="db", message=f"VM db: power on {n}", success=True)
+ for n in range(150))
+ compute = Compute.objects.create(name="host1", hostname="192.0.2.1", login="u", password="p", type=1)
+ self.vm = Instance.objects.create(compute=compute, name="web", uuid=UUID)
+
+ def page(self, **query):
+ return html.fromstring(self.client.get(reverse("admin:logs"), query).content)
+
+ def rows(self, page):
+ return [[" ".join(td.text_content().split()) for td in tr.xpath("td")] for tr in page.xpath("//tbody/tr")]
+
+ def test_a_search_finds_a_row_on_any_page(self):
+ for query in ("alice", "WEB: DELETE", "192.0.2.7", "host1"):
+ with self.subTest(query=query):
+ rows = self.rows(self.page(q=query))
+ self.assertEqual([row[2] for row in rows], ["alice"])
+
+ def test_a_search_is_text_not_a_pattern(self):
+ Logs.objects.create(user="carol", instance="", message="disk (vda) [full]", success=True)
+ self.assertEqual([row[2] for row in self.rows(self.page(q="(vda) ["))], ["carol"])
+
+ def test_the_failed_rows_only_and_marked(self):
+ page = self.page(failed="1")
+ self.assertEqual([row[2] for row in self.rows(page)], ["alice"])
+ self.assertIn("table-danger", page.xpath("//tbody/tr")[0].get("class"))
+
+ def test_the_pages_keep_the_search(self):
+ page = self.page(q="bob")
+ self.assertTrue(page.xpath("//a[contains(@href, 'page=2') and contains(@href, 'q=bob')]"))
+
+ def test_dates_have_the_year_and_the_time_zone(self):
+ page = self.page(q="alice")
+ self.assertIn("Date (UTC)", [" ".join(th.text_content().split()) for th in page.xpath("//thead//th")])
+ log = Logs.objects.get(user="alice")
+ self.assertEqual(self.rows(page)[0][1], log.date.strftime("%Y-%m-%d %H:%M:%S"))
+
+ def test_a_vm_row_links_the_vm_while_it_is_there(self):
+ page = self.page(q="alice")
+ self.assertEqual(page.xpath("//tbody/tr/td/a/@href"), [reverse("instances:instance", args=[self.vm.pk])])
+ self.vm.delete()
+ self.assertEqual(self.page(q="alice").xpath("//tbody/tr/td/a"), [])
+
+ def test_the_rows_shown_download_as_a_workbook(self):
+ Logs.objects.create(user="mallory", instance="", message="=HYPERLINK(1)", success=False)
+ response = self.client.get(reverse("admin:logs_export"), {"failed": "1"})
+ self.assertEqual(response["Content-Type"], "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
+ self.assertIn("attachment", response["Content-Disposition"])
+ sheet = load_workbook(BytesIO(response.content)).active
+ values = list(sheet.values)
+ self.assertEqual(values[0], ("Date (UTC)", "User", "Host", "VM", "VM UUID", "Message", "Result", "IP"))
+ self.assertEqual([row[1] for row in values[1:]], ["mallory", "alice"])
+ # a message starting with = stays text
+ self.assertEqual(sheet.cell(row=2, column=6).data_type, "s")
+ page = self.page(q="alice", failed="1")
+ self.assertEqual(page.xpath("//a[contains(@href, 'export')]/@href"),
+ [reverse("admin:logs_export") + "?q=alice&failed=1"])
+
+ def test_only_an_administrator_downloads(self):
+ self.client.force_login(get_user_model().objects.create_user("lg-user", password="pw"))
+ self.assertEqual(self.client.get(reverse("admin:logs_export")).status_code, 403)
diff --git a/admin/urls.py b/admin/urls.py
index 937a1349..c0ce1fe1 100644
--- a/admin/urls.py
+++ b/admin/urls.py
@@ -15,6 +15,7 @@ urlpatterns = [
path("users//block/", views.user_block, name="user_block"),
path("users//unblock/", views.user_unblock, name="user_unblock"),
path("logs/", views.logs, name="logs"),
+ path("logs/export.xlsx", views.logs_export, name="logs_export"),
path("removed-vms/", views.tombstone_list, name="tombstone_list"),
path("removed-vms//delete/", views.tombstone_delete, name="tombstone_delete"),
]
diff --git a/admin/views.py b/admin/views.py
index 10ec4156..76695c6a 100644
--- a/admin/views.py
+++ b/admin/views.py
@@ -7,14 +7,16 @@ from django.contrib import messages
from django.contrib.auth import update_session_auth_hash
from django.contrib.auth.models import Group, User
from django.core.paginator import Paginator
+from django.http import HttpResponse
from django.db import transaction
from django.db.models import Count, Q
from django.shortcuts import get_object_or_404, redirect, render
from django.utils import timezone
-from django.utils.http import url_has_allowed_host_and_scheme
+from django.utils.http import url_has_allowed_host_and_scheme, urlencode
from django.utils.translation import gettext_lazy as _
from django.views.decorators.http import require_POST
from instances.models import InstanceTombstone
+from instances.report import workbook
from logs.audit import audit, changed_fields, snapshot
from logs.models import Logs
@@ -254,11 +256,54 @@ def user_unblock(request, pk):
return redirect("admin:user_list")
+def _logs(request):
+ """The Logs rows the search and the failed-only box choose, newest first,
+ and the query that chooses them again."""
+ rows = Logs.objects.order_by("-date", "-id")
+ text = request.GET.get("q", "").strip()
+ failed = bool(request.GET.get("failed"))
+ if text:
+ rows = rows.filter(Q(user__icontains=text) | Q(host__icontains=text) | Q(instance__icontains=text)
+ | Q(instance_uuid__icontains=text) | Q(message__icontains=text) | Q(ip__icontains=text))
+ if failed:
+ rows = rows.filter(success=False)
+ query = urlencode([(key, value) for key, value in (("q", text), ("failed", "1" if failed else "")) if value])
+ return rows, text, failed, query
+
+
+def _result(success):
+ return _("Unknown") if success is None else _("Done") if success else _("Failed")
+
+
@superuser_only
def logs(request):
- paginator = Paginator(Logs.objects.order_by("-date", "-id"), int(app_settings.LOGS_PER_PAGE))
- logs = paginator.get_page(request.GET.get("page"))
- return render(request, "admin/logs.html", {"logs": logs})
+ rows, text, failed, query = _logs(request)
+ logs = Paginator(rows, int(app_settings.LOGS_PER_PAGE)).get_page(request.GET.get("page"))
+ # a VM's rows link to it while it is in the panel
+ vms = dict(Instance.objects.filter(uuid__in={log.instance_uuid for log in logs if log.instance_uuid})
+ .values_list("uuid", "pk"))
+ for log in logs:
+ log.vm_pk = vms.get(log.instance_uuid)
+ return render(request, "admin/logs.html", {"logs": logs, "q": text, "failed": failed, "query": query,
+ "time_zone": timezone.get_current_timezone_name()})
+
+
+@superuser_only
+def logs_export(request):
+ """The rows the Logs page shows for the same search, every page of them,
+ as an Excel workbook."""
+ rows, text, failed, _query = _logs(request)
+ zone = timezone.get_current_timezone_name()
+ header = [f"Date ({zone})", "User", "Host", "VM", "VM UUID", "Message", "Result", "IP"]
+ data = [[timezone.localtime(log.date).strftime("%Y-%m-%d %H:%M:%S"), log.user, log.host, log.instance,
+ log.instance_uuid, log.message, str(_result(log.success)), log.ip or ""] for log in rows.iterator()]
+ notes = [f"Search: {text}" if text else "Search: none", "Failed only" if failed else "All results",
+ f"Rows: {len(data)}", f"Times in {zone}"]
+ response = HttpResponse(workbook("Logs", header, data, "About", notes),
+ content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
+ name = f"webvirtcloud-logs-{timezone.localtime():%Y%m%d-%H%M%S}.xlsx"
+ response["Content-Disposition"] = f'attachment; filename="{name}"'
+ return response
@superuser_only