1 # -*- coding: utf-8 -*-
2 from __future__
import absolute_import
5 from .helper
import DashboardTestCase
31 'nodename': 'test-host0',
36 'nodename': 'test-host0',
41 'nodename': 'test-host0',
46 'nodename': 'test-host1',
54 class OrchestratorControllerTest(DashboardTestCase
):
56 AUTH_ROLES
= ['cluster-manager']
58 URL_STATUS
= '/api/orchestrator/status'
59 URL_INVENTORY
= '/api/orchestrator/inventory'
60 URL_OSD
= '/api/orchestrator/osd'
64 def test_data_inventory(self
):
65 return test_data
['inventory']
68 def test_data_daemons(self
):
69 return test_data
['daemons']
73 super(OrchestratorControllerTest
, cls
).setUpClass()
74 cls
._load
_module
('test_orchestrator')
75 cmd
= ['orch', 'set', 'backend', 'test_orchestrator']
76 cls
.mgr_cluster
.mon_manager
.raw_cluster_cmd(*cmd
)
78 cmd
= ['test_orchestrator', 'load_data', '-i', '-']
79 cls
.mgr_cluster
.mon_manager
.raw_cluster_cmd_result(*cmd
, stdin
=json
.dumps(test_data
))
82 def tearDownClass(cls
):
83 cmd
= ['test_orchestrator', 'load_data', '-i', '-']
84 cls
.mgr_cluster
.mon_manager
.raw_cluster_cmd_result(*cmd
, stdin
='{}')
86 def _validate_inventory(self
, data
, resp_data
):
87 self
.assertEqual(data
['name'], resp_data
['name'])
88 self
.assertEqual(len(data
['devices']), len(resp_data
['devices']))
90 if not data
['devices']:
92 test_devices
= sorted(data
['devices'], key
=lambda d
: d
['path'])
93 resp_devices
= sorted(resp_data
['devices'], key
=lambda d
: d
['path'])
95 for test
, resp
in zip(test_devices
, resp_devices
):
96 self
._validate
_device
(test
, resp
)
98 def _validate_device(self
, data
, resp_data
):
99 for key
, value
in data
.items():
100 self
.assertEqual(value
, resp_data
[key
])
102 def _validate_daemon(self
, data
, resp_data
):
103 for key
, value
in data
.items():
104 self
.assertEqual(value
, resp_data
[key
])
106 @DashboardTestCase.RunAs('test', 'test', ['block-manager'])
107 def test_access_permissions(self
):
108 self
._get
(self
.URL_STATUS
)
109 self
.assertStatus(200)
110 self
._get
(self
.URL_INVENTORY
)
111 self
.assertStatus(403)
113 def test_status_get(self
):
114 data
= self
._get
(self
.URL_STATUS
)
115 self
.assertTrue(data
['available'])
117 def test_inventory_list(self
):
119 data
= self
._get
(self
.URL_INVENTORY
)
120 self
.assertStatus(200)
122 sorting_key
= lambda node
: node
['name']
123 test_inventory
= sorted(self
.test_data_inventory
, key
=sorting_key
)
124 resp_inventory
= sorted(data
, key
=sorting_key
)
125 self
.assertEqual(len(test_inventory
), len(resp_inventory
))
126 for test
, resp
in zip(test_inventory
, resp_inventory
):
127 self
._validate
_inventory
(test
, resp
)
129 # get inventory by hostname
130 node
= self
.test_data_inventory
[-1]
131 data
= self
._get
('{}?hostname={}'.format(self
.URL_INVENTORY
, node
['name']))
132 self
.assertStatus(200)
133 self
.assertEqual(len(data
), 1)
134 self
._validate
_inventory
(node
, data
[0])