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
= "Jun 2017"
33 from argparse
import ArgumentParser
44 from pyvcloud
.vcloudair
import VCA
48 global test_config
# used for global variables with the test configuration
51 class test_base(unittest
.TestCase
):
57 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
60 def tearDownClass(cls
):
61 test_config
["test_number"] += 1
64 exec_info
= sys
.exc_info()
65 if exec_info
== (None, None, None):
66 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
68 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
69 logger
.critical("Traceback error",exc_info
=True)
72 def check_instance_scenario_active(uuid
):
73 instance
= test_config
["client"].get_instance(uuid
=uuid
)
75 for net
in instance
['nets']:
76 status
= net
['status']
77 if status
!= 'ACTIVE':
78 return (False, status
)
80 for vnf
in instance
['vnfs']:
83 if status
!= 'ACTIVE':
84 return (False, status
)
91 All unittest classes for code based tests must have prefix 'test_' in order to be taken into account for tests
93 class test_VIM_datacenter_tenant_operations(test_base
):
96 def test_000_create_RO_tenant(self
):
97 self
.__class
__.tenant_name
= _get_random_string(20)
98 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
99 inspect
.currentframe().f_code
.co_name
)
100 self
.__class
__.test_index
+= 1
101 tenant
= test_config
["client"].create_tenant(name
=self
.__class
__.tenant_name
,
102 description
=self
.__class
__.tenant_name
)
103 logger
.debug("{}".format(tenant
))
104 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.tenant_name
)
106 def test_010_list_RO_tenant(self
):
107 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
108 inspect
.currentframe().f_code
.co_name
)
109 self
.__class
__.test_index
+= 1
110 tenant
= test_config
["client"].get_tenant(name
=self
.__class
__.tenant_name
)
111 logger
.debug("{}".format(tenant
))
112 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.tenant_name
)
114 def test_020_delete_RO_tenant(self
):
115 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
116 inspect
.currentframe().f_code
.co_name
)
117 self
.__class
__.test_index
+= 1
118 tenant
= test_config
["client"].delete_tenant(name
=self
.__class
__.tenant_name
)
119 logger
.debug("{}".format(tenant
))
120 assert('deleted' in tenant
.get('result',""))
123 class test_VIM_datacenter_operations(test_base
):
124 datacenter_name
= None
126 def test_000_create_datacenter(self
):
127 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
128 inspect
.currentframe().f_code
.co_name
)
129 self
.__class
__.datacenter_name
= _get_random_string(20)
130 self
.__class
__.test_index
+= 1
131 self
.datacenter
= test_config
["client"].create_datacenter(name
=self
.__class
__.datacenter_name
,
132 vim_url
="http://fakeurl/fake")
133 logger
.debug("{}".format(self
.datacenter
))
134 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name',''), self
.__class
__.datacenter_name
)
136 def test_010_list_datacenter(self
):
137 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
138 inspect
.currentframe().f_code
.co_name
)
140 self
.__class
__.test_index
+= 1
141 self
.datacenter
= test_config
["client"].get_datacenter(all_tenants
=True, name
=self
.__class
__.datacenter_name
)
142 logger
.debug("{}".format(self
.datacenter
))
143 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name', ''), self
.__class
__.datacenter_name
)
145 def test_020_attach_datacenter(self
):
146 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
147 inspect
.currentframe().f_code
.co_name
)
149 self
.__class
__.test_index
+= 1
150 self
.datacenter
= test_config
["client"].attach_datacenter(name
=self
.__class
__.datacenter_name
,
151 vim_tenant_name
='fake')
152 logger
.debug("{}".format(self
.datacenter
))
153 assert ('vim_tenants' in self
.datacenter
.get('datacenter', {}))
155 def test_030_list_attached_datacenter(self
):
156 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
157 inspect
.currentframe().f_code
.co_name
)
159 self
.__class
__.test_index
+= 1
160 self
.datacenter
= test_config
["client"].get_datacenter(all_tenants
=False, name
=self
.__class
__.datacenter_name
)
161 logger
.debug("{}".format(self
.datacenter
))
162 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name', ''), self
.__class
__.datacenter_name
)
164 def test_040_detach_datacenter(self
):
165 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
166 inspect
.currentframe().f_code
.co_name
)
168 self
.__class
__.test_index
+= 1
169 self
.datacenter
= test_config
["client"].detach_datacenter(name
=self
.__class
__.datacenter_name
)
170 logger
.debug("{}".format(self
.datacenter
))
171 assert ('detached' in self
.datacenter
.get('result', ""))
173 def test_050_delete_datacenter(self
):
174 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
175 inspect
.currentframe().f_code
.co_name
)
177 self
.__class
__.test_index
+= 1
178 self
.datacenter
= test_config
["client"].delete_datacenter(name
=self
.__class
__.datacenter_name
)
179 logger
.debug("{}".format(self
.datacenter
))
180 assert('deleted' in self
.datacenter
.get('result',""))
183 class test_VIM_network_operations(test_base
):
184 vim_network_name
= None
185 vim_network_uuid
= None
187 def test_000_create_VIM_network(self
):
188 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
189 inspect
.currentframe().f_code
.co_name
)
190 self
.__class
__.vim_network_name
= _get_random_string(20)
191 self
.__class
__.test_index
+= 1
192 network
= test_config
["client"].vim_action("create", "networks", name
=self
.__class
__.vim_network_name
)
193 logger
.debug("{}".format(network
))
194 self
.__class
__.vim_network_uuid
= network
["network"]["id"]
195 self
.assertEqual(network
.get('network', {}).get('name', ''), self
.__class
__.vim_network_name
)
197 def test_010_list_VIM_networks(self
):
198 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
199 inspect
.currentframe().f_code
.co_name
)
200 self
.__class
__.test_index
+= 1
201 networks
= test_config
["client"].vim_action("list", "networks")
202 logger
.debug("{}".format(networks
))
204 def test_020_get_VIM_network_by_uuid(self
):
205 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
206 inspect
.currentframe().f_code
.co_name
)
208 self
.__class
__.test_index
+= 1
209 network
= test_config
["client"].vim_action("show", "networks", uuid
=self
.__class
__.vim_network_uuid
)
210 logger
.debug("{}".format(network
))
211 self
.assertEqual(network
.get('network', {}).get('name', ''), self
.__class
__.vim_network_name
)
213 def test_030_delete_VIM_network_by_uuid(self
):
214 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
215 inspect
.currentframe().f_code
.co_name
)
217 self
.__class
__.test_index
+= 1
218 network
= test_config
["client"].vim_action("delete", "networks", uuid
=self
.__class
__.vim_network_uuid
)
219 logger
.debug("{}".format(network
))
220 assert ('deleted' in network
.get('result', ""))
223 class test_VIM_image_operations(test_base
):
225 def test_000_list_VIM_images(self
):
226 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
227 inspect
.currentframe().f_code
.co_name
)
228 self
.__class
__.test_index
+= 1
229 images
= test_config
["client"].vim_action("list", "images")
230 logger
.debug("{}".format(images
))
233 The following is a non critical test that will fail most of the times.
234 In case of OpenStack datacenter these tests will only success if RO has access to the admin endpoint
235 This test will only be executed in case it is specifically requested by the user
237 class test_VIM_tenant_operations(test_base
):
238 vim_tenant_name
= None
239 vim_tenant_uuid
= None
243 test_base
.setUpClass(cls
)
244 logger
.warning("In case of OpenStack datacenter these tests will only success "
245 "if RO has access to the admin endpoint")
247 def test_000_create_VIM_tenant(self
):
248 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
249 inspect
.currentframe().f_code
.co_name
)
250 self
.__class
__.vim_tenant_name
= _get_random_string(20)
251 self
.__class
__.test_index
+= 1
252 tenant
= test_config
["client"].vim_action("create", "tenants", name
=self
.__class
__.vim_tenant_name
)
253 logger
.debug("{}".format(tenant
))
254 self
.__class
__.vim_tenant_uuid
= tenant
["tenant"]["id"]
255 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.vim_tenant_name
)
257 def test_010_list_VIM_tenants(self
):
258 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
259 inspect
.currentframe().f_code
.co_name
)
260 self
.__class
__.test_index
+= 1
261 tenants
= test_config
["client"].vim_action("list", "tenants")
262 logger
.debug("{}".format(tenants
))
264 def test_020_get_VIM_tenant_by_uuid(self
):
265 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
266 inspect
.currentframe().f_code
.co_name
)
268 self
.__class
__.test_index
+= 1
269 tenant
= test_config
["client"].vim_action("show", "tenants", uuid
=self
.__class
__.vim_tenant_uuid
)
270 logger
.debug("{}".format(tenant
))
271 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.vim_tenant_name
)
273 def test_030_delete_VIM_tenant_by_uuid(self
):
274 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
275 inspect
.currentframe().f_code
.co_name
)
277 self
.__class
__.test_index
+= 1
278 tenant
= test_config
["client"].vim_action("delete", "tenants", uuid
=self
.__class
__.vim_tenant_uuid
)
279 logger
.debug("{}".format(tenant
))
280 assert ('deleted' in tenant
.get('result', ""))
282 class test_vimconn_connect(test_base
):
287 # def setUpClass(cls):
288 # logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
291 # def tearDownClass(cls):
292 # test_config["test_number"] += 1
294 # def tearDown(self):
295 # exec_info = sys.exc_info()
296 # if exec_info == (None, None, None):
297 # logger.info(self.__class__.test_text+" -> TEST OK")
299 # logger.warning(self.__class__.test_text+" -> TEST NOK")
300 # logger.critical("Traceback error",exc_info=True)
302 def test_000_connect(self
):
303 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
304 self
.__class
__.test_index
,
305 inspect
.currentframe().f_code
.co_name
)
307 self
.__class
__.test_index
+= 1
308 if test_config
['vimtype'] == 'vmware':
309 vca_object
= test_config
["vim_conn"].connect()
310 logger
.debug("{}".format(vca_object
))
311 self
.assertIsInstance(vca_object
, VCA
)
314 class test_vimconn_new_network(test_base
):
320 # def setUpClass(cls):
321 # logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
324 # def tearDownClass(cls):
325 # test_config["test_number"] += 1
327 # def tearDown(self):
328 # exec_info = sys.exc_info()
329 # if exec_info == (None, None, None):
330 # logger.info(self.__class__.test_text+" -> TEST OK")
332 # logger.warning(self.__class__.test_text+" -> TEST NOK")
333 # logger.critical("Traceback error",exc_info=True)
335 def test_000_new_network(self
):
336 self
.__class
__.network_name
= _get_random_string(20)
337 network_type
= 'bridge'
339 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
340 self
.__class
__.test_index
, inspect
.currentframe().f_code
.co_name
)
341 self
.__class
__.test_index
+= 1
343 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
344 net_type
=network_type
)
345 self
.__class
__.network_id
= network
346 logger
.debug("{}".format(network
))
348 network_list
= test_config
["vim_conn"].get_vcd_network_list()
349 for net
in network_list
:
350 if self
.__class
__.network_name
in net
.get('name'):
351 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
352 self
.assertEqual(net
.get('type'), network_type
)
354 # Deleting created network
355 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
357 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
359 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
361 def test_010_new_network_by_types(self
):
363 network_types
= ['data','bridge','mgmt']
364 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
365 self
.__class
__.test_index
,
366 inspect
.currentframe().f_code
.co_name
)
367 self
.__class
__.test_index
+= 1
368 for net_type
in network_types
:
369 self
.__class
__.network_name
= _get_random_string(20)
370 network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
373 delete_net_ids
.append(network_id
)
374 logger
.debug("{}".format(network_id
))
376 network_list
= test_config
["vim_conn"].get_vcd_network_list()
377 for net
in network_list
:
378 if self
.__class
__.network_name
in net
.get('name'):
379 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
380 if net_type
in net
.get('type'):
381 self
.assertEqual(net
.get('type'), net_type
)
383 self
.assertNotEqual(net
.get('type'), net_type
)
385 # Deleting created network
386 for net_id
in delete_net_ids
:
387 result
= test_config
["vim_conn"].delete_network(net_id
)
389 logger
.info("Network id {} sucessfully deleted".format(net_id
))
391 logger
.info("Failed to delete network id {}".format(net_id
))
393 def test_020_new_network_by_ipprofile(self
):
394 test_directory_content
= os
.listdir(test_config
["test_directory"])
396 for dir_name
in test_directory_content
:
397 if dir_name
== 'simple_multi_vnfc':
398 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
399 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
402 for vnfd
in vnfd_files
:
403 with
open(vnfd
, 'r') as stream
:
404 vnf_descriptor
= yaml
.load(stream
)
406 internal_connections_list
= vnf_descriptor
['vnf']['internal-connections']
407 for item
in internal_connections_list
:
408 if 'ip-profile' in item
:
409 version
= item
['ip-profile']['ip-version']
410 dhcp_count
= item
['ip-profile']['dhcp']['count']
411 dhcp_enabled
= item
['ip-profile']['dhcp']['enabled']
413 self
.__class
__.network_name
= _get_random_string(20)
414 ip_profile
= {'dhcp_count': dhcp_count
,
415 'dhcp_enabled': dhcp_enabled
,
416 'ip_version': version
418 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
419 self
.__class
__.test_index
,
420 inspect
.currentframe().f_code
.co_name
)
421 self
.__class
__.test_index
+= 1
422 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
424 ip_profile
=ip_profile
)
425 self
.__class
__.network_id
= network
426 logger
.debug("{}".format(network
))
428 network_list
= test_config
["vim_conn"].get_vcd_network_list()
429 for net
in network_list
:
430 if self
.__class
__.network_name
in net
.get('name'):
431 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
433 # Deleting created network
434 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
436 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
438 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
440 def test_030_new_network_by_isshared(self
):
441 self
.__class
__.network_name
= _get_random_string(20)
443 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
444 self
.__class
__.test_index
,
445 inspect
.currentframe().f_code
.co_name
)
446 self
.__class
__.test_index
+= 1
447 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
450 self
.__class
__.network_id
= network
451 logger
.debug("{}".format(network
))
453 network_list
= test_config
["vim_conn"].get_vcd_network_list()
454 for net
in network_list
:
455 if self
.__class
__.network_name
in net
.get('name'):
456 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
457 self
.assertEqual(net
.get('shared'), shared
)
459 # Deleting created network
460 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
462 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
464 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
466 def test_040_new_network_by_negative(self
):
467 self
.__class
__.network_name
= _get_random_string(20)
468 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
469 self
.__class
__.test_index
,
470 inspect
.currentframe().f_code
.co_name
)
471 self
.__class
__.test_index
+= 1
472 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
473 net_type
='unknowntype')
474 self
.__class
__.network_id
= network
475 logger
.debug("{}".format(network
))
476 network_list
= test_config
["vim_conn"].get_vcd_network_list()
477 for net
in network_list
:
478 if self
.__class
__.network_name
in net
.get('name'):
479 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
481 # Deleting created network
482 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
484 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
486 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
488 class test_vimconn_get_network_list(test_base
):
494 # def setUpClass(cls):
495 # logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
498 # def tearDownClass(cls):
499 # test_config["test_number"] += 1
502 # creating new network
503 self
.__class
__.network_name
= _get_random_string(20)
504 self
.__class
__.net_type
= 'bridge'
505 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
506 net_type
=self
.__class
__.net_type
)
507 self
.__class
__.network_id
= network
508 logger
.debug("{}".format(network
))
511 test_base
.tearDown(self
)
512 # exec_info = sys.exc_info()
513 # if exec_info == (None, None, None):
514 # logger.info(self.__class__.test_text+" -> TEST OK")
516 # logger.warning(self.__class__.test_text+" -> TEST NOK")
517 # logger.critical("Traceback error",exc_info=True)
519 # Deleting created network
520 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
522 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
524 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
526 def test_000_get_network_list(self
):
527 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
528 self
.__class
__.test_index
,
529 inspect
.currentframe().f_code
.co_name
)
530 self
.__class
__.test_index
+= 1
532 network_list
= test_config
["vim_conn"].get_network_list()
533 for net
in network_list
:
534 if self
.__class
__.network_name
in net
.get('name'):
535 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
536 self
.assertEqual(net
.get('type'), self
.__class
__.net_type
)
537 self
.assertEqual(net
.get('status'), 'ACTIVE')
538 self
.assertEqual(net
.get('shared'), False)
540 def test_010_get_network_list_by_name(self
):
541 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
542 self
.__class
__.test_index
,
543 inspect
.currentframe().f_code
.co_name
)
544 self
.__class
__.test_index
+= 1
546 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
548 # find network from list by it's name
549 new_network_list
= test_config
["vim_conn"].get_network_list({'name': network_name
})
550 for list_item
in new_network_list
:
551 if self
.__class
__.network_name
in list_item
.get('name'):
552 self
.assertEqual(network_name
, list_item
.get('name'))
553 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
554 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
556 def test_020_get_network_list_by_id(self
):
557 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
558 self
.__class
__.test_index
,
559 inspect
.currentframe().f_code
.co_name
)
560 self
.__class
__.test_index
+= 1
562 # find network from list by it's id
563 new_network_list
= test_config
["vim_conn"].get_network_list({'id':self
.__class
__.network_id
})
564 for list_item
in new_network_list
:
565 if self
.__class
__.network_id
in list_item
.get('id'):
566 self
.assertEqual(self
.__class
__.network_id
, list_item
.get('id'))
567 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
568 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
570 def test_030_get_network_list_by_shared(self
):
572 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
573 self
.__class
__.test_index
,
574 inspect
.currentframe().f_code
.co_name
)
575 self
.__class
__.test_index
+= 1
577 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
578 # find network from list by it's shared value
579 new_network_list
= test_config
["vim_conn"].get_network_list({'shared':Shared
,
580 'name':network_name
})
581 for list_item
in new_network_list
:
582 if list_item
.get('shared') == Shared
:
583 self
.assertEqual(list_item
.get('shared'), Shared
)
584 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
585 self
.assertEqual(network_name
, list_item
.get('name'))
587 def test_040_get_network_list_by_tenant_id(self
):
588 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
589 self
.__class
__.test_index
,
590 inspect
.currentframe().f_code
.co_name
)
591 self
.__class
__.test_index
+= 1
593 tenant_list
= test_config
["vim_conn"].get_tenant_list()
594 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
596 for tenant_item
in tenant_list
:
597 if test_config
['tenant'] == tenant_item
.get('name'):
598 # find network from list by it's tenant id
599 tenant_id
= tenant_item
.get('id')
600 new_network_list
= test_config
["vim_conn"].get_network_list({'tenant_id':tenant_id
,
601 'name':network_name
})
602 for list_item
in new_network_list
:
603 self
.assertEqual(tenant_id
, list_item
.get('tenant_id'))
604 self
.assertEqual(network_name
, list_item
.get('name'))
605 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
606 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
608 def test_050_get_network_list_by_status(self
):
609 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
610 self
.__class
__.test_index
,
611 inspect
.currentframe().f_code
.co_name
)
612 self
.__class
__.test_index
+= 1
615 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
617 # find network from list by it's status
618 new_network_list
= test_config
["vim_conn"].get_network_list({'status':status
,
619 'name': network_name
})
620 for list_item
in new_network_list
:
621 self
.assertIn(self
.__class
__.network_name
, list_item
.get('name'))
622 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
623 self
.assertEqual(list_item
.get('status'), status
)
625 def test_060_get_network_list_by_negative(self
):
626 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
627 self
.__class
__.test_index
,
628 inspect
.currentframe().f_code
.co_name
)
629 self
.__class
__.test_index
+= 1
631 network_list
= test_config
["vim_conn"].get_network_list({'name': 'unknown_name'})
632 self
.assertEqual(network_list
, [])
634 class test_vimconn_get_network(test_base
):
640 # def setUpClass(cls):
641 # logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
644 # def tearDownClass(cls):
645 # test_config["test_number"] += 1
648 # creating new network
649 self
.__class
__.network_name
= _get_random_string(20)
650 self
.__class
__.net_type
= 'bridge'
651 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
652 net_type
=self
.__class
__.net_type
)
653 self
.__class
__.network_id
= network
654 logger
.debug("{}".format(network
))
657 test_base
.tearDown(self
)
658 # exec_info = sys.exc_info()
659 # if exec_info == (None, None, None):
660 # logger.info(self.__class__.test_text+" -> TEST OK")
662 # logger.warning(self.__class__.test_text+" -> TEST NOK")
663 # logger.critical("Traceback error",exc_info=True)
665 # Deleting created network
666 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
668 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
670 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
672 def test_000_get_network(self
):
673 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
674 self
.__class
__.test_index
,
675 inspect
.currentframe().f_code
.co_name
)
676 self
.__class
__.test_index
+= 1
678 network_info
= test_config
["vim_conn"].get_network(self
.__class
__.network_id
)
679 self
.assertEqual(network_info
.get('status'), 'ACTIVE')
680 self
.assertIn(self
.__class
__.network_name
, network_info
.get('name'))
681 self
.assertEqual(network_info
.get('type'), self
.__class
__.net_type
)
682 self
.assertEqual(network_info
.get('id'), self
.__class
__.network_id
)
684 def test_010_get_network_negative(self
):
685 Non_exist_id
= str(uuid
.uuid4())
686 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
687 self
.__class
__.test_index
,
688 inspect
.currentframe().f_code
.co_name
)
689 self
.__class
__.test_index
+= 1
691 network_info
= test_config
["vim_conn"].get_network(Non_exist_id
)
692 self
.assertEqual(network_info
, {})
694 class test_vimconn_delete_network(test_base
):
700 # def setUpClass(cls):
701 # logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
704 # def tearDownClass(cls):
705 # test_config["test_number"] += 1
707 # def tearDown(self):
708 # exec_info = sys.exc_info()
709 # if exec_info == (None, None, None):
710 # logger.info(self.__class__.test_text+" -> TEST OK")
712 # logger.warning(self.__class__.test_text+" -> TEST NOK")
713 # logger.critical("Traceback error",exc_info=True)
715 def test_000_delete_network(self
):
717 self
.__class
__.network_name
= _get_random_string(20)
718 self
.__class
__.net_type
= 'bridge'
719 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
720 net_type
=self
.__class
__.net_type
)
721 self
.__class
__.network_id
= network
722 logger
.debug("{}".format(network
))
724 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
725 self
.__class
__.test_index
,
726 inspect
.currentframe().f_code
.co_name
)
727 self
.__class
__.test_index
+= 1
729 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
731 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
733 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
735 # after deleting network we check in network list
736 network_list
= test_config
["vim_conn"].get_network_list({ 'id':self
.__class
__.network_id
})
737 self
.assertEqual(network_list
, [])
739 def test_010_delete_network_negative(self
):
740 Non_exist_id
= str(uuid
.uuid4())
742 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
743 self
.__class
__.test_index
,
744 inspect
.currentframe().f_code
.co_name
)
745 self
.__class
__.test_index
+= 1
747 with self
.assertRaises(Exception) as context
:
748 test_config
["vim_conn"].delete_network(Non_exist_id
)
750 self
.assertEqual((context
.exception
).http_code
, 400)
752 class test_vimconn_get_flavor(test_base
):
757 # def setUpClass(cls):
758 # logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
761 # def tearDownClass(cls):
762 # test_config["test_number"] += 1
764 # def tearDown(self):
765 # exec_info = sys.exc_info()
766 # if exec_info == (None, None, None):
767 # logger.info(self.__class__.test_text+" -> TEST OK")
769 # logger.warning(self.__class__.test_text+" -> TEST NOK")
770 # logger.critical("Traceback error",exc_info=True)
772 def test_000_get_flavor(self
):
773 test_directory_content
= os
.listdir(test_config
["test_directory"])
775 for dir_name
in test_directory_content
:
776 if dir_name
== 'simple_linux':
777 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
778 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
781 for vnfd
in vnfd_files
:
782 with
open(vnfd
, 'r') as stream
:
783 vnf_descriptor
= yaml
.load(stream
)
785 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
786 for item
in vnfc_list
:
787 if 'ram' in item
and 'vcpus' in item
and 'disk' in item
:
789 vcpus
= item
['vcpus']
792 flavor_data
= {'ram': ram
,
797 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
798 self
.__class
__.test_index
,
799 inspect
.currentframe().f_code
.co_name
)
800 self
.__class
__.test_index
+= 1
802 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
804 result
= test_config
["vim_conn"].get_flavor(flavor_id
)
805 self
.assertEqual(ram
, result
['ram'])
806 self
.assertEqual(vcpus
, result
['vcpus'])
807 self
.assertEqual(disk
, result
['disk'])
810 result
= test_config
["vim_conn"].delete_flavor(flavor_id
)
812 logger
.info("Flavor id {} sucessfully deleted".format(result
))
814 logger
.info("Failed to delete flavor id {}".format(result
))
816 def test_010_get_flavor_negative(self
):
817 Non_exist_flavor_id
= str(uuid
.uuid4())
819 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
820 self
.__class
__.test_index
,
821 inspect
.currentframe().f_code
.co_name
)
822 self
.__class
__.test_index
+= 1
824 with self
.assertRaises(Exception) as context
:
825 test_config
["vim_conn"].get_flavor(Non_exist_flavor_id
)
827 self
.assertEqual((context
.exception
).http_code
, 404)
829 class test_vimconn_new_flavor(test_base
):
832 def test_000_new_flavor(self
):
833 flavor_data
= {'ram': 1024, 'vpcus': 1, 'disk': 10}
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
841 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
842 self
.assertEqual(type(flavor_id
),str)
843 self
.assertIsInstance(uuid
.UUID(flavor_id
),uuid
.UUID
)
845 def test_010_delete_flavor(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
852 result
= test_config
["vim_conn"].delete_flavor(self
.__class
__.flavor_id
)
854 logger
.info("Flavor id {} sucessfully deleted".format(result
))
856 logger
.error("Failed to delete flavor id {}".format(result
))
857 raise Exception ("Failed to delete created flavor")
859 def test_020_new_flavor_negative(self
):
860 Invalid_flavor_data
= {'ram': '1024', 'vcpus': 2.0, 'disk': 2.0}
862 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
863 self
.__class
__.test_index
,
864 inspect
.currentframe().f_code
.co_name
)
865 self
.__class
__.test_index
+= 1
867 with self
.assertRaises(Exception) as context
:
868 test_config
["vim_conn"].new_flavor(Invalid_flavor_data
)
870 self
.assertEqual((context
.exception
).http_code
, 400)
872 def test_030_delete_flavor_negative(self
):
873 Non_exist_flavor_id
= str(uuid
.uuid4())
875 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
876 self
.__class
__.test_index
,
877 inspect
.currentframe().f_code
.co_name
)
878 self
.__class
__.test_index
+= 1
880 with self
.assertRaises(Exception) as context
:
881 test_config
["vim_conn"].delete_flavor(Non_exist_flavor_id
)
883 self
.assertEqual((context
.exception
).http_code
, 404)
885 class test_vimconn_new_image(test_base
):
887 def test_000_new_image(self
):
888 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
889 self
.__class
__.test_index
,
890 inspect
.currentframe().f_code
.co_name
)
891 self
.__class
__.test_index
+= 1
893 image_path
= test_config
['image_path']
895 image_id
= test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : image_path
})
896 self
.assertEqual(type(image_id
),str)
897 self
.assertIsInstance(uuid
.UUID(image_id
),uuid
.UUID
)
899 self
.skipTest("Skipping test as image file not present at RO container")
901 def test_010_new_image_negative(self
):
902 Non_exist_image_path
= '/temp1/cirros.ovf'
904 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
905 self
.__class
__.test_index
,
906 inspect
.currentframe().f_code
.co_name
)
907 self
.__class
__.test_index
+= 1
909 with self
.assertRaises(Exception) as context
:
910 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
912 self
.assertEqual((context
.exception
).http_code
, 400)
914 class test_vimconn_get_image_id_from_path(test_base
):
916 def test_000_get_image_id_from_path(self
):
917 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
918 self
.__class
__.test_index
,
919 inspect
.currentframe().f_code
.co_name
)
920 self
.__class
__.test_index
+= 1
922 image_path
= test_config
['image_path']
924 image_id
= test_config
["vim_conn"].get_image_id_from_path( image_path
)
925 self
.assertEqual(type(image_id
),str)
927 self
.skipTest("Skipping test as image file not present at RO container")
929 def test_010_get_image_id_from_path_negative(self
):
930 Non_exist_image_path
= '/temp1/cirros.ovf'
932 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
933 self
.__class
__.test_index
,
934 inspect
.currentframe().f_code
.co_name
)
935 self
.__class
__.test_index
+= 1
937 with self
.assertRaises(Exception) as context
:
938 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
940 self
.assertEqual((context
.exception
).http_code
, 400)
942 class test_vimconn_get_image_list(test_base
):
946 def test_000_get_image_list(self
):
947 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
948 self
.__class
__.test_index
,
949 inspect
.currentframe().f_code
.co_name
)
950 self
.__class
__.test_index
+= 1
951 image_list
= test_config
["vim_conn"].get_image_list()
953 for item
in image_list
:
955 self
.__class
__.image_name
= item
['name']
956 self
.__class
__.image_id
= item
['id']
957 self
.assertEqual(type(self
.__class
__.image_name
),str)
958 self
.assertEqual(type(self
.__class
__.image_id
),str)
960 def test_010_get_image_list_by_name(self
):
961 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
962 self
.__class
__.test_index
,
963 inspect
.currentframe().f_code
.co_name
)
964 self
.__class
__.test_index
+= 1
966 image_list
= test_config
["vim_conn"].get_image_list({'name': self
.__class
__.image_name
})
968 for item
in image_list
:
969 self
.assertEqual(type(item
['id']), str)
970 self
.assertEqual(item
['id'], self
.__class
__.image_id
)
971 self
.assertEqual(type(item
['name']), str)
972 self
.assertEqual(item
['name'], self
.__class
__.image_name
)
974 def test_020_get_image_list_by_id(self
):
975 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
976 self
.__class
__.test_index
,
977 inspect
.currentframe().f_code
.co_name
)
978 self
.__class
__.test_index
+= 1
980 filter_image_list
= test_config
["vim_conn"].get_image_list({'id': self
.__class
__.image_id
})
982 for item1
in filter_image_list
:
983 self
.assertEqual(type(item1
.get('id')), str)
984 self
.assertEqual(item1
.get('id'), self
.__class
__.image_id
)
985 self
.assertEqual(type(item1
.get('name')), str)
986 self
.assertEqual(item1
.get('name'), self
.__class
__.image_name
)
988 def test_030_get_image_list_negative(self
):
989 Non_exist_image_id
= uuid
.uuid4()
990 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
991 self
.__class
__.test_index
,
992 inspect
.currentframe().f_code
.co_name
)
993 self
.__class
__.test_index
+= 1
994 image_list
= test_config
["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id
})
996 self
.assertIsNotNone(image_list
, None)
997 self
.assertEqual(image_list
, [])
999 class test_vimconn_new_vminstance(test_base
):
1007 self
.__class
__.network_name
= _get_random_string(20)
1008 self
.__class
__.net_type
= 'bridge'
1010 self
.__class
__.network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
1011 net_type
=self
.__class
__.net_type
)
1014 test_base
.tearDown(self
)
1015 # Deleting created network
1016 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
1018 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
1020 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
1022 def test_000_new_vminstance(self
):
1023 vpci
= "0000:00:11.0"
1026 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1029 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1031 # find image name and image id
1032 if test_config
['image_name']:
1033 image_list
= test_config
['vim_conn'].get_image_list({'name': test_config
['image_name']})
1034 if len(image_list
) == 0:
1035 raise Exception("Image {} is not found at VIM".format(test_config
['image_name']))
1037 self
.__class
__.image_id
= image_list
[0]['id']
1039 image_list
= test_config
['vim_conn'].get_image_list()
1040 if len(image_list
) == 0:
1041 raise Exception("Not found any image at VIM")
1043 self
.__class
__.image_id
= image_list
[0]['id']
1045 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1046 self
.__class
__.test_index
,
1047 inspect
.currentframe().f_code
.co_name
)
1048 self
.__class
__.test_index
+= 1
1050 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
}]
1052 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
)
1054 self
.assertEqual(type(instance_id
),str)
1055 # Deleting created vm instance
1056 logger
.info("Deleting created vm intance")
1057 test_config
["vim_conn"].delete_vminstance(instance_id
)
1060 def test_010_new_vminstance_by_model(self
):
1061 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1062 model_name
= 'e1000'
1066 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1068 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1069 self
.__class
__.test_index
,
1070 inspect
.currentframe().f_code
.co_name
)
1071 self
.__class
__.test_index
+= 1
1073 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
}]
1075 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1076 flavor_id
=flavor_id
,
1078 self
.assertEqual(type(instance_id
),str)
1079 # Deleting created vm instance
1080 logger
.info("Deleting created vm intance")
1081 test_config
["vim_conn"].delete_vminstance(instance_id
)
1084 def test_020_new_vminstance_by_net_use(self
):
1085 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1090 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1092 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1093 self
.__class
__.test_index
,
1094 inspect
.currentframe().f_code
.co_name
)
1095 self
.__class
__.test_index
+= 1
1097 net_list
= [{'use': net_use
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1099 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1100 flavor_id
=flavor_id
,
1102 self
.assertEqual(type(instance_id
),str)
1103 # Deleting created vm instance
1104 logger
.info("Deleting created vm intance")
1105 test_config
["vim_conn"].delete_vminstance(instance_id
)
1108 def test_030_new_vminstance_by_net_type(self
):
1109 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1114 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1116 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1117 self
.__class
__.test_index
,
1118 inspect
.currentframe().f_code
.co_name
)
1119 self
.__class
__.test_index
+= 1
1121 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': _type
, 'net_id': self
.__class
__.network_id
}]
1123 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1124 flavor_id
=flavor_id
,
1126 self
.assertEqual(type(instance_id
),str)
1127 # Deleting created vm instance
1128 logger
.info("Deleting created vm intance")
1129 test_config
["vim_conn"].delete_vminstance(instance_id
)
1132 def test_040_new_vminstance_by_cloud_config(self
):
1133 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1135 user_name
= 'test_user'
1137 key_pairs
= ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com']
1139 users_data
= [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com'], 'name': user_name
}]
1141 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
}
1144 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1146 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1147 self
.__class
__.test_index
,
1148 inspect
.currentframe().f_code
.co_name
)
1149 self
.__class
__.test_index
+= 1
1151 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1153 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Cloud_vm', image_id
=self
.__class
__.image_id
,
1154 flavor_id
=flavor_id
,
1156 cloud_config
=cloud_data
)
1157 self
.assertEqual(type(instance_id
),str)
1158 # Deleting created vm instance
1159 logger
.info("Deleting created vm intance")
1160 test_config
["vim_conn"].delete_vminstance(instance_id
)
1163 def test_050_new_vminstance_by_disk_list(self
):
1164 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1167 device_data
= [{'image_id': self
.__class
__.image_id
, 'size': '5'}]
1170 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1172 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1173 self
.__class
__.test_index
,
1174 inspect
.currentframe().f_code
.co_name
)
1175 self
.__class
__.test_index
+= 1
1177 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1179 instance_id
= test_config
["vim_conn"].new_vminstance(name
='VM_test1', image_id
=self
.__class
__.image_id
,
1180 flavor_id
=flavor_id
,
1182 disk_list
=device_data
)
1183 self
.assertEqual(type(instance_id
),str)
1184 # Deleting created vm instance
1185 logger
.info("Deleting created vm intance")
1186 test_config
["vim_conn"].delete_vminstance(instance_id
)
1189 def test_060_new_vminstance_negative(self
):
1190 unknown_flavor_id
= str(uuid
.uuid4())
1191 unknown_image_id
= str(uuid
.uuid4())
1194 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1195 self
.__class
__.test_index
,
1196 inspect
.currentframe().f_code
.co_name
)
1197 self
.__class
__.test_index
+= 1
1199 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1201 with self
.assertRaises(Exception) as context
:
1202 test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=unknown_image_id
,
1203 flavor_id
=unknown_flavor_id
,
1205 self
.assertEqual((context
.exception
).http_code
, 404)
1209 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
1210 scenario based tests.
1212 class descriptor_based_scenario_test(test_base
):
1214 scenario_test_path
= None
1215 scenario_uuid
= None
1216 instance_scenario_uuid
= None
1220 def setUpClass(cls
):
1222 cls
.to_delete_list
= []
1223 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
1224 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
1227 def tearDownClass(cls
):
1228 test_config
["test_number"] += 1
1230 def test_000_load_scenario(self
):
1231 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1232 inspect
.currentframe().f_code
.co_name
,
1233 test_config
["test_folder"])
1234 self
.__class
__.test_index
+= 1
1235 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
1236 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
1237 if len(vnfd_files
) == 0 or len(scenario_file
) > 1:
1238 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
1239 test_config
["test_folder"]))
1242 for vnfd
in vnfd_files
:
1243 with
open(vnfd
, 'r') as stream
:
1244 vnf_descriptor
= yaml
.load(stream
)
1246 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
1247 for vnfc
in vnfc_list
:
1248 vnfc
['image name'] = test_config
["image_name"]
1249 devices
= vnfc
.get('devices',[])
1250 for device
in devices
:
1251 if device
['type'] == 'disk' and 'image name' in device
:
1252 device
['image name'] = test_config
["image_name"]
1254 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
1255 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
)
1257 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
1258 "params": {"uuid": vnf
['vnf']['uuid']}})
1260 #load the scenario definition
1261 with
open(scenario_file
[0], 'r') as stream
:
1262 scenario_descriptor
= yaml
.load(stream
)
1263 networks
= scenario_descriptor
['scenario']['networks']
1264 networks
[test_config
["mgmt_net"]] = networks
.pop('mgmt')
1265 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
1266 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
1267 logger
.debug(scenario
)
1268 self
.__class
__.to_delete_list
.insert(0,{"item": "scenario", "function": test_config
["client"].delete_scenario
,
1269 "params":{"uuid": scenario
['scenario']['uuid']} })
1270 self
.__class
__.scenario_uuid
= scenario
['scenario']['uuid']
1272 def test_010_instantiate_scenario(self
):
1273 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1274 inspect
.currentframe().f_code
.co_name
,
1275 test_config
["test_folder"])
1276 self
.__class
__.test_index
+= 1
1278 instance
= test_config
["client"].create_instance(scenario_id
=self
.__class
__.scenario_uuid
,
1279 name
=self
.__class
__.test_text
)
1280 self
.__class
__.instance_scenario_uuid
= instance
['uuid']
1281 logger
.debug(instance
)
1282 self
.__class
__.to_delete_list
.insert(0, {"item": "instance", "function": test_config
["client"].delete_instance
,
1283 "params": {"uuid": instance
['uuid']}})
1285 def test_020_check_deployent(self
):
1286 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1287 inspect
.currentframe().f_code
.co_name
,
1288 test_config
["test_folder"])
1289 self
.__class
__.test_index
+= 1
1291 if test_config
["manual"]:
1292 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1295 keep_waiting
= test_config
["timeout"]
1296 instance_active
= False
1298 result
= check_instance_scenario_active(self
.__class
__.instance_scenario_uuid
)
1301 elif 'ERROR' in result
[1]:
1302 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
1304 raise Exception(msg
)
1306 if keep_waiting
>= 5:
1309 elif keep_waiting
> 0:
1310 time
.sleep(keep_waiting
)
1313 msg
= 'Timeout reached while waiting instance scenario to get active'
1315 raise Exception(msg
)
1317 def test_030_clean_deployment(self
):
1318 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1319 inspect
.currentframe().f_code
.co_name
,
1320 test_config
["test_folder"])
1321 self
.__class
__.test_index
+= 1
1322 #At the moment if you delete an scenario right after creating it, in openstack datacenters
1323 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1325 for item
in self
.__class
__.to_delete_list
:
1326 response
= item
["function"](**item
["params"])
1327 logger
.debug(response
)
1330 def _get_random_string(maxLength
):
1331 '''generates a string with random characters string.letters and string.digits
1332 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1336 minLength
= min_string
- len(prefix
)
1337 if maxLength
< min_string
: maxLength
= min_string
1338 maxLength
-= len(prefix
)
1339 length
= random
.randint(minLength
,maxLength
)
1340 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
1343 def test_vimconnector(args
):
1345 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1346 test_config
['vimtype'] = args
.vimtype
1347 if args
.vimtype
== "vmware":
1348 import vimconn_vmware
as vim
1350 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1352 tenant_name
= args
.tenant_name
1353 test_config
['tenant'] = tenant_name
1354 config_params
= json
.loads(args
.config_param
)
1355 org_name
= config_params
.get('orgname')
1356 org_user
= config_params
.get('user')
1357 org_passwd
= config_params
.get('passwd')
1358 vim_url
= args
.endpoint_url
1359 test_config
['image_path'] = args
.image_path
1360 test_config
['image_name'] = args
.image_name
1362 # vmware connector obj
1363 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
)
1365 elif args
.vimtype
== "aws":
1366 import vimconn_aws
as vim
1367 elif args
.vimtype
== "openstack":
1368 import vimconn_openstack
as vim
1369 elif args
.vimtype
== "openvim":
1370 import vimconn_openvim
as vim
1372 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
1376 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1377 # If only want to obtain a tests list print it and exit
1380 for cls
in clsmembers
:
1381 if cls
[0].startswith('test_vimconnector'):
1382 tests_names
.append(cls
[0])
1384 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
1389 # Create the list of tests to be run
1390 code_based_tests
= []
1392 for test
in args
.tests
:
1393 for t
in test
.split(','):
1394 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1395 if len(matches_code_based_tests
) > 0:
1396 code_based_tests
.append(matches_code_based_tests
[0][1])
1398 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1400 if not code_based_tests
:
1402 for cls
in clsmembers
:
1403 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1404 if cls
[0].startswith('test_vimconnector'):
1405 code_based_tests
.append(cls
[1])
1407 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1409 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1410 # This is handled in the tests using logging.
1411 stream
= open('/dev/null', 'w')
1413 # Run code based tests
1414 basic_tests_suite
= unittest
.TestSuite()
1415 for test
in code_based_tests
:
1416 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1417 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1418 executed
+= result
.testsRun
1419 failed
+= len(result
.failures
) + len(result
.errors
)
1420 if failfast
and failed
:
1422 if len(result
.failures
) > 0:
1423 logger
.debug("failures : {}".format(result
.failures
))
1424 if len(result
.errors
) > 0:
1425 logger
.debug("errors : {}".format(result
.errors
))
1426 return executed
, failed
1431 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1432 import openmanoclient
1435 test_config
["client"] = openmanoclient
.openmanoclient(
1436 endpoint_url
=args
.endpoint_url
,
1437 tenant_name
=args
.tenant_name
,
1438 datacenter_name
=args
.datacenter
,
1439 debug
=args
.debug
, logger
=test_config
["logger_name"])
1440 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1441 # If only want to obtain a tests list print it and exit
1444 for cls
in clsmembers
:
1445 if cls
[0].startswith('test_VIM'):
1446 tests_names
.append(cls
[0])
1448 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
1449 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1450 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1455 # Create the list of tests to be run
1456 code_based_tests
= []
1458 for test
in args
.tests
:
1459 for t
in test
.split(','):
1460 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1461 if len(matches_code_based_tests
) > 0:
1462 code_based_tests
.append(matches_code_based_tests
[0][1])
1464 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1466 if not code_based_tests
:
1468 for cls
in clsmembers
:
1469 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1470 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
1471 code_based_tests
.append(cls
[1])
1473 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1475 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1476 # This is handled in the tests using logging.
1477 stream
= open('/dev/null', 'w')
1479 # Run code based tests
1480 basic_tests_suite
= unittest
.TestSuite()
1481 for test
in code_based_tests
:
1482 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1483 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1484 executed
+= result
.testsRun
1485 failed
+= len(result
.failures
) + len(result
.errors
)
1486 if failfast
and failed
:
1488 if len(result
.failures
) > 0:
1489 logger
.debug("failures : {}".format(result
.failures
))
1490 if len(result
.errors
) > 0:
1491 logger
.debug("errors : {}".format(result
.errors
))
1492 return executed
, failed
1495 def test_deploy(args
):
1497 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1498 import openmanoclient
1501 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1502 test_config
["image_name"] = args
.image_name
1503 test_config
["mgmt_net"] = args
.mgmt_net
1504 test_config
["manual"] = args
.manual
1505 test_directory_content
= os
.listdir(test_config
["test_directory"])
1506 # If only want to obtain a tests list print it and exit
1508 msg
= "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
1513 descriptor_based_tests
= []
1514 # Create the list of tests to be run
1515 code_based_tests
= []
1517 for test
in args
.tests
:
1518 for t
in test
.split(','):
1519 if t
in test_directory_content
:
1520 descriptor_based_tests
.append(t
)
1522 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1524 if not descriptor_based_tests
:
1526 descriptor_based_tests
= test_directory_content
1528 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1530 # import openmanoclient from relative path
1531 test_config
["client"] = openmanoclient
.openmanoclient(
1532 endpoint_url
=args
.endpoint_url
,
1533 tenant_name
=args
.tenant_name
,
1534 datacenter_name
=args
.datacenter
,
1535 debug
=args
.debug
, logger
=test_config
["logger_name"])
1537 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1538 # This is handled in the tests using logging.
1539 stream
= open('/dev/null', 'w')
1540 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1541 for test
in descriptor_based_tests
:
1542 test_config
["test_folder"] = test
1543 test_suite
= unittest
.TestSuite()
1544 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
1545 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
1546 executed
+= result
.testsRun
1547 failed
+= len(result
.failures
) + len(result
.errors
)
1548 if failfast
and failed
:
1550 if len(result
.failures
) > 0:
1551 logger
.debug("failures : {}".format(result
.failures
))
1552 if len(result
.errors
) > 0:
1553 logger
.debug("errors : {}".format(result
.errors
))
1555 return executed
, failed
1557 if __name__
=="__main__":
1559 parser
= ArgumentParser(description
='Test RO module')
1560 parser
.add_argument('-v','--version', action
='version', help="Show current version",
1561 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
1564 parent_parser
= ArgumentParser(add_help
=False)
1565 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1566 dest
='failfast', action
="store_true", default
=False)
1567 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1568 dest
='failed', action
="store_true", default
=False)
1569 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
1570 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
1572 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
1573 help='Set the logger file. By default '+default_logger_file
)
1574 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
1575 help="Set the openmano tenant to use for the test. By default 'osm'")
1576 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
1577 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1578 dest
='timeout', type=int, default
=300)
1579 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
1581 subparsers
= parser
.add_subparsers(help='test sets')
1583 # Deployment test set
1584 # -------------------
1585 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
1586 help="test deployment using descriptors at RO_test folder ")
1587 deploy_parser
.set_defaults(func
=test_deploy
)
1589 # Mandatory arguments
1590 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
1591 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1592 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
1593 help='Image name available at datacenter used for the tests')
1594 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
1595 help='Set the vim management network to use for tests')
1597 # Optional arguments
1598 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
1599 help='Pause execution once deployed to allow manual checking of the '
1600 'deployed instance scenario')
1601 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1602 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1605 # -------------------
1606 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
1607 vimconn_parser
.set_defaults(func
=test_vimconnector
)
1608 # Mandatory arguments
1609 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1610 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
1611 help='Set the vimconnector type to test')
1612 mandatory_arguments
.add_argument('-c', '--config', dest
='config_param', required
=True,
1613 help='Set the vimconnector specific config parameters in dictionary format')
1614 mandatory_arguments
.add_argument('-u', '--url', dest
='endpoint_url',required
=True, help="Set the vim connector url or Host IP")
1615 # Optional arguments
1616 vimconn_parser
.add_argument('-i', '--image-path', dest
='image_path', help="Provide image path present at RO container")
1617 vimconn_parser
.add_argument('-n', '--image-name', dest
='image_name', help="Provide image name for test")
1618 # TODO add optional arguments for vimconn tests
1619 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1621 # Datacenter test set
1622 # -------------------
1623 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
1624 vimconn_parser
.set_defaults(func
=test_vim
)
1626 # Mandatory arguments
1627 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1628 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1630 # Optional arguments
1631 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1632 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1634 argcomplete
.autocomplete(parser
)
1635 args
= parser
.parse_args()
1639 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1640 logger_level
= 'INFO'
1642 logger_level
= 'DEBUG'
1644 logger_level
= 'WARNING'
1645 logger_name
= os
.path
.basename(__file__
)
1646 test_config
["logger_name"] = logger_name
1647 logger
= logging
.getLogger(logger_name
)
1648 logger
.setLevel(logger_level
)
1649 failfast
= args
.failfast
1651 # Configure a logging handler to store in a logging file
1652 if args
.logger_file
:
1653 fileHandler
= logging
.FileHandler(args
.logger_file
)
1654 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1655 fileHandler
.setFormatter(formatter_fileHandler
)
1656 logger
.addHandler(fileHandler
)
1658 # Configure a handler to print to stdout
1659 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
1660 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1661 consoleHandler
.setFormatter(formatter_consoleHandler
)
1662 logger
.addHandler(consoleHandler
)
1664 logger
.debug('Program started with the following arguments: ' + str(args
))
1666 # set test config parameters
1667 test_config
["timeout"] = args
.timeout
1668 test_config
["test_number"] = 1
1670 executed
, failed
= args
.func(args
)
1673 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
1674 sys
.exit(1 if failed
else 0)