import ceph pacific 16.2.5

[ceph.git] / ceph / src / cephadm / tests / fixtures.py

diff --git a/ceph/src/cephadm/tests/fixtures.py b/ceph/src/cephadm/tests/fixtures.py
index a3f46253976e921a28d1d3640a814996efb65e86..2f07d6034805a5d9e3ecf1df897ede8790c59d5b 100644 (file)
--- a/ceph/src/cephadm/tests/fixtures.py
+++ b/ceph/src/cephadm/tests/fixtures.py
@@ -1,16 +1,32 @@
-
 import mock
-from mock import patch
-import pytest
-
 import os
+import pytest
 import time
 
-with patch('builtins.open', create=True):
+from contextlib import contextmanager
+from pyfakefs import fake_filesystem
+
+from typing import Callable, Dict, List, Optional
+
+
+with mock.patch('builtins.open', create=True):
     from importlib.machinery import SourceFileLoader
     cd = SourceFileLoader('cephadm', 'cephadm').load_module()
 
 
+def mock_docker():
+    docker = mock.Mock(cd.Docker)
+    docker.path = '/usr/bin/docker'
+    return docker
+
+
+def mock_podman():
+    podman = mock.Mock(cd.Podman)
+    podman.path = '/usr/bin/podman'
+    podman.version = (2, 1, 0)
+    return podman
+
+
 def _daemon_path():
     return os.getcwd()
 
@@ -40,3 +56,56 @@ def exporter():
         exporter = cd.CephadmDaemon(ctx, fsid='foobar', daemon_id='test')
         assert exporter.token == 'MyAccessToken' 
         yield exporter
+
+
+@pytest.fixture()
+def cephadm_fs(
+    fs: fake_filesystem.FakeFilesystem,
+):
+    """
+    use pyfakefs to stub filesystem calls
+    """
+    uid = os.getuid()
+    gid = os.getgid()
+
+    with mock.patch('os.fchown'), \
+         mock.patch('cephadm.extract_uid_gid', return_value=(uid, gid)):
+
+            fs.create_dir(cd.DATA_DIR)
+            fs.create_dir(cd.LOG_DIR)
+            fs.create_dir(cd.LOCK_DIR)
+            fs.create_dir(cd.LOGROTATE_DIR)
+            fs.create_dir(cd.UNIT_DIR)
+
+            yield fs
+
+
+@contextmanager
+def with_cephadm_ctx(
+    cmd: List[str],
+    container_engine: Callable = mock_podman(),
+    list_networks: Optional[Dict[str,Dict[str,List[str]]]] = None,
+    hostname: Optional[str] = None,
+):
+    """
+    :param cmd: cephadm command argv
+    :param container_engine: container engine mock (podman or docker)
+    :param list_networks: mock 'list-networks' return
+    :param hostname: mock 'socket.gethostname' return
+    """
+    if not list_networks:
+        list_networks = {}
+    if not hostname:
+        hostname = 'host1'
+
+    with mock.patch('cephadm.get_parm'), \
+         mock.patch('cephadm.attempt_bind'), \
+         mock.patch('cephadm.call', return_value=('', '', 0)), \
+         mock.patch('cephadm.find_executable', return_value='foo'), \
+         mock.patch('cephadm.is_available', return_value=True), \
+         mock.patch('cephadm.json_loads_retry', return_value={'epoch' : 1}), \
+         mock.patch('cephadm.list_networks', return_value=list_networks), \
+         mock.patch('socket.gethostname', return_value=hostname):
+             ctx: cd.CephadmContext = cd.cephadm_init_ctx(cmd)
+             ctx.container_engine = container_engine
+             yield ctx