Update check in SystemdService.exists and add tests

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):

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