Reverse logfile; proper validation; tests

Reversed the order of log entries.

Added proper validation for Mailcow responses. If extra fields are passed, or optional fields are missing, then Squire will operate as usual and only log a warning message.

Updated tests to reflect Mailcow's API changes.

core/admin_status/views.py

@@ -1,28 +1,28 @@
-from io import BufferedReader
 import os
 import re
 from django.conf import settings
 from django.views.generic import TemplateView
 
 from django.utils.safestring import SafeText
-from django.utils.html import escape, format_html
+from django.utils.html import escape
 
 from core.status_collective import AdminStatusViewMixin
 
 
 class LogFileView(AdminStatusViewMixin, TemplateView):
     """
-    TODO
+    A simple page that allows viewing the content of rotating log files in `/squire/logs/squire.log.<number>`
     """
 
     tags = {
+        "&lt;[a-z0-9_\-:\s]*&gt;": "text-danger",
+        "[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}(,[0-9]{3})?": "far fa-clock text-muted",
         re.escape("[debug]"): "text-secondary font-weight-bold",
         re.escape("[info]"): "text-info font-weight-bold",
         re.escape("[warning]"): "text-warning font-weight-bold",
         re.escape("[error]"): "text-danger font-weight-bold",
-        re.escape("TypeError"): "font-weight-bold",
+        re.escape("JSONParseError"): "font-weight-bold",
         re.escape("(mailcow_api)"): "far fa-envelope text-primary",
-        "[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2},[0-9]{3}": "far fa-clock text-muted",
     }
 
     template_name = "core/admin_status/log.html"
@@ -35,18 +35,21 @@ def __init__(self, *args, **kwargs) -> None:
         self._logfile1_name = None
         self._logfile2_name = None
 
+        # Open both logfiles, reverse the log's lines (most recent at the top),
+        #   and make sure the most recent rotating logfile appears first.
         try:
             with open(self.log_location, "r") as fl:
-                data = fl.read()
+                data = "".join(reversed(fl.readlines()))
+                # The log format is known; the first 23 chars are an ISO-timestamp
                 log1_peek = data[:23]
                 self._logfile1 = self._format_logfile(data)
                 self._logfile1_name = os.path.join("squire", "logs", "squire.log")
         except OSError:
             pass
         try:
             with open(self.log_location + ".1", "r") as fl:
-                data = fl.read()
-                if data[:23] < log1_peek:
+                data = "".join(reversed(fl.readlines()))
+                if data[:23] > log1_peek:
                     self._logfile2 = self._logfile1
                     self._logfile1 = self._format_logfile(data)
                     self._logfile1_name = os.path.join("squire", "logs", "squire.log.1")
@@ -58,9 +61,8 @@ def __init__(self, *args, **kwargs) -> None:
             pass
 
     def _format_logfile(self, data: str) -> SafeText:
-        """TODO"""
+        """Formats a logfile by adding 'syntax' highlighting to specific keywords defined in this class's `tags`."""
         text = escape(data)
-        text = re.sub("(\r\n|\r|\n)", "<br>", text)
         for string, classes in self.tags.items():
             text = SafeText(
                 re.sub(
@@ -69,16 +71,9 @@ def _format_logfile(self, data: str) -> SafeText:
                     text,
                 )
             )
+        text = SafeText(re.sub("(\r\n|\r|\n)", "<br>", text))
         return text
 
-    def _format_logline(self, line: str) -> SafeText:
-        """TODO"""
-        # format_html("{}")
-
-        print()
-
-        # xxx = "".replace(new RegExp(`(\\b)(${text})(\\b)`, 'gi'), `$1<span class='${styles[className]}'>$2<\/span>$3`)
-
     def get_context_data(self, **kwargs):
         context = super().get_context_data(**kwargs)
         context["log1"] = self._logfile1

core/templates/core/admin_status/log.html

@@ -8,10 +8,6 @@
   Logging
 {% endblock title %}
 
-{% block content-frame-class %}
-    wideContentFrame
-{% endblock %}
-
 {% block css %}
     {{ block.super }}
     <link rel="stylesheet" href="{% static 'css/bootstrap_tabs.css' %}">
@@ -20,12 +16,28 @@
 {% block content %}
     {% bootstrap_tabs tabs %}
     <h1>Logs</h1>
-    <h4>{{ log1_name }}</h4>
-    <samp class="text-break">{{ log1 }}</samp>
-    <hr>
-    <h4>{{ log2_name }}</h4>
-    <samp>{{ log2 }}</samp>
-    <hr>
+    <p>Logging created by various modules. Logs are written to two rotating files sizing (at most) 5MB each.</p>
+    {% if log1 is not None %}
+        <h4>{{ log1_name }}</h4>
+        {% if not log1 %}
+            <p class="font-italic">No logging written.</p>
+        {% else %}
+            <samp class="text-break">{{ log1 }}</samp>
+        {% endif %}
+        <hr>
+    {% endif %}
+    {% if log2 is not None %}
+        <h4>{{ log2_name }}</h4>
+        {% if not log2 %}
+            <p class="font-italic">No logging written.</p>
+        {% else %}
+            <samp class="text-break">{{ log2 }}</samp>
+        {% endif %}
+        <hr>
+    {% endif %}
+    {% if log1 is None and log2 is None %}
+        <p class="font-italic">No logging written.</p>
+    {% endif %}
 {% endblock content %}
 
 {% block js_bottom %}

mailcow_integration/api/client.py

@@ -101,12 +101,12 @@ def _make_request(
     def get_alias_all(self) -> Generator[Optional[MailcowAlias], None, None]:
         """Gets a list of all email aliases"""
         content = self._make_request(f"get/alias/all")
-        return map(lambda alias: MailcowAlias.from_json_safe(alias), content)
+        return map(lambda alias: MailcowAlias.from_json(alias), content)
 
     def get_alias(self, id: int) -> Optional[MailcowAlias]:
         """Gets an email alias with a specific id"""
         content = self._make_request(f"get/alias/{id}")
-        return MailcowAlias.from_json_safe(content)
+        return MailcowAlias.from_json(content)
 
     def update_alias(self, alias: MailcowAlias) -> dict:
         """Updates an alias"""
@@ -179,20 +179,20 @@ def delete_aliases(self, aliases: List[MailcowAlias]) -> dict:
     def get_mailbox_all(self) -> Generator[Optional[MailcowMailbox], None, None]:
         """Gets a list of all mailboxes"""
         content = self._make_request(f"get/mailbox/all")
-        return map(lambda mailbox: MailcowMailbox.from_json_safe(mailbox), content)
+        return map(lambda mailbox: MailcowMailbox.from_json(mailbox), content)
 
     ################
     # RSPAMD SETTINGS (undocumented API)
     ################
     def get_rspamd_setting_all(self) -> Generator[Optional[RspamdSettings], None, None]:
         """Gets all Rspamd settings maps"""
         content = self._make_request("get/rsetting/all")
-        return map(lambda rspamdsetting: RspamdSettings.from_json_safe(rspamdsetting), content)
+        return map(lambda rspamdsetting: RspamdSettings.from_json(rspamdsetting), content)
 
     def get_rspamd_setting(self, id: int) -> Optional[RspamdSettings]:
         """Gets an Rspamd settings map with a specific id"""
         content = self._make_request(f"get/rsetting/{id}")
-        return RspamdSettings.from_json_safe(content)
+        return RspamdSettings.from_json(content)
 
     def update_rspamd_setting(self, setting: RspamdSettings) -> dict:
         """Updates the RspamdSetting associated to the given ID with the given data"""

mailcow_integration/api/interface/alias.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import Optional, List
+from typing import Optional, List, Set
 
 from datetime import datetime
 from dataclasses import dataclass
@@ -38,6 +38,11 @@ class MailcowAlias(MailcowAPIResponse):
     created: Optional[datetime] = None
     modified: Optional[datetime] = None
 
+    _cleanable_bools = ("active", "is_catch_all", "sogo_visible")
+    _cleanable_ints = ("id", "active_int", "sogo_visible_int")
+    _cleanable_strings = ("in_primary_domain", "domain")
+    _cleanable_datetimes = ("created",)
+
     def __post_init__(self):
         if self.active_int is None:
             self.active_int = int(self.active)
@@ -54,17 +59,43 @@ def get_type(self) -> AliasType:
         return AliasType.NORMAL
 
     @classmethod
-    def from_json(cls, json: dict) -> "MailcowAlias":
-        json.update(
-            {
-                "goto": json["goto"].split(","),
-                "active": bool(json["active"]),
-                "is_catch_all": bool(json["is_catch_all"]),
-                "public_comment": json["public_comment"] or "",
-                "private_comment": json["private_comment"] or "",
-                "sogo_visible": bool(json["sogo_visible"]),
-                "created": datetime.fromisoformat(json["created"]),
-                "modified": datetime.fromisoformat(json["modified"]) if json["modified"] is not None else None,
-            }
-        )
-        return cls(**json)
+    def clean(cls, json: dict, extra_keys: Set[str] = None) -> dict:
+        address = json.get("address", None)
+        if address is None:
+            cls._issue_warning("address", address)
+            raise AttributeError(f"address was not provided when creating {cls.__name__}")
+
+        # Goto-addresses are comma-separated strings
+        goto = json.get("goto", None)
+        if goto is None or not isinstance(goto, str):
+            cls._issue_warning("goto", goto, "list")
+            raise AttributeError(f"goto was not provided or had an invalid value when creating {cls.__name__}")
+
+        new_json = {
+            "address": address,
+            "goto": goto.split(","),
+        }
+
+        # Modified can be returned as None
+        if "modified" not in json:
+            cls._issue_warning("modified", None, "ISO-datetime (or None)")
+        else:
+            modified = json.get("modified")
+            if modified is not None:
+                modified = cls._parse_as_dt("modified", json)
+            new_json["modified"] = modified
+
+        # Same for the private/public comment
+        if "public_comment" not in json:
+            cls._issue_warning("public_comment", None, "string (or None)")
+        else:
+            new_json["public_comment"] = str(json.get("public_comment") or "")
+
+        if "private_comment" not in json:
+            cls._issue_warning("private_comment", None, "string (or None)")
+        else:
+            new_json["private_comment"] = str(json.get("private_comment") or "")
+
+        extra_keys = extra_keys or set()
+        new_json.update(**super().clean(json, extra_keys=new_json.keys() | extra_keys))
+        return new_json