ceph/src/pybind/mgr/cephadm/tests/fixtures.py

   1 import datetime
   2 import time
   3 import fnmatch
   4 from contextlib import contextmanager
   5
   6 from ceph.deployment.service_spec import PlacementSpec, ServiceSpec
   7 from cephadm.module import CEPH_DATEFMT
   8
   9 try:
  10     from typing import Any
  11 except ImportError:
  12     pass
  13 import pytest
  14
  15 from cephadm import CephadmOrchestrator
  16 from cephadm.services.osd import RemoveUtil, OSD
  17 from orchestrator import raise_if_exception, Completion, HostSpec
  18 from tests import mock
  19
  20
  21 def get_ceph_option(_, key):
  22     return __file__
  23
  24
  25 def _run_cephadm(ret):
  26     def foo(*args, **kwargs):
  27         return ret, '', 0
  28     return foo
  29
  30
  31 def match_glob(val, pat):
  32     ok = fnmatch.fnmatchcase(val, pat)
  33     if not ok:
  34         assert pat in val
  35
  36
  37 @contextmanager
  38 def with_cephadm_module(module_options=None, store=None):
  39     """
  40     :param module_options: Set opts as if they were set before module.__init__ is called
  41     :param store: Set the store before module.__init__ is called
  42     """
  43     with mock.patch("cephadm.module.CephadmOrchestrator.get_ceph_option", get_ceph_option),\
  44             mock.patch("cephadm.services.osd.RemoveUtil._run_mon_cmd"), \
  45             mock.patch("cephadm.module.CephadmOrchestrator.get_osdmap"), \
  46             mock.patch("cephadm.services.osd.OSDService.get_osdspec_affinity", return_value='test_spec'), \
  47             mock.patch("cephadm.module.CephadmOrchestrator.remote"):
  48
  49         m = CephadmOrchestrator.__new__ (CephadmOrchestrator)
  50         if module_options is not None:
  51             for k, v in module_options.items():
  52                 m._ceph_set_module_option('cephadm', k, v)
  53         if store is None:
  54             store = {}
  55         if '_ceph_get/mon_map' not in store:
  56             m.mock_store_set('_ceph_get', 'mon_map', {
  57                 'modified': datetime.datetime.utcnow().strftime(CEPH_DATEFMT),
  58                 'fsid': 'foobar',
  59             })
  60         for k, v in store.items():
  61             m._ceph_set_store(k, v)
  62
  63         m.__init__('cephadm', 0, 0)
  64         m._cluster_fsid = "fsid"
  65         yield m
  66
  67
  68 @pytest.yield_fixture()
  69 def cephadm_module():
  70     with with_cephadm_module({}) as m:
  71         yield m
  72
  73
  74 @pytest.yield_fixture()
  75 def rm_util():
  76     with with_cephadm_module({}) as m:
  77         r = RemoveUtil.__new__(RemoveUtil)
  78         r.__init__(m)
  79         yield r
  80
  81
  82 @pytest.yield_fixture()
  83 def osd_obj():
  84     with mock.patch("cephadm.services.osd.RemoveUtil"):
  85         o = OSD(0, mock.MagicMock())
  86         yield o
  87
  88
  89 def wait(m, c):
  90     # type: (CephadmOrchestrator, Completion) -> Any
  91     m.process([c])
  92
  93     try:
  94         import pydevd  # if in debugger
  95         in_debug = True
  96     except ImportError:
  97         in_debug = False
  98
  99     if in_debug:
 100         while True:    # don't timeout
 101             if c.is_finished:
 102                 raise_if_exception(c)
 103                 return c.result
 104             time.sleep(0.1)
 105     else:
 106         for i in range(30):
 107             if i % 10 == 0:
 108                 m.process([c])
 109             if c.is_finished:
 110                 raise_if_exception(c)
 111                 return c.result
 112             time.sleep(0.1)
 113     assert False, "timeout" + str(c._state)
 114
 115
 116 @contextmanager
 117 def with_host(m:CephadmOrchestrator, name):
 118     # type: (CephadmOrchestrator, str) -> None
 119     wait(m, m.add_host(HostSpec(hostname=name)))
 120     yield
 121     wait(m, m.remove_host(name))
 122
 123
 124 def assert_rm_service(cephadm, srv_name):
 125     assert wait(cephadm, cephadm.remove_service(srv_name)) == f'Removed service {srv_name}'
 126     cephadm._apply_all_services()
 127
 128
 129 @contextmanager
 130 def with_service(cephadm_module: CephadmOrchestrator, spec: ServiceSpec, meth, host: str):
 131     if spec.placement.is_empty():
 132         spec.placement = PlacementSpec(hosts=[host], count=1)
 133     c = meth(cephadm_module, spec)
 134     assert wait(cephadm_module, c) == f'Scheduled {spec.service_name()} update...'
 135     specs = [d.spec for d in wait(cephadm_module, cephadm_module.describe_service())]
 136     assert spec in specs
 137
 138     cephadm_module._apply_all_services()
 139
 140     dds = wait(cephadm_module, cephadm_module.list_daemons())
 141     names = {dd.service_name() for dd in dds}
 142     assert spec.service_name() in names, dds
 143
 144     yield
 145
 146     assert_rm_service(cephadm_module, spec.service_name())