Skip to content

Commit

Permalink
Minimize code duplication around list-unit-files tests
Browse files Browse the repository at this point in the history 
Signed-off-by: Martin Perina <[email protected]>
  • Loading branch information
@mwperina
mwperina committed Aug 29, 2024
1 parent 249a847 commit 1df9cc0
Show file tree
Hide file tree
Showing 3 changed files with 101 additions and 145 deletions.
53 changes: 53 additions & 0 deletions tests/bluechi_test/systemd_lists.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
# Literals used during bluechictl/systemctl output parsing
_ATTR_NODE_NAME = "node_name"
_ATTR_UNIT_NAME = "unit_name"
_ATTR_UNIT_FILE = "unit_file"
_ATTR_PRESET = "preset"
_ATTR_STATE = "state"
_ATTR_SUB_STATE = "sub_state"
_ATTR_LOADED = "loaded"
Expand All @@ -25,6 +27,14 @@ class RegexPattern:
re.VERBOSE,
)

# Pattern to parse bluechictl list-unit-files line
BLUECHICTL_LIST_UNIT_FILES = re.compile(
rf"""\s*(?P<{_ATTR_NODE_NAME}>[\S]+)\s*\|
\s*((?:[^/]*/)*)(?P<{_ATTR_UNIT_FILE}>[\S]+)\s*\|
\s*(?P<{_ATTR_STATE}>[\S]+)\s*""",
re.VERBOSE,
)

# Pattern to parse systemctl list-units line
SYSTEMCTL_LIST_UNITS = re.compile(
rf"""\s*(?P<{_ATTR_UNIT_NAME}>\S+)
Expand All @@ -35,6 +45,14 @@ class RegexPattern:
re.VERBOSE,
)

# Pattern to parse systemctl list-unit-files line
SYSTEMCTL_LIST_UNIT_FILES = re.compile(
rf"""\s*(?P<{_ATTR_UNIT_FILE}>\S+)
\s+(?P<{_ATTR_STATE}>\S+)
\s+(?P<{_ATTR_PRESET}>\S+)""",
re.VERBOSE,
)

# Pattern to remove escape sequences appearing on some systemctl output lines
SC_LINE_ESC_SEQ = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")

Expand Down Expand Up @@ -106,6 +124,41 @@ def from_match(cls, match: re.Match):
)


class SystemdUnitFile(SystemdListItem):
def __init__(
self,
file_name: str,
state: str = None,
preset: str = None,
):
super().__init__(file_name)
self.state = state
self.preset = preset

def __str__(self):
return (
f"SystemdUnitFile("
f"file_name='{self.key}', "
f"state='{self.state}', "
f"preset='{self.preset}')"
)

def attributeEquals(self, other) -> bool:
if not isinstance(other, self.__class__):
return False
return self.state == other.state

@classmethod
def from_match(cls, match: re.Match):
return cls(
file_name=match.group(_ATTR_UNIT_FILE),
state=match.group(_ATTR_STATE),
preset=(
match.group(_ATTR_PRESET) if _ATTR_PRESET in match.groupdict() else ""
),
)


# Parse output of bluechictl list-* and return it as a dictionary containing list of items per each node
def parse_bluechictl_list_output(
content: str,
Expand Down
91 changes: 23 additions & 68 deletions ...s/tests/tier0/bluechi-list-unit-files-on-a-node/test_bluechi_list_unit_files_on_a_node.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,91 +3,46 @@
#
# SPDX-License-Identifier: LGPL-2.1-or-later

import re
from typing import Dict

from bluechi_test.config import BluechiAgentConfig, BluechiControllerConfig
from bluechi_test.machine import BluechiAgentMachine, BluechiControllerMachine
from bluechi_test.systemd_lists import (
RegexPattern,
SystemdUnitFile,
compare_lists,
parse_bluechictl_list_output,
parse_systemctl_list_output,
)
from bluechi_test.test import BluechiTest

node_foo_name = "node-foo"


def parse_bluechictl_output(output: str) -> Dict[str, Dict[str, str]]:
line_pat = re.compile(
r"""\s*(?P<node_name>[\S]+)\s*\|
\s*((?:[^/]*/)*)(?P<unit_file_path>[\S]+)\s*\|
\s*(?P<enablement_status>[\S]+)\s*""",
re.VERBOSE,
)
result = {}
for line in output.splitlines():
if line.startswith("NODE ") or line.startswith("===="):
continue

match = line_pat.match(line)
if not match:
raise Exception(
f"Error parsing bluechictl list-unit-files output, invalid line: '{line}'"
)

node_unit_files = result.get(match.group("node_name"))
if not node_unit_files:
node_unit_files = {}
result[match.group("node_name")] = node_unit_files

if match.group("unit_file_path") in node_unit_files:
raise Exception(
f"Error parsing bluechictl list-unit-files output, unit file already reported, line: '{line}'"
)

node_unit_files[match.group("unit_file_path")] = match.group(
"enablement_status"
)

return result


def verify_unit_files(all_unit_files: Dict[str, str], output: str, node_name: str):
esc_seq = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
line_pat = re.compile(
r"""\s*(?P<unit_file_path>\S+)
\s+(?P<enablement_status>\S+)
\s+.*$
""",
re.VERBOSE,
)
for line in output.splitlines():
line = esc_seq.sub("", line)

match = line_pat.match(line)
if not match:
raise Exception(
f"Error parsing systemctl list-unit-files output, invalid line: '{line}'"
)

found = all_unit_files.get(match.group("unit_file_path"))
if not found or match.group("enablement_status") != found:
raise Exception(
"Unit file '{}' with enablement status '{}' reported by systemctl"
" on node '{}', but not reported by bluechictl".format(
match.group("unit_file_path"),
match.group("enablement_status"),
node_name,
)
)


def exec(ctrl: BluechiControllerMachine, nodes: Dict[str, BluechiAgentMachine]):
node_foo = nodes[node_foo_name]

all_res, all_out = ctrl.bluechictl.list_unit_files(node_name=node_foo_name)
assert all_res == 0
all_unit_files = parse_bluechictl_output(all_out)
all_unit_files = parse_bluechictl_list_output(
content=all_out,
line_pattern=RegexPattern.BLUECHICTL_LIST_UNIT_FILES,
item_class=SystemdUnitFile,
)

foo_res, foo_out = node_foo.systemctl.list_unit_files()
assert foo_res == 0
verify_unit_files(all_unit_files[node_foo_name], foo_out, node_foo_name)
foo_unit_files = parse_systemctl_list_output(
content=foo_out,
line_pattern=RegexPattern.SYSTEMCTL_LIST_UNIT_FILES,
item_class=SystemdUnitFile,
)

compare_lists(
bc_items=all_unit_files[node_foo_name],
sc_items=foo_unit_files,
node_name=node_foo_name,
)


def test_bluechi_list_unit_files_on_a_node(
Expand Down
102 changes: 25 additions & 77 deletions ...s/tier0/bluechi-list-unit-files-on-all-nodes/test_bluechi_list_unit_files_on_all_nodes.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,97 +3,45 @@
#
# SPDX-License-Identifier: LGPL-2.1-or-later

import re
from typing import Dict

from bluechi_test.config import BluechiAgentConfig, BluechiControllerConfig
from bluechi_test.machine import BluechiAgentMachine, BluechiControllerMachine
from bluechi_test.systemd_lists import (
RegexPattern,
SystemdUnitFile,
compare_lists,
parse_bluechictl_list_output,
parse_systemctl_list_output,
)
from bluechi_test.test import BluechiTest

node_foo_name = "node-foo"
node_bar_name = "node-bar"


def parse_bluechictl_output(output: str) -> Dict[str, Dict[str, str]]:
line_pat = re.compile(
r"""\s*(?P<node_name>[\S]+)\s*\|
\s*((?:[^/]*/)*)(?P<unit_file_path>[\S]+)\s*\|
\s*(?P<enablement_status>[\S]+)\s*""",
re.VERBOSE,
)
result = {}
for line in output.splitlines():
if line.startswith("NODE ") or line.startswith("===="):
continue

match = line_pat.match(line)
if not match:
raise Exception(
f"Error parsing bluechictl list-unit-files output, invalid line: '{line}'"
)

node_unit_files = result.get(match.group("node_name"))
if not node_unit_files:
node_unit_files = {}
result[match.group("node_name")] = node_unit_files

if match.group("unit_file_path") in node_unit_files:
raise Exception(
f"Error parsing bluechictl list-unit-files output, unit file already reported, line: '{line}'"
)

node_unit_files[match.group("unit_file_path")] = match.group(
"enablement_status"
)

return result


def verify_unit_files(all_unit_files: Dict[str, str], output: str, node_name: str):
esc_seq = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
line_pat = re.compile(
r"""\s*(?P<unit_file_path>\S+)
\s+(?P<enablement_status>\S+)
\s+.*$
""",
re.VERBOSE,
)
for line in output.splitlines():
line = esc_seq.sub("", line)

match = line_pat.match(line)
if not match:
raise Exception(
f"Error parsing systemctl list-unit-files output, invalid line: '{line}'"
)

found = all_unit_files.get(match.group("unit_file_path"))
if not found or match.group("enablement_status") != found:
raise Exception(
"Unit file '{}' with enablement status '{}' reported by systemctl"
" on node '{}', but not reported by bluechictl".format(
match.group("unit_file_path"),
match.group("enablement_status"),
node_name,
)
)


def exec(ctrl: BluechiControllerMachine, nodes: Dict[str, BluechiAgentMachine]):
node_foo = nodes[node_foo_name]
node_bar = nodes[node_bar_name]

all_res, all_out = ctrl.bluechictl.list_unit_files()
assert all_res == 0
all_unit_files = parse_bluechictl_output(all_out)

foo_res, foo_out = node_foo.systemctl.list_unit_files()
assert foo_res == 0
verify_unit_files(all_unit_files[node_foo_name], foo_out, node_foo_name)
all_unit_files = parse_bluechictl_list_output(
content=all_out,
line_pattern=RegexPattern.BLUECHICTL_LIST_UNIT_FILES,
item_class=SystemdUnitFile,
)

bar_res, bar_out = node_bar.systemctl.list_unit_files()
assert bar_res == 0
verify_unit_files(all_unit_files[node_bar_name], bar_out, node_bar_name)
for node_name in nodes:
node_res, node_out = nodes[node_name].systemctl.list_unit_files()
assert node_res == 0
node_unit_files = parse_systemctl_list_output(
content=node_out,
line_pattern=RegexPattern.SYSTEMCTL_LIST_UNIT_FILES,
item_class=SystemdUnitFile,
)
compare_lists(
bc_items=all_unit_files[node_name],
sc_items=node_unit_files,
node_name=node_name,
)


def test_bluechi_list_unit_files_on_all_nodes(
Expand Down

0 comments on commit 1df9cc0

Please sign in to comment.