Introduce Service classes to minimize code duplication

Introduces Service and SimpleService to minimize code duplication around
systemd services used in integration tests.

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

tests/bluechi_test/machine.py

@@ -7,10 +7,11 @@
 
 from typing import Any, Iterator, Optional, Tuple, Union, List
 
+from bluechi_test.bluechictl import BluechiCtl
 from bluechi_test.client import Client
 from bluechi_test.config import BluechiAgentConfig, BluechiControllerConfig
+from bluechi_test.service import Service
 from bluechi_test.systemctl import SystemCtl
-from bluechi_test.bluechictl import BluechiCtl
 from bluechi_test.util import read_file, get_random_name
 
 LOGGER = logging.getLogger(__name__)
@@ -100,18 +101,15 @@ def systemctl_start_and_wait(self, service_name: str, sleep_after: float = 0.0)
 
         return result, output, wait_result
 
-    def copy_systemd_service(self, service_file_name: str, source_dir: str = "systemd",
-                             target_dir: str = os.path.join("/", "etc", "systemd", "system")):
-        source_path = os.path.join(source_dir, service_file_name)
-        content = read_file(source_path)
-
-        LOGGER.debug(f"Copy local systemd service '{source_path}' to container path '{target_dir}'\
-             with content:\n{content}")
-        self.create_file(target_dir, service_file_name, content)
+    def install_systemd_service(self, service: Service,
+                                target_dir: str = os.path.join("/", "etc", "systemd", "system")):
+        LOGGER.debug(f"Installing systemd service '{service.name}' to container path '{target_dir}'"
+                     f"with content:\n{service}")
+        self.create_file(target_dir, service.name, str(service))
         self.systemctl.daemon_reload()
 
         # keep track of created service file to potentially stop it in later cleanup
-        self.systemctl.tracked_services.append(service_file_name)
+        self.systemctl.tracked_services.append(service.name)
 
     def copy_container_script(self, script_file_name: str):
         curr_dir = os.getcwd()

tests/bluechi_test/service.py

@@ -0,0 +1,100 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+import configparser
+import enum
+import io
+
+from typing import Union
+
+
+@enum.unique
+class Section(enum.Enum):
+    # Available sections are described at https://www.freedesktop.org/software/systemd/man/latest/systemd.service.html
+    Install = "Install"
+    Service = "Service"
+    Unit = "Unit"
+
+
+@enum.unique
+class Option(enum.Enum):
+    # Available options for Unit section can be found at
+    # https://www.freedesktop.org/software/systemd/man/latest/systemd.unit.html#%5BUnit%5D%20Section%20Options
+    After = "After"
+    Description = "Description"
+    StopWhenUnneeded = "StopWhenUnneeded"
+    Wants = "Wants"
+
+    # Available options for Service section can be found at
+    # https://www.freedesktop.org/software/systemd/man/latest/systemd.service.html#Options
+    ExecReload = "ExecReload"
+    ExecStart = "ExecStart"
+    ExecStartPre = "ExecStartPre"
+    RemainAfterExit = "RemainAfterExit"
+    Type = "Type"
+
+    # Available options for Install section can be found at
+    # https://www.freedesktop.org/software/systemd/man/latest/systemd.unit.html#%5BInstall%5D%20Section%20Options
+    WantedBy = "WantedBy"
+
+
+class Service():
+    """An empty service, which only contains predefined sections"""
+    def __init__(self, name: str = None) -> None:
+        self.cp = configparser.ConfigParser()
+        # Enable case sensitivity
+        self.cp.optionxform = lambda option: option
+        self.name = name
+        for section in Section:
+            self.cp.add_section(section.value)
+
+    def __str__(self) -> str:
+        with io.StringIO() as content:
+            self.cp.write(content, space_around_delimiters=False)
+            return content.getvalue()
+
+    def set_option(self, section: Section, name: Option, value: str) -> None:
+        self.cp.set(section.value, name.value, value)
+
+    def get_option(self, section: Section, name: Option) -> Union[str, None]:
+        if self.cp.has_option(section.value, name.value):
+            return self.cp.get(section.value, name.value)
+
+        return None
+
+    def remove_option(self, section: Section, name: Option) -> None:
+        if self.cp.has_option(section.value, name.value):
+            self.cp.remove_option(section.value, name.value)
+
+
+class SimpleService(Service):
+    """A simple service, which is using /bin/true command"""
+    def __init__(self, name: str = "simple.service") -> None:
+        super(SimpleService, self).__init__(name)
+
+        self.set_option(Section.Unit, Option.Description, "A simple service")
+
+        self.set_option(Section.Service, Option.Type, "simple")
+        self.set_option(Section.Service, Option.ExecStart, "/bin/true")
+
+        self.set_option(Section.Install, Option.WantedBy, "multi-user.target")
+
+
+class SimpleRemainingService(SimpleService):
+    """A simple service, which adds RemainAfterExit enabled to SimpleService"""
+    def __init__(self, name: str = "simple.service") -> None:
+        super(SimpleRemainingService, self).__init__(name)
+
+        self.set_option(Section.Service, Option.RemainAfterExit, "yes")
+
+
+class SleepingService(Service):
+    """A simple service, which is using /bin/sleep command"""
+    def __init__(self, name: str = "sleeping.service") -> None:
+        super(SleepingService, self).__init__(name)
+
+        self.set_option(Section.Unit, Option.Description, "A sleeping service")
+
+        self.set_option(Section.Service, Option.Type, "simple")
+        self.set_option(Section.Service, Option.ExecStart, "/bin/sleep 60")
+
+        self.set_option(Section.Install, Option.WantedBy, "multi-user.target")

tests/tests/tier0/bluechi-daemon-reload/test_bluechi_deamon_reload.py

@@ -1,12 +1,13 @@
 # SPDX-License-Identifier: LGPL-2.1-or-later
 
-import os
 import logging
 
 from typing import Dict
-from bluechi_test.test import BluechiTest
-from bluechi_test.machine import BluechiControllerMachine, BluechiAgentMachine
+
 from bluechi_test.config import BluechiControllerConfig, BluechiAgentConfig
+from bluechi_test.machine import BluechiControllerMachine, BluechiAgentMachine
+from bluechi_test.service import Option, Section, SimpleRemainingService
+from bluechi_test.test import BluechiTest
 
 LOGGER = logging.getLogger(__name__)
 NODE_FOO = "node-foo"
@@ -15,17 +16,19 @@
 def exec(ctrl: BluechiControllerMachine, nodes: Dict[str, BluechiAgentMachine]):
 
     node_foo = nodes[NODE_FOO]
-    service_file = "simple.service"
-    service_path = os.path.join("/", "etc", "systemd", "system", service_file)
 
-    node_foo.copy_systemd_service(service_file)
-    ctrl.bluechictl.start_unit(NODE_FOO, service_file)
-    assert node_foo.wait_for_unit_state_to_be(service_file, "failed")
+    # Create a service which won't start due to nonexistent executable
+    service = SimpleRemainingService()
+    service.set_option(Section.Service, Option.ExecStart, "/s")
+
+    node_foo.install_systemd_service(service)
+    ctrl.bluechictl.start_unit(NODE_FOO, service.name)
+    assert node_foo.wait_for_unit_state_to_be(service.name, "failed")
 
-    node_foo.exec_run(f"sed -i '/ExecStart=/c\\ExecStart=/bin/true' {service_path}")
+    node_foo.exec_run(f"sed -i '/ExecStart=/c\\ExecStart=/bin/true' /etc/systemd/system/{service.name}")
     ctrl.bluechictl.daemon_reload_node(NODE_FOO)
-    ctrl.bluechictl.start_unit(NODE_FOO, service_file)
-    assert node_foo.wait_for_unit_state_to_be(service_file, "active")
+    ctrl.bluechictl.start_unit(NODE_FOO, service.name)
+    assert node_foo.wait_for_unit_state_to_be(service.name, "active")
 
 
 def test_bluechi_deamon_reload(

tests/tests/tier0/bluechi-enable-disable-service/test_bluechi_enable_disable_service.py

@@ -4,31 +4,32 @@
 
 from bluechi_test.config import BluechiControllerConfig, BluechiAgentConfig
 from bluechi_test.machine import BluechiControllerMachine, BluechiAgentMachine
+from bluechi_test.service import SimpleRemainingService
 from bluechi_test.test import BluechiTest
 
 
 node_foo_name = "node-foo"
-simple_service = "simple.service"
 
 
 def exec(ctrl: BluechiControllerMachine, nodes: Dict[str, BluechiAgentMachine]):
     foo = nodes[node_foo_name]
 
-    foo.copy_systemd_service(simple_service)
-    assert foo.wait_for_unit_state_to_be(simple_service, "inactive")
+    service = SimpleRemainingService()
+    foo.install_systemd_service(service)
+    assert foo.wait_for_unit_state_to_be(service.name, "inactive")
 
-    if not foo.systemctl.is_unit_disabled(simple_service, check_result=False):
-        raise Exception(f"Failed pre-check if unit {simple_service} is disabled")
+    if not foo.systemctl.is_unit_disabled(service.name, check_result=False):
+        raise Exception(f"Failed pre-check if unit {service.name} is disabled")
 
-    ctrl.bluechictl.enable_unit(node_foo_name, simple_service)
+    ctrl.bluechictl.enable_unit(node_foo_name, service.name)
 
-    if not foo.systemctl.is_unit_enabled(simple_service, check_result=False):
-        raise Exception(f"Unit {simple_service} expected to be enabled, but is not")
+    if not foo.systemctl.is_unit_enabled(service.name, check_result=False):
+        raise Exception(f"Unit {service.name} expected to be enabled, but is not")
 
-    ctrl.bluechictl.disable_unit(node_foo_name, simple_service)
+    ctrl.bluechictl.disable_unit(node_foo_name, service.name)
 
-    if not foo.systemctl.is_unit_disabled(simple_service, check_result=False):
-        raise Exception(f"Unit {simple_service} expected to be disabled, but is not")
+    if not foo.systemctl.is_unit_disabled(service.name, check_result=False):
+        raise Exception(f"Unit {service.name} expected to be disabled, but is not")
 
 
 def test_proxy_service_start(

tests/tests/tier0/bluechi-freeze-and-thaw-service/test_bluechi_freeze_and_thaw_service.py

@@ -4,41 +4,28 @@
 
 from bluechi_test.config import BluechiControllerConfig, BluechiAgentConfig
 from bluechi_test.machine import BluechiControllerMachine, BluechiAgentMachine
+from bluechi_test.service import SleepingService
 from bluechi_test.test import BluechiTest
 
 
 node_foo_name = "node-foo"
-simple_service = "simple.service"
-
-
-def verify_service_start(foo: BluechiAgentMachine):
-    assert foo.wait_for_unit_state_to_be(simple_service, "active")
-
-
-def verify_service_freeze(foo: BluechiAgentMachine):
-    output = foo.systemctl.get_unit_freezer_state(simple_service)
-    assert output == 'frozen'
-
-
-def verify_service_thaw(foo: BluechiAgentMachine):
-    output = foo.systemctl.get_unit_freezer_state(simple_service)
-    assert output == 'running'
 
 
 def exec(ctrl: BluechiControllerMachine, nodes: Dict[str, BluechiAgentMachine]):
     foo = nodes[node_foo_name]
+    service = SleepingService()
 
-    foo.copy_systemd_service(simple_service)
-    assert foo.wait_for_unit_state_to_be(simple_service, "inactive")
+    foo.install_systemd_service(service)
+    assert foo.wait_for_unit_state_to_be(service.name, "inactive")
 
-    ctrl.bluechictl.start_unit(node_foo_name, simple_service)
-    verify_service_start(foo)
+    ctrl.bluechictl.start_unit(node_foo_name, service.name)
+    assert foo.wait_for_unit_state_to_be(service.name, "active")
 
-    ctrl.bluechictl.freeze_unit(node_foo_name, simple_service)
-    verify_service_freeze(foo)
+    ctrl.bluechictl.freeze_unit(node_foo_name, service.name)
+    assert 'frozen' == foo.systemctl.get_unit_freezer_state(service.name)
 
-    ctrl.bluechictl.thaw_unit(node_foo_name, simple_service)
-    verify_service_thaw(foo)
+    ctrl.bluechictl.thaw_unit(node_foo_name, service.name)
+    assert 'running' == foo.systemctl.get_unit_freezer_state(service.name)
 
 
 def test_service_freeze_and_thaw(

tests/tests/tier0/bluechi-job-cancel/test_bluechi_job_cancel.py

@@ -5,21 +5,21 @@
 
 from bluechi_test.config import BluechiControllerConfig, BluechiAgentConfig
 from bluechi_test.machine import BluechiControllerMachine, BluechiAgentMachine
+from bluechi_test.service import Option, Section, SimpleRemainingService
 from bluechi_test.test import BluechiTest
 
 
 node_foo_name = "node-foo"
-simple_service = "simple.service"
 
 
 def exec(ctrl: BluechiControllerMachine, nodes: Dict[str, BluechiAgentMachine]):
     foo = nodes[node_foo_name]
 
-    source_dir = os.path.join(".", "systemd")
-    target_dir = os.path.join("/", "etc", "systemd", "system")
+    simple_service = SimpleRemainingService()
+    simple_service.set_option(Section.Service, Option.ExecStartPre, "/bin/sleep 20")
 
-    foo.copy_systemd_service(simple_service, source_dir, target_dir)
-    assert foo.wait_for_unit_state_to_be(simple_service, "inactive")
+    foo.install_systemd_service(simple_service)
+    assert foo.wait_for_unit_state_to_be(simple_service.name, "inactive")
 
     result, output = ctrl.run_python(os.path.join("python", "start_and_cancel.py"))
     if result != 0:

tests/tests/tier0/bluechi-monitor-service/test_bluechi_monitor_service.py

@@ -3,20 +3,21 @@
 import os
 from typing import Dict
 
-from bluechi_test.test import BluechiTest
-from bluechi_test.machine import BluechiControllerMachine, BluechiAgentMachine
 from bluechi_test.config import BluechiControllerConfig, BluechiAgentConfig
+from bluechi_test.machine import BluechiControllerMachine, BluechiAgentMachine
+from bluechi_test.service import SimpleRemainingService
+from bluechi_test.test import BluechiTest
 
 node_name_foo = "node-foo"
-service_simple = "simple.service"
 
 
 def exec(ctrl: BluechiControllerMachine, nodes: Dict[str, BluechiAgentMachine]):
-    nodes[node_name_foo].copy_systemd_service(service_simple)
-    assert nodes[node_name_foo].wait_for_unit_state_to_be(service_simple, "inactive")
+    service = SimpleRemainingService()
+    nodes[node_name_foo].install_systemd_service(service)
+    assert nodes[node_name_foo].wait_for_unit_state_to_be(service.name, "inactive")
 
-    ctrl.bluechictl.start_unit(node_name_foo, service_simple)
-    nodes[node_name_foo].wait_for_unit_state_to_be(service_simple, "active")
+    ctrl.bluechictl.start_unit(node_name_foo, service.name)
+    nodes[node_name_foo].wait_for_unit_state_to_be(service.name, "active")
 
     result, output = ctrl.run_python(os.path.join("python", "monitor.py"))
     if result != 0: