cephadm: move a pair of systemd unit status funcs to systemd.py

Signed-off-by: John Mulligan <jmulligan@redhat.com>

diff --git a/src/cephadm/cephadm.py b/src/cephadm/cephadm.py
index adcaa8ebe4b09deefda5583a8b4eba35cf84c1ce..de3b3fd08992651535b8f83913f1933668342f9d 100755 (executable)
--- a/src/cephadm/cephadm.py
+++ b/src/cephadm/cephadm.py
@@ -138,6 +138,7 @@ from cephadmlib.locking import FileLock
 from cephadmlib.daemon_identity import DaemonIdentity, DaemonSubIdentity
 from cephadmlib.packagers import create_packager, Packager
 from cephadmlib.logging import cephadm_init_logging, Highlight, LogDestination
+from cephadmlib.systemd import check_unit, check_units
 
 FuncT = TypeVar('FuncT', bound=Callable)
 
@@ -1905,59 +1906,6 @@ def get_unit_name_by_daemon_name(ctx: CephadmContext, fsid: str, name: str) -> s
         raise Error('Failed to get unit name for {}'.format(daemon))
 
 
-def check_unit(ctx, unit_name):
-    # type: (CephadmContext, str) -> Tuple[bool, str, bool]
-    # NOTE: we ignore the exit code here because systemctl outputs
-    # various exit codes based on the state of the service, but the
-    # string result is more explicit (and sufficient).
-    enabled = False
-    installed = False
-    try:
-        out, err, code = call(ctx, ['systemctl', 'is-enabled', unit_name],
-                              verbosity=CallVerbosity.QUIET)
-        if code == 0:
-            enabled = True
-            installed = True
-        elif 'disabled' in out:
-            installed = True
-    except Exception as e:
-        logger.warning('unable to run systemctl: %s' % e)
-        enabled = False
-        installed = False
-
-    state = 'unknown'
-    try:
-        out, err, code = call(ctx, ['systemctl', 'is-active', unit_name],
-                              verbosity=CallVerbosity.QUIET)
-        out = out.strip()
-        if out in ['active']:
-            state = 'running'
-        elif out in ['inactive']:
-            state = 'stopped'
-        elif out in ['failed', 'auto-restart']:
-            state = 'error'
-        else:
-            state = 'unknown'
-    except Exception as e:
-        logger.warning('unable to run systemctl: %s' % e)
-        state = 'unknown'
-    return (enabled, state, installed)
-
-
-def check_units(ctx, units, enabler=None):
-    # type: (CephadmContext, List[str], Optional[Packager]) -> bool
-    for u in units:
-        (enabled, state, installed) = check_unit(ctx, u)
-        if enabled and state == 'running':
-            logger.info('Unit %s is enabled and running' % u)
-            return True
-        if enabler is not None:
-            if installed:
-                logger.info('Enabling unit %s' % u)
-                enabler.enable_service(u)
-    return False
-
-
 def is_container_running(ctx: CephadmContext, c: 'CephContainer') -> bool:
     if ctx.name.split('.', 1)[0] in ['agent', 'cephadm-exporter']:
         # these are non-containerized daemon types

diff --git a/src/cephadm/cephadmlib/systemd.py b/src/cephadm/cephadmlib/systemd.py
new file mode 100644 (file)
index 0000000..e48f1dd
--- /dev/null
+++ b/src/cephadm/cephadmlib/systemd.py
@@ -0,0 +1,64 @@
+# systemd.py - general systemd related types and funcs
+
+import logging
+
+from typing import Tuple, List, Optional
+
+from .context import CephadmContext
+from .call_wrappers import call, CallVerbosity
+from .packagers import Packager
+
+logger = logging.getLogger()
+
+
+def check_unit(ctx, unit_name):
+    # type: (CephadmContext, str) -> Tuple[bool, str, bool]
+    # NOTE: we ignore the exit code here because systemctl outputs
+    # various exit codes based on the state of the service, but the
+    # string result is more explicit (and sufficient).
+    enabled = False
+    installed = False
+    try:
+        out, err, code = call(ctx, ['systemctl', 'is-enabled', unit_name],
+                              verbosity=CallVerbosity.QUIET)
+        if code == 0:
+            enabled = True
+            installed = True
+        elif 'disabled' in out:
+            installed = True
+    except Exception as e:
+        logger.warning('unable to run systemctl: %s' % e)
+        enabled = False
+        installed = False
+
+    state = 'unknown'
+    try:
+        out, err, code = call(ctx, ['systemctl', 'is-active', unit_name],
+                              verbosity=CallVerbosity.QUIET)
+        out = out.strip()
+        if out in ['active']:
+            state = 'running'
+        elif out in ['inactive']:
+            state = 'stopped'
+        elif out in ['failed', 'auto-restart']:
+            state = 'error'
+        else:
+            state = 'unknown'
+    except Exception as e:
+        logger.warning('unable to run systemctl: %s' % e)
+        state = 'unknown'
+    return (enabled, state, installed)
+
+
+def check_units(ctx, units, enabler=None):
+    # type: (CephadmContext, List[str], Optional[Packager]) -> bool
+    for u in units:
+        (enabled, state, installed) = check_unit(ctx, u)
+        if enabled and state == 'running':
+            logger.info('Unit %s is enabled and running' % u)
+            return True
+        if enabler is not None:
+            if installed:
+                logger.info('Enabling unit %s' % u)
+                enabler.enable_service(u)
+    return False

diff --git a/src/cephadm/tests/test_util_funcs.py b/src/cephadm/tests/test_util_funcs.py
index 6f71dd2d4291d78df6c3bfdfc2f6180ae6cae84c..ffcf3909c4ee15f670278c1784aecb26815abe78 100644 (file)
--- a/src/cephadm/tests/test_util_funcs.py
+++ b/src/cephadm/tests/test_util_funcs.py
@@ -411,11 +411,12 @@ def _mk_fake_call(enabled, active):
 )
 def test_check_unit(enabled_out, active_out, expected):
     with with_cephadm_ctx([]) as ctx:
-        _cephadm.call.side_effect = _mk_fake_call(
-            enabled=enabled_out,
-            active=active_out,
-        )
-        enabled, state, installed = _cephadm.check_unit(ctx, "foobar")
+        with mock.patch('cephadmlib.systemd.call') as _call:
+            _call.side_effect = _mk_fake_call(
+                enabled=enabled_out,
+                active=active_out,
+            )
+            enabled, state, installed = _cephadm.check_unit(ctx, "foobar")
     assert (enabled, state, installed) == expected
 
 
@@ -489,11 +490,12 @@ def test_check_time_sync(call_fn, enabler, expected):
     is enabled. It is also the only consumer of check_units.
     """
     with with_cephadm_ctx([]) as ctx:
-        _cephadm.call.side_effect = call_fn
-        result = _cephadm.check_time_sync(ctx, enabler=enabler)
-        assert result == expected
-        if enabler is not None:
-            enabler.check_expected()
+        with mock.patch('cephadmlib.systemd.call') as _call:
+            _call.side_effect = call_fn
+            result = _cephadm.check_time_sync(ctx, enabler=enabler)
+            assert result == expected
+            if enabler is not None:
+                enabler.check_expected()
 
 
 @pytest.mark.parametrize(