ceph/src/pybind/mgr/dashboard/tests/test_host.py

   1 import unittest
   2
   3 try:
   4     import mock
   5 except ImportError:
   6     from unittest import mock
   7
   8 from orchestrator import HostSpec
   9
  10 from . import ControllerTestCase
  11 from ..controllers.host import get_hosts, Host
  12 from .. import mgr
  13
  14
  15 class HostControllerTest(ControllerTestCase):
  16     URL_HOST = '/api/host'
  17
  18     @classmethod
  19     def setup_server(cls):
  20         # pylint: disable=protected-access
  21         Host._cp_config['tools.authenticate.on'] = False
  22         cls.setup_controllers([Host])
  23
  24     @mock.patch('dashboard.controllers.host.get_hosts')
  25     def test_host_list(self, mock_get_hosts):
  26         hosts = [
  27             {
  28                 'hostname': 'host-0',
  29                 'sources': {
  30                     'ceph': True, 'orchestrator': False
  31                 }
  32             },
  33             {
  34                 'hostname': 'host-1',
  35                 'sources': {
  36                     'ceph': False, 'orchestrator': True
  37                 }
  38             },
  39             {
  40                 'hostname': 'host-2',
  41                 'sources': {
  42                     'ceph': True, 'orchestrator': True
  43                 }
  44             }
  45         ]
  46
  47         def _get_hosts(from_ceph=True, from_orchestrator=True):
  48             _hosts = []
  49             if from_ceph:
  50                 _hosts.append(hosts[0])
  51             if from_orchestrator:
  52                 _hosts.append(hosts[1])
  53                 _hosts.append(hosts[2])
  54             return _hosts
  55         mock_get_hosts.side_effect = _get_hosts
  56
  57         self._get(self.URL_HOST)
  58         self.assertStatus(200)
  59         self.assertJsonBody(hosts)
  60
  61         self._get('{}?sources=ceph'.format(self.URL_HOST))
  62         self.assertStatus(200)
  63         self.assertJsonBody([hosts[0]])
  64
  65         self._get('{}?sources=orchestrator'.format(self.URL_HOST))
  66         self.assertStatus(200)
  67         self.assertJsonBody(hosts[1:])
  68
  69         self._get('{}?sources=ceph,orchestrator'.format(self.URL_HOST))
  70         self.assertStatus(200)
  71         self.assertJsonBody(hosts)
  72
  73
  74 class TestHosts(unittest.TestCase):
  75
  76     @mock.patch('dashboard.controllers.orchestrator.OrchClient.instance')
  77     def test_get_hosts(self, instance):
  78         mgr.list_servers.return_value = [{'hostname': 'node1'}, {'hostname': 'localhost'}]
  79
  80         fake_client = mock.Mock()
  81         fake_client.available.return_value = True
  82         fake_client.hosts.list.return_value = [
  83             HostSpec('node1'), HostSpec('node2')]
  84         instance.return_value = fake_client
  85
  86         hosts = get_hosts()
  87         self.assertEqual(len(hosts), 3)
  88         check_sources = {
  89             'localhost': {'ceph': True, 'orchestrator': False},
  90             'node1': {'ceph': True, 'orchestrator': True},
  91             'node2': {'ceph': False, 'orchestrator': True}
  92         }
  93         for host in hosts:
  94             hostname = host['hostname']
  95             sources = host['sources']
  96             self.assertDictEqual(sources, check_sources[hostname])