]>
git.proxmox.com Git - ceph.git/blob - ceph/qa/tasks/mgr/dashboard/test_host.py
407207133e3bac02230be87e2639e20d7780bc0d
1 # -*- coding: utf-8 -*-
2 from __future__
import absolute_import
5 from tasks
.mgr
.dashboard
.helper
import DashboardTestCase
, JList
, JObj
6 from tasks
.mgr
.dashboard
.test_orchestrator
import test_data
9 class HostControllerTest(DashboardTestCase
):
11 AUTH_ROLES
= ['read-only']
13 URL_HOST
= '/api/host'
17 super(HostControllerTest
, cls
).setUpClass()
18 cls
._load
_module
("test_orchestrator")
20 cmd
= ['orch', 'set', 'backend', 'test_orchestrator']
21 cls
.mgr_cluster
.mon_manager
.raw_cluster_cmd(*cmd
)
23 cmd
= ['test_orchestrator', 'load_data', '-i', '-']
24 cls
.mgr_cluster
.mon_manager
.raw_cluster_cmd_result(*cmd
, stdin
=json
.dumps(test_data
))
27 def tearDownClass(cls
):
28 cmd
= ['test_orchestrator', 'load_data', '-i', '-']
29 cls
.mgr_cluster
.mon_manager
.raw_cluster_cmd_result(*cmd
, stdin
='{}')
31 @DashboardTestCase.RunAs('test', 'test', ['block-manager'])
32 def test_access_permissions(self
):
33 self
._get
(self
.URL_HOST
)
34 self
.assertStatus(403)
36 def test_host_list(self
):
37 data
= self
._get
(self
.URL_HOST
)
38 self
.assertStatus(200)
40 orch_hostnames
= {inventory_node
['name'] for inventory_node
in test_data
['inventory']}
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'])
54 self
.assertIn('sources', server
)
55 in_ceph
, in_orchestrator
= server
['sources']['ceph'], server
['sources']['orchestrator']
57 self
.assertGreaterEqual(len(server
['services']), 1)
58 if not in_orchestrator
:
59 self
.assertNotIn(server
['hostname'], orch_hostnames
)
61 self
.assertEqual(len(server
['services']), 0)
62 self
.assertIn(server
['hostname'], orch_hostnames
)
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
)
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)
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'] != '']
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),
86 'location': JList(JObj({
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')
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')