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
})
855 self
.assertEqual(type(self
.__class
__.image_id
),str)
856 self
.assertIsInstance(uuid
.UUID(self
.__class
__.image_id
),uuid
.UUID
)
858 self
.skipTest("Skipping test as image file not present at RO container")
860 def test_010_new_image_negative(self
):
861 Non_exist_image_path
= '/temp1/cirros.ovf'
863 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
864 self
.__class
__.test_index
,
865 inspect
.currentframe().f_code
.co_name
)
866 self
.__class
__.test_index
+= 1
868 with self
.assertRaises(Exception) as context
:
869 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
871 self
.assertEqual((context
.exception
).http_code
, 400)
873 def test_020_delete_image(self
):
874 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
875 self
.__class
__.test_index
,
876 inspect
.currentframe().f_code
.co_name
)
877 self
.__class
__.test_index
+= 1
879 image_id
= test_config
["vim_conn"].delete_image(self
.__class
__.image_id
)
880 self
.assertEqual(type(image_id
),str)
882 def test_030_delete_image_negative(self
):
883 Non_exist_image_id
= str(uuid
.uuid4())
885 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
886 self
.__class
__.test_index
,
887 inspect
.currentframe().f_code
.co_name
)
888 self
.__class
__.test_index
+= 1
890 with self
.assertRaises(Exception) as context
:
891 test_config
["vim_conn"].delete_image(Non_exist_image_id
)
893 self
.assertEqual((context
.exception
).http_code
, 404)
895 class test_vimconn_get_image_id_from_path(test_base
):
897 def test_000_get_image_id_from_path(self
):
898 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
899 self
.__class
__.test_index
,
900 inspect
.currentframe().f_code
.co_name
)
901 self
.__class
__.test_index
+= 1
903 image_path
= test_config
['image_path']
905 image_id
= test_config
["vim_conn"].get_image_id_from_path( image_path
)
906 self
.assertEqual(type(image_id
),str)
908 self
.skipTest("Skipping test as image file not present at RO container")
910 def test_010_get_image_id_from_path_negative(self
):
911 Non_exist_image_path
= '/temp1/cirros.ovf'
913 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
914 self
.__class
__.test_index
,
915 inspect
.currentframe().f_code
.co_name
)
916 self
.__class
__.test_index
+= 1
918 with self
.assertRaises(Exception) as context
:
919 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
921 self
.assertEqual((context
.exception
).http_code
, 400)
923 class test_vimconn_get_image_list(test_base
):
927 def test_000_get_image_list(self
):
928 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
929 self
.__class
__.test_index
,
930 inspect
.currentframe().f_code
.co_name
)
931 self
.__class
__.test_index
+= 1
932 image_list
= test_config
["vim_conn"].get_image_list()
934 for item
in image_list
:
936 self
.__class
__.image_name
= item
['name']
937 self
.__class
__.image_id
= item
['id']
938 self
.assertEqual(type(self
.__class
__.image_name
),str)
939 self
.assertEqual(type(self
.__class
__.image_id
),str)
941 def test_010_get_image_list_by_name(self
):
942 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
943 self
.__class
__.test_index
,
944 inspect
.currentframe().f_code
.co_name
)
945 self
.__class
__.test_index
+= 1
947 image_list
= test_config
["vim_conn"].get_image_list({'name': self
.__class
__.image_name
})
949 for item
in image_list
:
950 self
.assertEqual(type(item
['id']), str)
951 self
.assertEqual(item
['id'], self
.__class
__.image_id
)
952 self
.assertEqual(type(item
['name']), str)
953 self
.assertEqual(item
['name'], self
.__class
__.image_name
)
955 def test_020_get_image_list_by_id(self
):
956 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
957 self
.__class
__.test_index
,
958 inspect
.currentframe().f_code
.co_name
)
959 self
.__class
__.test_index
+= 1
961 filter_image_list
= test_config
["vim_conn"].get_image_list({'id': self
.__class
__.image_id
})
963 for item1
in filter_image_list
:
964 self
.assertEqual(type(item1
.get('id')), str)
965 self
.assertEqual(item1
.get('id'), self
.__class
__.image_id
)
966 self
.assertEqual(type(item1
.get('name')), str)
967 self
.assertEqual(item1
.get('name'), self
.__class
__.image_name
)
969 def test_030_get_image_list_negative(self
):
970 Non_exist_image_id
= uuid
.uuid4()
971 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
972 self
.__class
__.test_index
,
973 inspect
.currentframe().f_code
.co_name
)
974 self
.__class
__.test_index
+= 1
975 image_list
= test_config
["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id
})
977 self
.assertIsNotNone(image_list
, None)
978 self
.assertEqual(image_list
, [])
980 class test_vimconn_new_vminstance(test_base
):
989 self
.__class
__.network_name
= _get_random_string(20)
990 self
.__class
__.net_type
= 'bridge'
992 self
.__class
__.network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
993 net_type
=self
.__class
__.net_type
)
996 test_base
.tearDown(self
)
997 # Deleting created network
998 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
1000 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
1002 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
1004 def test_000_new_vminstance(self
):
1005 vpci
= "0000:00:11.0"
1008 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1011 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1013 # find image name and image id
1014 if test_config
['image_name']:
1015 image_list
= test_config
['vim_conn'].get_image_list({'name': test_config
['image_name']})
1016 if len(image_list
) == 0:
1017 raise Exception("Image {} is not found at VIM".format(test_config
['image_name']))
1019 self
.__class
__.image_id
= image_list
[0]['id']
1021 image_list
= test_config
['vim_conn'].get_image_list()
1022 if len(image_list
) == 0:
1023 raise Exception("Not found any image at VIM")
1025 self
.__class
__.image_id
= image_list
[0]['id']
1027 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1028 self
.__class
__.test_index
,
1029 inspect
.currentframe().f_code
.co_name
)
1030 self
.__class
__.test_index
+= 1
1032 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
}]
1034 self
.__class
__.instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
, flavor_id
=flavor_id
, net_list
=net_list
)
1036 self
.assertEqual(type(self
.__class
__.instance_id
),str)
1038 def test_010_new_vminstance_by_model(self
):
1039 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1040 model_name
= 'e1000'
1044 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1046 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1047 self
.__class
__.test_index
,
1048 inspect
.currentframe().f_code
.co_name
)
1049 self
.__class
__.test_index
+= 1
1051 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
}]
1053 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1054 flavor_id
=flavor_id
,
1056 self
.assertEqual(type(instance_id
),str)
1057 # Deleting created vm instance
1058 logger
.info("Deleting created vm intance")
1059 test_config
["vim_conn"].delete_vminstance(instance_id
)
1062 def test_020_new_vminstance_by_net_use(self
):
1063 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1068 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1070 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1071 self
.__class
__.test_index
,
1072 inspect
.currentframe().f_code
.co_name
)
1073 self
.__class
__.test_index
+= 1
1075 net_list
= [{'use': net_use
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1077 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1078 flavor_id
=flavor_id
,
1080 self
.assertEqual(type(instance_id
),str)
1081 # Deleting created vm instance
1082 logger
.info("Deleting created vm intance")
1083 test_config
["vim_conn"].delete_vminstance(instance_id
)
1086 def test_030_new_vminstance_by_net_type(self
):
1087 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1092 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1094 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1095 self
.__class
__.test_index
,
1096 inspect
.currentframe().f_code
.co_name
)
1097 self
.__class
__.test_index
+= 1
1099 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': _type
, 'net_id': self
.__class
__.network_id
}]
1101 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1102 flavor_id
=flavor_id
,
1104 self
.assertEqual(type(instance_id
),str)
1105 # Deleting created vm instance
1106 logger
.info("Deleting created vm intance")
1107 test_config
["vim_conn"].delete_vminstance(instance_id
)
1110 def test_040_new_vminstance_by_cloud_config(self
):
1111 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1113 user_name
= 'test_user'
1115 key_pairs
= ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com']
1117 users_data
= [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com'], 'name': user_name
}]
1119 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
}
1122 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1124 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1125 self
.__class
__.test_index
,
1126 inspect
.currentframe().f_code
.co_name
)
1127 self
.__class
__.test_index
+= 1
1129 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1131 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Cloud_vm', image_id
=self
.__class
__.image_id
,
1132 flavor_id
=flavor_id
,
1134 cloud_config
=cloud_data
)
1135 self
.assertEqual(type(instance_id
),str)
1136 # Deleting created vm instance
1137 logger
.info("Deleting created vm intance")
1138 test_config
["vim_conn"].delete_vminstance(instance_id
)
1141 def test_050_new_vminstance_by_disk_list(self
):
1142 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1145 device_data
= [{'image_id': self
.__class
__.image_id
, 'size': '5'}]
1148 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1150 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1151 self
.__class
__.test_index
,
1152 inspect
.currentframe().f_code
.co_name
)
1153 self
.__class
__.test_index
+= 1
1155 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1157 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='VM_test1', image_id
=self
.__class
__.image_id
,
1158 flavor_id
=flavor_id
,
1160 disk_list
=device_data
)
1161 self
.assertEqual(type(instance_id
),str)
1162 # Deleting created vm instance
1163 logger
.info("Deleting created vm intance")
1164 test_config
["vim_conn"].delete_vminstance(instance_id
)
1167 def test_060_new_vminstance_negative(self
):
1168 unknown_flavor_id
= str(uuid
.uuid4())
1169 unknown_image_id
= str(uuid
.uuid4())
1172 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1173 self
.__class
__.test_index
,
1174 inspect
.currentframe().f_code
.co_name
)
1175 self
.__class
__.test_index
+= 1
1177 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1179 with self
.assertRaises(Exception) as context
:
1180 test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=unknown_image_id
,
1181 flavor_id
=unknown_flavor_id
,
1183 self
.assertEqual((context
.exception
).http_code
, 404)
1185 def test_070_get_vminstance(self
):
1186 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1187 self
.__class
__.test_index
,
1188 inspect
.currentframe().f_code
.co_name
)
1189 self
.__class
__.test_index
+= 1
1191 # Get instance by its id
1192 vm_info
= test_config
["vim_conn"].get_vminstance(self
.__class
__.instance_id
)
1194 if test_config
['vimtype'] == 'vmware':
1195 for attr
in vm_info
:
1196 if attr
== 'status':
1197 self
.assertEqual(vm_info
[attr
], 'ACTIVE')
1198 if attr
== 'hostId':
1199 self
.assertEqual(type(vm_info
[attr
]), str)
1200 if attr
== 'interfaces':
1201 self
.assertEqual(type(vm_info
[attr
]), list)
1202 self
.assertEqual(vm_info
[attr
][0]['IsConnected'], 'true')
1203 if attr
== 'IsEnabled':
1204 self
.assertEqual(vm_info
[attr
], 'true')
1206 def test_080_get_vminstance_negative(self
):
1207 unknown_instance_id
= str(uuid
.uuid4())
1209 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1210 self
.__class
__.test_index
,
1211 inspect
.currentframe().f_code
.co_name
)
1212 self
.__class
__.test_index
+= 1
1214 with self
.assertRaises(Exception) as context
:
1215 test_config
["vim_conn"].get_vminstance(unknown_instance_id
)
1217 self
.assertEqual((context
.exception
).http_code
, 404)
1219 def test_090_refresh_vms_status(self
):
1220 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1221 self
.__class
__.test_index
,
1222 inspect
.currentframe().f_code
.co_name
)
1223 self
.__class
__.test_index
+= 1
1225 vm_list
.append(self
.__class
__.instance_id
)
1228 vm_info
= test_config
["vim_conn"].refresh_vms_status(vm_list
)
1229 for attr
in vm_info
[self
.__class
__.instance_id
]:
1230 if attr
== 'status':
1231 self
.assertEqual(vm_info
[self
.__class
__.instance_id
][attr
], 'ACTIVE')
1232 if attr
== 'interfaces':
1233 self
.assertEqual(type(vm_info
[self
.__class
__.instance_id
][attr
]), list)
1235 def test_100_refresh_vms_status_negative(self
):
1236 unknown_id
= str(uuid
.uuid4())
1238 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1239 self
.__class
__.test_index
,
1240 inspect
.currentframe().f_code
.co_name
)
1241 self
.__class
__.test_index
+= 1
1243 vm_dict
= test_config
["vim_conn"].refresh_vms_status([unknown_id
])
1244 self
.assertEqual(vm_dict
, {})
1246 def test_110_action_vminstance(self
):
1247 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1248 self
.__class
__.test_index
,
1249 inspect
.currentframe().f_code
.co_name
)
1250 self
.__class
__.test_index
+= 1
1252 action_list
= ['shutdown','start','shutoff','rebuild','pause','resume']
1253 # various action on vminstace
1254 for action
in action_list
:
1255 instance_id
= test_config
["vim_conn"].action_vminstance(self
.__class
__.instance_id
,
1257 self
.assertEqual(instance_id
, self
.__class
__.instance_id
)
1259 def test_120_action_vminstance_negative(self
):
1260 non_exist_id
= str(uuid
.uuid4())
1261 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1262 self
.__class
__.test_index
,
1263 inspect
.currentframe().f_code
.co_name
)
1264 self
.__class
__.test_index
+= 1
1267 with self
.assertRaises(Exception) as context
:
1268 test_config
["vim_conn"].action_vminstance(non_exist_id
, { action
: None})
1270 self
.assertEqual((context
.exception
).http_code
, 400)
1272 def test_130_delete_vminstance(self
):
1273 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1274 self
.__class
__.test_index
,
1275 inspect
.currentframe().f_code
.co_name
)
1276 self
.__class
__.test_index
+= 1
1278 # Deleting created vm instance
1279 logger
.info("Deleting created vm instance")
1280 test_config
["vim_conn"].delete_vminstance(self
.__class
__.instance_id
)
1283 class test_vimconn_get_tenant_list(test_base
):
1286 def test_000_get_tenant_list(self
):
1287 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1288 self
.__class
__.test_index
,
1289 inspect
.currentframe().f_code
.co_name
)
1290 self
.__class
__.test_index
+= 1
1292 # Getting tenant list
1293 tenant_list
= test_config
["vim_conn"].get_tenant_list()
1295 for item
in tenant_list
:
1296 if test_config
['tenant'] == item
['name']:
1297 self
.__class
__.tenant_id
= item
['id']
1298 self
.assertEqual(type(item
['name']), str)
1299 self
.assertEqual(type(item
['id']), str)
1301 def test_010_get_tenant_list_by_id(self
):
1302 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1303 self
.__class
__.test_index
,
1304 inspect
.currentframe().f_code
.co_name
)
1305 self
.__class
__.test_index
+= 1
1307 # Getting filter tenant list by its id
1308 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'id': self
.__class
__.tenant_id
})
1310 for item
in filter_tenant_list
:
1311 self
.assertEqual(type(item
['id']), str)
1312 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1314 def test_020_get_tenant_list_by_name(self
):
1315 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1316 self
.__class
__.test_index
,
1317 inspect
.currentframe().f_code
.co_name
)
1318 self
.__class
__.test_index
+= 1
1320 # Getting filter tenant list by its name
1321 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant']})
1323 for item
in filter_tenant_list
:
1324 self
.assertEqual(type(item
['name']), str)
1325 self
.assertEqual(item
['name'], test_config
['tenant'])
1327 def test_030_get_tenant_list_by_name_and_id(self
):
1328 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1329 self
.__class
__.test_index
,
1330 inspect
.currentframe().f_code
.co_name
)
1331 self
.__class
__.test_index
+= 1
1333 # Getting filter tenant list by its name and id
1334 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant'],
1335 'id': self
.__class
__.tenant_id
})
1337 for item
in filter_tenant_list
:
1338 self
.assertEqual(type(item
['name']), str)
1339 self
.assertEqual(type(item
['id']), str)
1340 self
.assertEqual(item
['name'], test_config
['tenant'])
1341 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1343 def test_040_get_tenant_list_negative(self
):
1344 non_exist_tenant_name
= "Tenant_123"
1345 non_exist_tenant_id
= "kjhgrt456-45345kjhdfgnbdk-34dsfjdfg"
1346 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1347 self
.__class
__.test_index
,
1348 inspect
.currentframe().f_code
.co_name
)
1349 self
.__class
__.test_index
+= 1
1351 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': non_exist_tenant_name
,
1352 'id': non_exist_tenant_id
})
1354 self
.assertEqual(filter_tenant_list
, [])
1356 class test_vimconn_new_tenant(test_base
):
1359 def test_000_new_tenant(self
):
1360 tenant_name
= _get_random_string(20)
1361 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1362 self
.__class
__.test_index
,
1363 inspect
.currentframe().f_code
.co_name
)
1364 self
.__class
__.test_index
+= 1
1366 self
.__class
__.tenant_id
= test_config
["vim_conn"].new_tenant(tenant_name
)
1369 self
.assertEqual(type(self
.__class
__.tenant_id
), str)
1371 def test_010_new_tenant_negative(self
):
1372 Invalid_tenant_name
= 10121
1373 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1374 self
.__class
__.test_index
,
1375 inspect
.currentframe().f_code
.co_name
)
1376 self
.__class
__.test_index
+= 1
1378 with self
.assertRaises(Exception) as context
:
1379 test_config
["vim_conn"].new_tenant(Invalid_tenant_name
)
1381 self
.assertEqual((context
.exception
).http_code
, 400)
1383 def test_020_delete_tenant(self
):
1384 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1385 self
.__class
__.test_index
,
1386 inspect
.currentframe().f_code
.co_name
)
1387 self
.__class
__.test_index
+= 1
1389 tenant_id
= test_config
["vim_conn"].delete_tenant(self
.__class
__.tenant_id
)
1390 self
.assertEqual(type(tenant_id
), str)
1392 def test_030_delete_tenant_negative(self
):
1393 Non_exist_tenant_name
= 'Test_30_tenant'
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 with self
.assertRaises(Exception) as context
:
1400 test_config
["vim_conn"].delete_tenant(Non_exist_tenant_name
)
1402 self
.assertEqual((context
.exception
).http_code
, 404)
1407 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
1408 scenario based tests.
1410 class descriptor_based_scenario_test(test_base
):
1412 scenario_test_path
= None
1415 def setUpClass(cls
):
1417 cls
.to_delete_list
= []
1418 cls
.scenario_uuids
= []
1419 cls
.instance_scenario_uuids
= []
1420 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
1421 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
1424 def tearDownClass(cls
):
1425 test_config
["test_number"] += 1
1427 def test_000_load_scenario(self
):
1428 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1429 inspect
.currentframe().f_code
.co_name
,
1430 test_config
["test_folder"])
1431 self
.__class
__.test_index
+= 1
1433 descriptor_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/*.yaml')
1434 vnf_descriptors
= []
1435 scenario_descriptors
= []
1436 for descriptor_file
in descriptor_files
:
1437 with
open(descriptor_file
, 'r') as stream
:
1438 descriptor
= yaml
.load(stream
)
1439 if "vnf" in descriptor
or "vnfd:vnfd-catalog" in descriptor
or "vnfd-catalog" in descriptor
:
1440 vnf_descriptors
.append(descriptor
)
1442 scenario_descriptors
.append(descriptor
)
1444 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
1445 if not vnf_descriptors
or not scenario_descriptors
or len(scenario_descriptors
) > 1:
1446 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
1447 test_config
["test_folder"]))
1450 for vnf_descriptor
in vnf_descriptors
:
1451 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
1452 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
, image_name
=test_config
["image_name"])
1455 vnf_uuid
= vnf
['vnf']['uuid']
1457 vnf_uuid
= vnf
['vnfd'][0]['uuid']
1458 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
1459 "params": {"uuid": vnf_uuid
}})
1461 # load the scenario definition
1462 for scenario_descriptor
in scenario_descriptors
:
1463 # networks = scenario_descriptor['scenario']['networks']
1464 # networks[test_config["mgmt_net"]] = networks.pop('mgmt')
1465 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
1466 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
1467 logger
.debug(scenario
)
1468 if 'scenario' in scenario
:
1469 scenario_uuid
= scenario
['scenario']['uuid']
1471 scenario_uuid
= scenario
['nsd'][0]['uuid']
1472 self
.__class
__.to_delete_list
.insert(0, {"item": "scenario",
1473 "function": test_config
["client"].delete_scenario
,
1474 "params": {"uuid": scenario_uuid
}})
1475 self
.__class
__.scenario_uuids
.append(scenario_uuid
)
1477 def test_010_instantiate_scenario(self
):
1478 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1479 inspect
.currentframe().f_code
.co_name
,
1480 test_config
["test_folder"])
1481 self
.__class
__.test_index
+= 1
1482 for scenario_uuid
in self
.__class
__.scenario_uuids
:
1483 instance_descriptor
= {
1485 "name": self
.__class
__.test_text
,
1486 "scenario": scenario_uuid
,
1488 "mgmt": {"sites": [ { "netmap-use": test_config
["mgmt_net"]} ]}
1492 instance
= test_config
["client"].create_instance(instance_descriptor
)
1493 self
.__class
__.instance_scenario_uuids
.append(instance
['uuid'])
1494 logger
.debug(instance
)
1495 self
.__class
__.to_delete_list
.insert(0, {"item": "instance",
1496 "function": test_config
["client"].delete_instance
,
1497 "params": {"uuid": instance
['uuid']}})
1499 def test_020_check_deployent(self
):
1500 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1501 inspect
.currentframe().f_code
.co_name
,
1502 test_config
["test_folder"])
1503 self
.__class
__.test_index
+= 1
1505 if test_config
["manual"]:
1506 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1509 keep_waiting
= test_config
["timeout"]
1510 pending_instance_scenario_uuids
= list(self
.__class
__.instance_scenario_uuids
) # make a copy
1511 while pending_instance_scenario_uuids
:
1513 while index
< len(pending_instance_scenario_uuids
):
1514 result
= check_instance_scenario_active(pending_instance_scenario_uuids
[index
])
1516 del pending_instance_scenario_uuids
[index
]
1518 elif 'ERROR' in result
[1]:
1519 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
1521 raise Exception(msg
)
1524 if keep_waiting
>= 5:
1527 elif keep_waiting
> 0:
1528 time
.sleep(keep_waiting
)
1531 msg
= 'Timeout reached while waiting instance scenario to get active'
1533 raise Exception(msg
)
1535 def test_030_clean_deployment(self
):
1536 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1537 inspect
.currentframe().f_code
.co_name
,
1538 test_config
["test_folder"])
1539 self
.__class
__.test_index
+= 1
1540 #At the moment if you delete an scenario right after creating it, in openstack datacenters
1541 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1543 for item
in self
.__class
__.to_delete_list
:
1544 response
= item
["function"](**item
["params"])
1545 logger
.debug(response
)
1548 def _get_random_string(maxLength
):
1549 '''generates a string with random characters string.letters and string.digits
1550 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1554 minLength
= min_string
- len(prefix
)
1555 if maxLength
< min_string
: maxLength
= min_string
1556 maxLength
-= len(prefix
)
1557 length
= random
.randint(minLength
,maxLength
)
1558 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
1561 def test_vimconnector(args
):
1563 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1564 test_config
['vimtype'] = args
.vimtype
1565 if args
.vimtype
== "vmware":
1566 import vimconn_vmware
as vim
1568 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1570 tenant_name
= args
.tenant_name
1571 test_config
['tenant'] = tenant_name
1572 config_params
= yaml
.load(args
.config_param
)
1573 org_name
= config_params
.get('orgname')
1574 org_user
= config_params
.get('user')
1575 org_passwd
= config_params
.get('passwd')
1576 vim_url
= args
.endpoint_url
1577 test_config
['image_path'] = args
.image_path
1578 test_config
['image_name'] = args
.image_name
1580 # vmware connector obj
1581 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
)
1583 elif args
.vimtype
== "aws":
1584 import vimconn_aws
as vim
1585 elif args
.vimtype
== "openstack":
1586 import vimconn_openstack
as vim
1588 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1590 tenant_name
= args
.tenant_name
1591 test_config
['tenant'] = tenant_name
1592 config_params
= yaml
.load(args
.config_param
)
1593 os_user
= config_params
.get('user')
1594 os_passwd
= config_params
.get('passwd')
1595 vim_url
= args
.endpoint_url
1596 test_config
['image_path'] = args
.image_path
1597 test_config
['image_name'] = args
.image_name
1599 # openstack connector obj
1600 vim_persistent_info
= {}
1601 test_config
['vim_conn'] = vim
.vimconnector(
1602 uuid
="test-uuid-1", name
="VIO-openstack",
1603 tenant_id
=None, tenant_name
=tenant_name
,
1604 url
=vim_url
, url_admin
=None,
1605 user
=os_user
, passwd
=os_passwd
,
1606 config
=config_params
, persistent_info
=vim_persistent_info
1608 test_config
['vim_conn'].debug
= "true"
1610 elif args
.vimtype
== "openvim":
1611 import vimconn_openvim
as vim
1613 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
1617 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1618 # If only want to obtain a tests list print it and exit
1621 for cls
in clsmembers
:
1622 if cls
[0].startswith('test_vimconn'):
1623 tests_names
.append(cls
[0])
1625 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
1630 # Create the list of tests to be run
1631 code_based_tests
= []
1633 for test
in args
.tests
:
1634 for t
in test
.split(','):
1635 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1636 if len(matches_code_based_tests
) > 0:
1637 code_based_tests
.append(matches_code_based_tests
[0][1])
1639 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1641 if not code_based_tests
:
1643 for cls
in clsmembers
:
1644 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1645 if cls
[0].startswith('test_vimconn'):
1646 code_based_tests
.append(cls
[1])
1648 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1650 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1651 # This is handled in the tests using logging.
1652 stream
= open('/dev/null', 'w')
1654 # Run code based tests
1655 basic_tests_suite
= unittest
.TestSuite()
1656 for test
in code_based_tests
:
1657 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1658 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1659 executed
+= result
.testsRun
1660 failed
+= len(result
.failures
) + len(result
.errors
)
1661 if failfast
and failed
:
1663 if len(result
.failures
) > 0:
1664 logger
.debug("failures : {}".format(result
.failures
))
1665 if len(result
.errors
) > 0:
1666 logger
.debug("errors : {}".format(result
.errors
))
1667 return executed
, failed
1672 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1673 import openmanoclient
1676 test_config
["client"] = openmanoclient
.openmanoclient(
1677 endpoint_url
=args
.endpoint_url
,
1678 tenant_name
=args
.tenant_name
,
1679 datacenter_name
=args
.datacenter
,
1680 debug
=args
.debug
, logger
=test_config
["logger_name"])
1681 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1682 # If only want to obtain a tests list print it and exit
1685 for cls
in clsmembers
:
1686 if cls
[0].startswith('test_VIM'):
1687 tests_names
.append(cls
[0])
1689 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
1690 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1691 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1696 # Create the list of tests to be run
1697 code_based_tests
= []
1699 for test
in args
.tests
:
1700 for t
in test
.split(','):
1701 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1702 if len(matches_code_based_tests
) > 0:
1703 code_based_tests
.append(matches_code_based_tests
[0][1])
1705 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1707 if not code_based_tests
:
1709 for cls
in clsmembers
:
1710 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1711 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
1712 code_based_tests
.append(cls
[1])
1714 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1716 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1717 # This is handled in the tests using logging.
1718 stream
= open('/dev/null', 'w')
1720 # Run code based tests
1721 basic_tests_suite
= unittest
.TestSuite()
1722 for test
in code_based_tests
:
1723 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1724 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1725 executed
+= result
.testsRun
1726 failed
+= len(result
.failures
) + len(result
.errors
)
1727 if failfast
and failed
:
1729 if len(result
.failures
) > 0:
1730 logger
.debug("failures : {}".format(result
.failures
))
1731 if len(result
.errors
) > 0:
1732 logger
.debug("errors : {}".format(result
.errors
))
1733 return executed
, failed
1736 def test_deploy(args
):
1738 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1739 import openmanoclient
1742 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1743 test_config
["image_name"] = args
.image_name
1744 test_config
["mgmt_net"] = args
.mgmt_net
1745 test_config
["manual"] = args
.manual
1746 test_directory_content
= os
.listdir(test_config
["test_directory"])
1747 # If only want to obtain a tests list print it and exit
1749 msg
= "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
1754 descriptor_based_tests
= []
1755 # Create the list of tests to be run
1756 code_based_tests
= []
1758 for test
in args
.tests
:
1759 for t
in test
.split(','):
1760 if t
in test_directory_content
:
1761 descriptor_based_tests
.append(t
)
1763 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1765 if not descriptor_based_tests
:
1767 descriptor_based_tests
= test_directory_content
1769 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1771 # import openmanoclient from relative path
1772 test_config
["client"] = openmanoclient
.openmanoclient(
1773 endpoint_url
=args
.endpoint_url
,
1774 tenant_name
=args
.tenant_name
,
1775 datacenter_name
=args
.datacenter
,
1776 debug
=args
.debug
, logger
=test_config
["logger_name"])
1778 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1779 # This is handled in the tests using logging.
1780 stream
= open('/dev/null', 'w')
1781 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1782 for test
in descriptor_based_tests
:
1783 test_config
["test_folder"] = test
1784 test_suite
= unittest
.TestSuite()
1785 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
1786 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
1787 executed
+= result
.testsRun
1788 failed
+= len(result
.failures
) + len(result
.errors
)
1789 if failfast
and failed
:
1791 if len(result
.failures
) > 0:
1792 logger
.debug("failures : {}".format(result
.failures
))
1793 if len(result
.errors
) > 0:
1794 logger
.debug("errors : {}".format(result
.errors
))
1796 return executed
, failed
1798 if __name__
=="__main__":
1800 parser
= ArgumentParser(description
='Test RO module')
1801 parser
.add_argument('-v','--version', action
='version', help="Show current version",
1802 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
1805 parent_parser
= ArgumentParser(add_help
=False)
1806 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1807 dest
='failfast', action
="store_true", default
=False)
1808 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1809 dest
='failed', action
="store_true", default
=False)
1810 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
1811 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
1813 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
1814 help='Set the logger file. By default '+default_logger_file
)
1815 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
1816 help="Set the openmano tenant to use for the test. By default 'osm'")
1817 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
1818 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1819 dest
='timeout', type=int, default
=300)
1820 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
1822 subparsers
= parser
.add_subparsers(help='test sets')
1824 # Deployment test set
1825 # -------------------
1826 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
1827 help="test deployment using descriptors at RO_test folder ")
1828 deploy_parser
.set_defaults(func
=test_deploy
)
1830 # Mandatory arguments
1831 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
1832 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1833 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
1834 help='Image name available at datacenter used for the tests')
1835 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
1836 help='Set the vim management network to use for tests')
1838 # Optional arguments
1839 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
1840 help='Pause execution once deployed to allow manual checking of the '
1841 'deployed instance scenario')
1842 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1843 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1846 # -------------------
1847 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
1848 vimconn_parser
.set_defaults(func
=test_vimconnector
)
1849 # Mandatory arguments
1850 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1851 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
1852 help='Set the vimconnector type to test')
1853 mandatory_arguments
.add_argument('-c', '--config', dest
='config_param', required
=True,
1854 help='Set the vimconnector specific config parameters in dictionary format')
1855 mandatory_arguments
.add_argument('-u', '--url', dest
='endpoint_url',required
=True, help="Set the vim connector url or Host IP")
1856 # Optional arguments
1857 vimconn_parser
.add_argument('-i', '--image-path', dest
='image_path', help="Provide image path present at RO container")
1858 vimconn_parser
.add_argument('-n', '--image-name', dest
='image_name', help="Provide image name for test")
1859 # TODO add optional arguments for vimconn tests
1860 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1862 # Datacenter test set
1863 # -------------------
1864 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
1865 vimconn_parser
.set_defaults(func
=test_vim
)
1867 # Mandatory arguments
1868 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1869 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1871 # Optional arguments
1872 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1873 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1875 argcomplete
.autocomplete(parser
)
1876 args
= parser
.parse_args()
1880 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1881 logger_level
= 'INFO'
1883 logger_level
= 'DEBUG'
1885 logger_level
= 'WARNING'
1886 logger_name
= os
.path
.basename(__file__
)
1887 test_config
["logger_name"] = logger_name
1888 logger
= logging
.getLogger(logger_name
)
1889 logger
.setLevel(logger_level
)
1890 failfast
= args
.failfast
1892 # Configure a logging handler to store in a logging file
1893 if args
.logger_file
:
1894 fileHandler
= logging
.FileHandler(args
.logger_file
)
1895 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1896 fileHandler
.setFormatter(formatter_fileHandler
)
1897 logger
.addHandler(fileHandler
)
1899 # Configure a handler to print to stdout
1900 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
1901 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1902 consoleHandler
.setFormatter(formatter_consoleHandler
)
1903 logger
.addHandler(consoleHandler
)
1905 logger
.debug('Program started with the following arguments: ' + str(args
))
1907 # set test config parameters
1908 test_config
["timeout"] = args
.timeout
1909 test_config
["test_number"] = 1
1911 executed
, failed
= args
.func(args
)
1914 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
1915 sys
.exit(1 if failed
else 0)