diff --git a/test/test_modules.py b/test/test_modules.py index 1057c646..8ceeb325 100644 --- a/test/test_modules.py +++ b/test/test_modules.py @@ -33,26 +33,31 @@ ] ) +# content: ssh version, release shortcut, service name +ssh_pkg_info = { + "rockylinux9": ("8.", ".el9", "sshd"), + "debian_bookworm": ("1:9.2", None, "ssh"), +} + +# content: distribution, codename, architecture, release_regex +docker_image_info = { + "rockylinux9": ("rocky", None, "x86_64", r"^9.\d+$"), + "debian_bookworm": ("debian", "bookworm", "amd64", r"^12"), +} + @all_images def test_package(host, docker_image): assert not host.package("zsh").is_installed ssh = host.package("openssh-server") - version = { - "rockylinux9": "8.", - "debian_bookworm": "1:9.2", - }[docker_image] + ssh_version, sshd_release = ssh_pkg_info[docker_image][:2] assert ssh.is_installed - assert ssh.version.startswith(version) - release = { - "rockylinux9": ".el9", - "debian_bookworm": None, - }[docker_image] - if release is None: + assert ssh.version.startswith(ssh_version) + if sshd_release is None: with pytest.raises(NotImplementedError): ssh.release # noqa: B018 else: - assert release in ssh.release + assert sshd_release in ssh.release def test_held_package(host): @@ -102,38 +107,46 @@ def test_uninstalled_package_version(host): def test_systeminfo(host, docker_image): assert host.system_info.type == "linux" - release, distribution, codename, arch = { - "rockylinux9": (r"^9.\d+$", "rocky", None, "x86_64"), - "debian_bookworm": (r"^12", "debian", "bookworm", "x86_64"), - }[docker_image] - + distribution, codename, unused_arch, release_regex = docker_image_info[docker_image] assert host.system_info.distribution == distribution assert host.system_info.codename == codename - assert re.match(release, host.system_info.release) + assert re.match(release_regex, host.system_info.release) @all_images def test_ssh_service(host, docker_image): - name = "sshd" if docker_image == "rockylinux9" else "ssh" - ssh = host.service(name) + service_name = ssh_pkg_info[docker_image][2] + ssh_svc = host.service(service_name) # wait at max 10 seconds for ssh is running for _ in range(10): - if ssh.is_running: + if ssh_svc.is_running: break time.sleep(1) else: raise AssertionError("ssh is not running") - assert ssh.is_enabled + assert ssh_svc.is_enabled + + +@all_images +def test_systemdservice_exists(host, docker_image): + service_name = ssh_pkg_info[docker_image][2] + for name in [service_name, f"{service_name}.service"]: + ssh_svc = host.service(name) + assert ssh_svc.exists + + for name in ["non-existing", "non-existing.service", "non-existing.timer"]: + non_existing_service = host.service(name) + assert not non_existing_service.exists def test_service_systemd_mask(host): - ssh = host.service("ssh") - assert not ssh.is_masked + ssh_svc = host.service("ssh") + assert not ssh_svc.is_masked host.run("systemctl mask ssh") - assert ssh.is_masked + assert ssh_svc.is_masked host.run("systemctl unmask ssh") - assert not ssh.is_masked + assert not ssh_svc.is_masked def test_salt(host): diff --git a/testinfra/modules/service.py b/testinfra/modules/service.py index 24e666cb..fd83afe2 100644 --- a/testinfra/modules/service.py +++ b/testinfra/modules/service.py @@ -34,22 +34,22 @@ def __init__(self, name): super().__init__() @property - def exists(self): + def exists(self) -> bool: """Test if the service exists""" raise NotImplementedError @property - def is_running(self): + def is_running(self) -> bool: """Test if service is running""" raise NotImplementedError @property - def is_enabled(self): + def is_enabled(self) -> bool: """Test if service is enabled""" raise NotImplementedError @property - def is_valid(self): + def is_valid(self) -> bool: """Test if service is valid This method is only available in the systemd implementation, @@ -58,7 +58,7 @@ def is_valid(self): raise NotImplementedError @property - def is_masked(self): + def is_masked(self) -> bool: """Test if service is masked This method is only available in the systemd implementation, @@ -67,7 +67,7 @@ def is_masked(self): raise NotImplementedError @functools.cached_property - def systemd_properties(self): + def systemd_properties(self) -> dict[str, str]: """Properties of the service (unit). Return service properties as a `dict`, @@ -169,19 +169,25 @@ class SystemdService(SysvService): def _has_systemd_suffix(self): """ - Check if service name has a known systemd unit suffix + Check if the service name has a known systemd unit suffix """ unit_suffix = self.name.split(".")[-1] return unit_suffix in self.suffix_list @property def exists(self): - cmd = self.run_test('systemctl list-unit-files | grep -q "^%s"', self.name) - return cmd.rc == 0 + # systemctl return codes based on https://man7.org/linux/man-pages/man1/systemctl.1.html: + # 0: unit is active + # 1: unit not failed (used by is-failed) + # 2: unused + # 3: unit is not active + # 4: no such unit + cmd = self.run_expect([0, 1, 3, 4], "systemctl status %s", self.name) + return cmd.rc < 4 @property def is_running(self): - # based on https://man7.org/linux/man-pages/man1/systemctl.1.html + # systemctl return codes based on https://man7.org/linux/man-pages/man1/systemctl.1.html: # 0: program running # 1: program is dead and pid file exists # 3: not running and pid file does not exists