cephadm: move extract_uid_gid func to container_types module

While extract_uid_gid isn't a perfect fit for container_types it is a
fairly fundamental function for working with containers in cephadm and
doesn't require anything beyond types in containers_types and that
module's existing imports.  Moving extract_uid_gid should allow us to
more easily move other functions in the future.

Signed-off-by: John Mulligan <[email protected]>

Author: phlogistonjohn

src/cephadm/cephadm.py

@@ -153,6 +153,7 @@
     CephContainer,
     InitContainer,
     is_container_running,
+    extract_uid_gid,
 )
 from cephadmlib.decorators import (
     deprecated_command,
@@ -2828,37 +2829,6 @@ def _update_container_args_for_podman(
     )
 
 
-def extract_uid_gid(ctx, img='', file_path='/var/lib/ceph'):
-    # type: (CephadmContext, str, Union[str, List[str]]) -> Tuple[int, int]
-
-    if not img:
-        img = ctx.image
-
-    if isinstance(file_path, str):
-        paths = [file_path]
-    else:
-        paths = file_path
-
-    ex: Optional[Tuple[str, RuntimeError]] = None
-
-    for fp in paths:
-        try:
-            out = CephContainer(
-                ctx,
-                image=img,
-                entrypoint='stat',
-                args=['-c', '%u %g', fp]
-            ).run(verbosity=CallVerbosity.QUIET_UNLESS_ERROR)
-            uid, gid = out.split(' ')
-            return int(uid), int(gid)
-        except RuntimeError as e:
-            ex = (fp, e)
-    if ex:
-        raise Error(f'Failed to extract uid/gid for path {ex[0]}: {ex[1]}')
-
-    raise RuntimeError('uid/gid not found')
-
-
 def deploy_daemon(
     ctx: CephadmContext,
     ident: 'DaemonIdentity',

src/cephadm/cephadmlib/container_types.py

@@ -2,7 +2,7 @@
 
 import os
 
-from typing import Dict, List, Optional, Any
+from typing import Dict, List, Optional, Any, Union, Tuple
 
 from .call_wrappers import call, call_throws, CallVerbosity
 from .constants import DEFAULT_TIMEOUT
@@ -485,3 +485,34 @@ def get_running_container_name(
         if out.strip() == 'running':
             return name
     return None
+
+
+def extract_uid_gid(ctx, img='', file_path='/var/lib/ceph'):
+    # type: (CephadmContext, str, Union[str, List[str]]) -> Tuple[int, int]
+
+    if not img:
+        img = ctx.image
+
+    if isinstance(file_path, str):
+        paths = [file_path]
+    else:
+        paths = file_path
+
+    ex: Optional[Tuple[str, RuntimeError]] = None
+
+    for fp in paths:
+        try:
+            out = CephContainer(
+                ctx,
+                image=img,
+                entrypoint='stat',
+                args=['-c', '%u %g', fp]
+            ).run(verbosity=CallVerbosity.QUIET_UNLESS_ERROR)
+            uid, gid = out.split(' ')
+            return int(uid), int(gid)
+        except RuntimeError as e:
+            ex = (fp, e)
+    if ex:
+        raise Error(f'Failed to extract uid/gid for path {ex[0]}: {ex[1]}')
+
+    raise RuntimeError('uid/gid not found')

src/cephadm/tests/test_ingress.py

@@ -166,7 +166,7 @@ def test_haproxy_extract_uid_gid_haproxy():
             good_haproxy_json(),
             SAMPLE_HAPROXY_IMAGE,
         )
-        with mock.patch("cephadm.CephContainer") as cc:
+        with mock.patch("cephadmlib.container_types.CephContainer") as cc:
             cc.return_value.run.return_value = "500 500"
             uid, gid = hap.uid_gid(ctx)
             cc.return_value.run.assert_called()
@@ -329,7 +329,7 @@ def test_keepalived_extract_uid_gid_keepalived():
             good_keepalived_json(),
             SAMPLE_KEEPALIVED_IMAGE,
         )
-        with mock.patch("cephadm.CephContainer") as cc:
+        with mock.patch("cephadmlib.container_types.CephContainer") as cc:
             cc.return_value.run.return_value = "500 500"
             uid, gid = kad.uid_gid(ctx)
             cc.return_value.run.assert_called()