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
26 __author__
="Pablo Montes, Alfonso Tierno"
27 __date__
="$16-Feb-2017 17:08:16$"
29 version_date
="May 2017"
33 from argparse
import ArgumentParser
44 from pyvcloud
.vcloudair
import VCA
47 global test_config
# used for global variables with the test configuration
51 def check_instance_scenario_active(uuid
):
52 instance
= test_config
["client"].get_instance(uuid
=uuid
)
54 for net
in instance
['nets']:
55 status
= net
['status']
56 if status
!= 'ACTIVE':
57 return (False, status
)
59 for vnf
in instance
['vnfs']:
62 if status
!= 'ACTIVE':
63 return (False, status
)
70 All unittest classes for code based tests must have prefix 'test_' in order to be taken into account for tests
72 class test_VIM_datacenter_tenant_operations(unittest
.TestCase
):
79 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
82 def tearDownClass(cls
):
83 test_config
["test_number"] += 1
86 exec_info
= sys
.exc_info()
87 if exec_info
== (None, None, None):
88 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
90 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
91 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
93 for line
in error_trace
:
95 logger
.critical("{}".format(msg
))
97 def test_000_create_RO_tenant(self
):
98 self
.__class
__.tenant_name
= _get_random_string(20)
99 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
100 inspect
.currentframe().f_code
.co_name
)
101 self
.__class
__.test_index
+= 1
102 tenant
= test_config
["client"].create_tenant(name
=self
.__class
__.tenant_name
,
103 description
=self
.__class
__.tenant_name
)
104 logger
.debug("{}".format(tenant
))
105 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.tenant_name
)
107 def test_010_list_RO_tenant(self
):
108 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
109 inspect
.currentframe().f_code
.co_name
)
110 self
.__class
__.test_index
+= 1
111 tenant
= test_config
["client"].get_tenant(name
=self
.__class
__.tenant_name
)
112 logger
.debug("{}".format(tenant
))
113 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.tenant_name
)
115 def test_020_delete_RO_tenant(self
):
116 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
117 inspect
.currentframe().f_code
.co_name
)
118 self
.__class
__.test_index
+= 1
119 tenant
= test_config
["client"].delete_tenant(name
=self
.__class
__.tenant_name
)
120 logger
.debug("{}".format(tenant
))
121 assert('deleted' in tenant
.get('result',""))
124 class test_VIM_datacenter_operations(unittest
.TestCase
):
126 datacenter_name
= None
131 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
134 def tearDownClass(cls
):
135 test_config
["test_number"] += 1
138 exec_info
= sys
.exc_info()
139 if exec_info
== (None, None, None):
140 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
142 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
143 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
145 for line
in error_trace
:
147 logger
.critical("{}".format(msg
))
149 def test_000_create_datacenter(self
):
150 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
151 inspect
.currentframe().f_code
.co_name
)
152 self
.__class
__.datacenter_name
= _get_random_string(20)
153 self
.__class
__.test_index
+= 1
154 self
.datacenter
= test_config
["client"].create_datacenter(name
=self
.__class
__.datacenter_name
,
155 vim_url
="http://fakeurl/fake")
156 logger
.debug("{}".format(self
.datacenter
))
157 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name',''), self
.__class
__.datacenter_name
)
159 def test_010_list_datacenter(self
):
160 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
161 inspect
.currentframe().f_code
.co_name
)
163 self
.__class
__.test_index
+= 1
164 self
.datacenter
= test_config
["client"].get_datacenter(all_tenants
=True, name
=self
.__class
__.datacenter_name
)
165 logger
.debug("{}".format(self
.datacenter
))
166 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name', ''), self
.__class
__.datacenter_name
)
168 def test_020_attach_datacenter(self
):
169 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
170 inspect
.currentframe().f_code
.co_name
)
172 self
.__class
__.test_index
+= 1
173 self
.datacenter
= test_config
["client"].attach_datacenter(name
=self
.__class
__.datacenter_name
,
174 vim_tenant_name
='fake')
175 logger
.debug("{}".format(self
.datacenter
))
176 assert ('vim_tenants' in self
.datacenter
.get('datacenter', {}))
178 def test_030_list_attached_datacenter(self
):
179 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
180 inspect
.currentframe().f_code
.co_name
)
182 self
.__class
__.test_index
+= 1
183 self
.datacenter
= test_config
["client"].get_datacenter(all_tenants
=False, name
=self
.__class
__.datacenter_name
)
184 logger
.debug("{}".format(self
.datacenter
))
185 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name', ''), self
.__class
__.datacenter_name
)
187 def test_040_detach_datacenter(self
):
188 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
189 inspect
.currentframe().f_code
.co_name
)
191 self
.__class
__.test_index
+= 1
192 self
.datacenter
= test_config
["client"].detach_datacenter(name
=self
.__class
__.datacenter_name
)
193 logger
.debug("{}".format(self
.datacenter
))
194 assert ('detached' in self
.datacenter
.get('result', ""))
196 def test_050_delete_datacenter(self
):
197 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
198 inspect
.currentframe().f_code
.co_name
)
200 self
.__class
__.test_index
+= 1
201 self
.datacenter
= test_config
["client"].delete_datacenter(name
=self
.__class
__.datacenter_name
)
202 logger
.debug("{}".format(self
.datacenter
))
203 assert('deleted' in self
.datacenter
.get('result',""))
206 class test_VIM_network_operations(unittest
.TestCase
):
208 vim_network_name
= None
210 vim_network_uuid
= None
214 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
217 def tearDownClass(cls
):
218 test_config
["test_number"] += 1
221 exec_info
= sys
.exc_info()
222 if exec_info
== (None, None, None):
223 logger
.info(self
.__class
__.test_text
+ " -> TEST OK")
225 logger
.warning(self
.__class
__.test_text
+ " -> TEST NOK")
226 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
228 for line
in error_trace
:
230 logger
.critical("{}".format(msg
))
232 def test_000_create_VIM_network(self
):
233 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
234 inspect
.currentframe().f_code
.co_name
)
235 self
.__class
__.vim_network_name
= _get_random_string(20)
236 self
.__class
__.test_index
+= 1
237 network
= test_config
["client"].vim_action("create", "networks", name
=self
.__class
__.vim_network_name
)
238 logger
.debug("{}".format(network
))
239 self
.__class
__.vim_network_uuid
= network
["network"]["id"]
240 self
.assertEqual(network
.get('network', {}).get('name', ''), self
.__class
__.vim_network_name
)
242 def test_010_list_VIM_networks(self
):
243 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
244 inspect
.currentframe().f_code
.co_name
)
245 self
.__class
__.test_index
+= 1
246 networks
= test_config
["client"].vim_action("list", "networks")
247 logger
.debug("{}".format(networks
))
249 def test_020_get_VIM_network_by_uuid(self
):
250 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
251 inspect
.currentframe().f_code
.co_name
)
253 self
.__class
__.test_index
+= 1
254 network
= test_config
["client"].vim_action("show", "networks", uuid
=self
.__class
__.vim_network_uuid
)
255 logger
.debug("{}".format(network
))
256 self
.assertEqual(network
.get('network', {}).get('name', ''), self
.__class
__.vim_network_name
)
258 def test_030_delete_VIM_network_by_uuid(self
):
259 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
260 inspect
.currentframe().f_code
.co_name
)
262 self
.__class
__.test_index
+= 1
263 network
= test_config
["client"].vim_action("delete", "networks", uuid
=self
.__class
__.vim_network_uuid
)
264 logger
.debug("{}".format(network
))
265 assert ('deleted' in network
.get('result', ""))
268 class test_VIM_image_operations(unittest
.TestCase
):
274 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
277 def tearDownClass(cls
):
278 test_config
["test_number"] += 1
281 exec_info
= sys
.exc_info()
282 if exec_info
== (None, None, None):
283 logger
.info(self
.__class
__.test_text
+ " -> TEST OK")
285 logger
.warning(self
.__class
__.test_text
+ " -> TEST NOK")
286 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
288 for line
in error_trace
:
290 logger
.critical("{}".format(msg
))
292 def test_000_list_VIM_images(self
):
293 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
294 inspect
.currentframe().f_code
.co_name
)
295 self
.__class
__.test_index
+= 1
296 images
= test_config
["client"].vim_action("list", "images")
297 logger
.debug("{}".format(images
))
300 The following is a non critical test that will fail most of the times.
301 In case of OpenStack datacenter these tests will only success if RO has access to the admin endpoint
302 This test will only be executed in case it is specifically requested by the user
304 class test_VIM_tenant_operations(unittest
.TestCase
):
306 vim_tenant_name
= None
308 vim_tenant_uuid
= None
312 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
313 logger
.warning("In case of OpenStack datacenter these tests will only success "
314 "if RO has access to the admin endpoint")
317 def tearDownClass(cls
):
318 test_config
["test_number"] += 1
321 exec_info
= sys
.exc_info()
322 if exec_info
== (None, None, None):
323 logger
.info(self
.__class
__.test_text
+ " -> TEST OK")
325 logger
.warning(self
.__class
__.test_text
+ " -> TEST NOK")
326 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
328 for line
in error_trace
:
330 logger
.critical("{}".format(msg
))
332 def test_000_create_VIM_tenant(self
):
333 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
334 inspect
.currentframe().f_code
.co_name
)
335 self
.__class
__.vim_tenant_name
= _get_random_string(20)
336 self
.__class
__.test_index
+= 1
337 tenant
= test_config
["client"].vim_action("create", "tenants", name
=self
.__class
__.vim_tenant_name
)
338 logger
.debug("{}".format(tenant
))
339 self
.__class
__.vim_tenant_uuid
= tenant
["tenant"]["id"]
340 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.vim_tenant_name
)
342 def test_010_list_VIM_tenants(self
):
343 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
344 inspect
.currentframe().f_code
.co_name
)
345 self
.__class
__.test_index
+= 1
346 tenants
= test_config
["client"].vim_action("list", "tenants")
347 logger
.debug("{}".format(tenants
))
349 def test_020_get_VIM_tenant_by_uuid(self
):
350 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
351 inspect
.currentframe().f_code
.co_name
)
353 self
.__class
__.test_index
+= 1
354 tenant
= test_config
["client"].vim_action("show", "tenants", uuid
=self
.__class
__.vim_tenant_uuid
)
355 logger
.debug("{}".format(tenant
))
356 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.vim_tenant_name
)
358 def test_030_delete_VIM_tenant_by_uuid(self
):
359 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
360 inspect
.currentframe().f_code
.co_name
)
362 self
.__class
__.test_index
+= 1
363 tenant
= test_config
["client"].vim_action("delete", "tenants", uuid
=self
.__class
__.vim_tenant_uuid
)
364 logger
.debug("{}".format(tenant
))
365 assert ('deleted' in tenant
.get('result', ""))
367 class test_vimconn_connect(unittest
.TestCase
):
373 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
376 def tearDownClass(cls
):
377 test_config
["test_number"] += 1
380 exec_info
= sys
.exc_info()
381 if exec_info
== (None, None, None):
382 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
384 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
385 logger
.critical("Traceback error",exc_info
=True)
387 def test_000_connect(self
):
388 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
389 self
.__class
__.test_index
,
390 inspect
.currentframe().f_code
.co_name
)
392 self
.__class
__.test_index
+= 1
393 if test_config
['vimtype'] == 'vmware':
394 vca_object
= test_config
["vim_conn"].connect()
395 logger
.debug("{}".format(vca_object
))
396 self
.assertIsInstance(vca_object
, VCA
)
399 class test_vimconn_new_network(unittest
.TestCase
):
406 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
409 def tearDownClass(cls
):
410 test_config
["test_number"] += 1
413 exec_info
= sys
.exc_info()
414 if exec_info
== (None, None, None):
415 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
417 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
418 logger
.critical("Traceback error",exc_info
=True)
420 def test_000_new_network(self
):
421 self
.__class
__.network_name
= _get_random_string(20)
422 network_type
= 'bridge'
424 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
425 self
.__class
__.test_index
, inspect
.currentframe().f_code
.co_name
)
426 self
.__class
__.test_index
+= 1
428 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
429 net_type
=network_type
)
430 self
.__class
__.network_id
= network
431 logger
.debug("{}".format(network
))
433 network_list
= test_config
["vim_conn"].get_vcd_network_list()
434 for net
in network_list
:
435 if self
.__class
__.network_name
in net
.get('name'):
436 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
437 self
.assertEqual(net
.get('type'), network_type
)
439 # Deleting created network
440 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
442 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
444 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
446 def test_010_new_network_by_types(self
):
448 network_types
= ['data','bridge','mgmt']
449 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
450 self
.__class
__.test_index
,
451 inspect
.currentframe().f_code
.co_name
)
452 self
.__class
__.test_index
+= 1
453 for net_type
in network_types
:
454 self
.__class
__.network_name
= _get_random_string(20)
455 network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
458 delete_net_ids
.append(network_id
)
459 logger
.debug("{}".format(network_id
))
461 network_list
= test_config
["vim_conn"].get_vcd_network_list()
462 for net
in network_list
:
463 if self
.__class
__.network_name
in net
.get('name'):
464 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
465 if net_type
in net
.get('type'):
466 self
.assertEqual(net
.get('type'), net_type
)
468 self
.assertNotEqual(net
.get('type'), net_type
)
470 # Deleting created network
471 for net_id
in delete_net_ids
:
472 result
= test_config
["vim_conn"].delete_network(net_id
)
474 logger
.info("Network id {} sucessfully deleted".format(net_id
))
476 logger
.info("Failed to delete network id {}".format(net_id
))
478 def test_020_new_network_by_ipprofile(self
):
479 test_directory_content
= os
.listdir(test_config
["test_directory"])
481 for dir_name
in test_directory_content
:
482 if dir_name
== 'simple_multi_vnfc':
483 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
484 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
487 for vnfd
in vnfd_files
:
488 with
open(vnfd
, 'r') as stream
:
489 vnf_descriptor
= yaml
.load(stream
)
491 internal_connections_list
= vnf_descriptor
['vnf']['internal-connections']
492 for item
in internal_connections_list
:
493 if 'ip-profile' in item
:
494 version
= item
['ip-profile']['ip-version']
495 dhcp_count
= item
['ip-profile']['dhcp']['count']
496 dhcp_enabled
= item
['ip-profile']['dhcp']['enabled']
498 self
.__class
__.network_name
= _get_random_string(20)
499 ip_profile
= {'dhcp_count': dhcp_count
,
500 'dhcp_enabled': dhcp_enabled
,
501 'ip_version': version
503 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
504 self
.__class
__.test_index
,
505 inspect
.currentframe().f_code
.co_name
)
506 self
.__class
__.test_index
+= 1
507 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
509 ip_profile
=ip_profile
)
510 self
.__class
__.network_id
= network
511 logger
.debug("{}".format(network
))
513 network_list
= test_config
["vim_conn"].get_vcd_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'))
518 # Deleting created network
519 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
521 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
523 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
525 def test_030_new_network_by_isshared(self
):
526 self
.__class
__.network_name
= _get_random_string(20)
528 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
529 self
.__class
__.test_index
,
530 inspect
.currentframe().f_code
.co_name
)
531 self
.__class
__.test_index
+= 1
532 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
535 self
.__class
__.network_id
= network
536 logger
.debug("{}".format(network
))
538 network_list
= test_config
["vim_conn"].get_vcd_network_list()
539 for net
in network_list
:
540 if self
.__class
__.network_name
in net
.get('name'):
541 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
542 self
.assertEqual(net
.get('shared'), shared
)
544 # Deleting created network
545 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
547 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
549 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
551 def test_040_new_network_by_negative(self
):
552 self
.__class
__.network_name
= _get_random_string(20)
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
557 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
558 net_type
='unknowntype')
559 self
.__class
__.network_id
= network
560 logger
.debug("{}".format(network
))
561 network_list
= test_config
["vim_conn"].get_vcd_network_list()
562 for net
in network_list
:
563 if self
.__class
__.network_name
in net
.get('name'):
564 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
566 # Deleting created network
567 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
569 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
571 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
573 class test_vimconn_get_network_list(unittest
.TestCase
):
580 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
583 def tearDownClass(cls
):
584 test_config
["test_number"] += 1
587 # creating new network
588 self
.__class
__.network_name
= _get_random_string(20)
589 self
.__class
__.net_type
= 'bridge'
590 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
591 net_type
=self
.__class
__.net_type
)
592 self
.__class
__.network_id
= network
593 logger
.debug("{}".format(network
))
596 exec_info
= sys
.exc_info()
597 if exec_info
== (None, None, None):
598 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
600 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
601 logger
.critical("Traceback error",exc_info
=True)
603 # Deleting created network
604 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
606 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
608 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
610 def test_000_get_network_list(self
):
611 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
612 self
.__class
__.test_index
,
613 inspect
.currentframe().f_code
.co_name
)
614 self
.__class
__.test_index
+= 1
616 network_list
= test_config
["vim_conn"].get_network_list()
617 for net
in network_list
:
618 if self
.__class
__.network_name
in net
.get('name'):
619 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
620 self
.assertEqual(net
.get('type'), self
.__class
__.net_type
)
621 self
.assertEqual(net
.get('status'), 'ACTIVE')
622 self
.assertEqual(net
.get('shared'), False)
624 def test_010_get_network_list_by_name(self
):
625 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
626 self
.__class
__.test_index
,
627 inspect
.currentframe().f_code
.co_name
)
628 self
.__class
__.test_index
+= 1
630 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
632 # find network from list by it's name
633 new_network_list
= test_config
["vim_conn"].get_network_list({'name': network_name
})
634 for list_item
in new_network_list
:
635 if self
.__class
__.network_name
in list_item
.get('name'):
636 self
.assertEqual(network_name
, list_item
.get('name'))
637 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
638 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
640 def test_020_get_network_list_by_id(self
):
641 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
642 self
.__class
__.test_index
,
643 inspect
.currentframe().f_code
.co_name
)
644 self
.__class
__.test_index
+= 1
646 # find network from list by it's id
647 new_network_list
= test_config
["vim_conn"].get_network_list({'id':self
.__class
__.network_id
})
648 for list_item
in new_network_list
:
649 if self
.__class
__.network_id
in list_item
.get('id'):
650 self
.assertEqual(self
.__class
__.network_id
, list_item
.get('id'))
651 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
652 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
654 def test_030_get_network_list_by_shared(self
):
656 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
657 self
.__class
__.test_index
,
658 inspect
.currentframe().f_code
.co_name
)
659 self
.__class
__.test_index
+= 1
661 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
662 # find network from list by it's shared value
663 new_network_list
= test_config
["vim_conn"].get_network_list({'shared':Shared
,
664 'name':network_name
})
665 for list_item
in new_network_list
:
666 if list_item
.get('shared') == Shared
:
667 self
.assertEqual(list_item
.get('shared'), Shared
)
668 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
669 self
.assertEqual(network_name
, list_item
.get('name'))
671 def test_040_get_network_list_by_tenant_id(self
):
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 tenant_list
= test_config
["vim_conn"].get_tenant_list()
678 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
680 for tenant_item
in tenant_list
:
681 if test_config
['tenant'] == tenant_item
.get('name'):
682 # find network from list by it's tenant id
683 tenant_id
= tenant_item
.get('id')
684 new_network_list
= test_config
["vim_conn"].get_network_list({'tenant_id':tenant_id
,
685 'name':network_name
})
686 for list_item
in new_network_list
:
687 self
.assertEqual(tenant_id
, list_item
.get('tenant_id'))
688 self
.assertEqual(network_name
, list_item
.get('name'))
689 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
690 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
692 def test_050_get_network_list_by_status(self
):
693 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
694 self
.__class
__.test_index
,
695 inspect
.currentframe().f_code
.co_name
)
696 self
.__class
__.test_index
+= 1
699 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
701 # find network from list by it's status
702 new_network_list
= test_config
["vim_conn"].get_network_list({'status':status
,
703 'name': network_name
})
704 for list_item
in new_network_list
:
705 self
.assertIn(self
.__class
__.network_name
, list_item
.get('name'))
706 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
707 self
.assertEqual(list_item
.get('status'), status
)
709 def test_060_get_network_list_by_negative(self
):
710 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
711 self
.__class
__.test_index
,
712 inspect
.currentframe().f_code
.co_name
)
713 self
.__class
__.test_index
+= 1
715 network_list
= test_config
["vim_conn"].get_network_list({'name': 'unknown_name'})
716 self
.assertEqual(network_list
, [])
718 class test_vimconn_get_network(unittest
.TestCase
):
725 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
728 def tearDownClass(cls
):
729 test_config
["test_number"] += 1
732 # creating new network
733 self
.__class
__.network_name
= _get_random_string(20)
734 self
.__class
__.net_type
= 'bridge'
735 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
736 net_type
=self
.__class
__.net_type
)
737 self
.__class
__.network_id
= network
738 logger
.debug("{}".format(network
))
741 exec_info
= sys
.exc_info()
742 if exec_info
== (None, None, None):
743 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
745 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
746 logger
.critical("Traceback error",exc_info
=True)
748 # Deleting created network
749 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
751 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
753 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
755 def test_000_get_network(self
):
756 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
757 self
.__class
__.test_index
,
758 inspect
.currentframe().f_code
.co_name
)
759 self
.__class
__.test_index
+= 1
761 network_info
= test_config
["vim_conn"].get_network(self
.__class
__.network_id
)
762 self
.assertEqual(network_info
.get('status'), 'ACTIVE')
763 self
.assertIn(self
.__class
__.network_name
, network_info
.get('name'))
764 self
.assertEqual(network_info
.get('type'), self
.__class
__.net_type
)
765 self
.assertEqual(network_info
.get('id'), self
.__class
__.network_id
)
767 def test_010_get_network_negative(self
):
768 Non_exist_id
= str(uuid
.uuid4())
769 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
770 self
.__class
__.test_index
,
771 inspect
.currentframe().f_code
.co_name
)
772 self
.__class
__.test_index
+= 1
774 network_info
= test_config
["vim_conn"].get_network(Non_exist_id
)
775 self
.assertEqual(network_info
, {})
777 class test_vimconn_delete_network(unittest
.TestCase
):
784 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
787 def tearDownClass(cls
):
788 test_config
["test_number"] += 1
791 exec_info
= sys
.exc_info()
792 if exec_info
== (None, None, None):
793 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
795 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
796 logger
.critical("Traceback error",exc_info
=True)
798 def test_000_delete_network(self
):
800 self
.__class
__.network_name
= _get_random_string(20)
801 self
.__class
__.net_type
= 'bridge'
802 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
803 net_type
=self
.__class
__.net_type
)
804 self
.__class
__.network_id
= network
805 logger
.debug("{}".format(network
))
807 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
808 self
.__class
__.test_index
,
809 inspect
.currentframe().f_code
.co_name
)
810 self
.__class
__.test_index
+= 1
812 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
814 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
816 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
818 # after deleting network we check in network list
819 network_list
= test_config
["vim_conn"].get_network_list({ 'id':self
.__class
__.network_id
})
820 self
.assertEqual(network_list
, [])
822 def test_010_delete_network_negative(self
):
823 Non_exist_id
= str(uuid
.uuid4())
825 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
826 self
.__class
__.test_index
,
827 inspect
.currentframe().f_code
.co_name
)
828 self
.__class
__.test_index
+= 1
830 with self
.assertRaises(Exception) as context
:
831 test_config
["vim_conn"].delete_network(Non_exist_id
)
833 self
.assertEqual((context
.exception
).http_code
, 400)
835 class test_vimconn_get_flavor(unittest
.TestCase
):
841 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
844 def tearDownClass(cls
):
845 test_config
["test_number"] += 1
848 exec_info
= sys
.exc_info()
849 if exec_info
== (None, None, None):
850 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
852 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
853 logger
.critical("Traceback error",exc_info
=True)
855 def test_000_get_flavor(self
):
856 test_directory_content
= os
.listdir(test_config
["test_directory"])
858 for dir_name
in test_directory_content
:
859 if dir_name
== 'simple_linux':
860 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
861 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
864 for vnfd
in vnfd_files
:
865 with
open(vnfd
, 'r') as stream
:
866 vnf_descriptor
= yaml
.load(stream
)
868 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
869 for item
in vnfc_list
:
870 if 'ram' in item
and 'vcpus' in item
and 'disk' in item
:
872 vcpus
= item
['vcpus']
875 flavor_data
= {'ram': ram
,
880 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
881 self
.__class
__.test_index
,
882 inspect
.currentframe().f_code
.co_name
)
883 self
.__class
__.test_index
+= 1
885 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
887 result
= test_config
["vim_conn"].get_flavor(flavor_id
)
888 self
.assertEqual(ram
, result
['ram'])
889 self
.assertEqual(vcpus
, result
['vcpus'])
890 self
.assertEqual(disk
, result
['disk'])
893 result
= test_config
["vim_conn"].delete_flavor(flavor_id
)
895 logger
.info("Flavor id {} sucessfully deleted".format(result
))
897 logger
.info("Failed to delete flavor id {}".format(result
))
899 def test_010_get_flavor_negative(self
):
900 Non_exist_flavor_id
= str(uuid
.uuid4())
902 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
903 self
.__class
__.test_index
,
904 inspect
.currentframe().f_code
.co_name
)
905 self
.__class
__.test_index
+= 1
907 with self
.assertRaises(Exception) as context
:
908 test_config
["vim_conn"].get_flavor(Non_exist_flavor_id
)
910 self
.assertEqual((context
.exception
).http_code
, 404)
915 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
916 scenario based tests.
918 class descriptor_based_scenario_test(unittest
.TestCase
):
921 scenario_test_path
= None
923 instance_scenario_uuid
= None
929 cls
.to_delete_list
= []
930 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
931 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
934 def tearDownClass(cls
):
935 test_config
["test_number"] += 1
938 exec_info
= sys
.exc_info()
939 if exec_info
== (None, None, None):
940 logger
.info(self
.__class
__.test_text
+ " -> TEST OK")
942 logger
.warning(self
.__class
__.test_text
+ " -> TEST NOK")
943 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
945 for line
in error_trace
:
947 logger
.critical("{}".format(msg
))
950 def test_000_load_scenario(self
):
951 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
952 inspect
.currentframe().f_code
.co_name
,
953 test_config
["test_folder"])
954 self
.__class
__.test_index
+= 1
955 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
956 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
957 if len(vnfd_files
) == 0 or len(scenario_file
) > 1:
958 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
959 test_config
["test_folder"]))
962 for vnfd
in vnfd_files
:
963 with
open(vnfd
, 'r') as stream
:
964 vnf_descriptor
= yaml
.load(stream
)
966 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
967 for vnfc
in vnfc_list
:
968 vnfc
['image name'] = test_config
["image_name"]
969 devices
= vnfc
.get('devices',[])
970 for device
in devices
:
971 if device
['type'] == 'disk' and 'image name' in device
:
972 device
['image name'] = test_config
["image_name"]
974 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
975 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
)
977 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
978 "params": {"uuid": vnf
['vnf']['uuid']}})
980 #load the scenario definition
981 with
open(scenario_file
[0], 'r') as stream
:
982 scenario_descriptor
= yaml
.load(stream
)
983 networks
= scenario_descriptor
['scenario']['networks']
984 networks
[test_config
["mgmt_net"]] = networks
.pop('mgmt')
985 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
986 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
987 logger
.debug(scenario
)
988 self
.__class
__.to_delete_list
.insert(0,{"item": "scenario", "function": test_config
["client"].delete_scenario
,
989 "params":{"uuid": scenario
['scenario']['uuid']} })
990 self
.__class
__.scenario_uuid
= scenario
['scenario']['uuid']
992 def test_010_instantiate_scenario(self
):
993 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
994 inspect
.currentframe().f_code
.co_name
,
995 test_config
["test_folder"])
996 self
.__class
__.test_index
+= 1
998 instance
= test_config
["client"].create_instance(scenario_id
=self
.__class
__.scenario_uuid
,
999 name
=self
.__class
__.test_text
)
1000 self
.__class
__.instance_scenario_uuid
= instance
['uuid']
1001 logger
.debug(instance
)
1002 self
.__class
__.to_delete_list
.insert(0, {"item": "instance", "function": test_config
["client"].delete_instance
,
1003 "params": {"uuid": instance
['uuid']}})
1005 def test_020_check_deployent(self
):
1006 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1007 inspect
.currentframe().f_code
.co_name
,
1008 test_config
["test_folder"])
1009 self
.__class
__.test_index
+= 1
1011 if test_config
["manual"]:
1012 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1015 keep_waiting
= test_config
["timeout"]
1016 instance_active
= False
1018 result
= check_instance_scenario_active(self
.__class
__.instance_scenario_uuid
)
1021 elif 'ERROR' in result
[1]:
1022 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
1024 raise Exception(msg
)
1026 if keep_waiting
>= 5:
1029 elif keep_waiting
> 0:
1030 time
.sleep(keep_waiting
)
1033 msg
= 'Timeout reached while waiting instance scenario to get active'
1035 raise Exception(msg
)
1037 def test_030_clean_deployment(self
):
1038 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1039 inspect
.currentframe().f_code
.co_name
,
1040 test_config
["test_folder"])
1041 self
.__class
__.test_index
+= 1
1042 #At the moment if you delete an scenario right after creating it, in openstack datacenters
1043 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1045 for item
in self
.__class
__.to_delete_list
:
1046 response
= item
["function"](**item
["params"])
1047 logger
.debug(response
)
1050 def _get_random_string(maxLength
):
1051 '''generates a string with random characters string.letters and string.digits
1052 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1056 minLength
= min_string
- len(prefix
)
1057 if maxLength
< min_string
: maxLength
= min_string
1058 maxLength
-= len(prefix
)
1059 length
= random
.randint(minLength
,maxLength
)
1060 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
1063 def test_vimconnector(args
):
1065 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1066 test_config
['vimtype'] = args
.vimtype
1067 if args
.vimtype
== "vmware":
1068 import vimconn_vmware
as vim
1070 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1072 tenant_name
= args
.tenant_name
1073 test_config
['tenant'] = tenant_name
1074 config_params
= json
.loads(args
.config_param
)
1075 org_name
= config_params
.get('orgname')
1076 org_user
= config_params
.get('user')
1077 org_passwd
= config_params
.get('passwd')
1078 vim_url
= args
.endpoint_url
1080 # vmware connector obj
1081 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
)
1083 elif args
.vimtype
== "aws":
1084 import vimconn_aws
as vim
1085 elif args
.vimtype
== "openstack":
1086 import vimconn_openstack
as vim
1087 elif args
.vimtype
== "openvim":
1088 import vimconn_openvim
as vim
1090 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
1094 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1095 # If only want to obtain a tests list print it and exit
1098 for cls
in clsmembers
:
1099 if cls
[0].startswith('test_vimconnector'):
1100 tests_names
.append(cls
[0])
1102 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
1107 # Create the list of tests to be run
1108 code_based_tests
= []
1110 for test
in args
.tests
:
1111 for t
in test
.split(','):
1112 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1113 if len(matches_code_based_tests
) > 0:
1114 code_based_tests
.append(matches_code_based_tests
[0][1])
1116 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1118 if not code_based_tests
:
1120 for cls
in clsmembers
:
1121 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1122 if cls
[0].startswith('test_vimconnector'):
1123 code_based_tests
.append(cls
[1])
1125 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1127 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1128 # This is handled in the tests using logging.
1129 stream
= open('/dev/null', 'w')
1131 # Run code based tests
1132 basic_tests_suite
= unittest
.TestSuite()
1133 for test
in code_based_tests
:
1134 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1135 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1136 executed
+= result
.testsRun
1137 failed
+= len(result
.failures
) + len(result
.errors
)
1138 if failfast
and failed
:
1140 if len(result
.failures
) > 0:
1141 logger
.debug("failures : {}".format(result
.failures
))
1142 if len(result
.errors
) > 0:
1143 logger
.debug("errors : {}".format(result
.errors
))
1144 return executed
, failed
1149 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1150 import openmanoclient
1153 test_config
["client"] = openmanoclient
.openmanoclient(
1154 endpoint_url
=args
.endpoint_url
,
1155 tenant_name
=args
.tenant_name
,
1156 datacenter_name
=args
.datacenter
,
1157 debug
=args
.debug
, logger
=test_config
["logger_name"])
1158 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1159 # If only want to obtain a tests list print it and exit
1162 for cls
in clsmembers
:
1163 if cls
[0].startswith('test_VIM'):
1164 tests_names
.append(cls
[0])
1166 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
1167 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1168 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1173 # Create the list of tests to be run
1174 code_based_tests
= []
1176 for test
in args
.tests
:
1177 for t
in test
.split(','):
1178 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1179 if len(matches_code_based_tests
) > 0:
1180 code_based_tests
.append(matches_code_based_tests
[0][1])
1182 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1184 if not code_based_tests
:
1186 for cls
in clsmembers
:
1187 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1188 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
1189 code_based_tests
.append(cls
[1])
1191 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1193 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1194 # This is handled in the tests using logging.
1195 stream
= open('/dev/null', 'w')
1197 # Run code based tests
1198 basic_tests_suite
= unittest
.TestSuite()
1199 for test
in code_based_tests
:
1200 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1201 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1202 executed
+= result
.testsRun
1203 failed
+= len(result
.failures
) + len(result
.errors
)
1204 if failfast
and failed
:
1206 if len(result
.failures
) > 0:
1207 logger
.debug("failures : {}".format(result
.failures
))
1208 if len(result
.errors
) > 0:
1209 logger
.debug("errors : {}".format(result
.errors
))
1210 return executed
, failed
1213 def test_deploy(args
):
1215 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1216 import openmanoclient
1219 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1220 test_config
["image_name"] = args
.image_name
1221 test_config
["mgmt_net"] = args
.mgmt_net
1222 test_config
["manual"] = args
.manual
1223 test_directory_content
= os
.listdir(test_config
["test_directory"])
1224 # If only want to obtain a tests list print it and exit
1226 msg
= "he 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
1231 descriptor_based_tests
= []
1232 # Create the list of tests to be run
1233 code_based_tests
= []
1235 for test
in args
.tests
:
1236 for t
in test
.split(','):
1237 if t
in test_directory_content
:
1238 descriptor_based_tests
.append(t
)
1240 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1242 if not descriptor_based_tests
:
1244 descriptor_based_tests
= test_directory_content
1246 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1248 # import openmanoclient from relative path
1249 test_config
["client"] = openmanoclient
.openmanoclient(
1250 endpoint_url
=args
.endpoint_url
,
1251 tenant_name
=args
.tenant_name
,
1252 datacenter_name
=args
.datacenter
,
1253 debug
=args
.debug
, logger
=test_config
["logger_name"])
1255 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1256 # This is handled in the tests using logging.
1257 stream
= open('/dev/null', 'w')
1258 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1259 for test
in descriptor_based_tests
:
1260 test_config
["test_folder"] = test
1261 test_suite
= unittest
.TestSuite()
1262 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
1263 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
1264 executed
+= result
.testsRun
1265 failed
+= len(result
.failures
) + len(result
.errors
)
1266 if failfast
and failed
:
1268 if len(result
.failures
) > 0:
1269 logger
.debug("failures : {}".format(result
.failures
))
1270 if len(result
.errors
) > 0:
1271 logger
.debug("errors : {}".format(result
.errors
))
1273 return executed
, failed
1275 if __name__
=="__main__":
1277 parser
= ArgumentParser(description
='Test RO module')
1278 parser
.add_argument('-v','--version', action
='version', help="Show current version",
1279 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
1282 parent_parser
= ArgumentParser(add_help
=False)
1283 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1284 dest
='failfast', action
="store_true", default
=False)
1285 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1286 dest
='failed', action
="store_true", default
=False)
1287 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
1288 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
1290 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
1291 help='Set the logger file. By default '+default_logger_file
)
1292 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
1293 help="Set the openmano tenant to use for the test. By default 'osm'")
1294 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
1295 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1296 dest
='timeout', type=int, default
=300)
1297 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
1299 subparsers
= parser
.add_subparsers(help='test sets')
1301 # Deployment test set
1302 # -------------------
1303 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
1304 help="test deployment using descriptors at RO_test folder ")
1305 deploy_parser
.set_defaults(func
=test_deploy
)
1307 # Mandatory arguments
1308 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
1309 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1310 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
1311 help='Image name available at datacenter used for the tests')
1312 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
1313 help='Set the vim management network to use for tests')
1315 # Optional arguments
1316 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
1317 help='Pause execution once deployed to allow manual checking of the '
1318 'deployed instance scenario')
1319 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1320 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1323 # -------------------
1324 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
1325 vimconn_parser
.set_defaults(func
=test_vimconnector
)
1326 # Mandatory arguments
1327 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1328 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
1329 help='Set the vimconnector type to test')
1330 mandatory_arguments
.add_argument('-c', '--config', dest
='config_param', required
=True,
1331 help='Set the vimconnector specific config parameters in dictionary format')
1332 mandatory_arguments
.add_argument('-u', '--url', dest
='endpoint_url',required
=True, help="Set the vim connector url or Host IP")
1333 # Optional arguments
1334 # TODO add optional arguments for vimconn tests
1335 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1337 # Datacenter test set
1338 # -------------------
1339 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
1340 vimconn_parser
.set_defaults(func
=test_vim
)
1342 # Mandatory arguments
1343 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1344 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1346 # Optional arguments
1347 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1348 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1350 argcomplete
.autocomplete(parser
)
1351 args
= parser
.parse_args()
1355 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1356 logger_level
= 'INFO'
1358 logger_level
= 'DEBUG'
1360 logger_level
= 'WARNING'
1361 logger_name
= os
.path
.basename(__file__
)
1362 test_config
["logger_name"] = logger_name
1363 logger
= logging
.getLogger(logger_name
)
1364 logger
.setLevel(logger_level
)
1365 failfast
= args
.failfast
1367 # Configure a logging handler to store in a logging file
1368 if args
.logger_file
:
1369 fileHandler
= logging
.FileHandler(args
.logger_file
)
1370 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1371 fileHandler
.setFormatter(formatter_fileHandler
)
1372 logger
.addHandler(fileHandler
)
1374 # Configure a handler to print to stdout
1375 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
1376 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1377 consoleHandler
.setFormatter(formatter_consoleHandler
)
1378 logger
.addHandler(consoleHandler
)
1380 logger
.debug('Program started with the following arguments: ' + str(args
))
1382 # set test config parameters
1383 test_config
["timeout"] = args
.timeout
1384 test_config
["test_number"] = 1
1386 executed
, failed
= args
.func(args
)
1389 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
1390 sys
.exit(1 if failed
else 0)