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 def test_050_refresh_nets_status(self
):
489 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
490 self
.__class
__.test_index
,
491 inspect
.currentframe().f_code
.co_name
)
492 self
.__class
__.test_index
+= 1
493 # creating new network
494 network_name
= _get_random_string(20)
496 network_id
= test_config
["vim_conn"].new_network(net_name
=network_name
,
499 net_dict
= test_config
["vim_conn"].refresh_nets_status([network_id
])
500 for attr
in net_dict
[network_id
]:
502 self
.assertEqual(net_dict
[network_id
][attr
], 'ACTIVE')
504 # Deleting created network
505 result
= test_config
["vim_conn"].delete_network(network_id
)
507 logger
.info("Network id {} sucessfully deleted".format(network_id
))
509 logger
.info("Failed to delete network id {}".format(network_id
))
511 def test_060_refresh_nets_status_negative(self
):
512 unknown_net_id
= str(uuid
.uuid4())
513 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
514 self
.__class
__.test_index
,
515 inspect
.currentframe().f_code
.co_name
)
516 self
.__class
__.test_index
+= 1
519 net_dict
= test_config
["vim_conn"].refresh_nets_status([unknown_net_id
])
520 self
.assertEqual(net_dict
, {})
522 class test_vimconn_get_network_list(test_base
):
528 # def setUpClass(cls):
529 # logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
532 # def tearDownClass(cls):
533 # test_config["test_number"] += 1
536 # creating new network
537 self
.__class
__.network_name
= _get_random_string(20)
538 self
.__class
__.net_type
= 'bridge'
539 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
540 net_type
=self
.__class
__.net_type
)
541 self
.__class
__.network_id
= network
542 logger
.debug("{}".format(network
))
545 test_base
.tearDown(self
)
546 # exec_info = sys.exc_info()
547 # if exec_info == (None, None, None):
548 # logger.info(self.__class__.test_text+" -> TEST OK")
550 # logger.warning(self.__class__.test_text+" -> TEST NOK")
551 # logger.critical("Traceback error",exc_info=True)
553 # Deleting created network
554 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
556 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
558 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
560 def test_000_get_network_list(self
):
561 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
562 self
.__class
__.test_index
,
563 inspect
.currentframe().f_code
.co_name
)
564 self
.__class
__.test_index
+= 1
566 network_list
= test_config
["vim_conn"].get_network_list()
567 for net
in network_list
:
568 if self
.__class
__.network_name
in net
.get('name'):
569 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
570 self
.assertEqual(net
.get('type'), self
.__class
__.net_type
)
571 self
.assertEqual(net
.get('status'), 'ACTIVE')
572 self
.assertEqual(net
.get('shared'), False)
574 def test_010_get_network_list_by_name(self
):
575 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
576 self
.__class
__.test_index
,
577 inspect
.currentframe().f_code
.co_name
)
578 self
.__class
__.test_index
+= 1
580 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
582 # find network from list by it's name
583 new_network_list
= test_config
["vim_conn"].get_network_list({'name': network_name
})
584 for list_item
in new_network_list
:
585 if self
.__class
__.network_name
in list_item
.get('name'):
586 self
.assertEqual(network_name
, list_item
.get('name'))
587 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
588 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
590 def test_020_get_network_list_by_id(self
):
591 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
592 self
.__class
__.test_index
,
593 inspect
.currentframe().f_code
.co_name
)
594 self
.__class
__.test_index
+= 1
596 # find network from list by it's id
597 new_network_list
= test_config
["vim_conn"].get_network_list({'id':self
.__class
__.network_id
})
598 for list_item
in new_network_list
:
599 if self
.__class
__.network_id
in list_item
.get('id'):
600 self
.assertEqual(self
.__class
__.network_id
, list_item
.get('id'))
601 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
602 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
604 def test_030_get_network_list_by_shared(self
):
606 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
607 self
.__class
__.test_index
,
608 inspect
.currentframe().f_code
.co_name
)
609 self
.__class
__.test_index
+= 1
611 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
612 # find network from list by it's shared value
613 new_network_list
= test_config
["vim_conn"].get_network_list({'shared':Shared
,
614 'name':network_name
})
615 for list_item
in new_network_list
:
616 if list_item
.get('shared') == Shared
:
617 self
.assertEqual(list_item
.get('shared'), Shared
)
618 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
619 self
.assertEqual(network_name
, list_item
.get('name'))
621 def test_040_get_network_list_by_tenant_id(self
):
622 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
623 self
.__class
__.test_index
,
624 inspect
.currentframe().f_code
.co_name
)
625 self
.__class
__.test_index
+= 1
627 tenant_list
= test_config
["vim_conn"].get_tenant_list()
628 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
630 for tenant_item
in tenant_list
:
631 if test_config
['tenant'] == tenant_item
.get('name'):
632 # find network from list by it's tenant id
633 tenant_id
= tenant_item
.get('id')
634 new_network_list
= test_config
["vim_conn"].get_network_list({'tenant_id':tenant_id
,
635 'name':network_name
})
636 for list_item
in new_network_list
:
637 self
.assertEqual(tenant_id
, list_item
.get('tenant_id'))
638 self
.assertEqual(network_name
, list_item
.get('name'))
639 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
640 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
642 def test_050_get_network_list_by_status(self
):
643 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
644 self
.__class
__.test_index
,
645 inspect
.currentframe().f_code
.co_name
)
646 self
.__class
__.test_index
+= 1
649 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
651 # find network from list by it's status
652 new_network_list
= test_config
["vim_conn"].get_network_list({'status':status
,
653 'name': network_name
})
654 for list_item
in new_network_list
:
655 self
.assertIn(self
.__class
__.network_name
, list_item
.get('name'))
656 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
657 self
.assertEqual(list_item
.get('status'), status
)
659 def test_060_get_network_list_by_negative(self
):
660 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
661 self
.__class
__.test_index
,
662 inspect
.currentframe().f_code
.co_name
)
663 self
.__class
__.test_index
+= 1
665 network_list
= test_config
["vim_conn"].get_network_list({'name': 'unknown_name'})
666 self
.assertEqual(network_list
, [])
668 class test_vimconn_get_network(test_base
):
674 # def setUpClass(cls):
675 # logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
678 # def tearDownClass(cls):
679 # test_config["test_number"] += 1
682 # creating new network
683 self
.__class
__.network_name
= _get_random_string(20)
684 self
.__class
__.net_type
= 'bridge'
685 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
686 net_type
=self
.__class
__.net_type
)
687 self
.__class
__.network_id
= network
688 logger
.debug("{}".format(network
))
691 test_base
.tearDown(self
)
692 # exec_info = sys.exc_info()
693 # if exec_info == (None, None, None):
694 # logger.info(self.__class__.test_text+" -> TEST OK")
696 # logger.warning(self.__class__.test_text+" -> TEST NOK")
697 # logger.critical("Traceback error",exc_info=True)
699 # Deleting created network
700 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
702 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
704 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
706 def test_000_get_network(self
):
707 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
708 self
.__class
__.test_index
,
709 inspect
.currentframe().f_code
.co_name
)
710 self
.__class
__.test_index
+= 1
712 network_info
= test_config
["vim_conn"].get_network(self
.__class
__.network_id
)
713 self
.assertEqual(network_info
.get('status'), 'ACTIVE')
714 self
.assertIn(self
.__class
__.network_name
, network_info
.get('name'))
715 self
.assertEqual(network_info
.get('type'), self
.__class
__.net_type
)
716 self
.assertEqual(network_info
.get('id'), self
.__class
__.network_id
)
718 def test_010_get_network_negative(self
):
719 Non_exist_id
= str(uuid
.uuid4())
720 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
721 self
.__class
__.test_index
,
722 inspect
.currentframe().f_code
.co_name
)
723 self
.__class
__.test_index
+= 1
725 network_info
= test_config
["vim_conn"].get_network(Non_exist_id
)
726 self
.assertEqual(network_info
, {})
728 class test_vimconn_delete_network(test_base
):
734 # def setUpClass(cls):
735 # logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
738 # def tearDownClass(cls):
739 # test_config["test_number"] += 1
741 # def tearDown(self):
742 # exec_info = sys.exc_info()
743 # if exec_info == (None, None, None):
744 # logger.info(self.__class__.test_text+" -> TEST OK")
746 # logger.warning(self.__class__.test_text+" -> TEST NOK")
747 # logger.critical("Traceback error",exc_info=True)
749 def test_000_delete_network(self
):
751 self
.__class
__.network_name
= _get_random_string(20)
752 self
.__class
__.net_type
= 'bridge'
753 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
754 net_type
=self
.__class
__.net_type
)
755 self
.__class
__.network_id
= network
756 logger
.debug("{}".format(network
))
758 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
759 self
.__class
__.test_index
,
760 inspect
.currentframe().f_code
.co_name
)
761 self
.__class
__.test_index
+= 1
763 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
765 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
767 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
769 # after deleting network we check in network list
770 network_list
= test_config
["vim_conn"].get_network_list({ 'id':self
.__class
__.network_id
})
771 self
.assertEqual(network_list
, [])
773 def test_010_delete_network_negative(self
):
774 Non_exist_id
= str(uuid
.uuid4())
776 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
777 self
.__class
__.test_index
,
778 inspect
.currentframe().f_code
.co_name
)
779 self
.__class
__.test_index
+= 1
781 with self
.assertRaises(Exception) as context
:
782 test_config
["vim_conn"].delete_network(Non_exist_id
)
784 self
.assertEqual((context
.exception
).http_code
, 400)
786 class test_vimconn_get_flavor(test_base
):
791 # def setUpClass(cls):
792 # logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
795 # def tearDownClass(cls):
796 # test_config["test_number"] += 1
798 # def tearDown(self):
799 # exec_info = sys.exc_info()
800 # if exec_info == (None, None, None):
801 # logger.info(self.__class__.test_text+" -> TEST OK")
803 # logger.warning(self.__class__.test_text+" -> TEST NOK")
804 # logger.critical("Traceback error",exc_info=True)
806 def test_000_get_flavor(self
):
807 test_directory_content
= os
.listdir(test_config
["test_directory"])
809 for dir_name
in test_directory_content
:
810 if dir_name
== 'simple_linux':
811 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
812 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
815 for vnfd
in vnfd_files
:
816 with
open(vnfd
, 'r') as stream
:
817 vnf_descriptor
= yaml
.load(stream
)
819 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
820 for item
in vnfc_list
:
821 if 'ram' in item
and 'vcpus' in item
and 'disk' in item
:
823 vcpus
= item
['vcpus']
826 flavor_data
= {'ram': ram
,
831 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
832 self
.__class
__.test_index
,
833 inspect
.currentframe().f_code
.co_name
)
834 self
.__class
__.test_index
+= 1
836 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
838 result
= test_config
["vim_conn"].get_flavor(flavor_id
)
839 self
.assertEqual(ram
, result
['ram'])
840 self
.assertEqual(vcpus
, result
['vcpus'])
841 self
.assertEqual(disk
, result
['disk'])
844 result
= test_config
["vim_conn"].delete_flavor(flavor_id
)
846 logger
.info("Flavor id {} sucessfully deleted".format(result
))
848 logger
.info("Failed to delete flavor id {}".format(result
))
850 def test_010_get_flavor_negative(self
):
851 Non_exist_flavor_id
= str(uuid
.uuid4())
853 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
854 self
.__class
__.test_index
,
855 inspect
.currentframe().f_code
.co_name
)
856 self
.__class
__.test_index
+= 1
858 with self
.assertRaises(Exception) as context
:
859 test_config
["vim_conn"].get_flavor(Non_exist_flavor_id
)
861 self
.assertEqual((context
.exception
).http_code
, 404)
863 class test_vimconn_new_flavor(test_base
):
866 def test_000_new_flavor(self
):
867 flavor_data
= {'ram': 1024, 'vpcus': 1, 'disk': 10}
869 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
870 self
.__class
__.test_index
,
871 inspect
.currentframe().f_code
.co_name
)
872 self
.__class
__.test_index
+= 1
875 self
.__class
__.flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
876 self
.assertEqual(type(self
.__class
__.flavor_id
),str)
877 self
.assertIsInstance(uuid
.UUID(self
.__class
__.flavor_id
),uuid
.UUID
)
879 def test_010_delete_flavor(self
):
880 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
881 self
.__class
__.test_index
,
882 inspect
.currentframe().f_code
.co_name
)
883 self
.__class
__.test_index
+= 1
886 result
= test_config
["vim_conn"].delete_flavor(self
.__class
__.flavor_id
)
888 logger
.info("Flavor id {} sucessfully deleted".format(result
))
890 logger
.error("Failed to delete flavor id {}".format(result
))
891 raise Exception ("Failed to delete created flavor")
893 def test_020_new_flavor_negative(self
):
894 Invalid_flavor_data
= {'ram': '1024', 'vcpus': 2.0, 'disk': 2.0}
896 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
897 self
.__class
__.test_index
,
898 inspect
.currentframe().f_code
.co_name
)
899 self
.__class
__.test_index
+= 1
901 with self
.assertRaises(Exception) as context
:
902 test_config
["vim_conn"].new_flavor(Invalid_flavor_data
)
904 self
.assertEqual((context
.exception
).http_code
, 400)
906 def test_030_delete_flavor_negative(self
):
907 Non_exist_flavor_id
= str(uuid
.uuid4())
909 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
910 self
.__class
__.test_index
,
911 inspect
.currentframe().f_code
.co_name
)
912 self
.__class
__.test_index
+= 1
914 with self
.assertRaises(Exception) as context
:
915 test_config
["vim_conn"].delete_flavor(Non_exist_flavor_id
)
917 self
.assertEqual((context
.exception
).http_code
, 404)
919 class test_vimconn_new_image(test_base
):
921 def test_000_new_image(self
):
922 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
923 self
.__class
__.test_index
,
924 inspect
.currentframe().f_code
.co_name
)
925 self
.__class
__.test_index
+= 1
927 image_path
= test_config
['image_path']
929 image_id
= test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : image_path
})
930 self
.assertEqual(type(image_id
),str)
931 self
.assertIsInstance(uuid
.UUID(image_id
),uuid
.UUID
)
933 self
.skipTest("Skipping test as image file not present at RO container")
935 def test_010_new_image_negative(self
):
936 Non_exist_image_path
= '/temp1/cirros.ovf'
938 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
939 self
.__class
__.test_index
,
940 inspect
.currentframe().f_code
.co_name
)
941 self
.__class
__.test_index
+= 1
943 with self
.assertRaises(Exception) as context
:
944 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
946 self
.assertEqual((context
.exception
).http_code
, 400)
948 class test_vimconn_get_image_id_from_path(test_base
):
950 def test_000_get_image_id_from_path(self
):
951 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
952 self
.__class
__.test_index
,
953 inspect
.currentframe().f_code
.co_name
)
954 self
.__class
__.test_index
+= 1
956 image_path
= test_config
['image_path']
958 image_id
= test_config
["vim_conn"].get_image_id_from_path( image_path
)
959 self
.assertEqual(type(image_id
),str)
961 self
.skipTest("Skipping test as image file not present at RO container")
963 def test_010_get_image_id_from_path_negative(self
):
964 Non_exist_image_path
= '/temp1/cirros.ovf'
966 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
967 self
.__class
__.test_index
,
968 inspect
.currentframe().f_code
.co_name
)
969 self
.__class
__.test_index
+= 1
971 with self
.assertRaises(Exception) as context
:
972 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
974 self
.assertEqual((context
.exception
).http_code
, 400)
976 class test_vimconn_get_image_list(test_base
):
980 def test_000_get_image_list(self
):
981 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
982 self
.__class
__.test_index
,
983 inspect
.currentframe().f_code
.co_name
)
984 self
.__class
__.test_index
+= 1
985 image_list
= test_config
["vim_conn"].get_image_list()
987 for item
in image_list
:
989 self
.__class
__.image_name
= item
['name']
990 self
.__class
__.image_id
= item
['id']
991 self
.assertEqual(type(self
.__class
__.image_name
),str)
992 self
.assertEqual(type(self
.__class
__.image_id
),str)
994 def test_010_get_image_list_by_name(self
):
995 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
996 self
.__class
__.test_index
,
997 inspect
.currentframe().f_code
.co_name
)
998 self
.__class
__.test_index
+= 1
1000 image_list
= test_config
["vim_conn"].get_image_list({'name': self
.__class
__.image_name
})
1002 for item
in image_list
:
1003 self
.assertEqual(type(item
['id']), str)
1004 self
.assertEqual(item
['id'], self
.__class
__.image_id
)
1005 self
.assertEqual(type(item
['name']), str)
1006 self
.assertEqual(item
['name'], self
.__class
__.image_name
)
1008 def test_020_get_image_list_by_id(self
):
1009 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1010 self
.__class
__.test_index
,
1011 inspect
.currentframe().f_code
.co_name
)
1012 self
.__class
__.test_index
+= 1
1014 filter_image_list
= test_config
["vim_conn"].get_image_list({'id': self
.__class
__.image_id
})
1016 for item1
in filter_image_list
:
1017 self
.assertEqual(type(item1
.get('id')), str)
1018 self
.assertEqual(item1
.get('id'), self
.__class
__.image_id
)
1019 self
.assertEqual(type(item1
.get('name')), str)
1020 self
.assertEqual(item1
.get('name'), self
.__class
__.image_name
)
1022 def test_030_get_image_list_negative(self
):
1023 Non_exist_image_id
= uuid
.uuid4()
1024 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1025 self
.__class
__.test_index
,
1026 inspect
.currentframe().f_code
.co_name
)
1027 self
.__class
__.test_index
+= 1
1028 image_list
= test_config
["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id
})
1030 self
.assertIsNotNone(image_list
, None)
1031 self
.assertEqual(image_list
, [])
1033 class test_vimconn_new_vminstance(test_base
):
1042 self
.__class
__.network_name
= _get_random_string(20)
1043 self
.__class
__.net_type
= 'bridge'
1045 self
.__class
__.network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
1046 net_type
=self
.__class
__.net_type
)
1049 test_base
.tearDown(self
)
1050 # Deleting created network
1051 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
1053 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
1055 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
1057 def test_000_new_vminstance(self
):
1058 vpci
= "0000:00:11.0"
1061 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
1064 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1066 # find image name and image id
1067 if test_config
['image_name']:
1068 image_list
= test_config
['vim_conn'].get_image_list({'name': test_config
['image_name']})
1069 if len(image_list
) == 0:
1070 raise Exception("Image {} is not found at VIM".format(test_config
['image_name']))
1072 self
.__class
__.image_id
= image_list
[0]['id']
1074 image_list
= test_config
['vim_conn'].get_image_list()
1075 if len(image_list
) == 0:
1076 raise Exception("Not found any image at VIM")
1078 self
.__class
__.image_id
= image_list
[0]['id']
1080 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1081 self
.__class
__.test_index
,
1082 inspect
.currentframe().f_code
.co_name
)
1083 self
.__class
__.test_index
+= 1
1085 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
}]
1087 self
.__class
__.instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
, flavor_id
=flavor_id
, net_list
=net_list
)
1089 self
.assertEqual(type(self
.__class
__.instance_id
),str)
1091 def test_010_new_vminstance_by_model(self
):
1092 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1093 model_name
= 'e1000'
1097 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1099 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1100 self
.__class
__.test_index
,
1101 inspect
.currentframe().f_code
.co_name
)
1102 self
.__class
__.test_index
+= 1
1104 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
}]
1106 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1107 flavor_id
=flavor_id
,
1109 self
.assertEqual(type(instance_id
),str)
1110 # Deleting created vm instance
1111 logger
.info("Deleting created vm intance")
1112 test_config
["vim_conn"].delete_vminstance(instance_id
)
1115 def test_020_new_vminstance_by_net_use(self
):
1116 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1121 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1123 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1124 self
.__class
__.test_index
,
1125 inspect
.currentframe().f_code
.co_name
)
1126 self
.__class
__.test_index
+= 1
1128 net_list
= [{'use': net_use
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1130 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1131 flavor_id
=flavor_id
,
1133 self
.assertEqual(type(instance_id
),str)
1134 # Deleting created vm instance
1135 logger
.info("Deleting created vm intance")
1136 test_config
["vim_conn"].delete_vminstance(instance_id
)
1139 def test_030_new_vminstance_by_net_type(self
):
1140 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1145 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1147 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1148 self
.__class
__.test_index
,
1149 inspect
.currentframe().f_code
.co_name
)
1150 self
.__class
__.test_index
+= 1
1152 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': _type
, 'net_id': self
.__class
__.network_id
}]
1154 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1155 flavor_id
=flavor_id
,
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_040_new_vminstance_by_cloud_config(self
):
1164 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1166 user_name
= 'test_user'
1168 key_pairs
= ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com']
1170 users_data
= [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com'], 'name': user_name
}]
1172 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
}
1175 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1177 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1178 self
.__class
__.test_index
,
1179 inspect
.currentframe().f_code
.co_name
)
1180 self
.__class
__.test_index
+= 1
1182 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1184 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Cloud_vm', image_id
=self
.__class
__.image_id
,
1185 flavor_id
=flavor_id
,
1187 cloud_config
=cloud_data
)
1188 self
.assertEqual(type(instance_id
),str)
1189 # Deleting created vm instance
1190 logger
.info("Deleting created vm intance")
1191 test_config
["vim_conn"].delete_vminstance(instance_id
)
1194 def test_050_new_vminstance_by_disk_list(self
):
1195 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1198 device_data
= [{'image_id': self
.__class
__.image_id
, 'size': '5'}]
1201 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1203 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1204 self
.__class
__.test_index
,
1205 inspect
.currentframe().f_code
.co_name
)
1206 self
.__class
__.test_index
+= 1
1208 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1210 instance_id
= test_config
["vim_conn"].new_vminstance(name
='VM_test1', image_id
=self
.__class
__.image_id
,
1211 flavor_id
=flavor_id
,
1213 disk_list
=device_data
)
1214 self
.assertEqual(type(instance_id
),str)
1215 # Deleting created vm instance
1216 logger
.info("Deleting created vm intance")
1217 test_config
["vim_conn"].delete_vminstance(instance_id
)
1220 def test_060_new_vminstance_negative(self
):
1221 unknown_flavor_id
= str(uuid
.uuid4())
1222 unknown_image_id
= str(uuid
.uuid4())
1225 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1226 self
.__class
__.test_index
,
1227 inspect
.currentframe().f_code
.co_name
)
1228 self
.__class
__.test_index
+= 1
1230 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1232 with self
.assertRaises(Exception) as context
:
1233 test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=unknown_image_id
,
1234 flavor_id
=unknown_flavor_id
,
1236 self
.assertEqual((context
.exception
).http_code
, 404)
1238 def test_070_get_vminstance(self
):
1239 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1240 self
.__class
__.test_index
,
1241 inspect
.currentframe().f_code
.co_name
)
1242 self
.__class
__.test_index
+= 1
1244 # Get instance by its id
1245 vm_info
= test_config
["vim_conn"].get_vminstance(self
.__class
__.instance_id
)
1247 if test_config
['vimtype'] == 'vmware':
1248 for attr
in vm_info
:
1249 if attr
== 'status':
1250 self
.assertEqual(vm_info
[attr
], 'ACTIVE')
1251 if attr
== 'hostId':
1252 self
.assertEqual(type(vm_info
[attr
]), str)
1253 if attr
== 'interfaces':
1254 self
.assertEqual(type(vm_info
[attr
]), list)
1255 self
.assertEqual(vm_info
[attr
][0]['IsConnected'], 'true')
1256 if attr
== 'IsEnabled':
1257 self
.assertEqual(vm_info
[attr
], 'true')
1259 def test_080_get_vminstance_negative(self
):
1260 unknown_instance_id
= str(uuid
.uuid4())
1262 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1263 self
.__class
__.test_index
,
1264 inspect
.currentframe().f_code
.co_name
)
1265 self
.__class
__.test_index
+= 1
1267 with self
.assertRaises(Exception) as context
:
1268 test_config
["vim_conn"].get_vminstance(unknown_instance_id
)
1270 self
.assertEqual((context
.exception
).http_code
, 404)
1272 def test_090_refresh_vms_status(self
):
1273 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1274 self
.__class
__.test_index
,
1275 inspect
.currentframe().f_code
.co_name
)
1276 self
.__class
__.test_index
+= 1
1278 vm_list
.append(self
.__class
__.instance_id
)
1281 vm_info
= test_config
["vim_conn"].refresh_vms_status(vm_list
)
1282 for attr
in vm_info
[self
.__class
__.instance_id
]:
1283 if attr
== 'status':
1284 self
.assertEqual(vm_info
[self
.__class
__.instance_id
][attr
], 'ACTIVE')
1285 if attr
== 'interfaces':
1286 self
.assertEqual(type(vm_info
[self
.__class
__.instance_id
][attr
]), list)
1288 def test_100_refresh_vms_status_negative(self
):
1289 unknown_id
= str(uuid
.uuid4())
1291 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1292 self
.__class
__.test_index
,
1293 inspect
.currentframe().f_code
.co_name
)
1294 self
.__class
__.test_index
+= 1
1296 vm_dict
= test_config
["vim_conn"].refresh_vms_status([unknown_id
])
1297 self
.assertEqual(vm_dict
, {})
1299 def test_110_action_vminstance(self
):
1300 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1301 self
.__class
__.test_index
,
1302 inspect
.currentframe().f_code
.co_name
)
1303 self
.__class
__.test_index
+= 1
1305 action_list
= ['shutdown','start','shutoff','rebuild','pause','resume']
1306 # various action on vminstace
1307 for action
in action_list
:
1308 instance_id
= test_config
["vim_conn"].action_vminstance(self
.__class
__.instance_id
,
1310 self
.assertEqual(instance_id
, self
.__class
__.instance_id
)
1312 def test_120_action_vminstance_negative(self
):
1313 non_exist_id
= str(uuid
.uuid4())
1314 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1315 self
.__class
__.test_index
,
1316 inspect
.currentframe().f_code
.co_name
)
1317 self
.__class
__.test_index
+= 1
1320 with self
.assertRaises(Exception) as context
:
1321 test_config
["vim_conn"].action_vminstance(non_exist_id
, { action
: None})
1323 self
.assertEqual((context
.exception
).http_code
, 400)
1325 def test_130_delete_vminstance(self
):
1326 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1327 self
.__class
__.test_index
,
1328 inspect
.currentframe().f_code
.co_name
)
1329 self
.__class
__.test_index
+= 1
1331 # Deleting created vm instance
1332 logger
.info("Deleting created vm instance")
1333 test_config
["vim_conn"].delete_vminstance(self
.__class
__.instance_id
)
1338 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
1339 scenario based tests.
1341 class descriptor_based_scenario_test(test_base
):
1343 scenario_test_path
= None
1344 scenario_uuid
= None
1345 instance_scenario_uuid
= None
1349 def setUpClass(cls
):
1351 cls
.to_delete_list
= []
1352 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
1353 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
1356 def tearDownClass(cls
):
1357 test_config
["test_number"] += 1
1359 def test_000_load_scenario(self
):
1360 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1361 inspect
.currentframe().f_code
.co_name
,
1362 test_config
["test_folder"])
1363 self
.__class
__.test_index
+= 1
1364 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
1365 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
1366 if len(vnfd_files
) == 0 or len(scenario_file
) > 1:
1367 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
1368 test_config
["test_folder"]))
1371 for vnfd
in vnfd_files
:
1372 with
open(vnfd
, 'r') as stream
:
1373 vnf_descriptor
= yaml
.load(stream
)
1375 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
1376 for vnfc
in vnfc_list
:
1377 vnfc
['image name'] = test_config
["image_name"]
1378 devices
= vnfc
.get('devices',[])
1379 for device
in devices
:
1380 if device
['type'] == 'disk' and 'image name' in device
:
1381 device
['image name'] = test_config
["image_name"]
1383 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
1384 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
)
1386 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
1387 "params": {"uuid": vnf
['vnf']['uuid']}})
1389 #load the scenario definition
1390 with
open(scenario_file
[0], 'r') as stream
:
1391 scenario_descriptor
= yaml
.load(stream
)
1392 networks
= scenario_descriptor
['scenario']['networks']
1393 networks
[test_config
["mgmt_net"]] = networks
.pop('mgmt')
1394 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
1395 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
1396 logger
.debug(scenario
)
1397 self
.__class
__.to_delete_list
.insert(0,{"item": "scenario", "function": test_config
["client"].delete_scenario
,
1398 "params":{"uuid": scenario
['scenario']['uuid']} })
1399 self
.__class
__.scenario_uuid
= scenario
['scenario']['uuid']
1401 def test_010_instantiate_scenario(self
):
1402 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1403 inspect
.currentframe().f_code
.co_name
,
1404 test_config
["test_folder"])
1405 self
.__class
__.test_index
+= 1
1407 instance
= test_config
["client"].create_instance(scenario_id
=self
.__class
__.scenario_uuid
,
1408 name
=self
.__class
__.test_text
)
1409 self
.__class
__.instance_scenario_uuid
= instance
['uuid']
1410 logger
.debug(instance
)
1411 self
.__class
__.to_delete_list
.insert(0, {"item": "instance", "function": test_config
["client"].delete_instance
,
1412 "params": {"uuid": instance
['uuid']}})
1414 def test_020_check_deployent(self
):
1415 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1416 inspect
.currentframe().f_code
.co_name
,
1417 test_config
["test_folder"])
1418 self
.__class
__.test_index
+= 1
1420 if test_config
["manual"]:
1421 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1424 keep_waiting
= test_config
["timeout"]
1425 instance_active
= False
1427 result
= check_instance_scenario_active(self
.__class
__.instance_scenario_uuid
)
1430 elif 'ERROR' in result
[1]:
1431 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
1433 raise Exception(msg
)
1435 if keep_waiting
>= 5:
1438 elif keep_waiting
> 0:
1439 time
.sleep(keep_waiting
)
1442 msg
= 'Timeout reached while waiting instance scenario to get active'
1444 raise Exception(msg
)
1446 def test_030_clean_deployment(self
):
1447 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1448 inspect
.currentframe().f_code
.co_name
,
1449 test_config
["test_folder"])
1450 self
.__class
__.test_index
+= 1
1451 #At the moment if you delete an scenario right after creating it, in openstack datacenters
1452 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1454 for item
in self
.__class
__.to_delete_list
:
1455 response
= item
["function"](**item
["params"])
1456 logger
.debug(response
)
1459 def _get_random_string(maxLength
):
1460 '''generates a string with random characters string.letters and string.digits
1461 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1465 minLength
= min_string
- len(prefix
)
1466 if maxLength
< min_string
: maxLength
= min_string
1467 maxLength
-= len(prefix
)
1468 length
= random
.randint(minLength
,maxLength
)
1469 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
1472 def test_vimconnector(args
):
1474 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1475 test_config
['vimtype'] = args
.vimtype
1476 if args
.vimtype
== "vmware":
1477 import vimconn_vmware
as vim
1479 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1481 tenant_name
= args
.tenant_name
1482 test_config
['tenant'] = tenant_name
1483 config_params
= json
.loads(args
.config_param
)
1484 org_name
= config_params
.get('orgname')
1485 org_user
= config_params
.get('user')
1486 org_passwd
= config_params
.get('passwd')
1487 vim_url
= args
.endpoint_url
1488 test_config
['image_path'] = args
.image_path
1489 test_config
['image_name'] = args
.image_name
1491 # vmware connector obj
1492 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
)
1494 elif args
.vimtype
== "aws":
1495 import vimconn_aws
as vim
1496 elif args
.vimtype
== "openstack":
1497 import vimconn_openstack
as vim
1498 elif args
.vimtype
== "openvim":
1499 import vimconn_openvim
as vim
1501 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
1505 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1506 # If only want to obtain a tests list print it and exit
1509 for cls
in clsmembers
:
1510 if cls
[0].startswith('test_vimconnector'):
1511 tests_names
.append(cls
[0])
1513 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
1518 # Create the list of tests to be run
1519 code_based_tests
= []
1521 for test
in args
.tests
:
1522 for t
in test
.split(','):
1523 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1524 if len(matches_code_based_tests
) > 0:
1525 code_based_tests
.append(matches_code_based_tests
[0][1])
1527 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1529 if not code_based_tests
:
1531 for cls
in clsmembers
:
1532 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1533 if cls
[0].startswith('test_vimconnector'):
1534 code_based_tests
.append(cls
[1])
1536 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1538 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1539 # This is handled in the tests using logging.
1540 stream
= open('/dev/null', 'w')
1542 # Run code based tests
1543 basic_tests_suite
= unittest
.TestSuite()
1544 for test
in code_based_tests
:
1545 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1546 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1547 executed
+= result
.testsRun
1548 failed
+= len(result
.failures
) + len(result
.errors
)
1549 if failfast
and failed
:
1551 if len(result
.failures
) > 0:
1552 logger
.debug("failures : {}".format(result
.failures
))
1553 if len(result
.errors
) > 0:
1554 logger
.debug("errors : {}".format(result
.errors
))
1555 return executed
, failed
1560 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1561 import openmanoclient
1564 test_config
["client"] = openmanoclient
.openmanoclient(
1565 endpoint_url
=args
.endpoint_url
,
1566 tenant_name
=args
.tenant_name
,
1567 datacenter_name
=args
.datacenter
,
1568 debug
=args
.debug
, logger
=test_config
["logger_name"])
1569 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1570 # If only want to obtain a tests list print it and exit
1573 for cls
in clsmembers
:
1574 if cls
[0].startswith('test_VIM'):
1575 tests_names
.append(cls
[0])
1577 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
1578 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1579 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1584 # Create the list of tests to be run
1585 code_based_tests
= []
1587 for test
in args
.tests
:
1588 for t
in test
.split(','):
1589 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1590 if len(matches_code_based_tests
) > 0:
1591 code_based_tests
.append(matches_code_based_tests
[0][1])
1593 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1595 if not code_based_tests
:
1597 for cls
in clsmembers
:
1598 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1599 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
1600 code_based_tests
.append(cls
[1])
1602 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1604 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1605 # This is handled in the tests using logging.
1606 stream
= open('/dev/null', 'w')
1608 # Run code based tests
1609 basic_tests_suite
= unittest
.TestSuite()
1610 for test
in code_based_tests
:
1611 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1612 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1613 executed
+= result
.testsRun
1614 failed
+= len(result
.failures
) + len(result
.errors
)
1615 if failfast
and failed
:
1617 if len(result
.failures
) > 0:
1618 logger
.debug("failures : {}".format(result
.failures
))
1619 if len(result
.errors
) > 0:
1620 logger
.debug("errors : {}".format(result
.errors
))
1621 return executed
, failed
1624 def test_deploy(args
):
1626 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1627 import openmanoclient
1630 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1631 test_config
["image_name"] = args
.image_name
1632 test_config
["mgmt_net"] = args
.mgmt_net
1633 test_config
["manual"] = args
.manual
1634 test_directory_content
= os
.listdir(test_config
["test_directory"])
1635 # If only want to obtain a tests list print it and exit
1637 msg
= "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
1642 descriptor_based_tests
= []
1643 # Create the list of tests to be run
1644 code_based_tests
= []
1646 for test
in args
.tests
:
1647 for t
in test
.split(','):
1648 if t
in test_directory_content
:
1649 descriptor_based_tests
.append(t
)
1651 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1653 if not descriptor_based_tests
:
1655 descriptor_based_tests
= test_directory_content
1657 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1659 # import openmanoclient from relative path
1660 test_config
["client"] = openmanoclient
.openmanoclient(
1661 endpoint_url
=args
.endpoint_url
,
1662 tenant_name
=args
.tenant_name
,
1663 datacenter_name
=args
.datacenter
,
1664 debug
=args
.debug
, logger
=test_config
["logger_name"])
1666 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1667 # This is handled in the tests using logging.
1668 stream
= open('/dev/null', 'w')
1669 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1670 for test
in descriptor_based_tests
:
1671 test_config
["test_folder"] = test
1672 test_suite
= unittest
.TestSuite()
1673 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
1674 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
1675 executed
+= result
.testsRun
1676 failed
+= len(result
.failures
) + len(result
.errors
)
1677 if failfast
and failed
:
1679 if len(result
.failures
) > 0:
1680 logger
.debug("failures : {}".format(result
.failures
))
1681 if len(result
.errors
) > 0:
1682 logger
.debug("errors : {}".format(result
.errors
))
1684 return executed
, failed
1686 if __name__
=="__main__":
1688 parser
= ArgumentParser(description
='Test RO module')
1689 parser
.add_argument('-v','--version', action
='version', help="Show current version",
1690 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
1693 parent_parser
= ArgumentParser(add_help
=False)
1694 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1695 dest
='failfast', action
="store_true", default
=False)
1696 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1697 dest
='failed', action
="store_true", default
=False)
1698 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
1699 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
1701 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
1702 help='Set the logger file. By default '+default_logger_file
)
1703 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
1704 help="Set the openmano tenant to use for the test. By default 'osm'")
1705 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
1706 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1707 dest
='timeout', type=int, default
=300)
1708 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
1710 subparsers
= parser
.add_subparsers(help='test sets')
1712 # Deployment test set
1713 # -------------------
1714 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
1715 help="test deployment using descriptors at RO_test folder ")
1716 deploy_parser
.set_defaults(func
=test_deploy
)
1718 # Mandatory arguments
1719 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
1720 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1721 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
1722 help='Image name available at datacenter used for the tests')
1723 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
1724 help='Set the vim management network to use for tests')
1726 # Optional arguments
1727 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
1728 help='Pause execution once deployed to allow manual checking of the '
1729 'deployed instance scenario')
1730 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1731 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1734 # -------------------
1735 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
1736 vimconn_parser
.set_defaults(func
=test_vimconnector
)
1737 # Mandatory arguments
1738 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1739 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
1740 help='Set the vimconnector type to test')
1741 mandatory_arguments
.add_argument('-c', '--config', dest
='config_param', required
=True,
1742 help='Set the vimconnector specific config parameters in dictionary format')
1743 mandatory_arguments
.add_argument('-u', '--url', dest
='endpoint_url',required
=True, help="Set the vim connector url or Host IP")
1744 # Optional arguments
1745 vimconn_parser
.add_argument('-i', '--image-path', dest
='image_path', help="Provide image path present at RO container")
1746 vimconn_parser
.add_argument('-n', '--image-name', dest
='image_name', help="Provide image name for test")
1747 # TODO add optional arguments for vimconn tests
1748 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1750 # Datacenter test set
1751 # -------------------
1752 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
1753 vimconn_parser
.set_defaults(func
=test_vim
)
1755 # Mandatory arguments
1756 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1757 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1759 # Optional arguments
1760 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1761 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1763 argcomplete
.autocomplete(parser
)
1764 args
= parser
.parse_args()
1768 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1769 logger_level
= 'INFO'
1771 logger_level
= 'DEBUG'
1773 logger_level
= 'WARNING'
1774 logger_name
= os
.path
.basename(__file__
)
1775 test_config
["logger_name"] = logger_name
1776 logger
= logging
.getLogger(logger_name
)
1777 logger
.setLevel(logger_level
)
1778 failfast
= args
.failfast
1780 # Configure a logging handler to store in a logging file
1781 if args
.logger_file
:
1782 fileHandler
= logging
.FileHandler(args
.logger_file
)
1783 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1784 fileHandler
.setFormatter(formatter_fileHandler
)
1785 logger
.addHandler(fileHandler
)
1787 # Configure a handler to print to stdout
1788 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
1789 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1790 consoleHandler
.setFormatter(formatter_consoleHandler
)
1791 logger
.addHandler(consoleHandler
)
1793 logger
.debug('Program started with the following arguments: ' + str(args
))
1795 # set test config parameters
1796 test_config
["timeout"] = args
.timeout
1797 test_config
["test_number"] = 1
1799 executed
, failed
= args
.func(args
)
1802 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
1803 sys
.exit(1 if failed
else 0)