Adding debian version 0.143.debian/0.143debian

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 266 insertions, 0 deletions

diff --git a/debian/tests/check-log b/debian/tests/check-log
new file mode 100755
index 0000000..01b24b9
--- /dev/null
+++ b/debian/tests/check-log
@@ -0,0 +1,266 @@
+#!/usr/bin/python3
+
+# pylint: disable=invalid-name
+# pylint: enable=invalid-name
+
+"""Run given checks on the log output."""
+
+import argparse
+import json
+import pathlib
+import re
+import shlex
+import sys
+import typing
+
+
+class Check:
+    """The Check class contains all the checks that the caller can run."""
+
+    def __init__(self, log: str) -> None:
+        self.errors = 0
+        self.command_outputs = self._extract_command_outputs_from_log(log)
+
+    def _error(self, msg: str) -> None:
+        print("ERROR: " + msg)
+        self.errors += 1
+
+    @staticmethod
+    def _extract_command_outputs_from_log(log: str) -> dict[str, str]:
+        """Extract command outputs from the given log output.
+
+        The output must be framed by a header and a footer line. The header
+        line contains the key surrounded by 10 # characters. The footer
+        line consist of 40 # characters. Returns a mapping from the output
+        key to the command output.
+        """
+        marker = "#" * 10
+        footer = "#" * 40
+        matches = re.findall(
+            f"^{marker} ([^#]+) {marker}\n(.*?\n){footer}$",
+            log,
+            flags=re.DOTALL | re.MULTILINE,
+        )
+        return {m[0]: m[1] for m in matches}
+
+    def get_commands_from_ps_output(self) -> list[str]:
+        """Get list of command from `ps -ww aux` output."""
+        ps_output = self.command_outputs["ps -ww aux"]
+        lines = ps_output.strip().split("\n")[1:]
+        commands = []
+        for line in lines:
+            columns = re.split(r"\s+", line, maxsplit=10)
+            commands.append(columns[10])
+        return commands
+
+    def get_ip_addr(self) -> list[typing.Any]:
+        """Get IP address information from `ip addr` JSON output."""
+        ip_addr = json.loads(self.command_outputs["ip -json addr"])
+        assert isinstance(ip_addr, list)
+        return ip_addr
+
+    def get_ip_route(self) -> list[typing.Any]:
+        """Get IP route information from `ip route` JSON output."""
+        ip_route = json.loads(self.command_outputs["ip -json route"])
+        assert isinstance(ip_route, list)
+        return ip_route
+
+    def run_checks(self, args: list[str]) -> int:
+        """Run the checks and return the number of errors found.
+
+        The methods of this class can be u
+        """
+        if not args:
+            return self.errors
+
+        try:
+            check = getattr(self, args[0])
+        except AttributeError:
+            self._error(f"Check '{args[0]}' not found.")
+            return self.errors
+        check_args = []
+
+        for arg in args[1:]:
+            if not hasattr(self, arg):
+                check_args.append(arg)
+                continue
+
+            check(*check_args)
+            check = getattr(self, arg)
+            check_args = []
+
+        check(*check_args)
+        return self.errors
+
+    def _check_is_subset(self, actual: set[str], expected: set[str]) -> None:
+        """Check that the first dictionary is a subset of the second one.
+
+        Log errors if the sets are different.
+        """
+        unexpected = actual - expected
+        if unexpected:
+            self._error(f"Not expected entries: {unexpected}")
+
+    def _check_is_subdict(
+        self,
+        expected_dict: dict[str, str],
+        actual_dict: dict[str, str],
+        log_prefix: str,
+    ) -> None:
+        """Check that the first dictionary is a subset of the second one.
+
+        Log errors if differences are found.
+        """
+        missing_keys = set(expected_dict.keys()) - set(actual_dict.keys())
+        if missing_keys:
+            self._error(f"{log_prefix}Missing keys: {missing_keys}")
+        for key, expected_value in sorted(expected_dict.items()):
+            actual_value = actual_dict.get(key, "")
+            if expected_value != actual_value:
+                self._error(
+                    f"{log_prefix}Value for key '{key}' differs:"
+                    f" '{expected_value}' expected, but got '{actual_value}'"
+                )
+
+    # Below are all checks that the user might call.
+
+    def has_hostname(self, hostname_pattern: str) -> None:
+        """Check that the hostname matches the given regular expression."""
+        hostname = self.command_outputs["hostname"].strip()
+        if re.fullmatch(hostname_pattern, hostname):
+            print(f"hostname '{hostname}' matches pattern '{hostname_pattern}'")
+            return
+        self._error(
+            f"hostname '{hostname}' does not match"
+            f" expected pattern '{hostname_pattern}'"
+        )
+
+    def has_interface_mtu(self, device_pattern: str, expected_mtu: str) -> None:
+        """Check that a matching network device has the expected MTU set."""
+        for device in self.get_ip_addr():
+            if not re.fullmatch(device_pattern, device["ifname"]):
+                continue
+            if str(device["mtu"]) == expected_mtu:
+                print(f"device {device['ifname']} has MTU {device['mtu']}")
+                return
+            self._error(
+                f"device {device['ifname']} has MTU {device['mtu']}"
+                f" but expected {expected_mtu}"
+            )
+            return
+        self._error(f"no link found that matches '{device_pattern}'")
+
+    def has_ip_addr(self, family: str, addr_pattern: str, device_pattern: str) -> None:
+        """Check that a matching network device has a matching IP address."""
+        for device in self.get_ip_addr():
+            if not re.fullmatch(device_pattern, device["ifname"]):
+                continue
+            for addr in device["addr_info"]:
+                if addr["family"] != family or addr["scope"] != "global":
+                    continue
+                address = f"{addr['local']}/{addr['prefixlen']}"
+                if re.fullmatch(addr_pattern, address):
+                    print(f"found addr {address} for {device['ifname']}: {addr}")
+                    return
+                self._error(
+                    f"addr {address} for {device['ifname']}"
+                    f" does not match {addr_pattern}: {addr}"
+                )
+                return
+        name = {"inet": "IPv4", "inet6": "IPv6"}[family]
+        self._error(
+            f"no link found that matches '{device_pattern}' and has an {name} address"
+        )
+
+    def has_ipv4_addr(self, addr_pattern: str, device_pattern: str) -> None:
+        """Check that a matching network device has a matching IPv4 address."""
+        self.has_ip_addr("inet", addr_pattern, device_pattern)
+
+    def has_ipv6_addr(self, addr_pattern: str, device_pattern: str) -> None:
+        """Check that a matching network device has a matching IPv6 address."""
+        self.has_ip_addr("inet6", addr_pattern, device_pattern)
+
+    def has_ipv4_default_route(self, gateway_pattern: str, device_pattern: str) -> None:
+        """Check that the IPv4 default route is via a matching gateway and device."""
+        for route in self.get_ip_route():
+            if route["dst"] != "default":
+                continue
+            if not re.fullmatch(gateway_pattern, route["gateway"]) or not re.fullmatch(
+                device_pattern, route["dev"]
+            ):
+                self._error(
+                    f"Default IPv4 route does not match expected gateway pattern"
+                    f" '{gateway_pattern}' or dev pattern '{device_pattern}': {route}"
+                )
+                continue
+            print(
+                f"found IPv4 default route via {route['gateway']}"
+                f" for {route['dev']}: {route}"
+            )
+            return
+        self._error("no IPv4 default route found")
+
+    def has_net_conf(self, min_expected_files: str, *expected_net_confs: str) -> None:
+        """Compare the /run/net*.conf files.
+
+        There must be at least `min_expected_files` /run/net*.conf files
+        in the log output and no unexpected one. The format for
+        `expected_net_confs` is `<file name>=<expected content>`.
+        """
+
+        expected = dict(nc.split("=", maxsplit=1) for nc in expected_net_confs)
+        prog = re.compile(r"/run/net[^#]+\.conf")
+        got = {
+            key: value for key, value in self.command_outputs.items() if prog.match(key)
+        }
+
+        if len(got) < int(min_expected_files):
+            self._error(
+                f"Expected at least {min_expected_files} /run/net*.conf files,"
+                f" but got only {len(got)}: {set(got.keys())}"
+            )
+        self._check_is_subset(set(got.keys()), set(expected.keys()))
+
+        for net_dev in sorted(got.keys()):
+            log_prefix = f"{net_dev}: "
+            expected_net_conf = parse_net_conf(expected.get(net_dev, ""))
+            actual_net_conf = parse_net_conf(got.get(net_dev, ""))
+            self._check_is_subdict(expected_net_conf, actual_net_conf, log_prefix)
+            print(f"compared {len(expected_net_conf)} items from {net_dev}")
+
+    def has_no_running_processes(self) -> None:
+        """Check that there are no remaining running processes from the initrd."""
+        processes = drop_kernel_processes(self.get_commands_from_ps_output())
+        if len(processes) == 2:
+            print(f"found only expected init and ps process: {processes}")
+            return
+        self._error(
+            f"Expected only init and ps process, but got {len(processes)}: {processes}"
+        )
+
+
+def drop_kernel_processes(processes: list[str]) -> list[str]:
+    """Return a list of processes with the kernel processes dropped."""
+    return [p for p in processes if not p.startswith("[") or not p.endswith("]")]
+
+
+def parse_net_conf(net_conf: str) -> dict[str, str]:
+    """Parse /run/net*.conf file and return a key to value mapping."""
+    items = shlex.split(net_conf)
+    return dict(item.split("=", maxsplit=1) for item in items)
+
+
+def main(arguments: list[str]) -> int:
+    """Run given checks on the log output. Return number of errors."""
+    parser = argparse.ArgumentParser()
+    parser.add_argument("log_file", metavar="log-file")
+    parser.add_argument("checks", metavar="check", nargs="+")
+    args = parser.parse_args(arguments)
+
+    log = pathlib.Path(args.log_file).read_text(encoding="ascii")
+    check = Check(log)
+    return check.run_checks(args.checks)
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))