2 # -*- coding: utf-8 -*-
6 # This file is part of openmano
9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
10 # not use this file except in compliance with the License. You may obtain
11 # a copy of the License at
13 # http://www.apache.org/licenses/LICENSE-2.0
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
18 # License for the specific language governing permissions and limitations
24 Module for testing openmano functionality. It uses openmanoclient.py for invoking openmano
40 from argparse
import ArgumentParser
42 __author__
= "Pablo Montes, Alfonso Tierno"
43 __date__
= "$16-Feb-2017 17:08:16$"
45 version_date
= "Oct 2017"
47 test_config
= {} # used for global variables with the test configuration
50 class test_base(unittest
.TestCase
):
56 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
59 def tearDownClass(cls
):
60 test_config
["test_number"] += 1
63 exec_info
= sys
.exc_info()
64 if exec_info
== (None, None, None):
65 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
67 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
68 logger
.critical("Traceback error",exc_info
=True)
71 def check_instance_scenario_active(uuid
):
72 instance
= test_config
["client"].get_instance(uuid
=uuid
)
74 for net
in instance
['nets']:
75 status
= net
['status']
76 if status
!= 'ACTIVE':
77 return (False, status
)
79 for vnf
in instance
['vnfs']:
82 if status
!= 'ACTIVE':
83 return (False, status
)
90 All unittest classes for code based tests must have prefix 'test_' in order to be taken into account for tests
92 class test_VIM_datacenter_tenant_operations(test_base
):
95 def test_000_create_RO_tenant(self
):
96 self
.__class
__.tenant_name
= _get_random_string(20)
97 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
98 inspect
.currentframe().f_code
.co_name
)
99 self
.__class
__.test_index
+= 1
100 tenant
= test_config
["client"].create_tenant(name
=self
.__class
__.tenant_name
,
101 description
=self
.__class
__.tenant_name
)
102 logger
.debug("{}".format(tenant
))
103 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.tenant_name
)
105 def test_010_list_RO_tenant(self
):
106 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
107 inspect
.currentframe().f_code
.co_name
)
108 self
.__class
__.test_index
+= 1
109 tenant
= test_config
["client"].get_tenant(name
=self
.__class
__.tenant_name
)
110 logger
.debug("{}".format(tenant
))
111 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.tenant_name
)
113 def test_020_delete_RO_tenant(self
):
114 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
115 inspect
.currentframe().f_code
.co_name
)
116 self
.__class
__.test_index
+= 1
117 tenant
= test_config
["client"].delete_tenant(name
=self
.__class
__.tenant_name
)
118 logger
.debug("{}".format(tenant
))
119 assert('deleted' in tenant
.get('result',""))
122 class test_VIM_datacenter_operations(test_base
):
123 datacenter_name
= None
125 def test_000_create_datacenter(self
):
126 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
127 inspect
.currentframe().f_code
.co_name
)
128 self
.__class
__.datacenter_name
= _get_random_string(20)
129 self
.__class
__.test_index
+= 1
130 self
.datacenter
= test_config
["client"].create_datacenter(name
=self
.__class
__.datacenter_name
,
131 vim_url
="http://fakeurl/fake")
132 logger
.debug("{}".format(self
.datacenter
))
133 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name',''), self
.__class
__.datacenter_name
)
135 def test_010_list_datacenter(self
):
136 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
137 inspect
.currentframe().f_code
.co_name
)
139 self
.__class
__.test_index
+= 1
140 self
.datacenter
= test_config
["client"].get_datacenter(all_tenants
=True, name
=self
.__class
__.datacenter_name
)
141 logger
.debug("{}".format(self
.datacenter
))
142 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name', ''), self
.__class
__.datacenter_name
)
144 def test_020_attach_datacenter(self
):
145 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
146 inspect
.currentframe().f_code
.co_name
)
148 self
.__class
__.test_index
+= 1
149 self
.datacenter
= test_config
["client"].attach_datacenter(name
=self
.__class
__.datacenter_name
,
150 vim_tenant_name
='fake')
151 logger
.debug("{}".format(self
.datacenter
))
152 assert ('uuid' in self
.datacenter
.get('datacenter', {}))
154 def test_030_list_attached_datacenter(self
):
155 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
156 inspect
.currentframe().f_code
.co_name
)
158 self
.__class
__.test_index
+= 1
159 self
.datacenter
= test_config
["client"].get_datacenter(all_tenants
=False, name
=self
.__class
__.datacenter_name
)
160 logger
.debug("{}".format(self
.datacenter
))
161 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name', ''), self
.__class
__.datacenter_name
)
163 def test_040_detach_datacenter(self
):
164 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
165 inspect
.currentframe().f_code
.co_name
)
167 self
.__class
__.test_index
+= 1
168 self
.datacenter
= test_config
["client"].detach_datacenter(name
=self
.__class
__.datacenter_name
)
169 logger
.debug("{}".format(self
.datacenter
))
170 assert ('detached' in self
.datacenter
.get('result', ""))
172 def test_050_delete_datacenter(self
):
173 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
174 inspect
.currentframe().f_code
.co_name
)
176 self
.__class
__.test_index
+= 1
177 self
.datacenter
= test_config
["client"].delete_datacenter(name
=self
.__class
__.datacenter_name
)
178 logger
.debug("{}".format(self
.datacenter
))
179 assert('deleted' in self
.datacenter
.get('result',""))
182 class test_VIM_network_operations(test_base
):
183 vim_network_name
= None
184 vim_network_uuid
= None
186 def test_000_create_VIM_network(self
):
187 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
188 inspect
.currentframe().f_code
.co_name
)
189 self
.__class
__.vim_network_name
= _get_random_string(20)
190 self
.__class
__.test_index
+= 1
191 network
= test_config
["client"].vim_action("create", "networks", name
=self
.__class
__.vim_network_name
)
192 logger
.debug("{}".format(network
))
193 self
.__class
__.vim_network_uuid
= network
["network"]["id"]
194 self
.assertEqual(network
.get('network', {}).get('name', ''), self
.__class
__.vim_network_name
)
196 def test_010_list_VIM_networks(self
):
197 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
198 inspect
.currentframe().f_code
.co_name
)
199 self
.__class
__.test_index
+= 1
200 networks
= test_config
["client"].vim_action("list", "networks")
201 logger
.debug("{}".format(networks
))
203 def test_020_get_VIM_network_by_uuid(self
):
204 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
205 inspect
.currentframe().f_code
.co_name
)
207 self
.__class
__.test_index
+= 1
208 network
= test_config
["client"].vim_action("show", "networks", uuid
=self
.__class
__.vim_network_uuid
)
209 logger
.debug("{}".format(network
))
210 self
.assertEqual(network
.get('network', {}).get('name', ''), self
.__class
__.vim_network_name
)
212 def test_030_delete_VIM_network_by_uuid(self
):
213 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
214 inspect
.currentframe().f_code
.co_name
)
216 self
.__class
__.test_index
+= 1
217 network
= test_config
["client"].vim_action("delete", "networks", uuid
=self
.__class
__.vim_network_uuid
)
218 logger
.debug("{}".format(network
))
219 assert ('deleted' in network
.get('result', ""))
222 class test_VIM_image_operations(test_base
):
224 def test_000_list_VIM_images(self
):
225 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
226 inspect
.currentframe().f_code
.co_name
)
227 self
.__class
__.test_index
+= 1
228 images
= test_config
["client"].vim_action("list", "images")
229 logger
.debug("{}".format(images
))
232 The following is a non critical test that will fail most of the times.
233 In case of OpenStack datacenter these tests will only success if RO has access to the admin endpoint
234 This test will only be executed in case it is specifically requested by the user
236 class test_VIM_tenant_operations(test_base
):
237 vim_tenant_name
= None
238 vim_tenant_uuid
= None
242 test_base
.setUpClass(cls
)
243 logger
.warning("In case of OpenStack datacenter these tests will only success "
244 "if RO has access to the admin endpoint")
246 def test_000_create_VIM_tenant(self
):
247 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
248 inspect
.currentframe().f_code
.co_name
)
249 self
.__class
__.vim_tenant_name
= _get_random_string(20)
250 self
.__class
__.test_index
+= 1
251 tenant
= test_config
["client"].vim_action("create", "tenants", name
=self
.__class
__.vim_tenant_name
)
252 logger
.debug("{}".format(tenant
))
253 self
.__class
__.vim_tenant_uuid
= tenant
["tenant"]["id"]
254 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.vim_tenant_name
)
256 def test_010_list_VIM_tenants(self
):
257 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
258 inspect
.currentframe().f_code
.co_name
)
259 self
.__class
__.test_index
+= 1
260 tenants
= test_config
["client"].vim_action("list", "tenants")
261 logger
.debug("{}".format(tenants
))
263 def test_020_get_VIM_tenant_by_uuid(self
):
264 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
265 inspect
.currentframe().f_code
.co_name
)
267 self
.__class
__.test_index
+= 1
268 tenant
= test_config
["client"].vim_action("show", "tenants", uuid
=self
.__class
__.vim_tenant_uuid
)
269 logger
.debug("{}".format(tenant
))
270 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.vim_tenant_name
)
272 def test_030_delete_VIM_tenant_by_uuid(self
):
273 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
274 inspect
.currentframe().f_code
.co_name
)
276 self
.__class
__.test_index
+= 1
277 tenant
= test_config
["client"].vim_action("delete", "tenants", uuid
=self
.__class
__.vim_tenant_uuid
)
278 logger
.debug("{}".format(tenant
))
279 assert ('deleted' in tenant
.get('result', ""))
282 class test_vimconn_connect(test_base
):
284 def test_000_connect(self
):
285 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
286 self
.__class
__.test_index
,
287 inspect
.currentframe().f_code
.co_name
)
289 self
.__class
__.test_index
+= 1
290 if test_config
['vimtype'] == 'vmware':
291 vca_object
= test_config
["vim_conn"].connect()
292 logger
.debug("{}".format(vca_object
))
293 self
.assertIsNotNone(vca_object
)
294 elif test_config
['vimtype'] == 'openstack':
295 test_config
["vim_conn"]._reload
_connection
()
296 network_list
= test_config
["vim_conn"].get_network_list()
297 logger
.debug("{}".format(network_list
))
298 self
.assertIsNotNone(network_list
)
300 class test_vimconn_new_network(test_base
):
303 def test_000_new_network(self
):
304 self
.__class
__.network_name
= _get_random_string(20)
305 network_type
= 'bridge'
307 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
308 self
.__class
__.test_index
, inspect
.currentframe().f_code
.co_name
)
309 self
.__class
__.test_index
+= 1
311 network
, _
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
312 net_type
=network_type
)
313 self
.__class
__.network_id
= network
314 logger
.debug("{}".format(network
))
316 network_list
= test_config
["vim_conn"].get_network_list()
317 for net
in network_list
:
318 if self
.__class
__.network_name
in net
.get('name'):
319 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
320 self
.assertEqual(net
.get('type'), network_type
)
322 # Deleting created network
323 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
325 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
327 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
329 def test_010_new_network_by_types(self
):
331 network_types
= ['data','bridge','mgmt']
332 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
333 self
.__class
__.test_index
,
334 inspect
.currentframe().f_code
.co_name
)
335 self
.__class
__.test_index
+= 1
336 for net_type
in network_types
:
337 self
.__class
__.network_name
= _get_random_string(20)
338 network_id
, _
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
341 delete_net_ids
.append(network_id
)
342 logger
.debug("{}".format(network_id
))
344 network_list
= test_config
["vim_conn"].get_network_list()
345 for net
in network_list
:
346 if self
.__class
__.network_name
in net
.get('name'):
347 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
348 if net_type
in net
.get('type'):
349 self
.assertEqual(net
.get('type'), net_type
)
351 self
.assertNotEqual(net
.get('type'), net_type
)
353 # Deleting created network
354 for net_id
in delete_net_ids
:
355 result
= test_config
["vim_conn"].delete_network(net_id
)
357 logger
.info("Network id {} sucessfully deleted".format(net_id
))
359 logger
.info("Failed to delete network id {}".format(net_id
))
361 def test_020_new_network_by_ipprofile(self
):
362 test_directory_content
= os
.listdir(test_config
["test_directory"])
364 for dir_name
in test_directory_content
:
365 if dir_name
== 'simple_multi_vnfc':
366 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
367 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
370 for vnfd
in vnfd_files
:
371 with
open(vnfd
, 'r') as stream
:
372 vnf_descriptor
= yaml
.load(stream
)
374 internal_connections_list
= vnf_descriptor
['vnf']['internal-connections']
375 for item
in internal_connections_list
:
376 if 'ip-profile' in item
:
377 version
= item
['ip-profile']['ip-version']
378 dhcp_count
= item
['ip-profile']['dhcp']['count']
379 dhcp_enabled
= item
['ip-profile']['dhcp']['enabled']
380 dhcp_start_address
= item
['ip-profile']['dhcp']['start-address']
381 subnet_address
= item
['ip-profile']['subnet-address']
384 self
.__class
__.network_name
= _get_random_string(20)
385 ip_profile
= {'dhcp_count': dhcp_count
,
386 'dhcp_enabled': dhcp_enabled
,
387 'dhcp_start_address': dhcp_start_address
,
388 'ip_version': version
,
389 'subnet_address': subnet_address
391 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
392 self
.__class
__.test_index
,
393 inspect
.currentframe().f_code
.co_name
)
394 self
.__class
__.test_index
+= 1
395 network
, _
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
397 ip_profile
=ip_profile
)
398 self
.__class
__.network_id
= network
399 logger
.debug("{}".format(network
))
401 network_list
= test_config
["vim_conn"].get_network_list()
402 for net
in network_list
:
403 if self
.__class
__.network_name
in net
.get('name'):
404 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
406 # Deleting created network
407 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
409 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
411 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
413 def test_030_new_network_by_isshared(self
):
414 self
.__class
__.network_name
= _get_random_string(20)
416 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
417 self
.__class
__.test_index
,
418 inspect
.currentframe().f_code
.co_name
)
419 self
.__class
__.test_index
+= 1
420 network
, _
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
423 self
.__class
__.network_id
= network
424 logger
.debug("{}".format(network
))
426 network_list
= test_config
["vim_conn"].get_network_list()
427 for net
in network_list
:
428 if self
.__class
__.network_name
in net
.get('name'):
429 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
430 self
.assertEqual(net
.get('shared'), shared
)
432 # Deleting created network
433 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
435 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
437 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
439 def test_040_new_network_by_negative(self
):
440 self
.__class
__.network_name
= _get_random_string(20)
441 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
442 self
.__class
__.test_index
,
443 inspect
.currentframe().f_code
.co_name
)
444 self
.__class
__.test_index
+= 1
445 network
, _
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
446 net_type
='unknowntype')
447 self
.__class
__.network_id
= network
448 logger
.debug("{}".format(network
))
449 network_list
= test_config
["vim_conn"].get_network_list()
450 for net
in network_list
:
451 if self
.__class
__.network_name
in net
.get('name'):
452 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
454 # Deleting created network
455 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
457 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
459 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
461 def test_050_refresh_nets_status(self
):
462 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
463 self
.__class
__.test_index
,
464 inspect
.currentframe().f_code
.co_name
)
465 self
.__class
__.test_index
+= 1
466 # creating new network
467 network_name
= _get_random_string(20)
469 network_id
, _
= test_config
["vim_conn"].new_network(net_name
=network_name
,
472 net_dict
= test_config
["vim_conn"].refresh_nets_status([network_id
])
473 for attr
in net_dict
[network_id
]:
475 self
.assertEqual(net_dict
[network_id
][attr
], 'ACTIVE')
477 # Deleting created network
478 result
= test_config
["vim_conn"].delete_network(network_id
)
480 logger
.info("Network id {} sucessfully deleted".format(network_id
))
482 logger
.info("Failed to delete network id {}".format(network_id
))
484 def test_060_refresh_nets_status_negative(self
):
485 unknown_net_id
= str(uuid
.uuid4())
486 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
487 self
.__class
__.test_index
,
488 inspect
.currentframe().f_code
.co_name
)
489 self
.__class
__.test_index
+= 1
492 net_dict
= test_config
["vim_conn"].refresh_nets_status([unknown_net_id
])
493 if test_config
['vimtype'] == 'openstack':
494 self
.assertEqual(net_dict
[unknown_net_id
]['status'], 'DELETED')
496 # TODO : Fix vmware connector to return status DELETED as per vimconn.py
497 self
.assertEqual(net_dict
, {})
499 class test_vimconn_get_network_list(test_base
):
503 # creating new network
504 self
.__class
__.network_name
= _get_random_string(20)
505 self
.__class
__.net_type
= 'bridge'
506 network
, _
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
507 net_type
=self
.__class
__.net_type
)
508 self
.__class
__.network_id
= network
509 logger
.debug("{}".format(network
))
512 test_base
.tearDown(self
)
514 # Deleting created network
515 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
517 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
519 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
521 def test_000_get_network_list(self
):
522 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
523 self
.__class
__.test_index
,
524 inspect
.currentframe().f_code
.co_name
)
525 self
.__class
__.test_index
+= 1
527 network_list
= test_config
["vim_conn"].get_network_list()
528 for net
in network_list
:
529 if self
.__class
__.network_name
in net
.get('name'):
530 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
531 self
.assertEqual(net
.get('type'), self
.__class
__.net_type
)
532 self
.assertEqual(net
.get('status'), 'ACTIVE')
533 self
.assertEqual(net
.get('shared'), False)
535 def test_010_get_network_list_by_name(self
):
536 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
537 self
.__class
__.test_index
,
538 inspect
.currentframe().f_code
.co_name
)
539 self
.__class
__.test_index
+= 1
541 if test_config
['vimtype'] == 'openstack':
542 network_name
= test_config
['vim_conn'].get_network(self
.__class
__.network_id
)['name']
544 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
546 # find network from list by it's name
547 new_network_list
= test_config
["vim_conn"].get_network_list({'name': network_name
})
548 for list_item
in new_network_list
:
549 if self
.__class
__.network_name
in list_item
.get('name'):
550 self
.assertEqual(network_name
, list_item
.get('name'))
551 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
552 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
554 def test_020_get_network_list_by_id(self
):
555 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
556 self
.__class
__.test_index
,
557 inspect
.currentframe().f_code
.co_name
)
558 self
.__class
__.test_index
+= 1
560 # find network from list by it's id
561 new_network_list
= test_config
["vim_conn"].get_network_list({'id':self
.__class
__.network_id
})
562 for list_item
in new_network_list
:
563 if self
.__class
__.network_id
in list_item
.get('id'):
564 self
.assertEqual(self
.__class
__.network_id
, list_item
.get('id'))
565 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
566 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
568 def test_030_get_network_list_by_shared(self
):
570 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
571 self
.__class
__.test_index
,
572 inspect
.currentframe().f_code
.co_name
)
573 self
.__class
__.test_index
+= 1
575 if test_config
['vimtype'] == 'openstack':
576 network_name
= test_config
['vim_conn'].get_network(self
.__class
__.network_id
)['name']
578 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
579 # find network from list by it's shared value
580 new_network_list
= test_config
["vim_conn"].get_network_list({'shared':Shared
,
581 'name':network_name
})
582 for list_item
in new_network_list
:
583 if list_item
.get('shared') == Shared
:
584 self
.assertEqual(list_item
.get('shared'), Shared
)
585 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
586 self
.assertEqual(network_name
, list_item
.get('name'))
588 def test_040_get_network_list_by_tenant_id(self
):
589 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
590 self
.__class
__.test_index
,
591 inspect
.currentframe().f_code
.co_name
)
592 self
.__class
__.test_index
+= 1
594 tenant_list
= test_config
["vim_conn"].get_tenant_list()
595 if test_config
['vimtype'] == 'openstack':
596 network_name
= test_config
['vim_conn'].get_network(self
.__class
__.network_id
)['name']
598 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
600 for tenant_item
in tenant_list
:
601 if test_config
['tenant'] == tenant_item
.get('name'):
602 # find network from list by it's tenant id
603 tenant_id
= tenant_item
.get('id')
604 new_network_list
= test_config
["vim_conn"].get_network_list({'tenant_id':tenant_id
,
605 'name':network_name
})
606 for list_item
in new_network_list
:
607 self
.assertEqual(tenant_id
, list_item
.get('tenant_id'))
608 self
.assertEqual(network_name
, list_item
.get('name'))
609 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
610 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
612 def test_050_get_network_list_by_status(self
):
613 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
614 self
.__class
__.test_index
,
615 inspect
.currentframe().f_code
.co_name
)
616 self
.__class
__.test_index
+= 1
619 if test_config
['vimtype'] == 'openstack':
620 network_name
= test_config
['vim_conn'].get_network(self
.__class
__.network_id
)['name']
622 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
624 # find network from list by it's status
625 new_network_list
= test_config
["vim_conn"].get_network_list({'status':status
,
626 'name': network_name
})
627 for list_item
in new_network_list
:
628 self
.assertIn(self
.__class
__.network_name
, list_item
.get('name'))
629 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
630 self
.assertEqual(list_item
.get('status'), status
)
632 def test_060_get_network_list_by_negative(self
):
633 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
634 self
.__class
__.test_index
,
635 inspect
.currentframe().f_code
.co_name
)
636 self
.__class
__.test_index
+= 1
638 network_list
= test_config
["vim_conn"].get_network_list({'name': 'unknown_name'})
639 self
.assertEqual(network_list
, [])
641 class test_vimconn_get_network(test_base
):
645 # creating new network
646 self
.__class
__.network_name
= _get_random_string(20)
647 self
.__class
__.net_type
= 'bridge'
648 network
, _
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
649 net_type
=self
.__class
__.net_type
)
650 self
.__class
__.network_id
= network
651 logger
.debug("{}".format(network
))
654 test_base
.tearDown(self
)
656 # Deleting created network
657 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
659 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
661 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
663 def test_000_get_network(self
):
664 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
665 self
.__class
__.test_index
,
666 inspect
.currentframe().f_code
.co_name
)
667 self
.__class
__.test_index
+= 1
669 network_info
= test_config
["vim_conn"].get_network(self
.__class
__.network_id
)
670 self
.assertEqual(network_info
.get('status'), 'ACTIVE')
671 self
.assertIn(self
.__class
__.network_name
, network_info
.get('name'))
672 self
.assertEqual(network_info
.get('type'), self
.__class
__.net_type
)
673 self
.assertEqual(network_info
.get('id'), self
.__class
__.network_id
)
675 def test_010_get_network_negative(self
):
676 Non_exist_id
= str(uuid
.uuid4())
677 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
678 self
.__class
__.test_index
,
679 inspect
.currentframe().f_code
.co_name
)
680 self
.__class
__.test_index
+= 1
681 with self
.assertRaises(Exception) as context
:
682 test_config
["vim_conn"].get_network(Non_exist_id
)
684 self
.assertEqual((context
.exception
).http_code
, 404)
686 class test_vimconn_delete_network(test_base
):
689 def test_000_delete_network(self
):
691 self
.__class
__.network_name
= _get_random_string(20)
692 self
.__class
__.net_type
= 'bridge'
693 network
, _
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
694 net_type
=self
.__class
__.net_type
)
695 self
.__class
__.network_id
= network
696 logger
.debug("{}".format(network
))
698 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
699 self
.__class
__.test_index
,
700 inspect
.currentframe().f_code
.co_name
)
701 self
.__class
__.test_index
+= 1
703 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
705 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
707 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
709 # after deleting network we check in network list
710 network_list
= test_config
["vim_conn"].get_network_list({ 'id':self
.__class
__.network_id
})
711 self
.assertEqual(network_list
, [])
713 def test_010_delete_network_negative(self
):
714 Non_exist_id
= str(uuid
.uuid4())
716 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
717 self
.__class
__.test_index
,
718 inspect
.currentframe().f_code
.co_name
)
719 self
.__class
__.test_index
+= 1
721 with self
.assertRaises(Exception) as context
:
722 test_config
["vim_conn"].delete_network(Non_exist_id
)
724 self
.assertEqual((context
.exception
).http_code
, 404)
726 class test_vimconn_get_flavor(test_base
):
728 def test_000_get_flavor(self
):
729 test_directory_content
= os
.listdir(test_config
["test_directory"])
731 for dir_name
in test_directory_content
:
732 if dir_name
== 'simple_linux':
733 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
734 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
737 for vnfd
in vnfd_files
:
738 with
open(vnfd
, 'r') as stream
:
739 vnf_descriptor
= yaml
.load(stream
)
741 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
742 for item
in vnfc_list
:
743 if 'ram' in item
and 'vcpus' in item
and 'disk' in item
:
745 vcpus
= item
['vcpus']
749 'name' : _get_random_string(20),
755 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
756 self
.__class
__.test_index
,
757 inspect
.currentframe().f_code
.co_name
)
758 self
.__class
__.test_index
+= 1
760 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
762 result
= test_config
["vim_conn"].get_flavor(flavor_id
)
763 self
.assertEqual(ram
, result
['ram'])
764 self
.assertEqual(vcpus
, result
['vcpus'])
765 self
.assertEqual(disk
, result
['disk'])
768 result
= test_config
["vim_conn"].delete_flavor(flavor_id
)
770 logger
.info("Flavor id {} sucessfully deleted".format(result
))
772 logger
.info("Failed to delete flavor id {}".format(result
))
774 def test_010_get_flavor_negative(self
):
775 Non_exist_flavor_id
= str(uuid
.uuid4())
777 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
778 self
.__class
__.test_index
,
779 inspect
.currentframe().f_code
.co_name
)
780 self
.__class
__.test_index
+= 1
782 with self
.assertRaises(Exception) as context
:
783 test_config
["vim_conn"].get_flavor(Non_exist_flavor_id
)
785 self
.assertEqual((context
.exception
).http_code
, 404)
787 class test_vimconn_new_flavor(test_base
):
790 def test_000_new_flavor(self
):
791 flavor_data
= {'name': _get_random_string(20), 'ram': 1024, 'vpcus': 1, 'disk': 10}
793 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
794 self
.__class
__.test_index
,
795 inspect
.currentframe().f_code
.co_name
)
796 self
.__class
__.test_index
+= 1
799 self
.__class
__.flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
800 self
.assertIsInstance(self
.__class
__.flavor_id
, (str, unicode))
801 self
.assertIsInstance(uuid
.UUID(self
.__class
__.flavor_id
), uuid
.UUID
)
803 def test_010_delete_flavor(self
):
804 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
805 self
.__class
__.test_index
,
806 inspect
.currentframe().f_code
.co_name
)
807 self
.__class
__.test_index
+= 1
810 result
= test_config
["vim_conn"].delete_flavor(self
.__class
__.flavor_id
)
812 logger
.info("Flavor id {} sucessfully deleted".format(result
))
814 logger
.error("Failed to delete flavor id {}".format(result
))
815 raise Exception ("Failed to delete created flavor")
817 def test_020_new_flavor_negative(self
):
818 Invalid_flavor_data
= {'ram': '1024', 'vcpus': 2.0, 'disk': 2.0}
820 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
821 self
.__class
__.test_index
,
822 inspect
.currentframe().f_code
.co_name
)
823 self
.__class
__.test_index
+= 1
825 with self
.assertRaises(Exception) as context
:
826 test_config
["vim_conn"].new_flavor(Invalid_flavor_data
)
828 self
.assertEqual((context
.exception
).http_code
, 400)
830 def test_030_delete_flavor_negative(self
):
831 Non_exist_flavor_id
= str(uuid
.uuid4())
833 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
834 self
.__class
__.test_index
,
835 inspect
.currentframe().f_code
.co_name
)
836 self
.__class
__.test_index
+= 1
838 with self
.assertRaises(Exception) as context
:
839 test_config
["vim_conn"].delete_flavor(Non_exist_flavor_id
)
841 self
.assertEqual((context
.exception
).http_code
, 404)
843 class test_vimconn_new_image(test_base
):
845 def test_000_new_image(self
):
846 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
847 self
.__class
__.test_index
,
848 inspect
.currentframe().f_code
.co_name
)
849 self
.__class
__.test_index
+= 1
851 image_path
= test_config
['image_path']
853 self
.__class
__.image_id
= test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : image_path
, 'metadata': {'upload_location':None} })
856 self
.assertIsInstance(self
.__class
__.image_id
, (str, unicode))
857 self
.assertIsInstance(uuid
.UUID(self
.__class
__.image_id
), uuid
.UUID
)
859 self
.skipTest("Skipping test as image file not present at RO container")
861 def test_010_new_image_negative(self
):
862 Non_exist_image_path
= '/temp1/cirros.ovf'
864 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
865 self
.__class
__.test_index
,
866 inspect
.currentframe().f_code
.co_name
)
867 self
.__class
__.test_index
+= 1
869 with self
.assertRaises(Exception) as context
:
870 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
872 self
.assertEqual((context
.exception
).http_code
, 400)
874 def test_020_delete_image(self
):
875 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
876 self
.__class
__.test_index
,
877 inspect
.currentframe().f_code
.co_name
)
878 self
.__class
__.test_index
+= 1
880 image_id
= test_config
["vim_conn"].delete_image(self
.__class
__.image_id
)
882 self
.assertIsInstance(image_id
, (str, unicode))
884 def test_030_delete_image_negative(self
):
885 Non_exist_image_id
= str(uuid
.uuid4())
887 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
888 self
.__class
__.test_index
,
889 inspect
.currentframe().f_code
.co_name
)
890 self
.__class
__.test_index
+= 1
892 with self
.assertRaises(Exception) as context
:
893 test_config
["vim_conn"].delete_image(Non_exist_image_id
)
895 self
.assertEqual((context
.exception
).http_code
, 404)
897 class test_vimconn_get_image_id_from_path(test_base
):
899 def test_000_get_image_id_from_path(self
):
900 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
901 self
.__class
__.test_index
,
902 inspect
.currentframe().f_code
.co_name
)
903 self
.__class
__.test_index
+= 1
905 image_path
= test_config
['image_path']
907 image_id
= test_config
["vim_conn"].get_image_id_from_path( image_path
)
908 self
.assertEqual(type(image_id
),str)
910 self
.skipTest("Skipping test as image file not present at RO container")
912 def test_010_get_image_id_from_path_negative(self
):
913 Non_exist_image_path
= '/temp1/cirros.ovf'
915 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
916 self
.__class
__.test_index
,
917 inspect
.currentframe().f_code
.co_name
)
918 self
.__class
__.test_index
+= 1
920 with self
.assertRaises(Exception) as context
:
921 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
923 self
.assertEqual((context
.exception
).http_code
, 400)
925 class test_vimconn_get_image_list(test_base
):
929 def test_000_get_image_list(self
):
930 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
931 self
.__class
__.test_index
,
932 inspect
.currentframe().f_code
.co_name
)
933 self
.__class
__.test_index
+= 1
934 image_list
= test_config
["vim_conn"].get_image_list()
936 for item
in image_list
:
938 self
.__class
__.image_name
= item
['name']
939 self
.__class
__.image_id
= item
['id']
940 self
.assertIsInstance(self
.__class
__.image_name
, (str, unicode))
941 self
.assertIsInstance(self
.__class
__.image_id
, (str, unicode))
943 def test_010_get_image_list_by_name(self
):
944 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
945 self
.__class
__.test_index
,
946 inspect
.currentframe().f_code
.co_name
)
947 self
.__class
__.test_index
+= 1
949 image_list
= test_config
["vim_conn"].get_image_list({'name': self
.__class
__.image_name
})
951 for item
in image_list
:
952 self
.assertIsInstance(item
['id'], (str, unicode))
953 self
.assertIsInstance(item
['name'], (str, unicode))
954 self
.assertEqual(item
['id'], self
.__class
__.image_id
)
955 self
.assertEqual(item
['name'], self
.__class
__.image_name
)
957 def test_020_get_image_list_by_id(self
):
958 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
959 self
.__class
__.test_index
,
960 inspect
.currentframe().f_code
.co_name
)
961 self
.__class
__.test_index
+= 1
963 filter_image_list
= test_config
["vim_conn"].get_image_list({'id': self
.__class
__.image_id
})
965 for item1
in filter_image_list
:
966 self
.assertIsInstance(item1
['id'], (str, unicode))
967 self
.assertIsInstance(item1
['name'], (str, unicode))
968 self
.assertEqual(item1
['id'], self
.__class
__.image_id
)
969 self
.assertEqual(item1
['name'], self
.__class
__.image_name
)
971 def test_030_get_image_list_negative(self
):
972 Non_exist_image_id
= uuid
.uuid4()
973 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
974 self
.__class
__.test_index
,
975 inspect
.currentframe().f_code
.co_name
)
976 self
.__class
__.test_index
+= 1
977 image_list
= test_config
["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id
})
979 self
.assertIsNotNone(image_list
, None)
980 self
.assertEqual(image_list
, [])
982 class test_vimconn_new_vminstance(test_base
):
991 self
.__class
__.network_name
= _get_random_string(20)
992 self
.__class
__.net_type
= 'bridge'
994 self
.__class
__.network_id
, _
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
995 net_type
=self
.__class
__.net_type
)
996 # find image name and image id
997 if test_config
['image_name']:
998 image_list
= test_config
['vim_conn'].get_image_list({'name': test_config
['image_name']})
999 if len(image_list
) == 0:
1000 raise Exception("Image {} is not found at VIM".format(test_config
['image_name']))
1002 self
.__class
__.image_id
= image_list
[0]['id']
1004 image_list
= test_config
['vim_conn'].get_image_list()
1005 if len(image_list
) == 0:
1006 raise Exception("Not found any image at VIM")
1008 self
.__class
__.image_id
= image_list
[0]['id']
1011 test_base
.tearDown(self
)
1012 # Deleting created network
1013 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
1015 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
1017 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
1019 def test_000_new_vminstance(self
):
1020 vpci
= "0000:00:11.0"
1023 flavor_data
= {'name': _get_random_string(20), 'ram': 1024, 'vcpus': 1, 'disk': 10}
1026 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1029 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1030 self
.__class
__.test_index
,
1031 inspect
.currentframe().f_code
.co_name
)
1032 self
.__class
__.test_index
+= 1
1034 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'vpci': vpci
, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1036 self
.__class
__.instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', description
='', start
=False, image_id
=self
.__class
__.image_id
, flavor_id
=flavor_id
, net_list
=net_list
)
1038 self
.assertIsInstance(self
.__class
__.instance_id
, (str, unicode))
1040 def test_010_new_vminstance_by_model(self
):
1041 flavor_data
= {'name': _get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1042 model_name
= 'e1000'
1046 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1048 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1049 self
.__class
__.test_index
,
1050 inspect
.currentframe().f_code
.co_name
)
1051 self
.__class
__.test_index
+= 1
1053 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'model': model_name
, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1055 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', description
='', start
=False, image_id
=self
.__class
__.image_id
,flavor_id
=flavor_id
,net_list
=net_list
)
1057 self
.assertIsInstance(instance_id
, (str, unicode))
1059 # Deleting created vm instance
1060 logger
.info("Deleting created vm intance")
1061 test_config
["vim_conn"].delete_vminstance(instance_id
)
1064 def test_020_new_vminstance_by_net_use(self
):
1065 flavor_data
= {'name': _get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1070 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1072 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1073 self
.__class
__.test_index
,
1074 inspect
.currentframe().f_code
.co_name
)
1075 self
.__class
__.test_index
+= 1
1077 net_list
= [{'use': net_use
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1079 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', description
='', start
=False, image_id
=self
.__class
__.image_id
,disk_list
=None,
1080 flavor_id
=flavor_id
,
1082 self
.assertIsInstance(instance_id
, (str, unicode))
1084 # Deleting created vm instance
1085 logger
.info("Deleting created vm intance")
1086 test_config
["vim_conn"].delete_vminstance(instance_id
)
1089 def test_030_new_vminstance_by_net_type(self
):
1090 flavor_data
= {'name':_get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1095 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1097 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1098 self
.__class
__.test_index
,
1099 inspect
.currentframe().f_code
.co_name
)
1100 self
.__class
__.test_index
+= 1
1102 if test_config
['vimtype'] == 'vmware':
1103 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True,
1104 'type': _type
, 'net_id': self
.__class
__.network_id
}]
1106 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1107 flavor_id
=flavor_id
,
1109 self
.assertEqual(type(instance_id
),str)
1111 if test_config
['vimtype'] == 'openstack':
1112 # create network of type data
1113 network_name
= _get_random_string(20)
1116 network_id
, _
= test_config
["vim_conn"].new_network(net_name
=network_name
,
1118 net_list
= [{'use': net_type
, 'name': name
, 'floating_ip': False, 'port_security': True,
1119 'type': _type
, 'net_id': network_id
}]
1121 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', description
='', start
=False,
1122 image_id
=self
.__class
__.image_id
, disk_list
=None,
1123 flavor_id
=flavor_id
,
1126 self
.assertEqual(type(instance_id
), unicode)
1128 # delete created network
1129 result
= test_config
["vim_conn"].delete_network(network_id
)
1131 logger
.info("Network id {} sucessfully deleted".format(network_id
))
1133 logger
.info("Failed to delete network id {}".format(network_id
))
1135 # Deleting created vm instance
1136 logger
.info("Deleting created vm intance")
1137 test_config
["vim_conn"].delete_vminstance(instance_id
)
1140 def test_040_new_vminstance_by_cloud_config(self
):
1141 flavor_data
= {'name': _get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1143 user_name
= 'test_user'
1145 key_pairs
= ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com']
1147 users_data
= [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com'], 'name': user_name
}]
1149 cloud_data
= {'config-files': [{'content': 'auto enp0s3\niface enp0s3 inet dhcp\n', 'dest': '/etc/network/interfaces.d/enp0s3.cfg', 'owner': 'root:root', 'permissions': '0644'}, {'content': '#! /bin/bash\nls -al >> /var/log/osm.log\n', 'dest': '/etc/rc.local', 'permissions': '0755'}, {'content': 'file content', 'dest': '/etc/test_delete'}], 'boot-data-drive': True, 'key-pairs': key_pairs
, 'users': users_data
}
1152 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1154 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1155 self
.__class
__.test_index
,
1156 inspect
.currentframe().f_code
.co_name
)
1157 self
.__class
__.test_index
+= 1
1159 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1161 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Cloud_vm', description
='', start
=False,
1162 image_id
=self
.__class
__.image_id
, flavor_id
=flavor_id
,net_list
=net_list
,cloud_config
=cloud_data
)
1164 self
.assertIsInstance(instance_id
, (str, unicode))
1166 # Deleting created vm instance
1167 logger
.info("Deleting created vm intance")
1168 test_config
["vim_conn"].delete_vminstance(instance_id
)
1171 def test_050_new_vminstance_by_disk_list(self
):
1172 flavor_data
= {'name':_get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1175 device_data
= [{'image_id': self
.__class
__.image_id
, 'size': '10'}]
1178 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1180 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1181 self
.__class
__.test_index
,
1182 inspect
.currentframe().f_code
.co_name
)
1183 self
.__class
__.test_index
+= 1
1185 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1187 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='VM_test1', description
='', start
=False, image_id
=self
.__class
__.image_id
,
1188 flavor_id
=flavor_id
,
1190 disk_list
=device_data
)
1192 self
.assertIsInstance(instance_id
, (str, unicode))
1193 # Deleting created vm instance
1194 logger
.info("Deleting created vm intance")
1195 test_config
["vim_conn"].delete_vminstance(instance_id
)
1198 def test_060_new_vminstance_negative(self
):
1199 unknown_flavor_id
= str(uuid
.uuid4())
1200 unknown_image_id
= str(uuid
.uuid4())
1203 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1204 self
.__class
__.test_index
,
1205 inspect
.currentframe().f_code
.co_name
)
1206 self
.__class
__.test_index
+= 1
1208 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1210 with self
.assertRaises(Exception) as context
:
1211 test_config
["vim_conn"].new_vminstance(name
='Test1_vm', description
='', start
=False, image_id
=unknown_image_id
,
1212 flavor_id
=unknown_flavor_id
,
1215 self
.assertIn((context
.exception
).http_code
, (400, 404))
1218 def test_070_get_vminstance(self
):
1219 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1220 self
.__class
__.test_index
,
1221 inspect
.currentframe().f_code
.co_name
)
1222 self
.__class
__.test_index
+= 1
1224 # Get instance by its id
1225 vm_info
= test_config
["vim_conn"].get_vminstance(self
.__class
__.instance_id
)
1227 if test_config
['vimtype'] == 'vmware':
1228 for attr
in vm_info
:
1229 if attr
== 'status':
1230 self
.assertEqual(vm_info
[attr
], 'ACTIVE')
1231 if attr
== 'hostId':
1232 self
.assertEqual(type(vm_info
[attr
]), str)
1233 if attr
== 'interfaces':
1234 self
.assertEqual(type(vm_info
[attr
]), list)
1235 self
.assertEqual(vm_info
[attr
][0]['IsConnected'], 'true')
1236 if attr
== 'IsEnabled':
1237 self
.assertEqual(vm_info
[attr
], 'true')
1239 def test_080_get_vminstance_negative(self
):
1240 unknown_instance_id
= str(uuid
.uuid4())
1242 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1243 self
.__class
__.test_index
,
1244 inspect
.currentframe().f_code
.co_name
)
1245 self
.__class
__.test_index
+= 1
1247 with self
.assertRaises(Exception) as context
:
1248 test_config
["vim_conn"].get_vminstance(unknown_instance_id
)
1250 self
.assertEqual((context
.exception
).http_code
, 404)
1252 def test_090_refresh_vms_status(self
):
1253 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1254 self
.__class
__.test_index
,
1255 inspect
.currentframe().f_code
.co_name
)
1256 self
.__class
__.test_index
+= 1
1258 if test_config
['vimtype'] == 'vmware':
1260 vm_list
.append(self
.__class
__.instance_id
)
1263 vm_info
= test_config
["vim_conn"].refresh_vms_status(vm_list
)
1264 for attr
in vm_info
[self
.__class
__.instance_id
]:
1265 if attr
== 'status':
1266 self
.assertEqual(vm_info
[self
.__class
__.instance_id
][attr
], 'ACTIVE')
1267 if attr
== 'interfaces':
1268 self
.assertEqual(type(vm_info
[self
.__class
__.instance_id
][attr
]), list)
1270 if test_config
['vimtype'] == 'openstack':
1271 vpci
= "0000:00:11.0"
1274 flavor_data
= {'name': _get_random_string(20), 'ram': 1024, 'vcpus': 1, 'disk': 10}
1277 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1278 # create new vm instance
1279 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'vpci': vpci
, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1281 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', description
='', start
=False, image_id
=self
.__class
__.image_id
, flavor_id
=flavor_id
, net_list
=net_list
)
1285 vm_list
.append(instance_id
)
1288 vm_info
= test_config
["vim_conn"].refresh_vms_status(vm_list
)
1289 for attr
in vm_info
[instance_id
]:
1290 if attr
== 'status':
1291 self
.assertEqual(vm_info
[instance_id
][attr
], 'ACTIVE')
1292 if attr
== 'interfaces':
1293 self
.assertEqual(type(vm_info
[instance_id
][attr
]), list)
1295 #Deleting created vm instance
1296 logger
.info("Deleting created vm intance")
1297 test_config
["vim_conn"].delete_vminstance(instance_id
)
1301 def test_100_refresh_vms_status_negative(self
):
1302 unknown_id
= str(uuid
.uuid4())
1304 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1305 self
.__class
__.test_index
,
1306 inspect
.currentframe().f_code
.co_name
)
1307 self
.__class
__.test_index
+= 1
1309 vm_dict
= test_config
["vim_conn"].refresh_vms_status([unknown_id
])
1311 if test_config
['vimtype'] == 'vmware':
1312 self
.assertEqual(vm_dict
,{})
1314 if test_config
['vimtype'] == 'openstack':
1315 self
.assertEqual(vm_dict
[unknown_id
]['status'], 'DELETED')
1317 def test_110_action_vminstance(self
):
1318 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1319 self
.__class
__.test_index
,
1320 inspect
.currentframe().f_code
.co_name
)
1321 self
.__class
__.test_index
+= 1
1323 if test_config
['vimtype'] == 'vmware':
1324 action_list
= ['shutdown', 'start', 'shutoff', 'rebuild', 'pause', 'resume']
1325 # various action on vminstace
1326 for action
in action_list
:
1327 instance_id
= test_config
["vim_conn"].action_vminstance(self
.__class
__.instance_id
,
1329 self
.assertEqual(instance_id
, self
.__class
__.instance_id
)
1331 if test_config
['vimtype'] == 'openstack':
1332 # create new vm instance
1333 vpci
= "0000:00:11.0"
1336 flavor_data
= {'name': _get_random_string(20), 'ram': 1024, 'vcpus': 1, 'disk': 10}
1339 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1341 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'vpci': vpci
, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1343 new_instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', description
='', start
=False, image_id
=self
.__class
__.image_id
, flavor_id
=flavor_id
, net_list
=net_list
)
1345 action_list
= ['shutdown','start','shutoff','rebuild','start','pause','start']
1347 # various action on vminstace
1348 for action
in action_list
:
1349 # sleep for sometime till status is changed
1351 instance_id
= test_config
["vim_conn"].action_vminstance(new_instance_id
,
1354 self
.assertTrue(instance_id
is None)
1356 # Deleting created vm instance
1357 logger
.info("Deleting created vm intance")
1358 test_config
["vim_conn"].delete_vminstance(new_instance_id
)
1361 def test_120_action_vminstance_negative(self
):
1362 non_exist_id
= str(uuid
.uuid4())
1363 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1364 self
.__class
__.test_index
,
1365 inspect
.currentframe().f_code
.co_name
)
1366 self
.__class
__.test_index
+= 1
1369 with self
.assertRaises(Exception) as context
:
1370 test_config
["vim_conn"].action_vminstance(non_exist_id
, { action
: None})
1372 self
.assertEqual((context
.exception
).http_code
, 404)
1375 def test_130_delete_vminstance(self
):
1376 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1377 self
.__class
__.test_index
,
1378 inspect
.currentframe().f_code
.co_name
)
1379 self
.__class
__.test_index
+= 1
1381 # Deleting created vm instance
1382 logger
.info("Deleting created vm instance")
1383 test_config
["vim_conn"].delete_vminstance(self
.__class
__.instance_id
)
1386 def test_140_new_vminstance_sriov(self
):
1387 logger
.info("Testing creation of sriov vm instance using {}".format(test_config
['sriov_net_name']))
1388 flavor_data
= {'name': _get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1392 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1394 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1395 self
.__class
__.test_index
,
1396 inspect
.currentframe().f_code
.co_name
)
1397 self
.__class
__.test_index
+= 1
1399 sriov_net_name
= test_config
['sriov_net_name']
1400 new_network_list
= test_config
["vim_conn"].get_network_list({'name': sriov_net_name
})
1401 for list_item
in new_network_list
:
1402 self
.assertEqual(sriov_net_name
, list_item
.get('name'))
1403 self
.__class
__.sriov_network_id
= list_item
.get('id')
1405 net_list
= [{'use': 'data', 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'VF', 'net_id': self
.__class
__.sriov_network_id
}]
1407 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_sriov_vm', description
='', start
=False, image_id
=self
.__class
__.image_id
, flavor_id
=flavor_id
, net_list
=net_list
)
1409 self
.assertIsInstance(instance_id
, (str, unicode))
1411 logger
.info("Waiting for created sriov-vm intance")
1413 # Deleting created vm instance
1414 logger
.info("Deleting created sriov-vm intance")
1415 test_config
["vim_conn"].delete_vminstance(instance_id
)
1418 class test_vimconn_get_tenant_list(test_base
):
1421 def test_000_get_tenant_list(self
):
1422 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1423 self
.__class
__.test_index
,
1424 inspect
.currentframe().f_code
.co_name
)
1425 self
.__class
__.test_index
+= 1
1427 # Getting tenant list
1428 tenant_list
= test_config
["vim_conn"].get_tenant_list()
1430 for item
in tenant_list
:
1431 if test_config
['tenant'] == item
['name']:
1432 self
.__class
__.tenant_id
= item
['id']
1433 self
.assertIsInstance(item
['name'], (str, unicode))
1434 self
.assertIsInstance(item
['id'], (str, unicode))
1436 def test_010_get_tenant_list_by_id(self
):
1437 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1438 self
.__class
__.test_index
,
1439 inspect
.currentframe().f_code
.co_name
)
1440 self
.__class
__.test_index
+= 1
1442 # Getting filter tenant list by its id
1443 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'id': self
.__class
__.tenant_id
})
1445 for item
in filter_tenant_list
:
1446 self
.assertIsInstance(item
['id'], (str, unicode))
1447 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1449 def test_020_get_tenant_list_by_name(self
):
1450 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1451 self
.__class
__.test_index
,
1452 inspect
.currentframe().f_code
.co_name
)
1453 self
.__class
__.test_index
+= 1
1455 # Getting filter tenant list by its name
1456 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant']})
1458 for item
in filter_tenant_list
:
1459 self
.assertIsInstance(item
['name'], (str, unicode))
1460 self
.assertEqual(item
['name'], test_config
['tenant'])
1462 def test_030_get_tenant_list_by_name_and_id(self
):
1463 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1464 self
.__class
__.test_index
,
1465 inspect
.currentframe().f_code
.co_name
)
1466 self
.__class
__.test_index
+= 1
1468 # Getting filter tenant list by its name and id
1469 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant'],
1470 'id': self
.__class
__.tenant_id
})
1472 for item
in filter_tenant_list
:
1473 self
.assertIsInstance(item
['name'], (str, unicode))
1474 self
.assertIsInstance(item
['id'], (str, unicode))
1475 self
.assertEqual(item
['name'], test_config
['tenant'])
1476 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1478 def test_040_get_tenant_list_negative(self
):
1479 non_exist_tenant_name
= "Tenant_123"
1480 non_exist_tenant_id
= "kjhgrt456-45345kjhdfgnbdk-34dsfjdfg"
1481 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1482 self
.__class
__.test_index
,
1483 inspect
.currentframe().f_code
.co_name
)
1484 self
.__class
__.test_index
+= 1
1486 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': non_exist_tenant_name
,
1487 'id': non_exist_tenant_id
})
1489 self
.assertEqual(filter_tenant_list
, [])
1492 class test_vimconn_new_tenant(test_base
):
1495 def test_000_new_tenant(self
):
1496 tenant_name
= _get_random_string(20)
1497 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1498 self
.__class
__.test_index
,
1499 inspect
.currentframe().f_code
.co_name
)
1500 self
.__class
__.test_index
+= 1
1502 self
.__class
__.tenant_id
= test_config
["vim_conn"].new_tenant(tenant_name
, "")
1505 self
.assertIsInstance(self
.__class
__.tenant_id
, (str, unicode))
1508 def test_010_new_tenant_negative(self
):
1509 Invalid_tenant_name
= 10121
1510 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1511 self
.__class
__.test_index
,
1512 inspect
.currentframe().f_code
.co_name
)
1513 self
.__class
__.test_index
+= 1
1515 with self
.assertRaises(Exception) as context
:
1516 test_config
["vim_conn"].new_tenant(Invalid_tenant_name
, "")
1518 self
.assertEqual((context
.exception
).http_code
, 400)
1521 def test_020_delete_tenant(self
):
1522 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1523 self
.__class
__.test_index
,
1524 inspect
.currentframe().f_code
.co_name
)
1525 self
.__class
__.test_index
+= 1
1527 tenant_id
= test_config
["vim_conn"].delete_tenant(self
.__class
__.tenant_id
)
1529 self
.assertIsInstance(tenant_id
, (str, unicode))
1531 def test_030_delete_tenant_negative(self
):
1532 Non_exist_tenant_name
= 'Test_30_tenant'
1533 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1534 self
.__class
__.test_index
,
1535 inspect
.currentframe().f_code
.co_name
)
1536 self
.__class
__.test_index
+= 1
1538 with self
.assertRaises(Exception) as context
:
1539 test_config
["vim_conn"].delete_tenant(Non_exist_tenant_name
)
1541 self
.assertEqual((context
.exception
).http_code
, 404)
1545 if test_config
['image_name']:
1546 image_list
= test_config
['vim_conn'].get_image_list({'name': test_config
['image_name']})
1547 if len(image_list
) == 0:
1548 raise Exception("Image {} is not found at VIM".format(test_config
['image_name']))
1550 image_id
= image_list
[0]['id']
1552 image_list
= test_config
['vim_conn'].get_image_list()
1553 if len(image_list
) == 0:
1554 raise Exception("Not found any image at VIM")
1556 image_id
= image_list
[0]['id']
1560 class test_vimconn_vminstance_by_ip_address(test_base
):
1566 self
.network_name
= _get_random_string(20)
1568 self
.network_id
, _
= test_config
["vim_conn"].new_network(net_name
=self
.network_name
,
1572 test_base
.tearDown(self
)
1573 # Deleting created network
1574 result
= test_config
["vim_conn"].delete_network(self
.network_id
)
1576 logger
.info("Network id {} sucessfully deleted".format(self
.network_id
))
1578 logger
.info("Failed to delete network id {}".format(self
.network_id
))
1581 def test_000_vminstance_by_ip_address(self
):
1583 This test case will deploy VM with provided IP address
1584 Pre-requesite: provided IP address should be from IP pool range which has used for network creation
1587 # provide ip address
1590 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1593 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1596 image_id
= get_image_id()
1598 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1599 self
.__class
__.test_index
,
1600 inspect
.currentframe().f_code
.co_name
)
1601 self
.__class
__.test_index
+= 1
1603 net_list
= [{'use': 'bridge', 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual',
1604 'net_id': self
.network_id
, 'ip_address': ip_address
}]
1606 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=image_id
,
1607 flavor_id
=flavor_id
, net_list
=net_list
)
1609 self
.assertEqual(type(instance_id
),str)
1610 logger
.info("Deleting created vm instance")
1611 test_config
["vim_conn"].delete_vminstance(instance_id
)
1614 def test_010_vminstance_by_ip_address_negative(self
):
1616 # IP address not from subnet range
1617 invalid_ip_address
= '10.10.12.1'
1619 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1622 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1624 # find image name and image id
1625 image_id
= get_image_id()
1627 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1628 self
.__class
__.test_index
,
1629 inspect
.currentframe().f_code
.co_name
)
1630 self
.__class
__.test_index
+= 1
1632 net_list
= [{'use': 'bridge', 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual',
1633 'net_id': self
.network_id
, 'ip_address': invalid_ip_address
}]
1635 with self
.assertRaises(Exception) as context
:
1636 test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=image_id
,
1637 flavor_id
=flavor_id
,
1639 self
.assertEqual((context
.exception
).http_code
, 400)
1641 def test_020_vminstance_by_floating_ip(self
):
1643 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1646 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1648 # find image name and image id
1649 image_id
= get_image_id()
1651 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1652 self
.__class
__.test_index
,
1653 inspect
.currentframe().f_code
.co_name
)
1654 self
.__class
__.test_index
+= 1
1656 net_list
= [{'use': 'bridge', 'name': name
, 'floating_ip': True, 'port_security': True, 'type': 'virtual',
1657 'net_id': self
.network_id
}]
1659 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=image_id
,
1660 flavor_id
=flavor_id
, net_list
=net_list
)
1662 self
.assertEqual(type(instance_id
),str)
1663 logger
.info("Deleting created vm instance")
1664 test_config
["vim_conn"].delete_vminstance(instance_id
)
1667 def test_030_vminstance_by_mac_address(self
):
1669 mac_address
= "74:54:2f:21:da:8c"
1670 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1673 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1675 # find image name and image id
1676 image_id
= get_image_id()
1678 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1679 self
.__class
__.test_index
,
1680 inspect
.currentframe().f_code
.co_name
)
1681 self
.__class
__.test_index
+= 1
1683 net_list
= [{'use': 'bridge', 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual',
1684 'net_id': self
.network_id
,'mac_address': mac_address
}]
1686 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=image_id
,
1687 flavor_id
=flavor_id
, net_list
=net_list
)
1689 self
.assertEqual(type(instance_id
),str)
1690 logger
.info("Deleting created vm instance")
1691 test_config
["vim_conn"].delete_vminstance(instance_id
)
1694 class test_vimconn_vminstance_by_adding_10_nics(test_base
):
1702 self
.network_name
= _get_random_string(20)
1703 network_id
, _
= test_config
["vim_conn"].new_network(net_name
=self
.network_name
,
1705 self
.net_ids
.append(network_id
)
1708 test_base
.tearDown(self
)
1709 # Deleting created network
1710 for net_id
in self
.net_ids
:
1711 result
= test_config
["vim_conn"].delete_network(net_id
)
1713 logger
.info("Network id {} sucessfully deleted".format(net_id
))
1715 logger
.info("Failed to delete network id {}".format(net_id
))
1717 def test_000_vminstance_by_adding_10_nics(self
):
1718 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1721 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1723 # find image name and image id
1724 image_id
= get_image_id()
1726 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1727 self
.__class
__.test_index
,
1728 inspect
.currentframe().f_code
.co_name
)
1729 self
.__class
__.test_index
+= 1
1733 for net_id
in self
.net_ids
:
1734 name
= "eth{}".format(c
)
1735 net_list
.append({'use': 'bridge', 'name': name
, 'floating_ip': False,
1736 'port_security': True, 'type': 'virtual', 'net_id': net_id
})
1739 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=image_id
,
1740 flavor_id
=flavor_id
, net_list
=net_list
)
1742 self
.assertEqual(type(instance_id
),str)
1743 logger
.info("Deleting created vm instance")
1744 test_config
["vim_conn"].delete_vminstance(instance_id
)
1748 class test_vimconn_vminstance_by_existing_disk(test_base
):
1754 self
.network_name
= _get_random_string(20)
1755 self
.network_id
, _
= test_config
["vim_conn"].new_network(net_name
=self
.network_name
,
1759 test_base
.tearDown(self
)
1760 # Deleting created network
1761 result
= test_config
["vim_conn"].delete_network(self
.network_id
)
1763 logger
.info("Network id {} sucessfully deleted".format(self
.network_id
))
1765 logger
.info("Failed to delete network id {}".format(self
.network_id
))
1768 def test_000_vminstance_by_existing_disk(self
):
1769 """ This testcase will add existing disk only if given catalog/image is free
1770 means not used by any other VM
1773 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1777 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1779 # find image name and image id
1780 image_id
= get_image_id()
1781 cirros_image
= test_config
["vim_conn"].get_image_list({'name': 'cirros'})
1782 disk_list
= [{'image_id': cirros_image
[0]['id'],'size': 5}]
1784 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1785 self
.__class
__.test_index
,
1786 inspect
.currentframe().f_code
.co_name
)
1787 self
.__class
__.test_index
+= 1
1789 net_list
= [{'use': 'bridge', 'name': name
, 'floating_ip': False, 'port_security': True,
1790 'type': 'virtual', 'net_id': self
.network_id
}]
1792 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=image_id
,
1793 flavor_id
=flavor_id
, net_list
=net_list
,disk_list
=disk_list
)
1795 self
.assertEqual(type(instance_id
),str)
1796 logger
.info("Deleting created vm instance")
1797 test_config
["vim_conn"].delete_vminstance(instance_id
)
1800 def test_010_vminstance_by_new_disk(self
):
1801 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1805 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1807 # find image name and image id
1808 image_id
= get_image_id()
1809 disk_list
= [{'size': '5'}]
1811 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1812 self
.__class
__.test_index
,
1813 inspect
.currentframe().f_code
.co_name
)
1814 self
.__class
__.test_index
+= 1
1816 net_list
= [{'use': 'bridge', 'name': name
, 'floating_ip': False, 'port_security': True,
1817 'type': 'virtual', 'net_id': self
.network_id
}]
1819 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=image_id
,
1820 flavor_id
=flavor_id
, net_list
=net_list
,disk_list
=disk_list
)
1822 self
.assertEqual(type(instance_id
),str)
1823 logger
.info("Deleting created vm instance")
1824 test_config
["vim_conn"].delete_vminstance(instance_id
)
1827 def test_020_vminstance_by_CDROM(self
):
1828 """ This testcase will insert media file only if provided catalog
1829 has pre-created ISO media file into vCD
1831 flavor_data
={'ram': 1024, 'vcpus': 1, 'disk': 10}
1833 image_list
= test_config
["vim_conn"].get_image_list({'name':'Ubuntu'})
1834 disk_list
= [{'image_id':image_list
[0]['id'],'device_type':'cdrom'}]
1837 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1839 # find image name and image id
1840 image_id
= get_image_id()
1842 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1843 self
.__class
__.test_index
,
1844 inspect
.currentframe().f_code
.co_name
)
1845 self
.__class
__.test_index
+= 1
1847 net_list
= [{'use': 'bridge', 'name': name
, 'floating_ip': False, 'port_security': True,
1848 'type': 'virtual', 'net_id': self
.network_id
}]
1850 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=image_id
,
1851 flavor_id
=flavor_id
, net_list
=net_list
,disk_list
=disk_list
)
1853 self
.assertEqual(type(instance_id
),str)
1854 logger
.info("Deleting created vm instance")
1855 test_config
["vim_conn"].delete_vminstance(instance_id
)
1859 class test_vimconn_vminstance_by_affinity_anti_affinity(test_base
):
1865 self
.network_name
= _get_random_string(20)
1866 self
.network_id
, _
= test_config
["vim_conn"].new_network(net_name
=self
.network_name
,
1870 test_base
.tearDown(self
)
1871 # Deleting created network
1872 result
= test_config
["vim_conn"].delete_network(self
.network_id
)
1874 logger
.info("Network id {} sucessfully deleted".format(self
.network_id
))
1876 logger
.info("Failed to delete network id {}".format(self
.network_id
))
1878 def test_000_vminstance_by_affinity_anti_affinity(self
):
1879 """ This testcase will deploy VM into provided HOSTGROUP in VIM config
1880 Pre-requisites: User has created Hosh Groups in vCenter with respective Hosts to be used
1881 While creating VIM account user has to pass the Host Group names in availability_zone list
1883 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1887 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1889 # find image name and image id
1890 image_id
= get_image_id()
1892 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1893 self
.__class
__.test_index
,
1894 inspect
.currentframe().f_code
.co_name
)
1895 self
.__class
__.test_index
+= 1
1897 net_list
= [{'use': 'bridge', 'name': name
, 'floating_ip': False, 'port_security': True,
1898 'type': 'virtual', 'net_id': self
.network_id
}]
1900 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=image_id
,
1901 flavor_id
=flavor_id
, net_list
=net_list
,availability_zone_index
=1,
1902 availability_zone_list
=['HG_174','HG_175'])
1904 self
.assertEqual(type(instance_id
),str)
1906 logger
.info("Deleting created vm instance")
1907 test_config
["vim_conn"].delete_vminstance(instance_id
)
1909 class test_vimconn_vminstance_by_numa_affinity(test_base
):
1915 self
.network_name
= _get_random_string(20)
1916 self
.network_id
, _
= test_config
["vim_conn"].new_network(net_name
=self
.network_name
,
1920 test_base
.tearDown(self
)
1921 # Deleting created network
1922 result
= test_config
["vim_conn"].delete_network(self
.network_id
)
1924 logger
.info("Network id {} sucessfully deleted".format(self
.network_id
))
1926 logger
.info("Failed to delete network id {}".format(self
.network_id
))
1928 def test_000_vminstance_by_numa_affinity(self
):
1929 flavor_data
= {'extended': {'numas': [{'paired-threads-id': [['1', '3'], ['2', '4']],
1930 ' paired-threads': 2, 'memory': 1}]},
1931 'ram': 1024, 'vcpus': 1, 'disk': 10}
1935 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1937 # find image name and image id
1938 image_id
= get_image_id()
1940 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1941 self
.__class
__.test_index
,
1942 inspect
.currentframe().f_code
.co_name
)
1943 self
.__class
__.test_index
+= 1
1945 net_list
= [{'use': 'bridge', 'name': name
, 'floating_ip': False, 'port_security': True,
1946 'type': 'virtual', 'net_id': self
.network_id
}]
1948 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=image_id
,
1949 flavor_id
=flavor_id
, net_list
=net_list
)
1951 self
.assertEqual(type(instance_id
),str)
1952 logger
.info("Deleting created vm instance")
1953 test_config
["vim_conn"].delete_vminstance(instance_id
)
1959 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
1960 scenario based tests.
1962 class descriptor_based_scenario_test(test_base
):
1964 scenario_test_path
= None
1967 def setUpClass(cls
):
1969 cls
.to_delete_list
= []
1970 cls
.scenario_uuids
= []
1971 cls
.instance_scenario_uuids
= []
1972 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
1973 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
1976 def tearDownClass(cls
):
1977 test_config
["test_number"] += 1
1979 def test_000_load_scenario(self
):
1980 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1981 inspect
.currentframe().f_code
.co_name
,
1982 test_config
["test_folder"])
1983 self
.__class
__.test_index
+= 1
1985 descriptor_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/*.yaml')
1986 vnf_descriptors
= []
1987 scenario_descriptors
= []
1988 for descriptor_file
in descriptor_files
:
1989 with
open(descriptor_file
, 'r') as stream
:
1990 descriptor
= yaml
.load(stream
)
1991 if "vnf" in descriptor
or "vnfd:vnfd-catalog" in descriptor
or "vnfd-catalog" in descriptor
:
1992 vnf_descriptors
.append(descriptor
)
1994 scenario_descriptors
.append(descriptor
)
1996 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
1997 if not vnf_descriptors
or not scenario_descriptors
or len(scenario_descriptors
) > 1:
1998 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
1999 test_config
["test_folder"]))
2002 for vnf_descriptor
in vnf_descriptors
:
2003 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
2004 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
, image_name
=test_config
["image_name"])
2007 vnf_uuid
= vnf
['vnf']['uuid']
2009 vnf_uuid
= vnf
['vnfd'][0]['uuid']
2010 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
2011 "params": {"uuid": vnf_uuid
}})
2013 # load the scenario definition
2014 for scenario_descriptor
in scenario_descriptors
:
2015 # networks = scenario_descriptor['scenario']['networks']
2016 # networks[test_config["mgmt_net"]] = networks.pop('mgmt')
2017 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
2018 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
2019 logger
.debug(scenario
)
2020 if 'scenario' in scenario
:
2021 scenario_uuid
= scenario
['scenario']['uuid']
2023 scenario_uuid
= scenario
['nsd'][0]['uuid']
2024 self
.__class
__.to_delete_list
.insert(0, {"item": "scenario",
2025 "function": test_config
["client"].delete_scenario
,
2026 "params": {"uuid": scenario_uuid
}})
2027 self
.__class
__.scenario_uuids
.append(scenario_uuid
)
2029 def test_010_instantiate_scenario(self
):
2030 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
2031 inspect
.currentframe().f_code
.co_name
,
2032 test_config
["test_folder"])
2033 self
.__class
__.test_index
+= 1
2034 for scenario_uuid
in self
.__class
__.scenario_uuids
:
2035 instance_descriptor
= {
2037 "name": self
.__class
__.test_text
,
2038 "scenario": scenario_uuid
,
2040 "mgmt": {"sites": [ { "netmap-use": test_config
["mgmt_net"]} ]}
2044 instance
= test_config
["client"].create_instance(instance_descriptor
)
2045 self
.__class
__.instance_scenario_uuids
.append(instance
['uuid'])
2046 logger
.debug(instance
)
2047 self
.__class
__.to_delete_list
.insert(0, {"item": "instance",
2048 "function": test_config
["client"].delete_instance
,
2049 "params": {"uuid": instance
['uuid']}})
2051 def test_020_check_deployent(self
):
2052 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
2053 inspect
.currentframe().f_code
.co_name
,
2054 test_config
["test_folder"])
2055 self
.__class
__.test_index
+= 1
2057 if test_config
["manual"]:
2058 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
2061 keep_waiting
= test_config
["timeout"]
2062 pending_instance_scenario_uuids
= list(self
.__class
__.instance_scenario_uuids
) # make a copy
2063 while pending_instance_scenario_uuids
:
2065 while index
< len(pending_instance_scenario_uuids
):
2066 result
= check_instance_scenario_active(pending_instance_scenario_uuids
[index
])
2068 del pending_instance_scenario_uuids
[index
]
2070 elif 'ERROR' in result
[1]:
2071 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
2073 raise Exception(msg
)
2076 if keep_waiting
>= 5:
2079 elif keep_waiting
> 0:
2080 time
.sleep(keep_waiting
)
2083 msg
= 'Timeout reached while waiting instance scenario to get active'
2085 raise Exception(msg
)
2087 def test_030_clean_deployment(self
):
2088 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
2089 inspect
.currentframe().f_code
.co_name
,
2090 test_config
["test_folder"])
2091 self
.__class
__.test_index
+= 1
2092 #At the moment if you delete an scenario right after creating it, in openstack datacenters
2093 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
2095 for item
in self
.__class
__.to_delete_list
:
2096 response
= item
["function"](**item
["params"])
2097 logger
.debug(response
)
2100 def _get_random_string(maxLength
):
2101 '''generates a string with random characters string.letters and string.digits
2102 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
2106 minLength
= min_string
- len(prefix
)
2107 if maxLength
< min_string
: maxLength
= min_string
2108 maxLength
-= len(prefix
)
2109 length
= random
.randint(minLength
,maxLength
)
2110 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
2113 def test_vimconnector(args
):
2115 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
2116 test_config
['vimtype'] = args
.vimtype
2117 if args
.vimtype
== "vmware":
2118 import vimconn_vmware
as vim
2120 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
2122 tenant_name
= args
.tenant_name
2123 test_config
['tenant'] = tenant_name
2124 config_params
= yaml
.load(args
.config_param
)
2125 org_name
= config_params
.get('orgname')
2126 org_user
= config_params
.get('user')
2127 org_passwd
= config_params
.get('passwd')
2128 vim_url
= args
.endpoint_url
2129 test_config
['image_path'] = args
.image_path
2130 test_config
['image_name'] = args
.image_name
2131 test_config
['sriov_net_name'] = args
.sriov_net_name
2133 # vmware connector obj
2134 test_config
['vim_conn'] = vim
.vimconnector(name
=org_name
, tenant_name
=tenant_name
, user
=org_user
,passwd
=org_passwd
, url
=vim_url
, config
=config_params
)
2136 elif args
.vimtype
== "aws":
2137 import vimconn_aws
as vim
2138 elif args
.vimtype
== "openstack":
2139 import vimconn_openstack
as vim
2141 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
2143 tenant_name
= args
.tenant_name
2144 test_config
['tenant'] = tenant_name
2145 config_params
= yaml
.load(args
.config_param
)
2146 os_user
= config_params
.get('user')
2147 os_passwd
= config_params
.get('passwd')
2148 vim_url
= args
.endpoint_url
2149 test_config
['image_path'] = args
.image_path
2150 test_config
['image_name'] = args
.image_name
2151 test_config
['sriov_net_name'] = args
.sriov_net_name
2153 # openstack connector obj
2154 vim_persistent_info
= {}
2155 test_config
['vim_conn'] = vim
.vimconnector(
2156 uuid
="test-uuid-1", name
="VIO-openstack",
2157 tenant_id
=None, tenant_name
=tenant_name
,
2158 url
=vim_url
, url_admin
=None,
2159 user
=os_user
, passwd
=os_passwd
,
2160 config
=config_params
, persistent_info
=vim_persistent_info
2162 test_config
['vim_conn'].debug
= "true"
2164 elif args
.vimtype
== "openvim":
2165 import vimconn_openvim
as vim
2167 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
2171 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
2172 # If only want to obtain a tests list print it and exit
2175 for cls
in clsmembers
:
2176 if cls
[0].startswith('test_vimconn'):
2177 tests_names
.append(cls
[0])
2179 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
2184 # Create the list of tests to be run
2185 code_based_tests
= []
2187 for test
in args
.tests
:
2188 for t
in test
.split(','):
2189 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
2190 if len(matches_code_based_tests
) > 0:
2191 code_based_tests
.append(matches_code_based_tests
[0][1])
2193 logger
.critical("Test '{}' is not among the possible ones".format(t
))
2195 if not code_based_tests
:
2197 for cls
in clsmembers
:
2198 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
2199 if cls
[0].startswith('test_vimconn'):
2200 code_based_tests
.append(cls
[1])
2202 logger
.debug("tests to be executed: {}".format(code_based_tests
))
2204 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
2205 # This is handled in the tests using logging.
2206 stream
= open('/dev/null', 'w')
2208 # Run code based tests
2209 basic_tests_suite
= unittest
.TestSuite()
2210 for test
in code_based_tests
:
2211 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
2212 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
2213 executed
+= result
.testsRun
2214 failed
+= len(result
.failures
) + len(result
.errors
)
2215 if failfast
and failed
:
2217 if len(result
.failures
) > 0:
2218 logger
.debug("failures : {}".format(result
.failures
))
2219 if len(result
.errors
) > 0:
2220 logger
.debug("errors : {}".format(result
.errors
))
2221 return executed
, failed
2226 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
2227 import openmanoclient
2230 test_config
["client"] = openmanoclient
.openmanoclient(
2231 endpoint_url
=args
.endpoint_url
,
2232 tenant_name
=args
.tenant_name
,
2233 datacenter_name
=args
.datacenter
,
2234 debug
=args
.debug
, logger
=test_config
["logger_name"])
2235 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
2236 # If only want to obtain a tests list print it and exit
2239 for cls
in clsmembers
:
2240 if cls
[0].startswith('test_VIM'):
2241 tests_names
.append(cls
[0])
2243 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
2244 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
2245 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
2250 # Create the list of tests to be run
2251 code_based_tests
= []
2253 for test
in args
.tests
:
2254 for t
in test
.split(','):
2255 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
2256 if len(matches_code_based_tests
) > 0:
2257 code_based_tests
.append(matches_code_based_tests
[0][1])
2259 logger
.critical("Test '{}' is not among the possible ones".format(t
))
2261 if not code_based_tests
:
2263 for cls
in clsmembers
:
2264 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
2265 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
2266 code_based_tests
.append(cls
[1])
2268 logger
.debug("tests to be executed: {}".format(code_based_tests
))
2270 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
2271 # This is handled in the tests using logging.
2272 stream
= open('/dev/null', 'w')
2274 # Run code based tests
2275 basic_tests_suite
= unittest
.TestSuite()
2276 for test
in code_based_tests
:
2277 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
2278 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
2279 executed
+= result
.testsRun
2280 failed
+= len(result
.failures
) + len(result
.errors
)
2281 if failfast
and failed
:
2283 if len(result
.failures
) > 0:
2284 logger
.debug("failures : {}".format(result
.failures
))
2285 if len(result
.errors
) > 0:
2286 logger
.debug("errors : {}".format(result
.errors
))
2287 return executed
, failed
2292 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
2293 import openmanoclient
2296 test_config
["client"] = openmanoclient
.openmanoclient(
2297 endpoint_url
=args
.endpoint_url
,
2298 tenant_name
=args
.tenant_name
,
2299 datacenter_name
=args
.datacenter
,
2300 debug
=args
.debug
, logger
=test_config
["logger_name"])
2301 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
2302 # If only want to obtain a tests list print it and exit
2305 for cls
in clsmembers
:
2306 if cls
[0].startswith('test_WIM'):
2307 tests_names
.append(cls
[0])
2309 msg
= "The 'wim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
2310 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
2311 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
2316 # Create the list of tests to be run
2317 code_based_tests
= []
2319 for test
in args
.tests
:
2320 for t
in test
.split(','):
2321 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
2322 if len(matches_code_based_tests
) > 0:
2323 code_based_tests
.append(matches_code_based_tests
[0][1])
2325 logger
.critical("Test '{}' is not among the possible ones".format(t
))
2327 if not code_based_tests
:
2329 for cls
in clsmembers
:
2330 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
2331 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
2332 code_based_tests
.append(cls
[1])
2334 logger
.debug("tests to be executed: {}".format(code_based_tests
))
2336 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
2337 # This is handled in the tests using logging.
2338 stream
= open('/dev/null', 'w')
2340 # Run code based tests
2341 basic_tests_suite
= unittest
.TestSuite()
2342 for test
in code_based_tests
:
2343 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
2344 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
2345 executed
+= result
.testsRun
2346 failed
+= len(result
.failures
) + len(result
.errors
)
2347 if failfast
and failed
:
2349 if len(result
.failures
) > 0:
2350 logger
.debug("failures : {}".format(result
.failures
))
2351 if len(result
.errors
) > 0:
2352 logger
.debug("errors : {}".format(result
.errors
))
2353 return executed
, failed
2356 def test_deploy(args
):
2358 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
2359 import openmanoclient
2362 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
2363 test_config
["image_name"] = args
.image_name
2364 test_config
["mgmt_net"] = args
.mgmt_net
2365 test_config
["manual"] = args
.manual
2366 test_directory_content
= os
.listdir(test_config
["test_directory"])
2367 # If only want to obtain a tests list print it and exit
2369 msg
= "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
2374 descriptor_based_tests
= []
2375 # Create the list of tests to be run
2376 code_based_tests
= []
2378 for test
in args
.tests
:
2379 for t
in test
.split(','):
2380 if t
in test_directory_content
:
2381 descriptor_based_tests
.append(t
)
2383 logger
.critical("Test '{}' is not among the possible ones".format(t
))
2385 if not descriptor_based_tests
:
2387 descriptor_based_tests
= test_directory_content
2389 logger
.debug("tests to be executed: {}".format(code_based_tests
))
2391 # import openmanoclient from relative path
2392 test_config
["client"] = openmanoclient
.openmanoclient(
2393 endpoint_url
=args
.endpoint_url
,
2394 tenant_name
=args
.tenant_name
,
2395 datacenter_name
=args
.datacenter
,
2396 debug
=args
.debug
, logger
=test_config
["logger_name"])
2398 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
2399 # This is handled in the tests using logging.
2400 stream
= open('/dev/null', 'w')
2401 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
2402 for test
in descriptor_based_tests
:
2403 test_config
["test_folder"] = test
2404 test_suite
= unittest
.TestSuite()
2405 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
2406 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
2407 executed
+= result
.testsRun
2408 failed
+= len(result
.failures
) + len(result
.errors
)
2409 if failfast
and failed
:
2411 if len(result
.failures
) > 0:
2412 logger
.debug("failures : {}".format(result
.failures
))
2413 if len(result
.errors
) > 0:
2414 logger
.debug("errors : {}".format(result
.errors
))
2416 return executed
, failed
2418 if __name__
=="__main__":
2420 parser
= ArgumentParser(description
='Test RO module')
2421 parser
.add_argument('-v','--version', action
='version', help="Show current version",
2422 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
2425 parent_parser
= ArgumentParser(add_help
=False)
2426 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
2427 dest
='failfast', action
="store_true", default
=False)
2428 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
2429 dest
='failed', action
="store_true", default
=False)
2430 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
2431 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
2433 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
2434 help='Set the logger file. By default '+default_logger_file
)
2435 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
2436 help="Set the openmano tenant to use for the test. By default 'osm'")
2437 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
2438 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
2439 dest
='timeout', type=int, default
=300)
2440 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
2442 subparsers
= parser
.add_subparsers(help='test sets')
2444 # Deployment test set
2445 # -------------------
2446 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
2447 help="test deployment using descriptors at RO_test folder ")
2448 deploy_parser
.set_defaults(func
=test_deploy
)
2450 # Mandatory arguments
2451 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
2452 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
2453 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
2454 help='Image name available at datacenter used for the tests')
2455 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
2456 help='Set the vim management network to use for tests')
2458 # Optional arguments
2459 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
2460 help='Pause execution once deployed to allow manual checking of the '
2461 'deployed instance scenario')
2462 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
2463 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
2466 # -------------------
2467 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
2468 vimconn_parser
.set_defaults(func
=test_vimconnector
)
2469 # Mandatory arguments
2470 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
2471 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
2472 help='Set the vimconnector type to test')
2473 mandatory_arguments
.add_argument('-c', '--config', dest
='config_param', required
=True,
2474 help='Set the vimconnector specific config parameters in dictionary format')
2475 mandatory_arguments
.add_argument('-u', '--url', dest
='endpoint_url',required
=True, help="Set the vim connector url or Host IP")
2476 # Optional arguments
2477 vimconn_parser
.add_argument('-i', '--image-path', dest
='image_path', help="Provide image path present at RO container")
2478 vimconn_parser
.add_argument('-n', '--image-name', dest
='image_name', help="Provide image name for test")
2479 # TODO add optional arguments for vimconn tests
2480 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
2481 vimconn_parser
.add_argument('-s', '--sriov-net-name', dest
='sriov_net_name', help="Provide SRIOV network name for test")
2483 # Datacenter test set
2484 # -------------------
2485 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
2486 vimconn_parser
.set_defaults(func
=test_vim
)
2488 # Mandatory arguments
2489 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
2490 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
2492 # Optional arguments
2493 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
2494 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
2497 # -------------------
2498 vimconn_parser
= subparsers
.add_parser('wim', parents
=[parent_parser
], help="test wim")
2499 vimconn_parser
.set_defaults(func
=test_wim
)
2501 # Mandatory arguments
2502 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
2503 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
2505 # Optional arguments
2506 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
2507 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
2509 argcomplete
.autocomplete(parser
)
2510 args
= parser
.parse_args()
2514 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
2515 logger_level
= 'INFO'
2517 logger_level
= 'DEBUG'
2519 logger_level
= 'WARNING'
2520 logger_name
= os
.path
.basename(__file__
)
2521 test_config
["logger_name"] = logger_name
2522 logger
= logging
.getLogger(logger_name
)
2523 logger
.setLevel(logger_level
)
2524 failfast
= args
.failfast
2526 # Configure a logging handler to store in a logging file
2527 if args
.logger_file
:
2528 fileHandler
= logging
.FileHandler(args
.logger_file
)
2529 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
2530 fileHandler
.setFormatter(formatter_fileHandler
)
2531 logger
.addHandler(fileHandler
)
2533 # Configure a handler to print to stdout
2534 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
2535 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
2536 consoleHandler
.setFormatter(formatter_consoleHandler
)
2537 logger
.addHandler(consoleHandler
)
2539 logger
.debug('Program started with the following arguments: ' + str(args
))
2541 # set test config parameters
2542 test_config
["timeout"] = args
.timeout
2543 test_config
["test_number"] = 1
2545 executed
, failed
= args
.func(args
)
2548 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
2549 sys
.exit(1 if failed
else 0)