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
)
295 class test_vimconn_new_network(test_base
):
298 def test_000_new_network(self
):
299 self
.__class
__.network_name
= _get_random_string(20)
300 network_type
= 'bridge'
302 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
303 self
.__class
__.test_index
, inspect
.currentframe().f_code
.co_name
)
304 self
.__class
__.test_index
+= 1
306 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
307 net_type
=network_type
)
308 self
.__class
__.network_id
= network
309 logger
.debug("{}".format(network
))
311 network_list
= test_config
["vim_conn"].get_network_list()
312 for net
in network_list
:
313 if self
.__class
__.network_name
in net
.get('name'):
314 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
315 self
.assertEqual(net
.get('type'), network_type
)
317 # Deleting created network
318 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
320 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
322 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
324 def test_010_new_network_by_types(self
):
326 network_types
= ['data','bridge','mgmt']
327 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
328 self
.__class
__.test_index
,
329 inspect
.currentframe().f_code
.co_name
)
330 self
.__class
__.test_index
+= 1
331 for net_type
in network_types
:
332 self
.__class
__.network_name
= _get_random_string(20)
333 network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
336 delete_net_ids
.append(network_id
)
337 logger
.debug("{}".format(network_id
))
339 network_list
= test_config
["vim_conn"].get_network_list()
340 for net
in network_list
:
341 if self
.__class
__.network_name
in net
.get('name'):
342 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
343 if net_type
in net
.get('type'):
344 self
.assertEqual(net
.get('type'), net_type
)
346 self
.assertNotEqual(net
.get('type'), net_type
)
348 # Deleting created network
349 for net_id
in delete_net_ids
:
350 result
= test_config
["vim_conn"].delete_network(net_id
)
352 logger
.info("Network id {} sucessfully deleted".format(net_id
))
354 logger
.info("Failed to delete network id {}".format(net_id
))
356 def test_020_new_network_by_ipprofile(self
):
357 test_directory_content
= os
.listdir(test_config
["test_directory"])
359 for dir_name
in test_directory_content
:
360 if dir_name
== 'simple_multi_vnfc':
361 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
362 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
365 for vnfd
in vnfd_files
:
366 with
open(vnfd
, 'r') as stream
:
367 vnf_descriptor
= yaml
.load(stream
)
369 internal_connections_list
= vnf_descriptor
['vnf']['internal-connections']
370 for item
in internal_connections_list
:
371 if 'ip-profile' in item
:
372 version
= item
['ip-profile']['ip-version']
373 dhcp_count
= item
['ip-profile']['dhcp']['count']
374 dhcp_enabled
= item
['ip-profile']['dhcp']['enabled']
376 self
.__class
__.network_name
= _get_random_string(20)
377 ip_profile
= {'dhcp_count': dhcp_count
,
378 'dhcp_enabled': dhcp_enabled
,
379 'ip_version': version
381 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
382 self
.__class
__.test_index
,
383 inspect
.currentframe().f_code
.co_name
)
384 self
.__class
__.test_index
+= 1
385 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
387 ip_profile
=ip_profile
)
388 self
.__class
__.network_id
= network
389 logger
.debug("{}".format(network
))
391 network_list
= test_config
["vim_conn"].get_network_list()
392 for net
in network_list
:
393 if self
.__class
__.network_name
in net
.get('name'):
394 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
396 # Deleting created network
397 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
399 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
401 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
403 def test_030_new_network_by_isshared(self
):
404 self
.__class
__.network_name
= _get_random_string(20)
406 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
407 self
.__class
__.test_index
,
408 inspect
.currentframe().f_code
.co_name
)
409 self
.__class
__.test_index
+= 1
410 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
413 self
.__class
__.network_id
= network
414 logger
.debug("{}".format(network
))
416 network_list
= test_config
["vim_conn"].get_network_list()
417 for net
in network_list
:
418 if self
.__class
__.network_name
in net
.get('name'):
419 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
420 self
.assertEqual(net
.get('shared'), shared
)
422 # Deleting created network
423 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
425 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
427 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
429 def test_040_new_network_by_negative(self
):
430 self
.__class
__.network_name
= _get_random_string(20)
431 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
432 self
.__class
__.test_index
,
433 inspect
.currentframe().f_code
.co_name
)
434 self
.__class
__.test_index
+= 1
435 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
436 net_type
='unknowntype')
437 self
.__class
__.network_id
= network
438 logger
.debug("{}".format(network
))
439 network_list
= test_config
["vim_conn"].get_network_list()
440 for net
in network_list
:
441 if self
.__class
__.network_name
in net
.get('name'):
442 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
444 # Deleting created network
445 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
447 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
449 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
451 def test_050_refresh_nets_status(self
):
452 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
453 self
.__class
__.test_index
,
454 inspect
.currentframe().f_code
.co_name
)
455 self
.__class
__.test_index
+= 1
456 # creating new network
457 network_name
= _get_random_string(20)
459 network_id
= test_config
["vim_conn"].new_network(net_name
=network_name
,
462 net_dict
= test_config
["vim_conn"].refresh_nets_status([network_id
])
463 for attr
in net_dict
[network_id
]:
465 self
.assertEqual(net_dict
[network_id
][attr
], 'ACTIVE')
467 # Deleting created network
468 result
= test_config
["vim_conn"].delete_network(network_id
)
470 logger
.info("Network id {} sucessfully deleted".format(network_id
))
472 logger
.info("Failed to delete network id {}".format(network_id
))
474 def test_060_refresh_nets_status_negative(self
):
475 unknown_net_id
= str(uuid
.uuid4())
476 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
477 self
.__class
__.test_index
,
478 inspect
.currentframe().f_code
.co_name
)
479 self
.__class
__.test_index
+= 1
482 net_dict
= test_config
["vim_conn"].refresh_nets_status([unknown_net_id
])
483 self
.assertEqual(net_dict
, {})
485 class test_vimconn_get_network_list(test_base
):
489 # creating new network
490 self
.__class
__.network_name
= _get_random_string(20)
491 self
.__class
__.net_type
= 'bridge'
492 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
493 net_type
=self
.__class
__.net_type
)
494 self
.__class
__.network_id
= network
495 logger
.debug("{}".format(network
))
498 test_base
.tearDown(self
)
500 # Deleting created network
501 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
503 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
505 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
507 def test_000_get_network_list(self
):
508 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
509 self
.__class
__.test_index
,
510 inspect
.currentframe().f_code
.co_name
)
511 self
.__class
__.test_index
+= 1
513 network_list
= test_config
["vim_conn"].get_network_list()
514 for net
in network_list
:
515 if self
.__class
__.network_name
in net
.get('name'):
516 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
517 self
.assertEqual(net
.get('type'), self
.__class
__.net_type
)
518 self
.assertEqual(net
.get('status'), 'ACTIVE')
519 self
.assertEqual(net
.get('shared'), False)
521 def test_010_get_network_list_by_name(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_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
529 # find network from list by it's name
530 new_network_list
= test_config
["vim_conn"].get_network_list({'name': network_name
})
531 for list_item
in new_network_list
:
532 if self
.__class
__.network_name
in list_item
.get('name'):
533 self
.assertEqual(network_name
, list_item
.get('name'))
534 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
535 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
537 def test_020_get_network_list_by_id(self
):
538 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
539 self
.__class
__.test_index
,
540 inspect
.currentframe().f_code
.co_name
)
541 self
.__class
__.test_index
+= 1
543 # find network from list by it's id
544 new_network_list
= test_config
["vim_conn"].get_network_list({'id':self
.__class
__.network_id
})
545 for list_item
in new_network_list
:
546 if self
.__class
__.network_id
in list_item
.get('id'):
547 self
.assertEqual(self
.__class
__.network_id
, list_item
.get('id'))
548 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
549 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
551 def test_030_get_network_list_by_shared(self
):
553 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
554 self
.__class
__.test_index
,
555 inspect
.currentframe().f_code
.co_name
)
556 self
.__class
__.test_index
+= 1
558 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
559 # find network from list by it's shared value
560 new_network_list
= test_config
["vim_conn"].get_network_list({'shared':Shared
,
561 'name':network_name
})
562 for list_item
in new_network_list
:
563 if list_item
.get('shared') == Shared
:
564 self
.assertEqual(list_item
.get('shared'), Shared
)
565 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
566 self
.assertEqual(network_name
, list_item
.get('name'))
568 def test_040_get_network_list_by_tenant_id(self
):
569 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
570 self
.__class
__.test_index
,
571 inspect
.currentframe().f_code
.co_name
)
572 self
.__class
__.test_index
+= 1
574 tenant_list
= test_config
["vim_conn"].get_tenant_list()
575 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
577 for tenant_item
in tenant_list
:
578 if test_config
['tenant'] == tenant_item
.get('name'):
579 # find network from list by it's tenant id
580 tenant_id
= tenant_item
.get('id')
581 new_network_list
= test_config
["vim_conn"].get_network_list({'tenant_id':tenant_id
,
582 'name':network_name
})
583 for list_item
in new_network_list
:
584 self
.assertEqual(tenant_id
, list_item
.get('tenant_id'))
585 self
.assertEqual(network_name
, list_item
.get('name'))
586 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
587 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
589 def test_050_get_network_list_by_status(self
):
590 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
591 self
.__class
__.test_index
,
592 inspect
.currentframe().f_code
.co_name
)
593 self
.__class
__.test_index
+= 1
596 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
598 # find network from list by it's status
599 new_network_list
= test_config
["vim_conn"].get_network_list({'status':status
,
600 'name': network_name
})
601 for list_item
in new_network_list
:
602 self
.assertIn(self
.__class
__.network_name
, list_item
.get('name'))
603 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
604 self
.assertEqual(list_item
.get('status'), status
)
606 def test_060_get_network_list_by_negative(self
):
607 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
608 self
.__class
__.test_index
,
609 inspect
.currentframe().f_code
.co_name
)
610 self
.__class
__.test_index
+= 1
612 network_list
= test_config
["vim_conn"].get_network_list({'name': 'unknown_name'})
613 self
.assertEqual(network_list
, [])
615 class test_vimconn_get_network(test_base
):
619 # creating new network
620 self
.__class
__.network_name
= _get_random_string(20)
621 self
.__class
__.net_type
= 'bridge'
622 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
623 net_type
=self
.__class
__.net_type
)
624 self
.__class
__.network_id
= network
625 logger
.debug("{}".format(network
))
628 test_base
.tearDown(self
)
630 # Deleting created network
631 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
633 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
635 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
637 def test_000_get_network(self
):
638 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
639 self
.__class
__.test_index
,
640 inspect
.currentframe().f_code
.co_name
)
641 self
.__class
__.test_index
+= 1
643 network_info
= test_config
["vim_conn"].get_network(self
.__class
__.network_id
)
644 self
.assertEqual(network_info
.get('status'), 'ACTIVE')
645 self
.assertIn(self
.__class
__.network_name
, network_info
.get('name'))
646 self
.assertEqual(network_info
.get('type'), self
.__class
__.net_type
)
647 self
.assertEqual(network_info
.get('id'), self
.__class
__.network_id
)
649 def test_010_get_network_negative(self
):
650 Non_exist_id
= str(uuid
.uuid4())
651 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
652 self
.__class
__.test_index
,
653 inspect
.currentframe().f_code
.co_name
)
654 self
.__class
__.test_index
+= 1
655 with self
.assertRaises(Exception) as context
:
656 test_config
["vim_conn"].get_network(Non_exist_id
)
658 self
.assertEqual((context
.exception
).http_code
, 404)
660 class test_vimconn_delete_network(test_base
):
663 def test_000_delete_network(self
):
665 self
.__class
__.network_name
= _get_random_string(20)
666 self
.__class
__.net_type
= 'bridge'
667 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
668 net_type
=self
.__class
__.net_type
)
669 self
.__class
__.network_id
= network
670 logger
.debug("{}".format(network
))
672 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
673 self
.__class
__.test_index
,
674 inspect
.currentframe().f_code
.co_name
)
675 self
.__class
__.test_index
+= 1
677 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
679 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
681 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
683 # after deleting network we check in network list
684 network_list
= test_config
["vim_conn"].get_network_list({ 'id':self
.__class
__.network_id
})
685 self
.assertEqual(network_list
, [])
687 def test_010_delete_network_negative(self
):
688 Non_exist_id
= str(uuid
.uuid4())
690 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
691 self
.__class
__.test_index
,
692 inspect
.currentframe().f_code
.co_name
)
693 self
.__class
__.test_index
+= 1
695 with self
.assertRaises(Exception) as context
:
696 test_config
["vim_conn"].delete_network(Non_exist_id
)
698 self
.assertEqual((context
.exception
).http_code
, 404)
700 class test_vimconn_get_flavor(test_base
):
702 def test_000_get_flavor(self
):
703 test_directory_content
= os
.listdir(test_config
["test_directory"])
705 for dir_name
in test_directory_content
:
706 if dir_name
== 'simple_linux':
707 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
708 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
711 for vnfd
in vnfd_files
:
712 with
open(vnfd
, 'r') as stream
:
713 vnf_descriptor
= yaml
.load(stream
)
715 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
716 for item
in vnfc_list
:
717 if 'ram' in item
and 'vcpus' in item
and 'disk' in item
:
719 vcpus
= item
['vcpus']
722 flavor_data
= {'ram': ram
,
727 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
728 self
.__class
__.test_index
,
729 inspect
.currentframe().f_code
.co_name
)
730 self
.__class
__.test_index
+= 1
732 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
734 result
= test_config
["vim_conn"].get_flavor(flavor_id
)
735 self
.assertEqual(ram
, result
['ram'])
736 self
.assertEqual(vcpus
, result
['vcpus'])
737 self
.assertEqual(disk
, result
['disk'])
740 result
= test_config
["vim_conn"].delete_flavor(flavor_id
)
742 logger
.info("Flavor id {} sucessfully deleted".format(result
))
744 logger
.info("Failed to delete flavor id {}".format(result
))
746 def test_010_get_flavor_negative(self
):
747 Non_exist_flavor_id
= str(uuid
.uuid4())
749 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
750 self
.__class
__.test_index
,
751 inspect
.currentframe().f_code
.co_name
)
752 self
.__class
__.test_index
+= 1
754 with self
.assertRaises(Exception) as context
:
755 test_config
["vim_conn"].get_flavor(Non_exist_flavor_id
)
757 self
.assertEqual((context
.exception
).http_code
, 404)
759 class test_vimconn_new_flavor(test_base
):
762 def test_000_new_flavor(self
):
763 flavor_data
= {'ram': 1024, 'vpcus': 1, 'disk': 10}
765 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
766 self
.__class
__.test_index
,
767 inspect
.currentframe().f_code
.co_name
)
768 self
.__class
__.test_index
+= 1
771 self
.__class
__.flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
772 self
.assertEqual(type(self
.__class
__.flavor_id
),str)
773 self
.assertIsInstance(uuid
.UUID(self
.__class
__.flavor_id
),uuid
.UUID
)
775 def test_010_delete_flavor(self
):
776 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
777 self
.__class
__.test_index
,
778 inspect
.currentframe().f_code
.co_name
)
779 self
.__class
__.test_index
+= 1
782 result
= test_config
["vim_conn"].delete_flavor(self
.__class
__.flavor_id
)
784 logger
.info("Flavor id {} sucessfully deleted".format(result
))
786 logger
.error("Failed to delete flavor id {}".format(result
))
787 raise Exception ("Failed to delete created flavor")
789 def test_020_new_flavor_negative(self
):
790 Invalid_flavor_data
= {'ram': '1024', 'vcpus': 2.0, 'disk': 2.0}
792 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
793 self
.__class
__.test_index
,
794 inspect
.currentframe().f_code
.co_name
)
795 self
.__class
__.test_index
+= 1
797 with self
.assertRaises(Exception) as context
:
798 test_config
["vim_conn"].new_flavor(Invalid_flavor_data
)
800 self
.assertEqual((context
.exception
).http_code
, 400)
802 def test_030_delete_flavor_negative(self
):
803 Non_exist_flavor_id
= str(uuid
.uuid4())
805 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
806 self
.__class
__.test_index
,
807 inspect
.currentframe().f_code
.co_name
)
808 self
.__class
__.test_index
+= 1
810 with self
.assertRaises(Exception) as context
:
811 test_config
["vim_conn"].delete_flavor(Non_exist_flavor_id
)
813 self
.assertEqual((context
.exception
).http_code
, 404)
815 class test_vimconn_new_image(test_base
):
817 def test_000_new_image(self
):
818 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
819 self
.__class
__.test_index
,
820 inspect
.currentframe().f_code
.co_name
)
821 self
.__class
__.test_index
+= 1
823 image_path
= test_config
['image_path']
825 self
.__class
__.image_id
= test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : image_path
})
827 self
.assertEqual(type(self
.__class
__.image_id
),str)
828 self
.assertIsInstance(uuid
.UUID(self
.__class
__.image_id
),uuid
.UUID
)
830 self
.skipTest("Skipping test as image file not present at RO container")
832 def test_010_new_image_negative(self
):
833 Non_exist_image_path
= '/temp1/cirros.ovf'
835 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
836 self
.__class
__.test_index
,
837 inspect
.currentframe().f_code
.co_name
)
838 self
.__class
__.test_index
+= 1
840 with self
.assertRaises(Exception) as context
:
841 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
843 self
.assertEqual((context
.exception
).http_code
, 400)
845 def test_020_delete_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_id
= test_config
["vim_conn"].delete_image(self
.__class
__.image_id
)
852 self
.assertEqual(type(image_id
),str)
854 def test_030_delete_image_negative(self
):
855 Non_exist_image_id
= str(uuid
.uuid4())
857 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
858 self
.__class
__.test_index
,
859 inspect
.currentframe().f_code
.co_name
)
860 self
.__class
__.test_index
+= 1
862 with self
.assertRaises(Exception) as context
:
863 test_config
["vim_conn"].delete_image(Non_exist_image_id
)
865 self
.assertEqual((context
.exception
).http_code
, 404)
867 class test_vimconn_get_image_id_from_path(test_base
):
869 def test_000_get_image_id_from_path(self
):
870 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
871 self
.__class
__.test_index
,
872 inspect
.currentframe().f_code
.co_name
)
873 self
.__class
__.test_index
+= 1
875 image_path
= test_config
['image_path']
877 image_id
= test_config
["vim_conn"].get_image_id_from_path( image_path
)
878 self
.assertEqual(type(image_id
),str)
880 self
.skipTest("Skipping test as image file not present at RO container")
882 def test_010_get_image_id_from_path_negative(self
):
883 Non_exist_image_path
= '/temp1/cirros.ovf'
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"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
893 self
.assertEqual((context
.exception
).http_code
, 400)
895 class test_vimconn_get_image_list(test_base
):
899 def test_000_get_image_list(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
904 image_list
= test_config
["vim_conn"].get_image_list()
906 for item
in image_list
:
908 self
.__class
__.image_name
= item
['name']
909 self
.__class
__.image_id
= item
['id']
910 self
.assertEqual(type(self
.__class
__.image_name
),str)
911 self
.assertEqual(type(self
.__class
__.image_id
),str)
913 def test_010_get_image_list_by_name(self
):
914 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
915 self
.__class
__.test_index
,
916 inspect
.currentframe().f_code
.co_name
)
917 self
.__class
__.test_index
+= 1
919 image_list
= test_config
["vim_conn"].get_image_list({'name': self
.__class
__.image_name
})
921 for item
in image_list
:
922 self
.assertEqual(type(item
['id']), str)
923 self
.assertEqual(item
['id'], self
.__class
__.image_id
)
924 self
.assertEqual(type(item
['name']), str)
925 self
.assertEqual(item
['name'], self
.__class
__.image_name
)
927 def test_020_get_image_list_by_id(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
933 filter_image_list
= test_config
["vim_conn"].get_image_list({'id': self
.__class
__.image_id
})
935 for item1
in filter_image_list
:
936 self
.assertEqual(type(item1
.get('id')), str)
937 self
.assertEqual(item1
.get('id'), self
.__class
__.image_id
)
938 self
.assertEqual(type(item1
.get('name')), str)
939 self
.assertEqual(item1
.get('name'), self
.__class
__.image_name
)
941 def test_030_get_image_list_negative(self
):
942 Non_exist_image_id
= uuid
.uuid4()
943 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
944 self
.__class
__.test_index
,
945 inspect
.currentframe().f_code
.co_name
)
946 self
.__class
__.test_index
+= 1
947 image_list
= test_config
["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id
})
949 self
.assertIsNotNone(image_list
, None)
950 self
.assertEqual(image_list
, [])
952 class test_vimconn_new_vminstance(test_base
):
961 self
.__class
__.network_name
= _get_random_string(20)
962 self
.__class
__.net_type
= 'bridge'
964 self
.__class
__.network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
965 net_type
=self
.__class
__.net_type
)
968 test_base
.tearDown(self
)
969 # Deleting created network
970 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
972 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
974 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
976 def test_000_new_vminstance(self
):
977 vpci
= "0000:00:11.0"
980 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
983 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
985 # find image name and image id
986 if test_config
['image_name']:
987 image_list
= test_config
['vim_conn'].get_image_list({'name': test_config
['image_name']})
988 if len(image_list
) == 0:
989 raise Exception("Image {} is not found at VIM".format(test_config
['image_name']))
991 self
.__class
__.image_id
= image_list
[0]['id']
993 image_list
= test_config
['vim_conn'].get_image_list()
994 if len(image_list
) == 0:
995 raise Exception("Not found any image at VIM")
997 self
.__class
__.image_id
= image_list
[0]['id']
999 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1000 self
.__class
__.test_index
,
1001 inspect
.currentframe().f_code
.co_name
)
1002 self
.__class
__.test_index
+= 1
1004 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
}]
1006 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
)
1008 self
.assertEqual(type(self
.__class
__.instance_id
),str)
1010 def test_010_new_vminstance_by_model(self
):
1011 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1012 model_name
= 'e1000'
1016 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1018 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1019 self
.__class
__.test_index
,
1020 inspect
.currentframe().f_code
.co_name
)
1021 self
.__class
__.test_index
+= 1
1023 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
}]
1025 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1026 flavor_id
=flavor_id
,
1028 self
.assertEqual(type(instance_id
),str)
1029 # Deleting created vm instance
1030 logger
.info("Deleting created vm intance")
1031 test_config
["vim_conn"].delete_vminstance(instance_id
)
1034 def test_020_new_vminstance_by_net_use(self
):
1035 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1040 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1042 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1043 self
.__class
__.test_index
,
1044 inspect
.currentframe().f_code
.co_name
)
1045 self
.__class
__.test_index
+= 1
1047 net_list
= [{'use': net_use
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1049 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1050 flavor_id
=flavor_id
,
1052 self
.assertEqual(type(instance_id
),str)
1053 # Deleting created vm instance
1054 logger
.info("Deleting created vm intance")
1055 test_config
["vim_conn"].delete_vminstance(instance_id
)
1058 def test_030_new_vminstance_by_net_type(self
):
1059 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1064 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1066 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1067 self
.__class
__.test_index
,
1068 inspect
.currentframe().f_code
.co_name
)
1069 self
.__class
__.test_index
+= 1
1071 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': _type
, 'net_id': self
.__class
__.network_id
}]
1073 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1074 flavor_id
=flavor_id
,
1076 self
.assertEqual(type(instance_id
),str)
1077 # Deleting created vm instance
1078 logger
.info("Deleting created vm intance")
1079 test_config
["vim_conn"].delete_vminstance(instance_id
)
1082 def test_040_new_vminstance_by_cloud_config(self
):
1083 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1085 user_name
= 'test_user'
1087 key_pairs
= ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com']
1089 users_data
= [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com'], 'name': user_name
}]
1091 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
}
1094 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1096 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1097 self
.__class
__.test_index
,
1098 inspect
.currentframe().f_code
.co_name
)
1099 self
.__class
__.test_index
+= 1
1101 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1103 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Cloud_vm', image_id
=self
.__class
__.image_id
,
1104 flavor_id
=flavor_id
,
1106 cloud_config
=cloud_data
)
1107 self
.assertEqual(type(instance_id
),str)
1108 # Deleting created vm instance
1109 logger
.info("Deleting created vm intance")
1110 test_config
["vim_conn"].delete_vminstance(instance_id
)
1113 def test_050_new_vminstance_by_disk_list(self
):
1114 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1117 device_data
= [{'image_id': self
.__class
__.image_id
, 'size': '5'}]
1120 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1122 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1123 self
.__class
__.test_index
,
1124 inspect
.currentframe().f_code
.co_name
)
1125 self
.__class
__.test_index
+= 1
1127 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1129 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='VM_test1', image_id
=self
.__class
__.image_id
,
1130 flavor_id
=flavor_id
,
1132 disk_list
=device_data
)
1133 self
.assertEqual(type(instance_id
),str)
1134 # Deleting created vm instance
1135 logger
.info("Deleting created vm intance")
1136 test_config
["vim_conn"].delete_vminstance(instance_id
)
1139 def test_060_new_vminstance_negative(self
):
1140 unknown_flavor_id
= str(uuid
.uuid4())
1141 unknown_image_id
= str(uuid
.uuid4())
1144 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1145 self
.__class
__.test_index
,
1146 inspect
.currentframe().f_code
.co_name
)
1147 self
.__class
__.test_index
+= 1
1149 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1151 with self
.assertRaises(Exception) as context
:
1152 test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=unknown_image_id
,
1153 flavor_id
=unknown_flavor_id
,
1155 self
.assertEqual((context
.exception
).http_code
, 404)
1157 def test_070_get_vminstance(self
):
1158 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1159 self
.__class
__.test_index
,
1160 inspect
.currentframe().f_code
.co_name
)
1161 self
.__class
__.test_index
+= 1
1163 # Get instance by its id
1164 vm_info
= test_config
["vim_conn"].get_vminstance(self
.__class
__.instance_id
)
1166 if test_config
['vimtype'] == 'vmware':
1167 for attr
in vm_info
:
1168 if attr
== 'status':
1169 self
.assertEqual(vm_info
[attr
], 'ACTIVE')
1170 if attr
== 'hostId':
1171 self
.assertEqual(type(vm_info
[attr
]), str)
1172 if attr
== 'interfaces':
1173 self
.assertEqual(type(vm_info
[attr
]), list)
1174 self
.assertEqual(vm_info
[attr
][0]['IsConnected'], 'true')
1175 if attr
== 'IsEnabled':
1176 self
.assertEqual(vm_info
[attr
], 'true')
1178 def test_080_get_vminstance_negative(self
):
1179 unknown_instance_id
= str(uuid
.uuid4())
1181 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1182 self
.__class
__.test_index
,
1183 inspect
.currentframe().f_code
.co_name
)
1184 self
.__class
__.test_index
+= 1
1186 with self
.assertRaises(Exception) as context
:
1187 test_config
["vim_conn"].get_vminstance(unknown_instance_id
)
1189 self
.assertEqual((context
.exception
).http_code
, 404)
1191 def test_090_refresh_vms_status(self
):
1192 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1193 self
.__class
__.test_index
,
1194 inspect
.currentframe().f_code
.co_name
)
1195 self
.__class
__.test_index
+= 1
1197 vm_list
.append(self
.__class
__.instance_id
)
1200 vm_info
= test_config
["vim_conn"].refresh_vms_status(vm_list
)
1201 for attr
in vm_info
[self
.__class
__.instance_id
]:
1202 if attr
== 'status':
1203 self
.assertEqual(vm_info
[self
.__class
__.instance_id
][attr
], 'ACTIVE')
1204 if attr
== 'interfaces':
1205 self
.assertEqual(type(vm_info
[self
.__class
__.instance_id
][attr
]), list)
1207 def test_100_refresh_vms_status_negative(self
):
1208 unknown_id
= str(uuid
.uuid4())
1210 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1211 self
.__class
__.test_index
,
1212 inspect
.currentframe().f_code
.co_name
)
1213 self
.__class
__.test_index
+= 1
1215 vm_dict
= test_config
["vim_conn"].refresh_vms_status([unknown_id
])
1216 self
.assertEqual(vm_dict
, {})
1218 def test_110_action_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 action_list
= ['shutdown','start','shutoff','rebuild','pause','resume']
1225 # various action on vminstace
1226 for action
in action_list
:
1227 instance_id
= test_config
["vim_conn"].action_vminstance(self
.__class
__.instance_id
,
1229 self
.assertEqual(instance_id
, self
.__class
__.instance_id
)
1231 def test_120_action_vminstance_negative(self
):
1232 non_exist_id
= str(uuid
.uuid4())
1233 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1234 self
.__class
__.test_index
,
1235 inspect
.currentframe().f_code
.co_name
)
1236 self
.__class
__.test_index
+= 1
1239 with self
.assertRaises(Exception) as context
:
1240 test_config
["vim_conn"].action_vminstance(non_exist_id
, { action
: None})
1242 self
.assertEqual((context
.exception
).http_code
, 400)
1244 def test_130_delete_vminstance(self
):
1245 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1246 self
.__class
__.test_index
,
1247 inspect
.currentframe().f_code
.co_name
)
1248 self
.__class
__.test_index
+= 1
1250 # Deleting created vm instance
1251 logger
.info("Deleting created vm instance")
1252 test_config
["vim_conn"].delete_vminstance(self
.__class
__.instance_id
)
1255 class test_vimconn_get_tenant_list(test_base
):
1258 def test_000_get_tenant_list(self
):
1259 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1260 self
.__class
__.test_index
,
1261 inspect
.currentframe().f_code
.co_name
)
1262 self
.__class
__.test_index
+= 1
1264 # Getting tenant list
1265 tenant_list
= test_config
["vim_conn"].get_tenant_list()
1267 for item
in tenant_list
:
1268 if test_config
['tenant'] == item
['name']:
1269 self
.__class
__.tenant_id
= item
['id']
1270 self
.assertEqual(type(item
['name']), str)
1271 self
.assertEqual(type(item
['id']), str)
1273 def test_010_get_tenant_list_by_id(self
):
1274 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1275 self
.__class
__.test_index
,
1276 inspect
.currentframe().f_code
.co_name
)
1277 self
.__class
__.test_index
+= 1
1279 # Getting filter tenant list by its id
1280 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'id': self
.__class
__.tenant_id
})
1282 for item
in filter_tenant_list
:
1283 self
.assertEqual(type(item
['id']), str)
1284 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1286 def test_020_get_tenant_list_by_name(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 filter tenant list by its name
1293 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant']})
1295 for item
in filter_tenant_list
:
1296 self
.assertEqual(type(item
['name']), str)
1297 self
.assertEqual(item
['name'], test_config
['tenant'])
1299 def test_030_get_tenant_list_by_name_and_id(self
):
1300 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1301 self
.__class
__.test_index
,
1302 inspect
.currentframe().f_code
.co_name
)
1303 self
.__class
__.test_index
+= 1
1305 # Getting filter tenant list by its name and id
1306 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant'],
1307 'id': self
.__class
__.tenant_id
})
1309 for item
in filter_tenant_list
:
1310 self
.assertEqual(type(item
['name']), str)
1311 self
.assertEqual(type(item
['id']), str)
1312 self
.assertEqual(item
['name'], test_config
['tenant'])
1313 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1315 def test_040_get_tenant_list_negative(self
):
1316 non_exist_tenant_name
= "Tenant_123"
1317 non_exist_tenant_id
= "kjhgrt456-45345kjhdfgnbdk-34dsfjdfg"
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 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': non_exist_tenant_name
,
1324 'id': non_exist_tenant_id
})
1326 self
.assertEqual(filter_tenant_list
, [])
1328 class test_vimconn_new_tenant(test_base
):
1331 def test_000_new_tenant(self
):
1332 tenant_name
= _get_random_string(20)
1333 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1334 self
.__class
__.test_index
,
1335 inspect
.currentframe().f_code
.co_name
)
1336 self
.__class
__.test_index
+= 1
1338 self
.__class
__.tenant_id
= test_config
["vim_conn"].new_tenant(tenant_name
)
1341 self
.assertEqual(type(self
.__class
__.tenant_id
), str)
1343 def test_010_new_tenant_negative(self
):
1344 Invalid_tenant_name
= 10121
1345 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1346 self
.__class
__.test_index
,
1347 inspect
.currentframe().f_code
.co_name
)
1348 self
.__class
__.test_index
+= 1
1350 with self
.assertRaises(Exception) as context
:
1351 test_config
["vim_conn"].new_tenant(Invalid_tenant_name
)
1353 self
.assertEqual((context
.exception
).http_code
, 400)
1355 def test_020_delete_tenant(self
):
1356 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1357 self
.__class
__.test_index
,
1358 inspect
.currentframe().f_code
.co_name
)
1359 self
.__class
__.test_index
+= 1
1361 tenant_id
= test_config
["vim_conn"].delete_tenant(self
.__class
__.tenant_id
)
1362 self
.assertEqual(type(tenant_id
), str)
1364 def test_030_delete_tenant_negative(self
):
1365 Non_exist_tenant_name
= 'Test_30_tenant'
1366 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1367 self
.__class
__.test_index
,
1368 inspect
.currentframe().f_code
.co_name
)
1369 self
.__class
__.test_index
+= 1
1371 with self
.assertRaises(Exception) as context
:
1372 test_config
["vim_conn"].delete_tenant(Non_exist_tenant_name
)
1374 self
.assertEqual((context
.exception
).http_code
, 404)
1379 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
1380 scenario based tests.
1382 class descriptor_based_scenario_test(test_base
):
1384 scenario_test_path
= None
1387 def setUpClass(cls
):
1389 cls
.to_delete_list
= []
1390 cls
.scenario_uuids
= []
1391 cls
.instance_scenario_uuids
= []
1392 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
1393 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
1396 def tearDownClass(cls
):
1397 test_config
["test_number"] += 1
1399 def test_000_load_scenario(self
):
1400 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1401 inspect
.currentframe().f_code
.co_name
,
1402 test_config
["test_folder"])
1403 self
.__class
__.test_index
+= 1
1405 descriptor_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/*.yaml')
1406 vnf_descriptors
= []
1407 scenario_descriptors
= []
1408 for descriptor_file
in descriptor_files
:
1409 with
open(descriptor_file
, 'r') as stream
:
1410 descriptor
= yaml
.load(stream
)
1411 if "vnf" in descriptor
or "vnfd:vnfd-catalog" in descriptor
or "vnfd-catalog" in descriptor
:
1412 vnf_descriptors
.append(descriptor
)
1414 scenario_descriptors
.append(descriptor
)
1416 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
1417 if not vnf_descriptors
or not scenario_descriptors
or len(scenario_descriptors
) > 1:
1418 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
1419 test_config
["test_folder"]))
1422 for vnf_descriptor
in vnf_descriptors
:
1423 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
1424 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
, image_name
=test_config
["image_name"])
1427 vnf_uuid
= vnf
['vnf']['uuid']
1429 vnf_uuid
= vnf
['vnfd'][0]['uuid']
1430 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
1431 "params": {"uuid": vnf_uuid
}})
1433 # load the scenario definition
1434 for scenario_descriptor
in scenario_descriptors
:
1435 # networks = scenario_descriptor['scenario']['networks']
1436 # networks[test_config["mgmt_net"]] = networks.pop('mgmt')
1437 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
1438 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
1439 logger
.debug(scenario
)
1440 if 'scenario' in scenario
:
1441 scenario_uuid
= scenario
['scenario']['uuid']
1443 scenario_uuid
= scenario
['nsd'][0]['uuid']
1444 self
.__class
__.to_delete_list
.insert(0, {"item": "scenario",
1445 "function": test_config
["client"].delete_scenario
,
1446 "params": {"uuid": scenario_uuid
}})
1447 self
.__class
__.scenario_uuids
.append(scenario_uuid
)
1449 def test_010_instantiate_scenario(self
):
1450 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1451 inspect
.currentframe().f_code
.co_name
,
1452 test_config
["test_folder"])
1453 self
.__class
__.test_index
+= 1
1454 for scenario_uuid
in self
.__class
__.scenario_uuids
:
1455 instance_descriptor
= {
1457 "name": self
.__class
__.test_text
,
1458 "scenario": scenario_uuid
,
1460 "mgmt": {"sites": [ { "netmap-use": test_config
["mgmt_net"]} ]}
1464 instance
= test_config
["client"].create_instance(instance_descriptor
)
1465 self
.__class
__.instance_scenario_uuids
.append(instance
['uuid'])
1466 logger
.debug(instance
)
1467 self
.__class
__.to_delete_list
.insert(0, {"item": "instance",
1468 "function": test_config
["client"].delete_instance
,
1469 "params": {"uuid": instance
['uuid']}})
1471 def test_020_check_deployent(self
):
1472 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1473 inspect
.currentframe().f_code
.co_name
,
1474 test_config
["test_folder"])
1475 self
.__class
__.test_index
+= 1
1477 if test_config
["manual"]:
1478 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1481 keep_waiting
= test_config
["timeout"]
1482 pending_instance_scenario_uuids
= list(self
.__class
__.instance_scenario_uuids
) # make a copy
1483 while pending_instance_scenario_uuids
:
1485 while index
< len(pending_instance_scenario_uuids
):
1486 result
= check_instance_scenario_active(pending_instance_scenario_uuids
[index
])
1488 del pending_instance_scenario_uuids
[index
]
1490 elif 'ERROR' in result
[1]:
1491 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
1493 raise Exception(msg
)
1496 if keep_waiting
>= 5:
1499 elif keep_waiting
> 0:
1500 time
.sleep(keep_waiting
)
1503 msg
= 'Timeout reached while waiting instance scenario to get active'
1505 raise Exception(msg
)
1507 def test_030_clean_deployment(self
):
1508 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1509 inspect
.currentframe().f_code
.co_name
,
1510 test_config
["test_folder"])
1511 self
.__class
__.test_index
+= 1
1512 #At the moment if you delete an scenario right after creating it, in openstack datacenters
1513 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1515 for item
in self
.__class
__.to_delete_list
:
1516 response
= item
["function"](**item
["params"])
1517 logger
.debug(response
)
1520 def _get_random_string(maxLength
):
1521 '''generates a string with random characters string.letters and string.digits
1522 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1526 minLength
= min_string
- len(prefix
)
1527 if maxLength
< min_string
: maxLength
= min_string
1528 maxLength
-= len(prefix
)
1529 length
= random
.randint(minLength
,maxLength
)
1530 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
1533 def test_vimconnector(args
):
1535 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1536 test_config
['vimtype'] = args
.vimtype
1537 if args
.vimtype
== "vmware":
1538 import vimconn_vmware
as vim
1540 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1542 tenant_name
= args
.tenant_name
1543 test_config
['tenant'] = tenant_name
1544 config_params
= yaml
.load(args
.config_param
)
1545 org_name
= config_params
.get('orgname')
1546 org_user
= config_params
.get('user')
1547 org_passwd
= config_params
.get('passwd')
1548 vim_url
= args
.endpoint_url
1549 test_config
['image_path'] = args
.image_path
1550 test_config
['image_name'] = args
.image_name
1552 # vmware connector obj
1553 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
)
1555 elif args
.vimtype
== "aws":
1556 import vimconn_aws
as vim
1557 elif args
.vimtype
== "openstack":
1558 import vimconn_openstack
as vim
1560 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1562 tenant_name
= args
.tenant_name
1563 test_config
['tenant'] = tenant_name
1564 config_params
= yaml
.load(args
.config_param
)
1565 os_user
= config_params
.get('user')
1566 os_passwd
= config_params
.get('passwd')
1567 vim_url
= args
.endpoint_url
1568 test_config
['image_path'] = args
.image_path
1569 test_config
['image_name'] = args
.image_name
1571 # openstack connector obj
1572 vim_persistent_info
= {}
1573 test_config
['vim_conn'] = vim
.vimconnector(
1574 uuid
="test-uuid-1", name
="VIO-openstack",
1575 tenant_id
=None, tenant_name
=tenant_name
,
1576 url
=vim_url
, url_admin
=None,
1577 user
=os_user
, passwd
=os_passwd
,
1578 config
=config_params
, persistent_info
=vim_persistent_info
1580 test_config
['vim_conn'].debug
= "true"
1582 elif args
.vimtype
== "openvim":
1583 import vimconn_openvim
as vim
1585 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
1589 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1590 # If only want to obtain a tests list print it and exit
1593 for cls
in clsmembers
:
1594 if cls
[0].startswith('test_vimconn'):
1595 tests_names
.append(cls
[0])
1597 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
1602 # Create the list of tests to be run
1603 code_based_tests
= []
1605 for test
in args
.tests
:
1606 for t
in test
.split(','):
1607 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1608 if len(matches_code_based_tests
) > 0:
1609 code_based_tests
.append(matches_code_based_tests
[0][1])
1611 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1613 if not code_based_tests
:
1615 for cls
in clsmembers
:
1616 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1617 if cls
[0].startswith('test_vimconn'):
1618 code_based_tests
.append(cls
[1])
1620 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1622 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1623 # This is handled in the tests using logging.
1624 stream
= open('/dev/null', 'w')
1626 # Run code based tests
1627 basic_tests_suite
= unittest
.TestSuite()
1628 for test
in code_based_tests
:
1629 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1630 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1631 executed
+= result
.testsRun
1632 failed
+= len(result
.failures
) + len(result
.errors
)
1633 if failfast
and failed
:
1635 if len(result
.failures
) > 0:
1636 logger
.debug("failures : {}".format(result
.failures
))
1637 if len(result
.errors
) > 0:
1638 logger
.debug("errors : {}".format(result
.errors
))
1639 return executed
, failed
1644 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1645 import openmanoclient
1648 test_config
["client"] = openmanoclient
.openmanoclient(
1649 endpoint_url
=args
.endpoint_url
,
1650 tenant_name
=args
.tenant_name
,
1651 datacenter_name
=args
.datacenter
,
1652 debug
=args
.debug
, logger
=test_config
["logger_name"])
1653 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1654 # If only want to obtain a tests list print it and exit
1657 for cls
in clsmembers
:
1658 if cls
[0].startswith('test_VIM'):
1659 tests_names
.append(cls
[0])
1661 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
1662 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1663 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1668 # Create the list of tests to be run
1669 code_based_tests
= []
1671 for test
in args
.tests
:
1672 for t
in test
.split(','):
1673 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1674 if len(matches_code_based_tests
) > 0:
1675 code_based_tests
.append(matches_code_based_tests
[0][1])
1677 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1679 if not code_based_tests
:
1681 for cls
in clsmembers
:
1682 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1683 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
1684 code_based_tests
.append(cls
[1])
1686 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1688 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1689 # This is handled in the tests using logging.
1690 stream
= open('/dev/null', 'w')
1692 # Run code based tests
1693 basic_tests_suite
= unittest
.TestSuite()
1694 for test
in code_based_tests
:
1695 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1696 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1697 executed
+= result
.testsRun
1698 failed
+= len(result
.failures
) + len(result
.errors
)
1699 if failfast
and failed
:
1701 if len(result
.failures
) > 0:
1702 logger
.debug("failures : {}".format(result
.failures
))
1703 if len(result
.errors
) > 0:
1704 logger
.debug("errors : {}".format(result
.errors
))
1705 return executed
, failed
1708 def test_deploy(args
):
1710 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1711 import openmanoclient
1714 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1715 test_config
["image_name"] = args
.image_name
1716 test_config
["mgmt_net"] = args
.mgmt_net
1717 test_config
["manual"] = args
.manual
1718 test_directory_content
= os
.listdir(test_config
["test_directory"])
1719 # If only want to obtain a tests list print it and exit
1721 msg
= "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
1726 descriptor_based_tests
= []
1727 # Create the list of tests to be run
1728 code_based_tests
= []
1730 for test
in args
.tests
:
1731 for t
in test
.split(','):
1732 if t
in test_directory_content
:
1733 descriptor_based_tests
.append(t
)
1735 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1737 if not descriptor_based_tests
:
1739 descriptor_based_tests
= test_directory_content
1741 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1743 # import openmanoclient from relative path
1744 test_config
["client"] = openmanoclient
.openmanoclient(
1745 endpoint_url
=args
.endpoint_url
,
1746 tenant_name
=args
.tenant_name
,
1747 datacenter_name
=args
.datacenter
,
1748 debug
=args
.debug
, logger
=test_config
["logger_name"])
1750 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1751 # This is handled in the tests using logging.
1752 stream
= open('/dev/null', 'w')
1753 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1754 for test
in descriptor_based_tests
:
1755 test_config
["test_folder"] = test
1756 test_suite
= unittest
.TestSuite()
1757 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
1758 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
1759 executed
+= result
.testsRun
1760 failed
+= len(result
.failures
) + len(result
.errors
)
1761 if failfast
and failed
:
1763 if len(result
.failures
) > 0:
1764 logger
.debug("failures : {}".format(result
.failures
))
1765 if len(result
.errors
) > 0:
1766 logger
.debug("errors : {}".format(result
.errors
))
1768 return executed
, failed
1770 if __name__
=="__main__":
1772 parser
= ArgumentParser(description
='Test RO module')
1773 parser
.add_argument('-v','--version', action
='version', help="Show current version",
1774 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
1777 parent_parser
= ArgumentParser(add_help
=False)
1778 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1779 dest
='failfast', action
="store_true", default
=False)
1780 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1781 dest
='failed', action
="store_true", default
=False)
1782 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
1783 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
1785 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
1786 help='Set the logger file. By default '+default_logger_file
)
1787 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
1788 help="Set the openmano tenant to use for the test. By default 'osm'")
1789 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
1790 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1791 dest
='timeout', type=int, default
=300)
1792 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
1794 subparsers
= parser
.add_subparsers(help='test sets')
1796 # Deployment test set
1797 # -------------------
1798 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
1799 help="test deployment using descriptors at RO_test folder ")
1800 deploy_parser
.set_defaults(func
=test_deploy
)
1802 # Mandatory arguments
1803 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
1804 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1805 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
1806 help='Image name available at datacenter used for the tests')
1807 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
1808 help='Set the vim management network to use for tests')
1810 # Optional arguments
1811 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
1812 help='Pause execution once deployed to allow manual checking of the '
1813 'deployed instance scenario')
1814 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1815 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1818 # -------------------
1819 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
1820 vimconn_parser
.set_defaults(func
=test_vimconnector
)
1821 # Mandatory arguments
1822 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1823 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
1824 help='Set the vimconnector type to test')
1825 mandatory_arguments
.add_argument('-c', '--config', dest
='config_param', required
=True,
1826 help='Set the vimconnector specific config parameters in dictionary format')
1827 mandatory_arguments
.add_argument('-u', '--url', dest
='endpoint_url',required
=True, help="Set the vim connector url or Host IP")
1828 # Optional arguments
1829 vimconn_parser
.add_argument('-i', '--image-path', dest
='image_path', help="Provide image path present at RO container")
1830 vimconn_parser
.add_argument('-n', '--image-name', dest
='image_name', help="Provide image name for test")
1831 # TODO add optional arguments for vimconn tests
1832 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1834 # Datacenter test set
1835 # -------------------
1836 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
1837 vimconn_parser
.set_defaults(func
=test_vim
)
1839 # Mandatory arguments
1840 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1841 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1843 # Optional arguments
1844 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1845 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1847 argcomplete
.autocomplete(parser
)
1848 args
= parser
.parse_args()
1852 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1853 logger_level
= 'INFO'
1855 logger_level
= 'DEBUG'
1857 logger_level
= 'WARNING'
1858 logger_name
= os
.path
.basename(__file__
)
1859 test_config
["logger_name"] = logger_name
1860 logger
= logging
.getLogger(logger_name
)
1861 logger
.setLevel(logger_level
)
1862 failfast
= args
.failfast
1864 # Configure a logging handler to store in a logging file
1865 if args
.logger_file
:
1866 fileHandler
= logging
.FileHandler(args
.logger_file
)
1867 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1868 fileHandler
.setFormatter(formatter_fileHandler
)
1869 logger
.addHandler(fileHandler
)
1871 # Configure a handler to print to stdout
1872 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
1873 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1874 consoleHandler
.setFormatter(formatter_consoleHandler
)
1875 logger
.addHandler(consoleHandler
)
1877 logger
.debug('Program started with the following arguments: ' + str(args
))
1879 # set test config parameters
1880 test_config
["timeout"] = args
.timeout
1881 test_config
["test_number"] = 1
1883 executed
, failed
= args
.func(args
)
1886 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
1887 sys
.exit(1 if failed
else 0)