Skip to content
Merged
Show file tree
Hide file tree
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
11 changes: 11 additions & 0 deletions .github/dependabot.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
---
version: 2
updates:
- package-ecosystem: "github-actions"
directory: "/"
schedule:
interval: "daily"
- package-ecosystem: "gitsubmodule"
directory: "/"
schedule:
interval: "daily"
4 changes: 1 addition & 3 deletions .github/workflows/ci.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,4 @@ on:

jobs:
call_ci:
uses: EffectiveRange/ci-workflows/.github/workflows/python-ci.yaml@test
with:
coverage-threshold: 95
uses: EffectiveRange/ci-workflows/.github/workflows/python-ci.yaml@v4
12 changes: 12 additions & 0 deletions systemd_dbus/systemd.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,9 @@ def get_service_properties(self, service_name: str) -> Any:
def get_service_file_properties(self, service_name: str) -> Any:
raise NotImplementedError()

def list_service_names(self, states: Optional[list[str]], patterns: Optional[list[str]]) -> list[str]:
raise NotImplementedError()

def reload_daemon(self) -> bool:
raise NotImplementedError()

Expand Down Expand Up @@ -204,6 +207,15 @@ def get_service_properties(self, service_name: str) -> Any:
def get_service_file_properties(self, service_name: str) -> Any:
return self._get_service_properties(service_name, self.SYSTEMD_UNIT_INTERFACE)

def list_service_names(self, states: Optional[list[str]] = None, patterns: Optional[list[str]] = None) -> list[str]:
try:
interface = self._get_interface()
units = interface.ListUnitsByPatterns(states or [], patterns or [])
return [str(unit[0]) for unit in units]
except DBusException as error:
log.error('Failed to list service names', reason=error)
return []

def reload_daemon(self) -> bool:
method = 'Reload'

Expand Down
53 changes: 53 additions & 0 deletions tests/SystemdDbusTest.py
Original file line number Diff line number Diff line change
Expand Up @@ -493,6 +493,59 @@ def test_returns_service_file_properties(self):
system_bus.get_object().get_dbus_method.assert_called_with('GetAll', 'org.freedesktop.DBus.Properties')
system_bus.get_object().get_dbus_method().assert_called_with('org.freedesktop.systemd1.Unit')

def test_returns_service_names(self):
# Given
system_bus = MagicMock(spec=dbus.SystemBus)
system_bus.get_object().get_dbus_method().return_value = dbus.Array([
dbus.Struct([dbus.String('test1.service'), dbus.String('Test1 Service'), dbus.String('active')]),
dbus.Struct([dbus.String('test2.service'), dbus.String('Test2 Service'), dbus.String('inactive')]),
], signature='s')

systemd = SystemdDbus(system_bus)

# When
result = systemd.list_service_names()

# Then
self.assertEqual(['test1.service', 'test2.service'], result)
system_bus.get_object().get_dbus_method.assert_called_with('ListUnitsByPatterns',
'org.freedesktop.systemd1.Manager')
system_bus.get_object().get_dbus_method().assert_called_with([], [])

def test_returns_service_names_when_filtered(self):
# Given
system_bus = MagicMock(spec=dbus.SystemBus)
system_bus.get_object().get_dbus_method().return_value = dbus.Array([
dbus.Struct([dbus.String('test1.service'), dbus.String('Test1 Service'), dbus.String('active')])
], signature='s')

systemd = SystemdDbus(system_bus)

# When
result = systemd.list_service_names(['active'], ['test*.service'])

# Then
self.assertEqual(['test1.service'], result)
system_bus.get_object().get_dbus_method.assert_called_with('ListUnitsByPatterns',
'org.freedesktop.systemd1.Manager')
system_bus.get_object().get_dbus_method().assert_called_with(['active'], ['test*.service'])

def test_returns_empty_list_when_fails_to_list_units(self):
# Given
system_bus = MagicMock(spec=dbus.SystemBus)
system_bus.get_object().get_dbus_method().side_effect = DBusException('Failure')

systemd = SystemdDbus(system_bus)

# When
result = systemd.list_service_names()

# Then
self.assertEqual([], result)
system_bus.get_object().get_dbus_method.assert_called_with('ListUnitsByPatterns',
'org.freedesktop.systemd1.Manager')
system_bus.get_object().get_dbus_method().assert_called_with([], [])

def test_returns_true_when_systemd_daemon_is_reloaded(self):
# Given
system_bus = MagicMock(spec=dbus.SystemBus)
Expand Down
Loading