Test list-units on a specific node

Verifies that bluechictl list-units reports the same units for a
specific node as systemctl list-units executed on this node.

Signed-off-by: Martin Perina <[email protected]>

tests/tests/tier0/bluechi-list-units-on-a-node/main.fmf

@@ -0,0 +1,3 @@
+summary: Test if bluechi list-nodes returns the same list of units from a specific
+    node which can be gathered by running systemctl on the node
+id: 62225c45-9654-4f19-8175-c6de1a0a70e0

tests/tests/tier0/bluechi-list-units-on-a-node/test_bluechi_list_units_on_a_node.py

@@ -0,0 +1,118 @@
+#
+# Copyright Contributors to the Eclipse BlueChi project
+#
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+import re
+from typing import Dict, Tuple
+
+from bluechi_test.config import BluechiAgentConfig, BluechiControllerConfig
+from bluechi_test.machine import BluechiAgentMachine, BluechiControllerMachine
+from bluechi_test.test import BluechiTest
+
+node_foo_name = "node-foo"
+
+
+def parse_bluechictl_output(output: str) -> Dict[str, Dict[str, Tuple[str, str]]]:
+    line_pat = re.compile(
+        r"""\s*(?P<node_name>[\S]+)\s*\|
+                              \s*(?P<unit_name>[\S]+)\s*\|
+                              \s*(?P<state>[\S]+)\s*\|
+                              \s*(?P<sub_state>[\S]+)\s*""",
+        re.VERBOSE,
+    )
+    result = {}
+    for line in output.splitlines():
+        if line.startswith("NODE ") or line.startswith("===="):
+            # Ignore header lines
+            continue
+
+        match = line_pat.match(line)
+        if not match:
+            raise Exception(
+                f"Error parsing bluechictl list-units output, invalid line: '{line}'"
+            )
+
+        node_units = result.get(match.group("node_name"))
+        if not node_units:
+            node_units = {}
+            result[match.group("node_name")] = node_units
+
+        if match.group("unit_name") in node_units:
+            raise Exception(
+                f"Error parsing bluechictl list-units output, unit already reported, line: '{line}'"
+            )
+
+        node_units[match.group("unit_name")] = (
+            match.group("state"),
+            match.group("sub_state"),
+        )
+
+    return result
+
+
+def verify_units(all_units: Dict[str, Tuple[str, str]], output: str, node_name: str):
+    esc_seq = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
+    line_pat = re.compile(
+        r"""\s*(?P<unit_name>\S+)
+                              .*loaded
+                              \s+(?P<state>\S+)
+                              \s+(?P<sub_state>\S+)
+                              \s+.*$
+                          """,
+        re.VERBOSE,
+    )
+    for line in output.splitlines():
+        # Some systemctl output contains ANSI sequences, which we need to remove before matching
+        line = esc_seq.sub("", line)
+
+        match = line_pat.match(line)
+        if not match:
+            raise Exception(
+                f"Error parsing systemctl list-units output, invalid line: '{line}'"
+            )
+
+        found = all_units.get(match.group("unit_name"))
+        if (
+            not found
+            or match.group("state") != found[0]
+            or match.group("sub_state") != found[1]
+        ):
+            raise Exception(
+                "Unit '{}' with state '{}' and substate '{}' reported by systemctl"
+                " on node '{}', but not reported by bluechictl".format(
+                    match.group("unit_name"),
+                    match.group("state"),
+                    match.group("sub_state"),
+                    node_name,
+                )
+            )
+
+
+def exec(ctrl: BluechiControllerMachine, nodes: Dict[str, BluechiAgentMachine]):
+    node_foo = nodes[node_foo_name]
+
+    bc_res, bc_out = ctrl.bluechictl.list_units(node_name=node_foo_name)
+    assert bc_res == 0
+    bc_units = parse_bluechictl_output(bc_out)
+
+    foo_res, foo_out = node_foo.systemctl.list_units()
+    assert foo_res == 0
+    verify_units(bc_units[node_foo_name], foo_out, node_foo_name)
+
+
+def test_bluechi_list_units_on_a_node(
+    bluechi_test: BluechiTest,
+    bluechi_ctrl_default_config: BluechiControllerConfig,
+    bluechi_node_default_config: BluechiAgentConfig,
+):
+
+    node_foo_cfg = bluechi_node_default_config.deep_copy()
+    node_foo_cfg.node_name = node_foo_name
+
+    bluechi_ctrl_default_config.allowed_node_names = [node_foo_name]
+
+    bluechi_test.set_bluechi_controller_config(bluechi_ctrl_default_config)
+    bluechi_test.add_bluechi_agent_config(node_foo_cfg)
+
+    bluechi_test.run(exec)