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
45 global test_config
# used for global variables with the test configuration
49 def check_instance_scenario_active(uuid
):
50 instance
= test_config
["client"].get_instance(uuid
=uuid
)
52 for net
in instance
['nets']:
53 status
= net
['status']
54 if status
!= 'ACTIVE':
55 return (False, status
)
57 for vnf
in instance
['vnfs']:
60 if status
!= 'ACTIVE':
61 return (False, status
)
68 All unittest classes for code based tests must have prefix 'test_' in order to be taken into account for tests
70 class test_VIM_datacenter_tenant_operations(unittest
.TestCase
):
77 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
80 def tearDownClass(cls
):
81 test_config
["test_number"] += 1
84 exec_info
= sys
.exc_info()
85 if exec_info
== (None, None, None):
86 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
88 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
89 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
91 for line
in error_trace
:
93 logger
.critical("{}".format(msg
))
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(unittest
.TestCase
):
124 datacenter_name
= None
129 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
132 def tearDownClass(cls
):
133 test_config
["test_number"] += 1
136 exec_info
= sys
.exc_info()
137 if exec_info
== (None, None, None):
138 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
140 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
141 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
143 for line
in error_trace
:
145 logger
.critical("{}".format(msg
))
147 def test_000_create_datacenter(self
):
148 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
149 inspect
.currentframe().f_code
.co_name
)
150 self
.__class
__.datacenter_name
= _get_random_string(20)
151 self
.__class
__.test_index
+= 1
152 self
.datacenter
= test_config
["client"].create_datacenter(name
=self
.__class
__.datacenter_name
,
153 vim_url
="http://fakeurl/fake")
154 logger
.debug("{}".format(self
.datacenter
))
155 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name',''), self
.__class
__.datacenter_name
)
157 def test_010_list_datacenter(self
):
158 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
159 inspect
.currentframe().f_code
.co_name
)
161 self
.__class
__.test_index
+= 1
162 self
.datacenter
= test_config
["client"].get_datacenter(all_tenants
=True, name
=self
.__class
__.datacenter_name
)
163 logger
.debug("{}".format(self
.datacenter
))
164 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name', ''), self
.__class
__.datacenter_name
)
166 def test_020_attach_datacenter(self
):
167 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
168 inspect
.currentframe().f_code
.co_name
)
170 self
.__class
__.test_index
+= 1
171 self
.datacenter
= test_config
["client"].attach_datacenter(name
=self
.__class
__.datacenter_name
,
172 vim_tenant_name
='fake')
173 logger
.debug("{}".format(self
.datacenter
))
174 assert ('vim_tenants' in self
.datacenter
.get('datacenter', {}))
176 def test_030_list_attached_datacenter(self
):
177 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
178 inspect
.currentframe().f_code
.co_name
)
180 self
.__class
__.test_index
+= 1
181 self
.datacenter
= test_config
["client"].get_datacenter(all_tenants
=False, name
=self
.__class
__.datacenter_name
)
182 logger
.debug("{}".format(self
.datacenter
))
183 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name', ''), self
.__class
__.datacenter_name
)
185 def test_040_detach_datacenter(self
):
186 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
187 inspect
.currentframe().f_code
.co_name
)
189 self
.__class
__.test_index
+= 1
190 self
.datacenter
= test_config
["client"].detach_datacenter(name
=self
.__class
__.datacenter_name
)
191 logger
.debug("{}".format(self
.datacenter
))
192 assert ('detached' in self
.datacenter
.get('result', ""))
194 def test_050_delete_datacenter(self
):
195 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
196 inspect
.currentframe().f_code
.co_name
)
198 self
.__class
__.test_index
+= 1
199 self
.datacenter
= test_config
["client"].delete_datacenter(name
=self
.__class
__.datacenter_name
)
200 logger
.debug("{}".format(self
.datacenter
))
201 assert('deleted' in self
.datacenter
.get('result',""))
204 class test_VIM_network_operations(unittest
.TestCase
):
206 vim_network_name
= None
208 vim_network_uuid
= None
212 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
215 def tearDownClass(cls
):
216 test_config
["test_number"] += 1
219 exec_info
= sys
.exc_info()
220 if exec_info
== (None, None, None):
221 logger
.info(self
.__class
__.test_text
+ " -> TEST OK")
223 logger
.warning(self
.__class
__.test_text
+ " -> TEST NOK")
224 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
226 for line
in error_trace
:
228 logger
.critical("{}".format(msg
))
230 def test_000_create_VIM_network(self
):
231 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
232 inspect
.currentframe().f_code
.co_name
)
233 self
.__class
__.vim_network_name
= _get_random_string(20)
234 self
.__class
__.test_index
+= 1
235 network
= test_config
["client"].vim_action("create", "networks", name
=self
.__class
__.vim_network_name
)
236 logger
.debug("{}".format(network
))
237 self
.__class
__.vim_network_uuid
= network
["network"]["id"]
238 self
.assertEqual(network
.get('network', {}).get('name', ''), self
.__class
__.vim_network_name
)
240 def test_010_list_VIM_networks(self
):
241 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
242 inspect
.currentframe().f_code
.co_name
)
243 self
.__class
__.test_index
+= 1
244 networks
= test_config
["client"].vim_action("list", "networks")
245 logger
.debug("{}".format(networks
))
247 def test_020_get_VIM_network_by_uuid(self
):
248 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
249 inspect
.currentframe().f_code
.co_name
)
251 self
.__class
__.test_index
+= 1
252 network
= test_config
["client"].vim_action("show", "networks", uuid
=self
.__class
__.vim_network_uuid
)
253 logger
.debug("{}".format(network
))
254 self
.assertEqual(network
.get('network', {}).get('name', ''), self
.__class
__.vim_network_name
)
256 def test_030_delete_VIM_network_by_uuid(self
):
257 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
258 inspect
.currentframe().f_code
.co_name
)
260 self
.__class
__.test_index
+= 1
261 network
= test_config
["client"].vim_action("delete", "networks", uuid
=self
.__class
__.vim_network_uuid
)
262 logger
.debug("{}".format(network
))
263 assert ('deleted' in network
.get('result', ""))
266 class test_VIM_image_operations(unittest
.TestCase
):
272 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
275 def tearDownClass(cls
):
276 test_config
["test_number"] += 1
279 exec_info
= sys
.exc_info()
280 if exec_info
== (None, None, None):
281 logger
.info(self
.__class
__.test_text
+ " -> TEST OK")
283 logger
.warning(self
.__class
__.test_text
+ " -> TEST NOK")
284 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
286 for line
in error_trace
:
288 logger
.critical("{}".format(msg
))
290 def test_000_list_VIM_images(self
):
291 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
292 inspect
.currentframe().f_code
.co_name
)
293 self
.__class
__.test_index
+= 1
294 images
= test_config
["client"].vim_action("list", "images")
295 logger
.debug("{}".format(images
))
298 The following is a non critical test that will fail most of the times.
299 In case of OpenStack datacenter these tests will only success if RO has access to the admin endpoint
300 This test will only be executed in case it is specifically requested by the user
302 class test_VIM_tenant_operations(unittest
.TestCase
):
304 vim_tenant_name
= None
306 vim_tenant_uuid
= None
310 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
311 logger
.warning("In case of OpenStack datacenter these tests will only success "
312 "if RO has access to the admin endpoint")
315 def tearDownClass(cls
):
316 test_config
["test_number"] += 1
319 exec_info
= sys
.exc_info()
320 if exec_info
== (None, None, None):
321 logger
.info(self
.__class
__.test_text
+ " -> TEST OK")
323 logger
.warning(self
.__class
__.test_text
+ " -> TEST NOK")
324 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
326 for line
in error_trace
:
328 logger
.critical("{}".format(msg
))
330 def test_000_create_VIM_tenant(self
):
331 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
332 inspect
.currentframe().f_code
.co_name
)
333 self
.__class
__.vim_tenant_name
= _get_random_string(20)
334 self
.__class
__.test_index
+= 1
335 tenant
= test_config
["client"].vim_action("create", "tenants", name
=self
.__class
__.vim_tenant_name
)
336 logger
.debug("{}".format(tenant
))
337 self
.__class
__.vim_tenant_uuid
= tenant
["tenant"]["id"]
338 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.vim_tenant_name
)
340 def test_010_list_VIM_tenants(self
):
341 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
342 inspect
.currentframe().f_code
.co_name
)
343 self
.__class
__.test_index
+= 1
344 tenants
= test_config
["client"].vim_action("list", "tenants")
345 logger
.debug("{}".format(tenants
))
347 def test_020_get_VIM_tenant_by_uuid(self
):
348 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
349 inspect
.currentframe().f_code
.co_name
)
351 self
.__class
__.test_index
+= 1
352 tenant
= test_config
["client"].vim_action("show", "tenants", uuid
=self
.__class
__.vim_tenant_uuid
)
353 logger
.debug("{}".format(tenant
))
354 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.vim_tenant_name
)
356 def test_030_delete_VIM_tenant_by_uuid(self
):
357 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
358 inspect
.currentframe().f_code
.co_name
)
360 self
.__class
__.test_index
+= 1
361 tenant
= test_config
["client"].vim_action("delete", "tenants", uuid
=self
.__class
__.vim_tenant_uuid
)
362 logger
.debug("{}".format(tenant
))
363 assert ('deleted' in tenant
.get('result', ""))
367 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
368 scenario based tests.
370 class descriptor_based_scenario_test(unittest
.TestCase
):
373 scenario_test_path
= None
375 instance_scenario_uuid
= None
381 cls
.to_delete_list
= []
382 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
383 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
386 def tearDownClass(cls
):
387 test_config
["test_number"] += 1
390 exec_info
= sys
.exc_info()
391 if exec_info
== (None, None, None):
392 logger
.info(self
.__class
__.test_text
+ " -> TEST OK")
394 logger
.warning(self
.__class
__.test_text
+ " -> TEST NOK")
395 error_trace
= traceback
.format_exception(exec_info
[0], exec_info
[1], exec_info
[2])
397 for line
in error_trace
:
399 logger
.critical("{}".format(msg
))
402 def test_000_load_scenario(self
):
403 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
404 inspect
.currentframe().f_code
.co_name
,
405 test_config
["test_folder"])
406 self
.__class
__.test_index
+= 1
407 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
408 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
409 if len(vnfd_files
) == 0 or len(scenario_file
) > 1:
410 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
411 test_config
["test_folder"]))
414 for vnfd
in vnfd_files
:
415 with
open(vnfd
, 'r') as stream
:
416 vnf_descriptor
= yaml
.load(stream
)
418 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
419 for vnfc
in vnfc_list
:
420 vnfc
['image name'] = test_config
["image_name"]
421 devices
= vnfc
.get('devices',[])
422 for device
in devices
:
423 if device
['type'] == 'disk' and 'image name' in device
:
424 device
['image name'] = test_config
["image_name"]
426 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
427 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
)
429 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
430 "params": {"uuid": vnf
['vnf']['uuid']}})
432 #load the scenario definition
433 with
open(scenario_file
[0], 'r') as stream
:
434 scenario_descriptor
= yaml
.load(stream
)
435 networks
= scenario_descriptor
['scenario']['networks']
436 networks
[test_config
["mgmt_net"]] = networks
.pop('mgmt')
437 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
438 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
439 logger
.debug(scenario
)
440 self
.__class
__.to_delete_list
.insert(0,{"item": "scenario", "function": test_config
["client"].delete_scenario
,
441 "params":{"uuid": scenario
['scenario']['uuid']} })
442 self
.__class
__.scenario_uuid
= scenario
['scenario']['uuid']
444 def test_010_instantiate_scenario(self
):
445 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
446 inspect
.currentframe().f_code
.co_name
,
447 test_config
["test_folder"])
448 self
.__class
__.test_index
+= 1
450 instance
= test_config
["client"].create_instance(scenario_id
=self
.__class
__.scenario_uuid
,
451 name
=self
.__class
__.test_text
)
452 self
.__class
__.instance_scenario_uuid
= instance
['uuid']
453 logger
.debug(instance
)
454 self
.__class
__.to_delete_list
.insert(0, {"item": "instance", "function": test_config
["client"].delete_instance
,
455 "params": {"uuid": instance
['uuid']}})
457 def test_020_check_deployent(self
):
458 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
459 inspect
.currentframe().f_code
.co_name
,
460 test_config
["test_folder"])
461 self
.__class
__.test_index
+= 1
463 if test_config
["manual"]:
464 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
467 keep_waiting
= test_config
["timeout"]
468 instance_active
= False
470 result
= check_instance_scenario_active(self
.__class
__.instance_scenario_uuid
)
473 elif 'ERROR' in result
[1]:
474 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
478 if keep_waiting
>= 5:
481 elif keep_waiting
> 0:
482 time
.sleep(keep_waiting
)
485 msg
= 'Timeout reached while waiting instance scenario to get active'
489 def test_030_clean_deployment(self
):
490 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
491 inspect
.currentframe().f_code
.co_name
,
492 test_config
["test_folder"])
493 self
.__class
__.test_index
+= 1
494 #At the moment if you delete an scenario right after creating it, in openstack datacenters
495 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
497 for item
in self
.__class
__.to_delete_list
:
498 response
= item
["function"](**item
["params"])
499 logger
.debug(response
)
502 def _get_random_string(maxLength
):
503 '''generates a string with random characters string.letters and string.digits
504 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
508 minLength
= min_string
- len(prefix
)
509 if maxLength
< min_string
: maxLength
= min_string
510 maxLength
-= len(prefix
)
511 length
= random
.randint(minLength
,maxLength
)
512 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
515 def test_vimconnector(args
):
517 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
518 if args
.vimtype
== "vmware":
519 import vimconn_vmware
as vim
520 elif args
.vimtype
== "aws":
521 import vimconn_aws
as vim
522 elif args
.vimtype
== "openstack":
523 import vimconn_openstack
as vim
524 elif args
.vimtype
== "openvim":
525 import vimconn_openvim
as vim
527 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
531 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
532 # If only want to obtain a tests list print it and exit
535 for cls
in clsmembers
:
536 if cls
[0].startswith('test_vimconnector'):
537 tests_names
.append(cls
[0])
539 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
544 # Create the list of tests to be run
545 code_based_tests
= []
547 for test
in args
.tests
:
548 for t
in test
.split(','):
549 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
550 if len(matches_code_based_tests
) > 0:
551 code_based_tests
.append(matches_code_based_tests
[0][1])
553 logger
.critical("Test '{}' is not among the possible ones".format(t
))
555 if not code_based_tests
:
557 for cls
in clsmembers
:
558 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
559 if cls
[0].startswith('test_vimconnector'):
560 code_based_tests
.append(cls
[1])
562 logger
.debug("tests to be executed: {}".format(code_based_tests
))
564 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
565 # This is handled in the tests using logging.
566 stream
= open('/dev/null', 'w')
568 # Run code based tests
569 basic_tests_suite
= unittest
.TestSuite()
570 for test
in code_based_tests
:
571 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
572 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
573 executed
+= result
.testsRun
574 failed
+= len(result
.failures
) + len(result
.errors
)
575 if failfast
and failed
:
577 if len(result
.failures
) > 0:
578 logger
.debug("failures : {}".format(result
.failures
))
579 if len(result
.errors
) > 0:
580 logger
.debug("errors : {}".format(result
.errors
))
581 return executed
, failed
586 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
587 import openmanoclient
590 test_config
["client"] = openmanoclient
.openmanoclient(
591 endpoint_url
=args
.endpoint_url
,
592 tenant_name
=args
.tenant_name
,
593 datacenter_name
=args
.datacenter
,
594 debug
=args
.debug
, logger
=test_config
["logger_name"])
595 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
596 # If only want to obtain a tests list print it and exit
599 for cls
in clsmembers
:
600 if cls
[0].startswith('test_VIM'):
601 tests_names
.append(cls
[0])
603 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
604 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
605 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
610 # Create the list of tests to be run
611 code_based_tests
= []
613 for test
in args
.tests
:
614 for t
in test
.split(','):
615 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
616 if len(matches_code_based_tests
) > 0:
617 code_based_tests
.append(matches_code_based_tests
[0][1])
619 logger
.critical("Test '{}' is not among the possible ones".format(t
))
621 if not code_based_tests
:
623 for cls
in clsmembers
:
624 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
625 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
626 code_based_tests
.append(cls
[1])
628 logger
.debug("tests to be executed: {}".format(code_based_tests
))
630 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
631 # This is handled in the tests using logging.
632 stream
= open('/dev/null', 'w')
634 # Run code based tests
635 basic_tests_suite
= unittest
.TestSuite()
636 for test
in code_based_tests
:
637 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
638 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
639 executed
+= result
.testsRun
640 failed
+= len(result
.failures
) + len(result
.errors
)
641 if failfast
and failed
:
643 if len(result
.failures
) > 0:
644 logger
.debug("failures : {}".format(result
.failures
))
645 if len(result
.errors
) > 0:
646 logger
.debug("errors : {}".format(result
.errors
))
647 return executed
, failed
650 def test_deploy(args
):
652 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
653 import openmanoclient
656 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
657 test_config
["image_name"] = args
.image_name
658 test_config
["mgmt_net"] = args
.mgmt_net
659 test_config
["manual"] = args
.manual
660 test_directory_content
= os
.listdir(test_config
["test_directory"])
661 # If only want to obtain a tests list print it and exit
663 msg
= "he 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
668 descriptor_based_tests
= []
669 # Create the list of tests to be run
670 code_based_tests
= []
672 for test
in args
.tests
:
673 for t
in test
.split(','):
674 if t
in test_directory_content
:
675 descriptor_based_tests
.append(t
)
677 logger
.critical("Test '{}' is not among the possible ones".format(t
))
679 if not descriptor_based_tests
:
681 descriptor_based_tests
= test_directory_content
683 logger
.debug("tests to be executed: {}".format(code_based_tests
))
685 # import openmanoclient from relative path
686 test_config
["client"] = openmanoclient
.openmanoclient(
687 endpoint_url
=args
.endpoint_url
,
688 tenant_name
=args
.tenant_name
,
689 datacenter_name
=args
.datacenter
,
690 debug
=args
.debug
, logger
=test_config
["logger_name"])
692 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
693 # This is handled in the tests using logging.
694 stream
= open('/dev/null', 'w')
695 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
696 for test
in descriptor_based_tests
:
697 test_config
["test_folder"] = test
698 test_suite
= unittest
.TestSuite()
699 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
700 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
701 executed
+= result
.testsRun
702 failed
+= len(result
.failures
) + len(result
.errors
)
703 if failfast
and failed
:
705 if len(result
.failures
) > 0:
706 logger
.debug("failures : {}".format(result
.failures
))
707 if len(result
.errors
) > 0:
708 logger
.debug("errors : {}".format(result
.errors
))
710 return executed
, failed
712 if __name__
=="__main__":
714 parser
= ArgumentParser(description
='Test RO module')
715 parser
.add_argument('-v','--version', action
='version', help="Show current version",
716 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
719 parent_parser
= ArgumentParser(add_help
=False)
720 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
721 dest
='failfast', action
="store_true", default
=False)
722 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
723 dest
='failed', action
="store_true", default
=False)
724 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
725 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
727 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
728 help='Set the logger file. By default '+default_logger_file
)
729 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
730 help="Set the openmano tenant to use for the test. By default 'osm'")
731 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
732 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
733 dest
='timeout', type=int, default
=300)
734 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
736 subparsers
= parser
.add_subparsers(help='test sets')
738 # Deployment test set
739 # -------------------
740 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
741 help="test deployment using descriptors at RO_test folder ")
742 deploy_parser
.set_defaults(func
=test_deploy
)
744 # Mandatory arguments
745 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
746 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
747 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
748 help='Image name available at datacenter used for the tests')
749 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
750 help='Set the vim management network to use for tests')
753 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
754 help='Pause execution once deployed to allow manual checking of the '
755 'deployed instance scenario')
756 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
757 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
760 # -------------------
761 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
762 vimconn_parser
.set_defaults(func
=test_vimconnector
)
763 # Mandatory arguments
764 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
765 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
766 help='Set the vimconnector type to test')
767 # TODO add mandatory arguments for vimconn test
768 # mandatory_arguments.add_argument('-c', '--config', dest='config_param', required=True, help='<HELP>')
771 # TODO add optional arguments for vimconn tests
772 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
774 # Datacenter test set
775 # -------------------
776 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
777 vimconn_parser
.set_defaults(func
=test_vim
)
779 # Mandatory arguments
780 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
781 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
784 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
785 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
787 argcomplete
.autocomplete(parser
)
788 args
= parser
.parse_args()
792 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
793 logger_level
= 'INFO'
795 logger_level
= 'DEBUG'
797 logger_level
= 'WARNING'
798 logger_name
= os
.path
.basename(__file__
)
799 test_config
["logger_name"] = logger_name
800 logger
= logging
.getLogger(logger_name
)
801 logger
.setLevel(logger_level
)
802 failfast
= args
.failfast
804 # Configure a logging handler to store in a logging file
806 fileHandler
= logging
.FileHandler(args
.logger_file
)
807 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
808 fileHandler
.setFormatter(formatter_fileHandler
)
809 logger
.addHandler(fileHandler
)
811 # Configure a handler to print to stdout
812 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
813 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
814 consoleHandler
.setFormatter(formatter_consoleHandler
)
815 logger
.addHandler(consoleHandler
)
817 logger
.debug('Program started with the following arguments: ' + str(args
))
819 # set test config parameters
820 test_config
["timeout"] = args
.timeout
821 test_config
["test_number"] = 1
823 executed
, failed
= args
.func(args
)
826 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
827 sys
.exit(1 if failed
else 0)