]>
git.proxmox.com Git - ceph.git/blob - ceph/src/pybind/mgr/cephadm/tests/fixtures.py
3 from contextlib
import contextmanager
5 from ceph
.deployment
.service_spec
import PlacementSpec
, ServiceSpec
6 from ceph
.utils
import datetime_to_str
, datetime_now
7 from cephadm
.serve
import CephadmServe
10 from typing
import Any
, Iterator
, List
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
21 def get_ceph_option(_
, key
):
25 def _run_cephadm(ret
):
26 def foo(*args
, **kwargs
):
31 def match_glob(val
, pat
):
32 ok
= fnmatch
.fnmatchcase(val
, pat
)
38 def with_cephadm_module(module_options
=None, store
=None):
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
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"):
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
)
55 if '_ceph_get/mon_map' not in store
:
56 m
.mock_store_set('_ceph_get', 'mon_map', {
57 'modified': datetime_to_str(datetime_now()),
60 for k
, v
in store
.items():
61 m
._ceph
_set
_store
(k
, v
)
63 m
.__init
__('cephadm', 0, 0)
64 m
._cluster
_fsid
= "fsid"
68 @pytest.yield_fixture()
70 with
with_cephadm_module({}) as m
:
74 @pytest.yield_fixture()
76 with
with_cephadm_module({}) as m
:
77 r
= RemoveUtil
.__new
__(RemoveUtil
)
82 @pytest.yield_fixture()
84 with mock
.patch("cephadm.services.osd.RemoveUtil"):
85 o
= OSD(0, mock
.MagicMock())
90 # type: (CephadmOrchestrator, Completion) -> Any
94 import pydevd
# if in debugger
100 while True: # don't timeout
102 raise_if_exception(c
)
110 raise_if_exception(c
)
113 assert False, "timeout" + str(c
._state
)
117 def with_host(m
: CephadmOrchestrator
, name
, refresh_hosts
=True):
118 # type: (CephadmOrchestrator, str) -> None
119 wait(m
, m
.add_host(HostSpec(hostname
=name
)))
121 CephadmServe(m
)._refresh
_hosts
_and
_daemons
()
123 wait(m
, m
.remove_host(name
))
126 def assert_rm_service(cephadm
, srv_name
):
127 assert wait(cephadm
, cephadm
.remove_service(srv_name
)) == f
'Removed service {srv_name}'
128 CephadmServe(cephadm
)._apply
_all
_services
()
132 def with_service(cephadm_module
: CephadmOrchestrator
, spec
: ServiceSpec
, meth
, host
: str) -> Iterator
[List
[str]]:
133 if spec
.placement
.is_empty():
134 spec
.placement
= PlacementSpec(hosts
=[host
], count
=1)
135 c
= meth(cephadm_module
, spec
)
136 assert wait(cephadm_module
, c
) == f
'Scheduled {spec.service_name()} update...'
137 specs
= [d
.spec
for d
in wait(cephadm_module
, cephadm_module
.describe_service())]
140 CephadmServe(cephadm_module
)._apply
_all
_services
()
142 dds
= wait(cephadm_module
, cephadm_module
.list_daemons())
143 own_dds
= [dd
for dd
in dds
if dd
.service_name() == spec
.service_name()]
146 yield [dd
.name() for dd
in own_dds
]
148 assert_rm_service(cephadm_module
, spec
.service_name())