ceph/src/cephadm/tests/fixtures.py

   1 from importlib.resources import contents
   2 import mock
   3 import os
   4 import pytest
   5 import time
   6
   7 from contextlib import contextmanager
   8 from pyfakefs import fake_filesystem
   9
  10 from typing import Dict, List, Optional
  11
  12
  13 def import_cephadm():
  14     """Import cephadm as a module."""
  15     import cephadm as _cephadm
  16
  17     return _cephadm
  18
  19
  20 def mock_docker():
  21     _cephadm = import_cephadm()
  22     docker = mock.Mock(_cephadm.Docker)
  23     docker.path = '/usr/bin/docker'
  24     return docker
  25
  26
  27 def mock_podman():
  28     _cephadm = import_cephadm()
  29     podman = mock.Mock(_cephadm.Podman)
  30     podman.path = '/usr/bin/podman'
  31     podman.version = (2, 1, 0)
  32     return podman
  33
  34
  35 def _daemon_path():
  36     return os.getcwd()
  37
  38
  39 def mock_bad_firewalld():
  40     def raise_bad_firewalld():
  41         raise Exception('Called bad firewalld')
  42
  43     _cephadm = import_cephadm()
  44     f = mock.Mock(_cephadm.Firewalld)
  45     f.enable_service_for = lambda _: raise_bad_firewalld()
  46     f.apply_rules = lambda: raise_bad_firewalld()
  47     f.open_ports = lambda _: raise_bad_firewalld()
  48
  49
  50 def _mock_scrape_host(obj, interval):
  51     try:
  52         raise ValueError("wah")
  53     except Exception as e:
  54         obj._handle_thread_exception(e, 'host')
  55
  56
  57 def _mock_run(obj):
  58     t = obj._create_thread(obj._scrape_host_facts, 'host', 5)
  59     time.sleep(1)
  60     if not t.is_alive():
  61         obj.cephadm_cache.update_health('host', "inactive", "host thread stopped")
  62
  63
  64 @pytest.fixture()
  65 def cephadm_fs(
  66     fs: fake_filesystem.FakeFilesystem,
  67 ):
  68     """
  69     use pyfakefs to stub filesystem calls
  70     """
  71     uid = os.getuid()
  72     gid = os.getgid()
  73
  74     _cephadm = import_cephadm()
  75     with mock.patch('os.fchown'), \
  76          mock.patch('os.fchmod'), \
  77          mock.patch('platform.processor', return_value='x86_64'), \
  78          mock.patch('cephadm.extract_uid_gid', return_value=(uid, gid)):
  79
  80         try:
  81             if not fake_filesystem.is_root():
  82                 fake_filesystem.set_uid(0)
  83         except AttributeError:
  84             pass
  85
  86         fs.create_dir(_cephadm.DATA_DIR)
  87         fs.create_dir(_cephadm.LOG_DIR)
  88         fs.create_dir(_cephadm.LOCK_DIR)
  89         fs.create_dir(_cephadm.LOGROTATE_DIR)
  90         fs.create_dir(_cephadm.UNIT_DIR)
  91         fs.create_dir('/sys/block')
  92
  93         yield fs
  94
  95
  96 @pytest.fixture()
  97 def host_sysfs(fs: fake_filesystem.FakeFilesystem):
  98     """Create a fake filesystem to represent sysfs"""
  99     enc_path = '/sys/class/scsi_generic/sg2/device/enclosure/0:0:1:0'
 100     dev_path = '/sys/class/scsi_generic/sg2/device'
 101     slot_count = 12
 102     fs.create_dir(dev_path)
 103     fs.create_file(os.path.join(dev_path, 'vendor'), contents="EnclosuresInc")
 104     fs.create_file(os.path.join(dev_path, 'model'), contents="D12")
 105     fs.create_file(os.path.join(enc_path, 'id'), contents='1')
 106     fs.create_file(os.path.join(enc_path, 'components'), contents=str(slot_count))
 107     for slot_num in range(slot_count):
 108         slot_dir = os.path.join(enc_path, str(slot_num))
 109         fs.create_file(os.path.join(slot_dir, 'locate'), contents='0')
 110         fs.create_file(os.path.join(slot_dir, 'fault'), contents='0')
 111         fs.create_file(os.path.join(slot_dir, 'slot'), contents=str(slot_num))
 112         if slot_num < 6:
 113             fs.create_file(os.path.join(slot_dir, 'status'), contents='Ok')
 114             slot_dev = os.path.join(slot_dir, 'device')
 115             fs.create_dir(slot_dev)
 116             fs.create_file(os.path.join(slot_dev, 'vpd_pg80'), contents=f'fake{slot_num:0>3}')
 117         else:
 118             fs.create_file(os.path.join(slot_dir, 'status'), contents='not installed')
 119
 120     yield fs
 121
 122
 123 @contextmanager
 124 def with_cephadm_ctx(
 125     cmd: List[str],
 126     list_networks: Optional[Dict[str, Dict[str, List[str]]]] = None,
 127     hostname: Optional[str] = None,
 128 ):
 129     """
 130     :param cmd: cephadm command argv
 131     :param list_networks: mock 'list-networks' return
 132     :param hostname: mock 'socket.gethostname' return
 133     """
 134     if not hostname:
 135         hostname = 'host1'
 136
 137     _cephadm = import_cephadm()
 138     with mock.patch('cephadm.attempt_bind'), \
 139          mock.patch('cephadm.call', return_value=('', '', 0)), \
 140          mock.patch('cephadm.call_timeout', return_value=0), \
 141          mock.patch('cephadm.find_executable', return_value='foo'), \
 142          mock.patch('cephadm.get_container_info', return_value=None), \
 143          mock.patch('cephadm.is_available', return_value=True), \
 144          mock.patch('cephadm.json_loads_retry', return_value={'epoch' : 1}), \
 145          mock.patch('cephadm.logger'), \
 146          mock.patch('socket.gethostname', return_value=hostname):
 147         ctx: _cephadm.CephadmContext = _cephadm.cephadm_init_ctx(cmd)
 148         ctx.container_engine = mock_podman()
 149         if list_networks is not None:
 150             with mock.patch('cephadm.list_networks', return_value=list_networks):
 151                 yield ctx
 152         else:
 153             yield ctx
 154