LOGROTATE_DIR,
LOG_DIR,
LOG_DIR_MODE,
- MIN_PODMAN_VERSION,
NO_DEPRECATED,
PIDS_LIMIT_UNLIMITED_PODMAN_VERSION,
QUIET_LOG_LEVEL,
call_timeout,
concurrent_tasks,
)
-from cephadmlib.container_engine_base import ContainerEngine
+from cephadmlib.container_engines import (
+ Docker,
+ Podman,
+ check_container_engine,
+ find_container_engine,
+)
FuncT = TypeVar('FuncT', bound=Callable)
RECONFIG = 'Reconfig'
-class Podman(ContainerEngine):
- EXE = 'podman'
-
- def __init__(self) -> None:
- super().__init__()
- self._version: Optional[Tuple[int, ...]] = None
-
- @property
- def version(self) -> Tuple[int, ...]:
- if self._version is None:
- raise RuntimeError('Please call `get_version` first')
- return self._version
-
- def get_version(self, ctx: CephadmContext) -> None:
- out, _, _ = call_throws(ctx, [self.path, 'version', '--format', '{{.Client.Version}}'], verbosity=CallVerbosity.QUIET)
- self._version = _parse_podman_version(out)
-
- def __str__(self) -> str:
- version = '.'.join(map(str, self.version))
- return f'{self.EXE} ({self.path}) version {version}'
-
-
-class Docker(ContainerEngine):
- EXE = 'docker'
-
-
-CONTAINER_PREFERENCE = (Podman, Docker) # prefer podman to docker
-
-
# During normal cephadm operations (cephadm ls, gather-facts, etc ) we use:
# stdout: for JSON output only
# stderr: for error, debug, info, etc
return None
-def _parse_podman_version(version_str):
- # type: (str) -> Tuple[int, ...]
- def to_int(val: str, org_e: Optional[Exception] = None) -> int:
- if not val and org_e:
- raise org_e
- try:
- return int(val)
- except ValueError as e:
- return to_int(val[0:-1], org_e or e)
-
- return tuple(map(to_int, version_str.split('.')))
-
-
def get_hostname():
# type: () -> str
return socket.gethostname()
os.chown(os.path.join(dirpath, filename), uid, gid)
-def find_container_engine(ctx: CephadmContext) -> Optional[ContainerEngine]:
- if ctx.docker:
- return Docker()
- else:
- for i in CONTAINER_PREFERENCE:
- try:
- return i()
- except Exception:
- pass
- return None
-
-
-def check_container_engine(ctx: CephadmContext) -> ContainerEngine:
- engine = ctx.container_engine
- if not isinstance(engine, CONTAINER_PREFERENCE):
- # See https://github.com/python/mypy/issues/8993
- exes: List[str] = [i.EXE for i in CONTAINER_PREFERENCE] # type: ignore
- raise Error('No container engine binary found ({}). Try run `apt/dnf/yum/zypper install <container engine>`'.format(' or '.join(exes)))
- elif isinstance(engine, Podman):
- engine.get_version(ctx)
- if engine.version < MIN_PODMAN_VERSION:
- raise Error('podman version %d.%d.%d or later is required' % MIN_PODMAN_VERSION)
- return engine
-
-
def get_unit_name(fsid, daemon_type, daemon_id=None):
# type: (str, str, Optional[Union[int, str]]) -> str
# accept either name or type + id
--- /dev/null
+# container_engines.py - container engine types and selection funcs
+
+
+from typing import Tuple, List, Optional
+
+from .call_wrappers import call_throws, CallVerbosity
+from .context import CephadmContext
+from .container_engine_base import ContainerEngine
+from .constants import MIN_PODMAN_VERSION
+from .exceptions import Error
+
+
+class Podman(ContainerEngine):
+ EXE = 'podman'
+
+ def __init__(self) -> None:
+ super().__init__()
+ self._version: Optional[Tuple[int, ...]] = None
+
+ @property
+ def version(self) -> Tuple[int, ...]:
+ if self._version is None:
+ raise RuntimeError('Please call `get_version` first')
+ return self._version
+
+ def get_version(self, ctx: CephadmContext) -> None:
+ out, _, _ = call_throws(ctx, [self.path, 'version', '--format', '{{.Client.Version}}'], verbosity=CallVerbosity.QUIET)
+ self._version = _parse_podman_version(out)
+
+ def __str__(self) -> str:
+ version = '.'.join(map(str, self.version))
+ return f'{self.EXE} ({self.path}) version {version}'
+
+
+class Docker(ContainerEngine):
+ EXE = 'docker'
+
+
+CONTAINER_PREFERENCE = (Podman, Docker) # prefer podman to docker
+
+
+def find_container_engine(ctx: CephadmContext) -> Optional[ContainerEngine]:
+ if ctx.docker:
+ return Docker()
+ else:
+ for i in CONTAINER_PREFERENCE:
+ try:
+ return i()
+ except Exception:
+ pass
+ return None
+
+
+def check_container_engine(ctx: CephadmContext) -> ContainerEngine:
+ engine = ctx.container_engine
+ if not isinstance(engine, CONTAINER_PREFERENCE):
+ # See https://github.com/python/mypy/issues/8993
+ exes: List[str] = [i.EXE for i in CONTAINER_PREFERENCE] # type: ignore
+ raise Error('No container engine binary found ({}). Try run `apt/dnf/yum/zypper install <container engine>`'.format(' or '.join(exes)))
+ elif isinstance(engine, Podman):
+ engine.get_version(ctx)
+ if engine.version < MIN_PODMAN_VERSION:
+ raise Error('podman version %d.%d.%d or later is required' % MIN_PODMAN_VERSION)
+ return engine
+
+
+def _parse_podman_version(version_str):
+ # type: (str) -> Tuple[int, ...]
+ def to_int(val: str, org_e: Optional[Exception] = None) -> int:
+ if not val and org_e:
+ raise org_e
+ try:
+ return int(val)
+ except ValueError as e:
+ return to_int(val[0:-1], org_e or e)
+
+ return tuple(map(to_int, version_str.split('.')))
("1.6.2-stable2", (1,6,2)),
])
def test_parse_podman_version(self, test_input, expected):
- assert _cephadm._parse_podman_version(test_input) == expected
+ from cephadmlib.container_engines import _parse_podman_version
+
+ assert _parse_podman_version(test_input) == expected
def test_parse_podman_version_invalid(self):
+ from cephadmlib.container_engines import _parse_podman_version
+
with pytest.raises(ValueError) as res:
- _cephadm._parse_podman_version('inval.id')
+ _parse_podman_version('inval.id')
assert 'inval' in str(res.value)
@mock.patch('cephadm.logger')
_find_program_loc = 'cephadmlib.container_engine_base.find_program'
+_call_throws_loc = 'cephadmlib.container_engines.call_throws'
def test_container_engine():
+ from cephadmlib.container_engine_base import ContainerEngine
+
with pytest.raises(NotImplementedError):
- _cephadm.ContainerEngine()
+ ContainerEngine()
- class PhonyContainerEngine(_cephadm.ContainerEngine):
+ class PhonyContainerEngine(ContainerEngine):
EXE = "true"
with mock.patch(_find_program_loc) as find_program:
find_program.assert_called()
with pytest.raises(RuntimeError):
pm.version
- with mock.patch("cephadm.call_throws") as call_throws:
+ with mock.patch(_call_throws_loc) as call_throws:
call_throws.return_value = ("4.9.9", None, None)
with with_cephadm_ctx([]) as ctx:
pm.get_version(ctx)
find_program.return_value = "/usr/bin/podman"
pm = _cephadm.Podman()
find_program.assert_called()
- with mock.patch("cephadm.call_throws") as call_throws:
+ with mock.patch(_call_throws_loc) as call_throws:
call_throws.return_value = ("4.10.beta2", None, None)
with with_cephadm_ctx([]) as ctx:
with pytest.raises(ValueError):
projects / ceph-ci.git / commitdiff