Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feat: add scoped systemd service support #762

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 53 additions & 7 deletions testinfra/modules/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import enum
import functools

from testinfra.modules.base import Module
Expand Down Expand Up @@ -147,6 +147,11 @@ def is_enabled(self):
)


class SystemdServiceScope(enum.Enum):
SYSTEM = "--system"
USER = "--user"


class SystemdService(SysvService):
suffix_list = [
"service",
Expand All @@ -167,6 +172,23 @@ class SystemdService(SysvService):
See systemd.unit(5) for more details
"""

def __init__(self, name,
service_scope=SystemdServiceScope.SYSTEM,
user="",
):
self.name = name
machine = ""
if user:
systemctl_user_cmd_line = "-M {}@{}".format(user, machine)
else:
systemctl_user_cmd_line = ""
self.systemctl_scope_cmd_line = "{} {}".format(
service_scope.value,
systemctl_user_cmd_line
)

super().__init__(name)

def _has_systemd_suffix(self):
"""
Check if service name has a known systemd unit suffix
Expand All @@ -176,20 +198,34 @@ def _has_systemd_suffix(self):

@property
def exists(self):
cmd = self.run_test('systemctl list-unit-files | grep -q"^%s"', self.name)
cmd = self.run_test(
'systemctl {} list-unit-files | grep -q "^{}"'.format(
self.systemctl_scope_cmd_line,
self.name)
)
return cmd.rc == 0

@property
def is_running(self):
out = self.run_expect([0, 1, 3], "systemctl is-active %s", self.name)
out = self.run_expect(
[0, 1, 3],
"systemctl {} is-active {}".format(
self.systemctl_scope_cmd_line,
self.name)
)
if out.rc == 1:
# Failed to connect to bus: No such file or directory
return super().is_running
return out.rc == 0

@property
def is_enabled(self):
cmd = self.run_test("systemctl is-enabled %s", self.name)
cmd = self.run_test(
"systemctl {} is-enabled {}".format(
self.systemctl_scope_cmd_line,
self.name,
)
)
if cmd.rc == 0:
return True
if cmd.stdout.strip() == "disabled":
Expand All @@ -211,7 +247,10 @@ def is_valid(self):
name = self.name
else:
name = self.name + ".service"
cmd = self.run("systemd-analyze verify %s", name)
cmd = self.run("systemd-analyze {} verify {}".format(
self.systemctl_scope_cmd_line,
name)
)
# A bad unit file still returns a rc of 0, so check the
# stdout for anything. Nothing means no warns/errors.
# Docs at https://www.freedesktop.org/software/systemd/man/systemd
Expand All @@ -221,12 +260,19 @@ def is_valid(self):

@property
def is_masked(self):
cmd = self.run_test("systemctl is-enabled %s", self.name)
cmd = self.run_test("systemctl {} is-enabled {}".format(
self.systemctl_scope_cmd_line,
self.name)
)
return cmd.stdout.strip() == "masked"

@functools.cached_property
def systemd_properties(self):
out = self.check_output("systemctl show %s", self.name)
out = self.check_output("systemctl {} show {}".format(
self.systemctl_scope_cmd_line,
self.name
)
)
out_d = {}
if out:
# maxsplit is required because values can contain `=`
Expand Down