Add a new service plugin

Signed-off-by: Ronan Abhamon <[email protected]>

SOURCES/etc/xapi.d/plugins/service.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import json
+import sys
+import XenAPIPlugin
+
+sys.path.append('.')
+from xcpngutils import configure_logging, run_command, error_wrapped
+
+def run_service_command(cmd_name, args):
+    service = args.get('service')
+    if not service:
+        raise Exception('Missing or empty argument `service`')
+    run_command(['systemctl', cmd_name, service])
+
+@error_wrapped
+def start_service(session, args):
+    run_service_command('start', args)
+    return json.dumps(True)
+
+@error_wrapped
+def stop_service(session, args):
+    run_service_command('stop', args)
+    return json.dumps(True)
+
+@error_wrapped
+def restart_service(session, args):
+    run_service_command('restart', args)
+    return json.dumps(True)
+
+@error_wrapped
+def try_restart_service(session, args):
+    run_service_command('try-start', args)
+    return json.dumps(True)
+
+_LOGGER = configure_logging('service')
+if __name__ == "__main__":
+    XenAPIPlugin.dispatch({
+        'start_service': start_service,
+        'stop_service': stop_service,
+        'restart_service': restart_service,
+        'try_restart_service': try_restart_service
+    })

tests/test_service.py

@@ -0,0 +1,43 @@
+import mock
+import pytest
+import XenAPIPlugin
+
+from service import start_service, stop_service, restart_service
+
+@mock.patch("service.run_command", autospec=True)
+class TestService:
+    SERVICE = 'linstor-satellite'
+
+    def _test_command(self, cmd, cmd_name, run_command):
+        cmd(mock.MagicMock(), {'service': self.SERVICE})
+        run_command.assert_called_once_with(['systemctl', cmd_name, self.SERVICE])
+
+    def _test_command_without_service(self, cmd):
+        with pytest.raises(XenAPIPlugin.Failure) as e:
+            start_service(mock.MagicMock(), {})
+        assert e.value.params[0] == '-1'
+        assert e.value.params[1] == 'Missing or empty argument `service`'
+
+    def test_start(self, run_command):
+        self._test_command(start_service, 'start', run_command)
+
+    def test_stop(self, run_command):
+        self._test_command(stop_service, 'stop', run_command)
+
+    def test_restart(self, run_command):
+        self._test_command(restart_service, 'restart', run_command)
+
+    def test_try_restart(self, run_command):
+        self._test_command(try_restart_service, 'try-restart', run_command)
+
+    def test_start_without_service(self, run_command):
+        self._test_command_without_service(start_service)
+
+    def test_stop_without_service(self, run_command):
+        self._test_command_without_service(stop_service)
+
+    def test_restart_without_service(self, run_command):
+        self._test_command_without_service(restart_service)
+
+    def test_try_restart_without_service(self, run_command):
+        self._test_command_without_service(try_restart_service)