Coverage for manila/tests/compute/test_nova.py: 97%
178 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright 2014 Mirantis Inc.
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may
4# not use this file except in compliance with the License. You may obtain
5# a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations
13# under the License.
15from unittest import mock
17import ddt
18from novaclient import exceptions as nova_exception
19from novaclient import utils
20from novaclient.v2 import servers as nova_servers
22from manila.compute import nova
23from manila import context
24from manila import exception
25from manila import test
26from manila.tests import utils as test_utils
29class Volume(object):
30 def __init__(self, volume_id):
31 self.id = volume_id
32 self.volumeId = volume_id
35class Network(object):
36 def __init__(self, net_id):
37 self.id = net_id
38 self.label = 'fake_label_%s' % net_id
41class FakeNovaClient(object):
42 class Servers(object):
43 def get(self, instance_id):
44 return {'id': instance_id}
46 def list(self, *args, **kwargs):
47 return [{'id': 'id1'}, {'id': 'id2'}]
49 def create(self, *args, **kwargs):
50 return {'id': 'created_id'}
52 def __getattr__(self, item):
53 return None
55 class Volumes(object):
56 def get(self, volume_id):
57 return Volume(volume_id)
59 def list(self, detailed, *args, **kwargs):
60 return [{'id': 'id1'}, {'id': 'id2'}]
62 def create(self, *args, **kwargs):
63 return {'id': 'created_id'}
65 def __getattr__(self, item):
66 return None
68 class Networks(object):
69 def get(self, net_id):
70 return Network(net_id)
72 def __init__(self):
73 self.servers = self.Servers()
74 self.volumes = self.Volumes()
75 self.keypairs = self.servers
76 self.networks = self.Networks()
79@nova.translate_server_exception
80def decorated_by_translate_server_exception(self, context, instance_id, exc):
81 if exc:
82 raise exc(instance_id)
83 else:
84 return 'OK'
87@ddt.ddt
88class TranslateServerExceptionTestCase(test.TestCase):
90 def test_translate_server_exception(self):
91 result = decorated_by_translate_server_exception(
92 'foo_self', 'foo_ctxt', 'foo_instance_id', None)
93 self.assertEqual('OK', result)
95 def test_translate_server_exception_not_found(self):
96 self.assertRaises(
97 exception.InstanceNotFound,
98 decorated_by_translate_server_exception,
99 'foo_self', 'foo_ctxt', 'foo_instance_id', nova_exception.NotFound)
101 def test_translate_server_exception_bad_request(self):
102 self.assertRaises(
103 exception.InvalidInput,
104 decorated_by_translate_server_exception,
105 'foo_self', 'foo_ctxt', 'foo_instance_id',
106 nova_exception.BadRequest)
108 @ddt.data(
109 nova_exception.HTTPNotImplemented,
110 nova_exception.RetryAfterException,
111 nova_exception.Unauthorized,
112 nova_exception.Forbidden,
113 nova_exception.MethodNotAllowed,
114 nova_exception.OverLimit,
115 nova_exception.RateLimit,
116 )
117 def test_translate_server_exception_other_exception(self, exc):
118 self.assertRaises(
119 exception.ManilaException,
120 decorated_by_translate_server_exception,
121 'foo_self', 'foo_ctxt', 'foo_instance_id', exc)
124def get_fake_auth_obj():
125 return type('FakeAuthObj', (object, ), {'get_client': mock.Mock()})
128class NovaclientTestCase(test.TestCase):
130 @mock.patch('manila.compute.nova.AUTH_OBJ', None)
131 def test_no_auth_obj(self):
132 mock_client_loader = self.mock_object(
133 nova.client_auth, 'AuthClientLoader')
134 fake_context = 'fake_context'
135 data = {
136 'nova': {
137 'api_microversion': 'foo_api_microversion',
138 'endpoint_type': 'internalURL',
139 'region_name': 'foo_region_name',
140 }
141 }
143 with test_utils.create_temp_config_with_opts(data):
144 nova.novaclient(fake_context)
146 mock_client_loader.assert_called_once_with(
147 client_class=nova.nova_client.Client,
148 cfg_group=nova.NOVA_GROUP
149 )
150 mock_client_loader.return_value.get_client.assert_called_once_with(
151 fake_context,
152 version=data['nova']['api_microversion'],
153 endpoint_type=data['nova']['endpoint_type'],
154 region_name=data['nova']['region_name'],
155 )
157 @mock.patch('manila.compute.nova.AUTH_OBJ', get_fake_auth_obj())
158 def test_with_auth_obj(self):
159 fake_context = 'fake_context'
160 data = {
161 'nova': {
162 'api_microversion': 'foo_api_microversion',
163 'endpoint_type': 'internalURL',
164 'region_name': 'foo_region_name',
165 }
166 }
168 with test_utils.create_temp_config_with_opts(data):
169 nova.novaclient(fake_context)
171 nova.AUTH_OBJ.get_client.assert_called_once_with(
172 fake_context,
173 version=data['nova']['api_microversion'],
174 endpoint_type=data['nova']['endpoint_type'],
175 region_name=data['nova']['region_name'],
176 )
179@ddt.ddt
180class NovaApiTestCase(test.TestCase):
181 def setUp(self):
182 super(NovaApiTestCase, self).setUp()
184 self.api = nova.API()
185 self.novaclient = FakeNovaClient()
186 self.ctx = context.get_admin_context()
187 self.mock_object(nova, 'novaclient',
188 mock.Mock(return_value=self.novaclient))
189 self.mock_object(nova, '_untranslate_server_summary_view',
190 lambda server: server)
192 def test_server_create(self):
193 result = self.api.server_create(self.ctx, 'server_name', 'fake_image',
194 'fake_flavor', None, None, None)
195 self.assertEqual('created_id', result['id'])
197 def test_server_delete(self):
198 self.mock_object(self.novaclient.servers, 'delete')
199 self.api.server_delete(self.ctx, 'id1')
200 self.novaclient.servers.delete.assert_called_once_with('id1')
202 def test_server_get(self):
203 instance_id = 'instance_id1'
204 result = self.api.server_get(self.ctx, instance_id)
205 self.assertEqual(instance_id, result['id'])
207 def test_server_get_by_name_or_id(self):
208 instance_id = 'instance_id1'
209 server = {'id': instance_id, 'fake_key': 'fake_value'}
210 self.mock_object(utils, 'find_resource',
211 mock.Mock(return_value=server))
213 result = self.api.server_get_by_name_or_id(self.ctx, instance_id)
215 self.assertEqual(instance_id, result['id'])
216 utils.find_resource.assert_called_once_with(mock.ANY, instance_id)
218 def test_server_get_by_name_or_id_failed(self):
219 instance_id = 'instance_id1'
220 server = {'id': instance_id, 'fake_key': 'fake_value'}
221 self.mock_object(utils, 'find_resource',
222 mock.Mock(return_value=server,
223 side_effect=nova_exception.CommandError))
225 self.assertRaises(exception.ManilaException,
226 self.api.server_get_by_name_or_id,
227 self.ctx, instance_id)
228 utils.find_resource.assert_any_call(mock.ANY, instance_id)
229 utils.find_resource.assert_called_with(mock.ANY, instance_id,
230 all_tenants=True)
232 @ddt.data(
233 {'nova_e': nova_exception.NotFound(404),
234 'manila_e': exception.InstanceNotFound},
235 {'nova_e': nova_exception.BadRequest(400),
236 'manila_e': exception.InvalidInput},
237 )
238 @ddt.unpack
239 def test_server_get_failed(self, nova_e, manila_e):
240 nova.novaclient.side_effect = nova_e
241 instance_id = 'instance_id'
242 self.assertRaises(manila_e, self.api.server_get, self.ctx, instance_id)
244 def test_server_reboot_hard(self):
245 self.mock_object(self.novaclient.servers, 'reboot')
246 self.api.server_reboot(self.ctx, 'id1')
247 self.novaclient.servers.reboot.assert_called_once_with(
248 'id1', nova_servers.REBOOT_HARD)
250 def test_server_reboot_soft(self):
251 self.mock_object(self.novaclient.servers, 'reboot')
252 self.api.server_reboot(self.ctx, 'id1', True)
253 self.novaclient.servers.reboot.assert_called_once_with(
254 'id1', nova_servers.REBOOT_SOFT)
256 def test_instance_volume_attach(self):
257 self.mock_object(self.novaclient.volumes, 'create_server_volume')
258 self.api.instance_volume_attach(self.ctx, 'instance_id',
259 'vol_id', 'device')
260 (self.novaclient.volumes.create_server_volume.
261 assert_called_once_with('instance_id', 'vol_id', 'device'))
263 def test_instance_volume_detach(self):
264 self.mock_object(self.novaclient.volumes, 'delete_server_volume')
265 self.api.instance_volume_detach(self.ctx, 'instance_id',
266 'att_id')
267 (self.novaclient.volumes.delete_server_volume.
268 assert_called_once_with('instance_id', 'att_id'))
270 def test_instance_volumes_list(self):
271 self.mock_object(
272 self.novaclient.volumes, 'get_server_volumes',
273 mock.Mock(return_value=[Volume('id1'), Volume('id2')]))
274 result = self.api.instance_volumes_list(self.ctx, 'instance_id')
275 self.assertEqual(['id1', 'id2'], result)
277 def test_server_update(self):
278 self.mock_object(self.novaclient.servers, 'update')
279 self.api.server_update(self.ctx, 'id1', 'new_name')
280 self.novaclient.servers.update.assert_called_once_with('id1',
281 name='new_name')
283 def test_keypair_import(self):
284 self.mock_object(self.novaclient.keypairs, 'create')
285 self.api.keypair_import(self.ctx, 'keypair_name', 'fake_pub_key')
286 (self.novaclient.keypairs.create.
287 assert_called_once_with('keypair_name', 'fake_pub_key'))
289 def test_keypair_delete(self):
290 self.mock_object(self.novaclient.keypairs, 'delete')
291 self.api.keypair_delete(self.ctx, 'fake_keypair_id')
292 (self.novaclient.keypairs.delete.
293 assert_called_once_with('fake_keypair_id'))
295 def test_keypair_list(self):
296 self.assertEqual([{'id': 'id1'}, {'id': 'id2'}],
297 self.api.keypair_list(self.ctx))
300class ToDictTestCase(test.TestCase):
302 def test_dict_provided(self):
303 fake_dict = {'foo_key': 'foo_value', 'bar_key': 'bar_value'}
304 result = nova._to_dict(fake_dict)
305 self.assertEqual(fake_dict, result)
307 def test_obj_provided_with_to_dict_method(self):
308 expected = {'foo': 'bar'}
310 class FakeObj(object):
311 def __init__(self):
312 self.fake_attr = 'fake_attr_value'
314 def to_dict(self):
315 return expected
317 fake_obj = FakeObj()
318 result = nova._to_dict(fake_obj)
319 self.assertEqual(expected, result)
321 def test_obj_provided_without_to_dict_method(self):
322 expected = {'foo': 'bar'}
324 class FakeObj(object):
325 def __init__(self):
326 self.foo = expected['foo']
328 fake_obj = FakeObj()
329 result = nova._to_dict(fake_obj)
330 self.assertEqual(expected, result)