]>
Commit | Line | Data |
---|---|---|
11fdf7f2 TL |
1 | # -*- coding: utf-8 -*- |
2 | from __future__ import absolute_import | |
9f95a23c | 3 | import json |
11fdf7f2 | 4 | |
f6b5b4d7 TL |
5 | from .helper import DashboardTestCase, JList, JObj |
6 | from .test_orchestrator import test_data | |
11fdf7f2 TL |
7 | |
8 | ||
9 | class HostControllerTest(DashboardTestCase): | |
10 | ||
11 | AUTH_ROLES = ['read-only'] | |
12 | ||
9f95a23c TL |
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 | ||
11fdf7f2 TL |
31 | @DashboardTestCase.RunAs('test', 'test', ['block-manager']) |
32 | def test_access_permissions(self): | |
9f95a23c | 33 | self._get(self.URL_HOST) |
11fdf7f2 TL |
34 | self.assertStatus(403) |
35 | ||
36 | def test_host_list(self): | |
9f95a23c | 37 | data = self._get(self.URL_HOST) |
11fdf7f2 TL |
38 | self.assertStatus(200) |
39 | ||
9f95a23c TL |
40 | orch_hostnames = {inventory_node['name'] for inventory_node in test_data['inventory']} |
41 | ||
11fdf7f2 TL |
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']) | |
11fdf7f2 TL |
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']) | |
9f95a23c TL |
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 | ||
f6b5b4d7 TL |
93 | def test_host_daemons(self): |
94 | hosts = self._get('{}'.format(self.URL_HOST)) | |
95 | hosts = [host['hostname'] for host in hosts if host['hostname'] != ''] | |
96 | assert hosts[0] | |
97 | data = self._get('{}/daemons'.format('{}/{}'.format(self.URL_HOST, hosts[0]))) | |
98 | self.assertStatus(200) | |
99 | self.assertSchema(data, JList(JObj({ | |
100 | 'hostname': str, | |
101 | 'daemon_id': str, | |
102 | 'daemon_type': str | |
103 | }))) | |
104 | ||
105 | def test_host_smart(self): | |
106 | hosts = self._get('{}'.format(self.URL_HOST)) | |
107 | hosts = [host['hostname'] for host in hosts if host['hostname'] != ''] | |
108 | assert hosts[0] | |
109 | self._get('{}/smart'.format('{}/{}'.format(self.URL_HOST, hosts[0]))) | |
110 | self.assertStatus(200) | |
111 | ||
9f95a23c TL |
112 | |
113 | class HostControllerNoOrchestratorTest(DashboardTestCase): | |
114 | def test_host_create(self): | |
115 | self._post('/api/host?hostname=foo') | |
116 | self.assertStatus(503) | |
117 | self.assertError(code='orchestrator_status_unavailable', | |
118 | component='orchestrator') | |
119 | ||
120 | def test_host_delete(self): | |
121 | self._delete('/api/host/bar') | |
122 | self.assertStatus(503) | |
123 | self.assertError(code='orchestrator_status_unavailable', | |
124 | component='orchestrator') |