ceph/qa/tasks/mgr/dashboard/test_host.py

   1 # -*- coding: utf-8 -*-
   2 from __future__ import absolute_import
   3 import json
   4
   5 from .helper import DashboardTestCase, JList, JObj
   6 from .test_orchestrator import test_data
   7
   8
   9 class HostControllerTest(DashboardTestCase):
  10
  11     AUTH_ROLES = ['read-only']
  12
  13     URL_HOST = '/api/host'
  14
  15     @classmethod
  16     def setUpClass(cls):
  17         super(HostControllerTest, cls).setUpClass()
  18         cls._load_module("test_orchestrator")
  19
  20         cmd = ['orch', 'set', 'backend', 'test_orchestrator']
  21         cls.mgr_cluster.mon_manager.raw_cluster_cmd(*cmd)
  22
  23         cmd = ['test_orchestrator', 'load_data', '-i', '-']
  24         cls.mgr_cluster.mon_manager.raw_cluster_cmd_result(*cmd, stdin=json.dumps(test_data))
  25
  26     @classmethod
  27     def tearDownClass(cls):
  28         cmd = ['test_orchestrator', 'load_data', '-i', '-']
  29         cls.mgr_cluster.mon_manager.raw_cluster_cmd_result(*cmd, stdin='{}')
  30
  31     @DashboardTestCase.RunAs('test', 'test', ['block-manager'])
  32     def test_access_permissions(self):
  33         self._get(self.URL_HOST)
  34         self.assertStatus(403)
  35
  36     def test_host_list(self):
  37         data = self._get(self.URL_HOST)
  38         self.assertStatus(200)
  39
  40         orch_hostnames = {inventory_node['name'] for inventory_node in test_data['inventory']}
  41
  42         for server in data:
  43             self.assertIn('services', server)
  44             self.assertIn('hostname', server)
  45             self.assertIn('ceph_version', server)
  46             self.assertIsNotNone(server['hostname'])
  47             self.assertIsNotNone(server['ceph_version'])
  48             for service in server['services']:
  49                 self.assertIn('type', service)
  50                 self.assertIn('id', service)
  51                 self.assertIsNotNone(service['type'])
  52                 self.assertIsNotNone(service['id'])
  53
  54             self.assertIn('sources', server)
  55             in_ceph, in_orchestrator = server['sources']['ceph'], server['sources']['orchestrator']
  56             if in_ceph:
  57                 self.assertGreaterEqual(len(server['services']), 1)
  58                 if not in_orchestrator:
  59                     self.assertNotIn(server['hostname'], orch_hostnames)
  60             if in_orchestrator:
  61                 self.assertEqual(len(server['services']), 0)
  62                 self.assertIn(server['hostname'], orch_hostnames)
  63
  64     def test_host_list_with_sources(self):
  65         data = self._get('{}?sources=orchestrator'.format(self.URL_HOST))
  66         self.assertStatus(200)
  67         test_hostnames = {inventory_node['name'] for inventory_node in test_data['inventory']}
  68         resp_hostnames = {host['hostname'] for host in data}
  69         self.assertEqual(test_hostnames, resp_hostnames)
  70
  71         data = self._get('{}?sources=ceph'.format(self.URL_HOST))
  72         self.assertStatus(200)
  73         test_hostnames = {inventory_node['name'] for inventory_node in test_data['inventory']}
  74         resp_hostnames = {host['hostname'] for host in data}
  75         self.assertEqual(len(test_hostnames.intersection(resp_hostnames)), 0)
  76
  77     def test_host_devices(self):
  78         hosts = self._get('{}'.format(self.URL_HOST))
  79         hosts = [host['hostname'] for host in hosts if host['hostname'] != '']
  80         assert hosts[0]
  81         data = self._get('{}/devices'.format('{}/{}'.format(self.URL_HOST, hosts[0])))
  82         self.assertStatus(200)
  83         self.assertSchema(data, JList(JObj({
  84             'daemons': JList(str),
  85             'devid': str,
  86             'location': JList(JObj({
  87                 'host': str,
  88                 'dev': str,
  89                 'path': str
  90             }))
  91         })))
  92
  93
  94 class HostControllerNoOrchestratorTest(DashboardTestCase):
  95     def test_host_create(self):
  96         self._post('/api/host?hostname=foo')
  97         self.assertStatus(503)
  98         self.assertError(code='orchestrator_status_unavailable',
  99                          component='orchestrator')
 100
 101     def test_host_delete(self):
 102         self._delete('/api/host/bar')
 103         self.assertStatus(503)
 104         self.assertError(code='orchestrator_status_unavailable',
 105                          component='orchestrator')