+
+
+class Stub:
+ """Test class for returning predefined values"""
+
+ @classmethod
+ def get_mgr_no_services(cls):
+ mgr.get = Mock(return_value={})
+
+
+class RgwStub(Stub):
+
+ @classmethod
+ def get_daemons(cls):
+ mgr.get = Mock(return_value={'services': {'rgw': {'daemons': {
+ '5297': {
+ 'addr': '192.168.178.3:49774/1534999298',
+ 'metadata': {
+ 'frontend_config#0': 'beast port=8000',
+ 'id': 'daemon1',
+ 'realm_name': 'realm1',
+ 'zonegroup_name': 'zonegroup1',
+ 'zone_name': 'zone1',
+ 'hostname': 'daemon1.server.lan'
+ }
+ },
+ '5398': {
+ 'addr': '[2001:db8:85a3::8a2e:370:7334]:49774/1534999298',
+ 'metadata': {
+ 'frontend_config#0': 'civetweb port=8002',
+ 'id': 'daemon2',
+ 'realm_name': 'realm2',
+ 'zonegroup_name': 'zonegroup2',
+ 'zone_name': 'zone2',
+ 'hostname': 'daemon2.server.lan'
+ }
+ }
+ }}}})
+
+ @classmethod
+ def get_settings(cls):
+ settings = {
+ 'RGW_API_ACCESS_KEY': 'fake-access-key',
+ 'RGW_API_SECRET_KEY': 'fake-secret-key',
+ }
+ mgr.get_module_option = Mock(side_effect=settings.get)
+
+
+# pylint: disable=protected-access
+class Waiter(threading.Thread):
+ def __init__(self, task_name, task_metadata, tc, version):
+ super(Waiter, self).__init__()
+ self.task_name = task_name
+ self.task_metadata = task_metadata
+ self.ev = threading.Event()
+ self.abort = False
+ self.res_task = None
+ self.tc = tc
+ self.version = version
+
+ def run(self):
+ running = True
+ while running and not self.abort:
+ logger.info("task (%s, %s) is still executing", self.task_name,
+ self.task_metadata)
+ time.sleep(1)
+ self.tc._get('/api/task?name={}'.format(self.task_name), version=self.version)
+ res = self.tc.json_body()
+ for task in res['finished_tasks']:
+ if task['metadata'] == self.task_metadata:
+ # task finished
+ running = False
+ self.res_task = task
+ self.ev.set()
+
+
+@contextlib.contextmanager
+def patch_orch(available: bool, missing_features: Optional[List[str]] = None,
+ hosts: Optional[List[HostSpec]] = None,
+ inventory: Optional[List[dict]] = None,
+ daemons: Optional[List[DaemonDescription]] = None):
+ with mock.patch('dashboard.controllers.orchestrator.OrchClient.instance') as instance:
+ fake_client = mock.Mock()
+ fake_client.available.return_value = available
+ fake_client.get_missing_features.return_value = missing_features
+
+ if not daemons:
+ daemons = [
+ DaemonDescription(
+ daemon_type='mon',
+ daemon_id='a',
+ hostname='node0'
+ )
+ ]
+ fake_client.services.list_daemons.return_value = daemons
+ if hosts is not None:
+ fake_client.hosts.list.return_value = hosts
+
+ if inventory is not None:
+ def _list_inventory(hosts=None, refresh=False): # pylint: disable=unused-argument
+ inv_hosts = []
+ for inv_host in inventory:
+ if hosts is None or inv_host['name'] in hosts:
+ inv_hosts.append(InventoryHost.from_json(inv_host))
+ return inv_hosts
+ fake_client.inventory.list.side_effect = _list_inventory
+
+ instance.return_value = fake_client
+ yield fake_client