From 52ad060bb818f171455f2a3fc92915240a5418e5 Mon Sep 17 00:00:00 2001 From: Carsten Grohmann Date: Mon, 7 Aug 2023 12:40:11 +0200 Subject: [PATCH 1/5] Extract constants from functions to file level --- test/test_modules.py | 35 ++++++++++++++++++----------------- 1 file changed, 18 insertions(+), 17 deletions(-) diff --git a/test/test_modules.py b/test/test_modules.py index 1057c646..265f869c 100644 --- a/test/test_modules.py +++ b/test/test_modules.py @@ -33,26 +33,31 @@ ] ) +# content: ssh version, release shortcut, +ssh_pkg_info = { + "rockylinux9": ("8.", ".el9"), + "debian_bookworm": ("1:9.2", None), +} + +# 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] 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,14 +107,10 @@ 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 From 77234ec1f08e7eec156f0a6c4fae370753eff1be Mon Sep 17 00:00:00 2001 From: Carsten Grohmann Date: Mon, 22 Jan 2024 15:14:35 +0100 Subject: [PATCH 2/5] Move ssh service name to ssh_pkg_info variable --- test/test_modules.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/test/test_modules.py b/test/test_modules.py index 265f869c..f57ec9a5 100644 --- a/test/test_modules.py +++ b/test/test_modules.py @@ -33,10 +33,10 @@ ] ) -# content: ssh version, release shortcut, +# content: ssh version, release shortcut, service name ssh_pkg_info = { - "rockylinux9": ("8.", ".el9"), - "debian_bookworm": ("1:9.2", None), + "rockylinux9": ("8.", ".el9", "sshd"), + "debian_bookworm": ("1:9.2", None, "ssh"), } # content: distribution, codename, architecture, release_regex @@ -50,7 +50,7 @@ def test_package(host, docker_image): assert not host.package("zsh").is_installed ssh = host.package("openssh-server") - ssh_version, sshd_release = ssh_pkg_info[docker_image] + ssh_version, sshd_release = ssh_pkg_info[docker_image][:2] assert ssh.is_installed assert ssh.version.startswith(ssh_version) if sshd_release is None: @@ -115,8 +115,8 @@ def test_systeminfo(host, docker_image): @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 = host.service(service_name) # wait at max 10 seconds for ssh is running for _ in range(10): if ssh.is_running: From e7a99829646c1a651a5ccd9dea775b1e3d1abd4d Mon Sep 17 00:00:00 2001 From: Carsten Grohmann Date: Tue, 23 Jan 2024 08:07:38 +0100 Subject: [PATCH 3/5] Add test for SystemdService.exists --- test/test_modules.py | 26 +++++++++++++++++++------- 1 file changed, 19 insertions(+), 7 deletions(-) diff --git a/test/test_modules.py b/test/test_modules.py index f57ec9a5..8ceeb325 100644 --- a/test/test_modules.py +++ b/test/test_modules.py @@ -116,25 +116,37 @@ def test_systeminfo(host, docker_image): @all_images def test_ssh_service(host, docker_image): service_name = ssh_pkg_info[docker_image][2] - ssh = host.service(service_name) + 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): From b179062e3dda635d63d55bb9a1497c024e6525e3 Mon Sep 17 00:00:00 2001 From: Carsten Grohmann Date: Tue, 23 Jan 2024 08:36:54 +0100 Subject: [PATCH 4/5] Update SystemdService.exists to find all unit files --- testinfra/modules/service.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/testinfra/modules/service.py b/testinfra/modules/service.py index 24e666cb..2eea3fdf 100644 --- a/testinfra/modules/service.py +++ b/testinfra/modules/service.py @@ -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 From 9b09891595b6e9b626bce6c6a1d7af05ec4dbb51 Mon Sep 17 00:00:00 2001 From: Carsten Grohmann Date: Fri, 4 Apr 2025 17:50:53 +0200 Subject: [PATCH 5/5] Add type hints to Service properties --- testinfra/modules/service.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/testinfra/modules/service.py b/testinfra/modules/service.py index 2eea3fdf..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`,