ceph/src/cephadm/tests/fixtures.py

   1 import mock
   2 import os
   3 import pytest
   4 import time
   5
   6 from contextlib import contextmanager
   7 from pyfakefs import fake_filesystem
   8
   9 from typing import Callable, Dict, List, Optional
  10
  11
  12 with mock.patch('builtins.open', create=True):
  13     from importlib.machinery import SourceFileLoader
  14     cd = SourceFileLoader('cephadm', 'cephadm').load_module()
  15
  16
  17 def mock_docker():
  18     docker = mock.Mock(cd.Docker)
  19     docker.path = '/usr/bin/docker'
  20     return docker
  21
  22
  23 def mock_podman():
  24     podman = mock.Mock(cd.Podman)
  25     podman.path = '/usr/bin/podman'
  26     podman.version = (2, 1, 0)
  27     return podman
  28
  29
  30 def _daemon_path():
  31     return os.getcwd()
  32
  33
  34 def mock_bad_firewalld():
  35     def raise_bad_firewalld():
  36         raise Exception('Called bad firewalld')
  37     f = mock.Mock(cd.Firewalld)
  38     f.enable_service_for = lambda _ : raise_bad_firewalld()
  39     f.apply_rules = lambda : raise_bad_firewalld()
  40     f.open_ports = lambda _ : raise_bad_firewalld()
  41
  42 def _mock_scrape_host(obj, interval):
  43     try:
  44         raise ValueError("wah")
  45     except Exception as e:
  46         obj._handle_thread_exception(e, 'host')
  47
  48
  49 def _mock_run(obj):
  50     t = obj._create_thread(obj._scrape_host_facts, 'host', 5)
  51     time.sleep(1)
  52     if not t.is_alive():
  53         obj.cephadm_cache.update_health('host', "inactive", "host thread stopped")
  54
  55
  56 @pytest.fixture()
  57 def cephadm_fs(
  58     fs: fake_filesystem.FakeFilesystem,
  59 ):
  60     """
  61     use pyfakefs to stub filesystem calls
  62     """
  63     uid = os.getuid()
  64     gid = os.getgid()
  65
  66     with mock.patch('os.fchown'), \
  67          mock.patch('os.fchmod'), \
  68          mock.patch('platform.processor', return_value='x86_64'), \
  69          mock.patch('cephadm.extract_uid_gid', return_value=(uid, gid)):
  70
  71             fs.create_dir(cd.DATA_DIR)
  72             fs.create_dir(cd.LOG_DIR)
  73             fs.create_dir(cd.LOCK_DIR)
  74             fs.create_dir(cd.LOGROTATE_DIR)
  75             fs.create_dir(cd.UNIT_DIR)
  76
  77             yield fs
  78
  79
  80 @contextmanager
  81 def with_cephadm_ctx(
  82     cmd: List[str],
  83     container_engine: Callable = mock_podman(),
  84     list_networks: Optional[Dict[str,Dict[str,List[str]]]] = None,
  85     hostname: Optional[str] = None,
  86 ):
  87     """
  88     :param cmd: cephadm command argv
  89     :param container_engine: container engine mock (podman or docker)
  90     :param list_networks: mock 'list-networks' return
  91     :param hostname: mock 'socket.gethostname' return
  92     """
  93     if not hostname:
  94         hostname = 'host1'
  95
  96     with mock.patch('cephadm.attempt_bind'), \
  97          mock.patch('cephadm.call', return_value=('', '', 0)), \
  98          mock.patch('cephadm.call_timeout', return_value=0), \
  99          mock.patch('cephadm.find_executable', return_value='foo'), \
 100          mock.patch('cephadm.is_available', return_value=True), \
 101          mock.patch('cephadm.get_container_info', return_value=None), \
 102          mock.patch('cephadm.json_loads_retry', return_value={'epoch' : 1}), \
 103          mock.patch('socket.gethostname', return_value=hostname):
 104         ctx: cd.CephadmContext = cd.cephadm_init_ctx(cmd)
 105         ctx.container_engine = container_engine
 106         if list_networks is not None:
 107             with mock.patch('cephadm.list_networks', return_value=list_networks):
 108                 yield ctx
 109         else:
 110             yield ctx
 111