import 15.2.0 Octopus source

[ceph.git] / ceph / qa / tasks / mgr / dashboard / test_host.py

diff --git a/ceph/qa/tasks/mgr/dashboard/test_host.py b/ceph/qa/tasks/mgr/dashboard/test_host.py
index f6a8d25ccb4101b75476af0567ea1d00af5cdbe9..d608dc247ed565244b1797f100d324409e1597cd 100644 (file)
--- a/ceph/qa/tasks/mgr/dashboard/test_host.py
+++ b/ceph/qa/tasks/mgr/dashboard/test_host.py
@@ -1,31 +1,105 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import
+import json
 
-from .helper import DashboardTestCase
+from .helper import DashboardTestCase, JList, JObj
+from .test_orchestrator import test_data
 
 
 class HostControllerTest(DashboardTestCase):
 
     AUTH_ROLES = ['read-only']
 
+    URL_HOST = '/api/host'
+
+    @classmethod
+    def setUpClass(cls):
+        super(HostControllerTest, cls).setUpClass()
+        cls._load_module("test_orchestrator")
+
+        cmd = ['orch', 'set', 'backend', 'test_orchestrator']
+        cls.mgr_cluster.mon_manager.raw_cluster_cmd(*cmd)
+
+        cmd = ['test_orchestrator', 'load_data', '-i', '-']
+        cls.mgr_cluster.mon_manager.raw_cluster_cmd_result(*cmd, stdin=json.dumps(test_data))
+
+    @classmethod
+    def tearDownClass(cls):
+        cmd = ['test_orchestrator', 'load_data', '-i', '-']
+        cls.mgr_cluster.mon_manager.raw_cluster_cmd_result(*cmd, stdin='{}')
+
     @DashboardTestCase.RunAs('test', 'test', ['block-manager'])
     def test_access_permissions(self):
-        self._get('/api/host')
+        self._get(self.URL_HOST)
         self.assertStatus(403)
 
     def test_host_list(self):
-        data = self._get('/api/host')
+        data = self._get(self.URL_HOST)
         self.assertStatus(200)
 
+        orch_hostnames = {inventory_node['name'] for inventory_node in test_data['inventory']}
+
         for server in data:
             self.assertIn('services', server)
             self.assertIn('hostname', server)
             self.assertIn('ceph_version', server)
             self.assertIsNotNone(server['hostname'])
             self.assertIsNotNone(server['ceph_version'])
-            self.assertGreaterEqual(len(server['services']), 1)
             for service in server['services']:
                 self.assertIn('type', service)
                 self.assertIn('id', service)
                 self.assertIsNotNone(service['type'])
                 self.assertIsNotNone(service['id'])
+
+            self.assertIn('sources', server)
+            in_ceph, in_orchestrator = server['sources']['ceph'], server['sources']['orchestrator']
+            if in_ceph:
+                self.assertGreaterEqual(len(server['services']), 1)
+                if not in_orchestrator:
+                    self.assertNotIn(server['hostname'], orch_hostnames)
+            if in_orchestrator:
+                self.assertEqual(len(server['services']), 0)
+                self.assertIn(server['hostname'], orch_hostnames)
+
+    def test_host_list_with_sources(self):
+        data = self._get('{}?sources=orchestrator'.format(self.URL_HOST))
+        self.assertStatus(200)
+        test_hostnames = {inventory_node['name'] for inventory_node in test_data['inventory']}
+        resp_hostnames = {host['hostname'] for host in data}
+        self.assertEqual(test_hostnames, resp_hostnames)
+
+        data = self._get('{}?sources=ceph'.format(self.URL_HOST))
+        self.assertStatus(200)
+        test_hostnames = {inventory_node['name'] for inventory_node in test_data['inventory']}
+        resp_hostnames = {host['hostname'] for host in data}
+        self.assertEqual(len(test_hostnames.intersection(resp_hostnames)), 0)
+
+    def test_host_devices(self):
+        hosts = self._get('{}'.format(self.URL_HOST))
+        hosts = [host['hostname'] for host in hosts if host['hostname'] != '']
+        assert hosts[0]
+        data = self._get('{}/devices'.format('{}/{}'.format(self.URL_HOST, hosts[0])))
+        self.assertStatus(200)
+        self.assertSchema(data, JList(JObj({
+            'daemons': JList(str),
+            'devid': str,
+            'location': JList(JObj({
+                'host': str,
+                'dev': str,
+                'path': str
+            }))
+        })))
+
+
+class HostControllerNoOrchestratorTest(DashboardTestCase):
+    def test_host_create(self):
+        self._post('/api/host?hostname=foo')
+        self.assertStatus(503)
+        self.assertError(code='orchestrator_status_unavailable',
+                         component='orchestrator')
+
+    def test_host_delete(self):
+        self._delete('/api/host/bar')
+        self.assertStatus(503)
+        self.assertError(code='orchestrator_status_unavailable',
+                         component='orchestrator')