]>
git.proxmox.com Git - ceph.git/blob - ceph/src/pybind/mgr/cephadm/tests/fixtures.py
4 from contextlib
import contextmanager
6 from ceph
.deployment
.service_spec
import PlacementSpec
, ServiceSpec
7 from cephadm
.module
import CEPH_DATEFMT
8 from cephadm
.serve
import CephadmServe
11 from typing
import Any
, Iterator
, List
16 from cephadm
import CephadmOrchestrator
17 from cephadm
.services
.osd
import RemoveUtil
, OSD
18 from orchestrator
import raise_if_exception
, Completion
, HostSpec
19 from tests
import mock
22 def get_ceph_option(_
, key
):
26 def _run_cephadm(ret
):
27 def foo(*args
, **kwargs
):
32 def match_glob(val
, pat
):
33 ok
= fnmatch
.fnmatchcase(val
, pat
)
39 def with_cephadm_module(module_options
=None, store
=None):
41 :param module_options: Set opts as if they were set before module.__init__ is called
42 :param store: Set the store before module.__init__ is called
44 with mock
.patch("cephadm.module.CephadmOrchestrator.get_ceph_option", get_ceph_option
),\
45 mock
.patch("cephadm.services.osd.RemoveUtil._run_mon_cmd"), \
46 mock
.patch("cephadm.module.CephadmOrchestrator.get_osdmap"), \
47 mock
.patch("cephadm.services.osd.OSDService.get_osdspec_affinity", return_value
='test_spec'), \
48 mock
.patch("cephadm.module.CephadmOrchestrator.remote"):
50 m
= CephadmOrchestrator
.__new
__(CephadmOrchestrator
)
51 if module_options
is not None:
52 for k
, v
in module_options
.items():
53 m
._ceph
_set
_module
_option
('cephadm', k
, v
)
56 if '_ceph_get/mon_map' not in store
:
57 m
.mock_store_set('_ceph_get', 'mon_map', {
58 'modified': datetime
.datetime
.utcnow().strftime(CEPH_DATEFMT
),
61 for k
, v
in store
.items():
62 m
._ceph
_set
_store
(k
, v
)
64 m
.__init
__('cephadm', 0, 0)
65 m
._cluster
_fsid
= "fsid"
69 @pytest.yield_fixture()
71 with
with_cephadm_module({}) as m
:
75 @pytest.yield_fixture()
77 with
with_cephadm_module({}) as m
:
78 r
= RemoveUtil
.__new
__(RemoveUtil
)
83 @pytest.yield_fixture()
85 with mock
.patch("cephadm.services.osd.RemoveUtil"):
86 o
= OSD(0, mock
.MagicMock())
91 # type: (CephadmOrchestrator, Completion) -> Any
95 import pydevd
# if in debugger
101 while True: # don't timeout
103 raise_if_exception(c
)
111 raise_if_exception(c
)
114 assert False, "timeout" + str(c
._state
)
118 def with_host(m
: CephadmOrchestrator
, name
, refresh_hosts
=True):
119 # type: (CephadmOrchestrator, str) -> None
120 wait(m
, m
.add_host(HostSpec(hostname
=name
)))
122 CephadmServe(m
)._refresh
_hosts
_and
_daemons
()
124 wait(m
, m
.remove_host(name
))
127 def assert_rm_service(cephadm
, srv_name
):
128 assert wait(cephadm
, cephadm
.remove_service(srv_name
)) == f
'Removed service {srv_name}'
129 CephadmServe(cephadm
)._apply
_all
_services
()
133 def with_service(cephadm_module
: CephadmOrchestrator
, spec
: ServiceSpec
, meth
, host
: str) -> Iterator
[List
[str]]:
134 if spec
.placement
.is_empty():
135 spec
.placement
= PlacementSpec(hosts
=[host
], count
=1)
136 c
= meth(cephadm_module
, spec
)
137 assert wait(cephadm_module
, c
) == f
'Scheduled {spec.service_name()} update...'
138 specs
= [d
.spec
for d
in wait(cephadm_module
, cephadm_module
.describe_service())]
141 CephadmServe(cephadm_module
)._apply
_all
_services
()
143 dds
= wait(cephadm_module
, cephadm_module
.list_daemons())
144 own_dds
= [dd
for dd
in dds
if dd
.service_name() == spec
.service_name()]
147 yield [dd
.name() for dd
in own_dds
]
149 assert_rm_service(cephadm_module
, spec
.service_name())