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
):
284 def test_000_connect(self
):
285 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
286 self
.__class
__.test_index
,
287 inspect
.currentframe().f_code
.co_name
)
289 self
.__class
__.test_index
+= 1
290 if test_config
['vimtype'] == 'vmware':
291 vca_object
= test_config
["vim_conn"].connect()
292 logger
.debug("{}".format(vca_object
))
293 self
.assertIsInstance(vca_object
, VCA
)
296 class test_vimconn_new_network(test_base
):
299 def test_000_new_network(self
):
300 self
.__class
__.network_name
= _get_random_string(20)
301 network_type
= 'bridge'
303 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
304 self
.__class
__.test_index
, inspect
.currentframe().f_code
.co_name
)
305 self
.__class
__.test_index
+= 1
307 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
308 net_type
=network_type
)
309 self
.__class
__.network_id
= network
310 logger
.debug("{}".format(network
))
312 network_list
= test_config
["vim_conn"].get_network_list()
313 for net
in network_list
:
314 if self
.__class
__.network_name
in net
.get('name'):
315 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
316 self
.assertEqual(net
.get('type'), network_type
)
318 # Deleting created network
319 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
321 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
323 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
325 def test_010_new_network_by_types(self
):
327 network_types
= ['data','bridge','mgmt']
328 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
329 self
.__class
__.test_index
,
330 inspect
.currentframe().f_code
.co_name
)
331 self
.__class
__.test_index
+= 1
332 for net_type
in network_types
:
333 self
.__class
__.network_name
= _get_random_string(20)
334 network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
337 delete_net_ids
.append(network_id
)
338 logger
.debug("{}".format(network_id
))
340 network_list
= test_config
["vim_conn"].get_network_list()
341 for net
in network_list
:
342 if self
.__class
__.network_name
in net
.get('name'):
343 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
344 if net_type
in net
.get('type'):
345 self
.assertEqual(net
.get('type'), net_type
)
347 self
.assertNotEqual(net
.get('type'), net_type
)
349 # Deleting created network
350 for net_id
in delete_net_ids
:
351 result
= test_config
["vim_conn"].delete_network(net_id
)
353 logger
.info("Network id {} sucessfully deleted".format(net_id
))
355 logger
.info("Failed to delete network id {}".format(net_id
))
357 def test_020_new_network_by_ipprofile(self
):
358 test_directory_content
= os
.listdir(test_config
["test_directory"])
360 for dir_name
in test_directory_content
:
361 if dir_name
== 'simple_multi_vnfc':
362 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
363 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
366 for vnfd
in vnfd_files
:
367 with
open(vnfd
, 'r') as stream
:
368 vnf_descriptor
= yaml
.load(stream
)
370 internal_connections_list
= vnf_descriptor
['vnf']['internal-connections']
371 for item
in internal_connections_list
:
372 if 'ip-profile' in item
:
373 version
= item
['ip-profile']['ip-version']
374 dhcp_count
= item
['ip-profile']['dhcp']['count']
375 dhcp_enabled
= item
['ip-profile']['dhcp']['enabled']
377 self
.__class
__.network_name
= _get_random_string(20)
378 ip_profile
= {'dhcp_count': dhcp_count
,
379 'dhcp_enabled': dhcp_enabled
,
380 'ip_version': version
382 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
383 self
.__class
__.test_index
,
384 inspect
.currentframe().f_code
.co_name
)
385 self
.__class
__.test_index
+= 1
386 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
388 ip_profile
=ip_profile
)
389 self
.__class
__.network_id
= network
390 logger
.debug("{}".format(network
))
392 network_list
= test_config
["vim_conn"].get_network_list()
393 for net
in network_list
:
394 if self
.__class
__.network_name
in net
.get('name'):
395 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
397 # Deleting created network
398 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
400 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
402 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
404 def test_030_new_network_by_isshared(self
):
405 self
.__class
__.network_name
= _get_random_string(20)
407 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
408 self
.__class
__.test_index
,
409 inspect
.currentframe().f_code
.co_name
)
410 self
.__class
__.test_index
+= 1
411 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
414 self
.__class
__.network_id
= network
415 logger
.debug("{}".format(network
))
417 network_list
= test_config
["vim_conn"].get_network_list()
418 for net
in network_list
:
419 if self
.__class
__.network_name
in net
.get('name'):
420 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
421 self
.assertEqual(net
.get('shared'), shared
)
423 # Deleting created network
424 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
426 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
428 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
430 def test_040_new_network_by_negative(self
):
431 self
.__class
__.network_name
= _get_random_string(20)
432 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
433 self
.__class
__.test_index
,
434 inspect
.currentframe().f_code
.co_name
)
435 self
.__class
__.test_index
+= 1
436 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
437 net_type
='unknowntype')
438 self
.__class
__.network_id
= network
439 logger
.debug("{}".format(network
))
440 network_list
= test_config
["vim_conn"].get_network_list()
441 for net
in network_list
:
442 if self
.__class
__.network_name
in net
.get('name'):
443 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
445 # Deleting created network
446 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
448 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
450 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
452 def test_050_refresh_nets_status(self
):
453 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
454 self
.__class
__.test_index
,
455 inspect
.currentframe().f_code
.co_name
)
456 self
.__class
__.test_index
+= 1
457 # creating new network
458 network_name
= _get_random_string(20)
460 network_id
= test_config
["vim_conn"].new_network(net_name
=network_name
,
463 net_dict
= test_config
["vim_conn"].refresh_nets_status([network_id
])
464 for attr
in net_dict
[network_id
]:
466 self
.assertEqual(net_dict
[network_id
][attr
], 'ACTIVE')
468 # Deleting created network
469 result
= test_config
["vim_conn"].delete_network(network_id
)
471 logger
.info("Network id {} sucessfully deleted".format(network_id
))
473 logger
.info("Failed to delete network id {}".format(network_id
))
475 def test_060_refresh_nets_status_negative(self
):
476 unknown_net_id
= str(uuid
.uuid4())
477 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
478 self
.__class
__.test_index
,
479 inspect
.currentframe().f_code
.co_name
)
480 self
.__class
__.test_index
+= 1
483 net_dict
= test_config
["vim_conn"].refresh_nets_status([unknown_net_id
])
484 self
.assertEqual(net_dict
, {})
486 class test_vimconn_get_network_list(test_base
):
490 # creating new network
491 self
.__class
__.network_name
= _get_random_string(20)
492 self
.__class
__.net_type
= 'bridge'
493 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
494 net_type
=self
.__class
__.net_type
)
495 self
.__class
__.network_id
= network
496 logger
.debug("{}".format(network
))
499 test_base
.tearDown(self
)
501 # Deleting created network
502 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
504 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
506 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
508 def test_000_get_network_list(self
):
509 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
510 self
.__class
__.test_index
,
511 inspect
.currentframe().f_code
.co_name
)
512 self
.__class
__.test_index
+= 1
514 network_list
= test_config
["vim_conn"].get_network_list()
515 for net
in network_list
:
516 if self
.__class
__.network_name
in net
.get('name'):
517 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
518 self
.assertEqual(net
.get('type'), self
.__class
__.net_type
)
519 self
.assertEqual(net
.get('status'), 'ACTIVE')
520 self
.assertEqual(net
.get('shared'), False)
522 def test_010_get_network_list_by_name(self
):
523 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
524 self
.__class
__.test_index
,
525 inspect
.currentframe().f_code
.co_name
)
526 self
.__class
__.test_index
+= 1
528 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
530 # find network from list by it's name
531 new_network_list
= test_config
["vim_conn"].get_network_list({'name': network_name
})
532 for list_item
in new_network_list
:
533 if self
.__class
__.network_name
in list_item
.get('name'):
534 self
.assertEqual(network_name
, list_item
.get('name'))
535 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
536 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
538 def test_020_get_network_list_by_id(self
):
539 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
540 self
.__class
__.test_index
,
541 inspect
.currentframe().f_code
.co_name
)
542 self
.__class
__.test_index
+= 1
544 # find network from list by it's id
545 new_network_list
= test_config
["vim_conn"].get_network_list({'id':self
.__class
__.network_id
})
546 for list_item
in new_network_list
:
547 if self
.__class
__.network_id
in list_item
.get('id'):
548 self
.assertEqual(self
.__class
__.network_id
, list_item
.get('id'))
549 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
550 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
552 def test_030_get_network_list_by_shared(self
):
554 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
555 self
.__class
__.test_index
,
556 inspect
.currentframe().f_code
.co_name
)
557 self
.__class
__.test_index
+= 1
559 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
560 # find network from list by it's shared value
561 new_network_list
= test_config
["vim_conn"].get_network_list({'shared':Shared
,
562 'name':network_name
})
563 for list_item
in new_network_list
:
564 if list_item
.get('shared') == Shared
:
565 self
.assertEqual(list_item
.get('shared'), Shared
)
566 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
567 self
.assertEqual(network_name
, list_item
.get('name'))
569 def test_040_get_network_list_by_tenant_id(self
):
570 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
571 self
.__class
__.test_index
,
572 inspect
.currentframe().f_code
.co_name
)
573 self
.__class
__.test_index
+= 1
575 tenant_list
= test_config
["vim_conn"].get_tenant_list()
576 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
578 for tenant_item
in tenant_list
:
579 if test_config
['tenant'] == tenant_item
.get('name'):
580 # find network from list by it's tenant id
581 tenant_id
= tenant_item
.get('id')
582 new_network_list
= test_config
["vim_conn"].get_network_list({'tenant_id':tenant_id
,
583 'name':network_name
})
584 for list_item
in new_network_list
:
585 self
.assertEqual(tenant_id
, list_item
.get('tenant_id'))
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_050_get_network_list_by_status(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
597 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
599 # find network from list by it's status
600 new_network_list
= test_config
["vim_conn"].get_network_list({'status':status
,
601 'name': network_name
})
602 for list_item
in new_network_list
:
603 self
.assertIn(self
.__class
__.network_name
, list_item
.get('name'))
604 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
605 self
.assertEqual(list_item
.get('status'), status
)
607 def test_060_get_network_list_by_negative(self
):
608 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
609 self
.__class
__.test_index
,
610 inspect
.currentframe().f_code
.co_name
)
611 self
.__class
__.test_index
+= 1
613 network_list
= test_config
["vim_conn"].get_network_list({'name': 'unknown_name'})
614 self
.assertEqual(network_list
, [])
616 class test_vimconn_get_network(test_base
):
620 # creating new network
621 self
.__class
__.network_name
= _get_random_string(20)
622 self
.__class
__.net_type
= 'bridge'
623 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
624 net_type
=self
.__class
__.net_type
)
625 self
.__class
__.network_id
= network
626 logger
.debug("{}".format(network
))
629 test_base
.tearDown(self
)
631 # Deleting created network
632 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
634 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
636 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
638 def test_000_get_network(self
):
639 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
640 self
.__class
__.test_index
,
641 inspect
.currentframe().f_code
.co_name
)
642 self
.__class
__.test_index
+= 1
644 network_info
= test_config
["vim_conn"].get_network(self
.__class
__.network_id
)
645 self
.assertEqual(network_info
.get('status'), 'ACTIVE')
646 self
.assertIn(self
.__class
__.network_name
, network_info
.get('name'))
647 self
.assertEqual(network_info
.get('type'), self
.__class
__.net_type
)
648 self
.assertEqual(network_info
.get('id'), self
.__class
__.network_id
)
650 def test_010_get_network_negative(self
):
651 Non_exist_id
= str(uuid
.uuid4())
652 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
653 self
.__class
__.test_index
,
654 inspect
.currentframe().f_code
.co_name
)
655 self
.__class
__.test_index
+= 1
657 network_info
= test_config
["vim_conn"].get_network(Non_exist_id
)
658 self
.assertEqual(network_info
, {})
660 class test_vimconn_delete_network(test_base
):
663 def test_000_delete_network(self
):
665 self
.__class
__.network_name
= _get_random_string(20)
666 self
.__class
__.net_type
= 'bridge'
667 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
668 net_type
=self
.__class
__.net_type
)
669 self
.__class
__.network_id
= network
670 logger
.debug("{}".format(network
))
672 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
673 self
.__class
__.test_index
,
674 inspect
.currentframe().f_code
.co_name
)
675 self
.__class
__.test_index
+= 1
677 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
679 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
681 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
683 # after deleting network we check in network list
684 network_list
= test_config
["vim_conn"].get_network_list({ 'id':self
.__class
__.network_id
})
685 self
.assertEqual(network_list
, [])
687 def test_010_delete_network_negative(self
):
688 Non_exist_id
= str(uuid
.uuid4())
690 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
691 self
.__class
__.test_index
,
692 inspect
.currentframe().f_code
.co_name
)
693 self
.__class
__.test_index
+= 1
695 with self
.assertRaises(Exception) as context
:
696 test_config
["vim_conn"].delete_network(Non_exist_id
)
698 self
.assertEqual((context
.exception
).http_code
, 400)
700 class test_vimconn_get_flavor(test_base
):
702 def test_000_get_flavor(self
):
703 test_directory_content
= os
.listdir(test_config
["test_directory"])
705 for dir_name
in test_directory_content
:
706 if dir_name
== 'simple_linux':
707 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
708 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
711 for vnfd
in vnfd_files
:
712 with
open(vnfd
, 'r') as stream
:
713 vnf_descriptor
= yaml
.load(stream
)
715 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
716 for item
in vnfc_list
:
717 if 'ram' in item
and 'vcpus' in item
and 'disk' in item
:
719 vcpus
= item
['vcpus']
722 flavor_data
= {'ram': ram
,
727 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
728 self
.__class
__.test_index
,
729 inspect
.currentframe().f_code
.co_name
)
730 self
.__class
__.test_index
+= 1
732 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
734 result
= test_config
["vim_conn"].get_flavor(flavor_id
)
735 self
.assertEqual(ram
, result
['ram'])
736 self
.assertEqual(vcpus
, result
['vcpus'])
737 self
.assertEqual(disk
, result
['disk'])
740 result
= test_config
["vim_conn"].delete_flavor(flavor_id
)
742 logger
.info("Flavor id {} sucessfully deleted".format(result
))
744 logger
.info("Failed to delete flavor id {}".format(result
))
746 def test_010_get_flavor_negative(self
):
747 Non_exist_flavor_id
= str(uuid
.uuid4())
749 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
750 self
.__class
__.test_index
,
751 inspect
.currentframe().f_code
.co_name
)
752 self
.__class
__.test_index
+= 1
754 with self
.assertRaises(Exception) as context
:
755 test_config
["vim_conn"].get_flavor(Non_exist_flavor_id
)
757 self
.assertEqual((context
.exception
).http_code
, 404)
759 class test_vimconn_new_flavor(test_base
):
762 def test_000_new_flavor(self
):
763 flavor_data
= {'ram': 1024, 'vpcus': 1, 'disk': 10}
765 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
766 self
.__class
__.test_index
,
767 inspect
.currentframe().f_code
.co_name
)
768 self
.__class
__.test_index
+= 1
771 self
.__class
__.flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
772 self
.assertEqual(type(self
.__class
__.flavor_id
),str)
773 self
.assertIsInstance(uuid
.UUID(self
.__class
__.flavor_id
),uuid
.UUID
)
775 def test_010_delete_flavor(self
):
776 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
777 self
.__class
__.test_index
,
778 inspect
.currentframe().f_code
.co_name
)
779 self
.__class
__.test_index
+= 1
782 result
= test_config
["vim_conn"].delete_flavor(self
.__class
__.flavor_id
)
784 logger
.info("Flavor id {} sucessfully deleted".format(result
))
786 logger
.error("Failed to delete flavor id {}".format(result
))
787 raise Exception ("Failed to delete created flavor")
789 def test_020_new_flavor_negative(self
):
790 Invalid_flavor_data
= {'ram': '1024', 'vcpus': 2.0, 'disk': 2.0}
792 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
793 self
.__class
__.test_index
,
794 inspect
.currentframe().f_code
.co_name
)
795 self
.__class
__.test_index
+= 1
797 with self
.assertRaises(Exception) as context
:
798 test_config
["vim_conn"].new_flavor(Invalid_flavor_data
)
800 self
.assertEqual((context
.exception
).http_code
, 400)
802 def test_030_delete_flavor_negative(self
):
803 Non_exist_flavor_id
= str(uuid
.uuid4())
805 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
806 self
.__class
__.test_index
,
807 inspect
.currentframe().f_code
.co_name
)
808 self
.__class
__.test_index
+= 1
810 with self
.assertRaises(Exception) as context
:
811 test_config
["vim_conn"].delete_flavor(Non_exist_flavor_id
)
813 self
.assertEqual((context
.exception
).http_code
, 404)
815 class test_vimconn_new_image(test_base
):
817 def test_000_new_image(self
):
818 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
819 self
.__class
__.test_index
,
820 inspect
.currentframe().f_code
.co_name
)
821 self
.__class
__.test_index
+= 1
823 image_path
= test_config
['image_path']
825 image_id
= test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : image_path
})
827 self
.assertEqual(type(image_id
),str)
828 self
.assertIsInstance(uuid
.UUID(image_id
),uuid
.UUID
)
830 self
.skipTest("Skipping test as image file not present at RO container")
832 def test_010_new_image_negative(self
):
833 Non_exist_image_path
= '/temp1/cirros.ovf'
835 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
836 self
.__class
__.test_index
,
837 inspect
.currentframe().f_code
.co_name
)
838 self
.__class
__.test_index
+= 1
840 with self
.assertRaises(Exception) as context
:
841 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
843 self
.assertEqual((context
.exception
).http_code
, 400)
845 def test_020_delete_image(self
):
846 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
847 self
.__class
__.test_index
,
848 inspect
.currentframe().f_code
.co_name
)
849 self
.__class
__.test_index
+= 1
851 image_id
= test_config
["vim_conn"].delete_image(self
.__class
__.image_id
)
852 self
.assertEqual(type(image_id
),str)
854 def test_030_delete_image_negative(self
):
855 Non_exist_image_id
= str(uuid
.uuid4())
857 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
858 self
.__class
__.test_index
,
859 inspect
.currentframe().f_code
.co_name
)
860 self
.__class
__.test_index
+= 1
862 with self
.assertRaises(Exception) as context
:
863 test_config
["vim_conn"].delete_image(Non_exist_image_id
)
865 self
.assertEqual((context
.exception
).http_code
, 404)
867 class test_vimconn_get_image_id_from_path(test_base
):
869 def test_000_get_image_id_from_path(self
):
870 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
871 self
.__class
__.test_index
,
872 inspect
.currentframe().f_code
.co_name
)
873 self
.__class
__.test_index
+= 1
875 image_path
= test_config
['image_path']
877 image_id
= test_config
["vim_conn"].get_image_id_from_path( image_path
)
878 self
.assertEqual(type(image_id
),str)
880 self
.skipTest("Skipping test as image file not present at RO container")
882 def test_010_get_image_id_from_path_negative(self
):
883 Non_exist_image_path
= '/temp1/cirros.ovf'
885 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
886 self
.__class
__.test_index
,
887 inspect
.currentframe().f_code
.co_name
)
888 self
.__class
__.test_index
+= 1
890 with self
.assertRaises(Exception) as context
:
891 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
893 self
.assertEqual((context
.exception
).http_code
, 400)
895 class test_vimconn_get_image_list(test_base
):
899 def test_000_get_image_list(self
):
900 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
901 self
.__class
__.test_index
,
902 inspect
.currentframe().f_code
.co_name
)
903 self
.__class
__.test_index
+= 1
904 image_list
= test_config
["vim_conn"].get_image_list()
906 for item
in image_list
:
908 self
.__class
__.image_name
= item
['name']
909 self
.__class
__.image_id
= item
['id']
910 self
.assertEqual(type(self
.__class
__.image_name
),str)
911 self
.assertEqual(type(self
.__class
__.image_id
),str)
913 def test_010_get_image_list_by_name(self
):
914 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
915 self
.__class
__.test_index
,
916 inspect
.currentframe().f_code
.co_name
)
917 self
.__class
__.test_index
+= 1
919 image_list
= test_config
["vim_conn"].get_image_list({'name': self
.__class
__.image_name
})
921 for item
in image_list
:
922 self
.assertEqual(type(item
['id']), str)
923 self
.assertEqual(item
['id'], self
.__class
__.image_id
)
924 self
.assertEqual(type(item
['name']), str)
925 self
.assertEqual(item
['name'], self
.__class
__.image_name
)
927 def test_020_get_image_list_by_id(self
):
928 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
929 self
.__class
__.test_index
,
930 inspect
.currentframe().f_code
.co_name
)
931 self
.__class
__.test_index
+= 1
933 filter_image_list
= test_config
["vim_conn"].get_image_list({'id': self
.__class
__.image_id
})
935 for item1
in filter_image_list
:
936 self
.assertEqual(type(item1
.get('id')), str)
937 self
.assertEqual(item1
.get('id'), self
.__class
__.image_id
)
938 self
.assertEqual(type(item1
.get('name')), str)
939 self
.assertEqual(item1
.get('name'), self
.__class
__.image_name
)
941 def test_030_get_image_list_negative(self
):
942 Non_exist_image_id
= uuid
.uuid4()
943 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
944 self
.__class
__.test_index
,
945 inspect
.currentframe().f_code
.co_name
)
946 self
.__class
__.test_index
+= 1
947 image_list
= test_config
["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id
})
949 self
.assertIsNotNone(image_list
, None)
950 self
.assertEqual(image_list
, [])
952 class test_vimconn_new_vminstance(test_base
):
961 self
.__class
__.network_name
= _get_random_string(20)
962 self
.__class
__.net_type
= 'bridge'
964 self
.__class
__.network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
965 net_type
=self
.__class
__.net_type
)
968 test_base
.tearDown(self
)
969 # Deleting created network
970 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
972 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
974 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
976 def test_000_new_vminstance(self
):
977 vpci
= "0000:00:11.0"
980 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
983 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
985 # find image name and image id
986 if test_config
['image_name']:
987 image_list
= test_config
['vim_conn'].get_image_list({'name': test_config
['image_name']})
988 if len(image_list
) == 0:
989 raise Exception("Image {} is not found at VIM".format(test_config
['image_name']))
991 self
.__class
__.image_id
= image_list
[0]['id']
993 image_list
= test_config
['vim_conn'].get_image_list()
994 if len(image_list
) == 0:
995 raise Exception("Not found any image at VIM")
997 self
.__class
__.image_id
= image_list
[0]['id']
999 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1000 self
.__class
__.test_index
,
1001 inspect
.currentframe().f_code
.co_name
)
1002 self
.__class
__.test_index
+= 1
1004 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'vpci': vpci
, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1006 self
.__class
__.instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
, flavor_id
=flavor_id
, net_list
=net_list
)
1008 self
.assertEqual(type(self
.__class
__.instance_id
),str)
1010 def test_010_new_vminstance_by_model(self
):
1011 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1012 model_name
= 'e1000'
1016 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1018 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1019 self
.__class
__.test_index
,
1020 inspect
.currentframe().f_code
.co_name
)
1021 self
.__class
__.test_index
+= 1
1023 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'model': model_name
, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1025 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1026 flavor_id
=flavor_id
,
1028 self
.assertEqual(type(instance_id
),str)
1029 # Deleting created vm instance
1030 logger
.info("Deleting created vm intance")
1031 test_config
["vim_conn"].delete_vminstance(instance_id
)
1034 def test_020_new_vminstance_by_net_use(self
):
1035 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1040 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1042 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1043 self
.__class
__.test_index
,
1044 inspect
.currentframe().f_code
.co_name
)
1045 self
.__class
__.test_index
+= 1
1047 net_list
= [{'use': net_use
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1049 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1050 flavor_id
=flavor_id
,
1052 self
.assertEqual(type(instance_id
),str)
1053 # Deleting created vm instance
1054 logger
.info("Deleting created vm intance")
1055 test_config
["vim_conn"].delete_vminstance(instance_id
)
1058 def test_030_new_vminstance_by_net_type(self
):
1059 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1064 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1066 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1067 self
.__class
__.test_index
,
1068 inspect
.currentframe().f_code
.co_name
)
1069 self
.__class
__.test_index
+= 1
1071 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': _type
, 'net_id': self
.__class
__.network_id
}]
1073 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1074 flavor_id
=flavor_id
,
1076 self
.assertEqual(type(instance_id
),str)
1077 # Deleting created vm instance
1078 logger
.info("Deleting created vm intance")
1079 test_config
["vim_conn"].delete_vminstance(instance_id
)
1082 def test_040_new_vminstance_by_cloud_config(self
):
1083 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1085 user_name
= 'test_user'
1087 key_pairs
= ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com']
1089 users_data
= [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com'], 'name': user_name
}]
1091 cloud_data
= {'config-files': [{'content': 'auto enp0s3\niface enp0s3 inet dhcp\n', 'dest': '/etc/network/interfaces.d/enp0s3.cfg', 'owner': 'root:root', 'permissions': '0644'}, {'content': '#! /bin/bash\nls -al >> /var/log/osm.log\n', 'dest': '/etc/rc.local', 'permissions': '0755'}, {'content': 'file content', 'dest': '/etc/test_delete'}], 'boot-data-drive': True, 'key-pairs': key_pairs
, 'users': users_data
}
1094 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1096 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1097 self
.__class
__.test_index
,
1098 inspect
.currentframe().f_code
.co_name
)
1099 self
.__class
__.test_index
+= 1
1101 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1103 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Cloud_vm', image_id
=self
.__class
__.image_id
,
1104 flavor_id
=flavor_id
,
1106 cloud_config
=cloud_data
)
1107 self
.assertEqual(type(instance_id
),str)
1108 # Deleting created vm instance
1109 logger
.info("Deleting created vm intance")
1110 test_config
["vim_conn"].delete_vminstance(instance_id
)
1113 def test_050_new_vminstance_by_disk_list(self
):
1114 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1117 device_data
= [{'image_id': self
.__class
__.image_id
, 'size': '5'}]
1120 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1122 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1123 self
.__class
__.test_index
,
1124 inspect
.currentframe().f_code
.co_name
)
1125 self
.__class
__.test_index
+= 1
1127 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1129 instance_id
= test_config
["vim_conn"].new_vminstance(name
='VM_test1', image_id
=self
.__class
__.image_id
,
1130 flavor_id
=flavor_id
,
1132 disk_list
=device_data
)
1133 self
.assertEqual(type(instance_id
),str)
1134 # Deleting created vm instance
1135 logger
.info("Deleting created vm intance")
1136 test_config
["vim_conn"].delete_vminstance(instance_id
)
1139 def test_060_new_vminstance_negative(self
):
1140 unknown_flavor_id
= str(uuid
.uuid4())
1141 unknown_image_id
= str(uuid
.uuid4())
1144 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1145 self
.__class
__.test_index
,
1146 inspect
.currentframe().f_code
.co_name
)
1147 self
.__class
__.test_index
+= 1
1149 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1151 with self
.assertRaises(Exception) as context
:
1152 test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=unknown_image_id
,
1153 flavor_id
=unknown_flavor_id
,
1155 self
.assertEqual((context
.exception
).http_code
, 404)
1157 def test_070_get_vminstance(self
):
1158 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1159 self
.__class
__.test_index
,
1160 inspect
.currentframe().f_code
.co_name
)
1161 self
.__class
__.test_index
+= 1
1163 # Get instance by its id
1164 vm_info
= test_config
["vim_conn"].get_vminstance(self
.__class
__.instance_id
)
1166 if test_config
['vimtype'] == 'vmware':
1167 for attr
in vm_info
:
1168 if attr
== 'status':
1169 self
.assertEqual(vm_info
[attr
], 'ACTIVE')
1170 if attr
== 'hostId':
1171 self
.assertEqual(type(vm_info
[attr
]), str)
1172 if attr
== 'interfaces':
1173 self
.assertEqual(type(vm_info
[attr
]), list)
1174 self
.assertEqual(vm_info
[attr
][0]['IsConnected'], 'true')
1175 if attr
== 'IsEnabled':
1176 self
.assertEqual(vm_info
[attr
], 'true')
1178 def test_080_get_vminstance_negative(self
):
1179 unknown_instance_id
= str(uuid
.uuid4())
1181 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1182 self
.__class
__.test_index
,
1183 inspect
.currentframe().f_code
.co_name
)
1184 self
.__class
__.test_index
+= 1
1186 with self
.assertRaises(Exception) as context
:
1187 test_config
["vim_conn"].get_vminstance(unknown_instance_id
)
1189 self
.assertEqual((context
.exception
).http_code
, 404)
1191 def test_090_refresh_vms_status(self
):
1192 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1193 self
.__class
__.test_index
,
1194 inspect
.currentframe().f_code
.co_name
)
1195 self
.__class
__.test_index
+= 1
1197 vm_list
.append(self
.__class
__.instance_id
)
1200 vm_info
= test_config
["vim_conn"].refresh_vms_status(vm_list
)
1201 for attr
in vm_info
[self
.__class
__.instance_id
]:
1202 if attr
== 'status':
1203 self
.assertEqual(vm_info
[self
.__class
__.instance_id
][attr
], 'ACTIVE')
1204 if attr
== 'interfaces':
1205 self
.assertEqual(type(vm_info
[self
.__class
__.instance_id
][attr
]), list)
1207 def test_100_refresh_vms_status_negative(self
):
1208 unknown_id
= str(uuid
.uuid4())
1210 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1211 self
.__class
__.test_index
,
1212 inspect
.currentframe().f_code
.co_name
)
1213 self
.__class
__.test_index
+= 1
1215 vm_dict
= test_config
["vim_conn"].refresh_vms_status([unknown_id
])
1216 self
.assertEqual(vm_dict
, {})
1218 def test_110_action_vminstance(self
):
1219 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1220 self
.__class
__.test_index
,
1221 inspect
.currentframe().f_code
.co_name
)
1222 self
.__class
__.test_index
+= 1
1224 action_list
= ['shutdown','start','shutoff','rebuild','pause','resume']
1225 # various action on vminstace
1226 for action
in action_list
:
1227 instance_id
= test_config
["vim_conn"].action_vminstance(self
.__class
__.instance_id
,
1229 self
.assertEqual(instance_id
, self
.__class
__.instance_id
)
1231 def test_120_action_vminstance_negative(self
):
1232 non_exist_id
= str(uuid
.uuid4())
1233 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1234 self
.__class
__.test_index
,
1235 inspect
.currentframe().f_code
.co_name
)
1236 self
.__class
__.test_index
+= 1
1239 with self
.assertRaises(Exception) as context
:
1240 test_config
["vim_conn"].action_vminstance(non_exist_id
, { action
: None})
1242 self
.assertEqual((context
.exception
).http_code
, 400)
1244 def test_130_delete_vminstance(self
):
1245 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1246 self
.__class
__.test_index
,
1247 inspect
.currentframe().f_code
.co_name
)
1248 self
.__class
__.test_index
+= 1
1250 # Deleting created vm instance
1251 logger
.info("Deleting created vm instance")
1252 test_config
["vim_conn"].delete_vminstance(self
.__class
__.instance_id
)
1255 class test_vimconn_get_tenant_list(test_base
):
1258 def test_000_get_tenant_list(self
):
1259 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1260 self
.__class
__.test_index
,
1261 inspect
.currentframe().f_code
.co_name
)
1262 self
.__class
__.test_index
+= 1
1264 # Getting tenant list
1265 tenant_list
= test_config
["vim_conn"].get_tenant_list()
1267 for item
in tenant_list
:
1268 if test_config
['tenant'] == item
['name']:
1269 self
.__class
__.tenant_id
= item
['id']
1270 self
.assertEqual(type(item
['name']), str)
1271 self
.assertEqual(type(item
['id']), str)
1273 def test_010_get_tenant_list_by_id(self
):
1274 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1275 self
.__class
__.test_index
,
1276 inspect
.currentframe().f_code
.co_name
)
1277 self
.__class
__.test_index
+= 1
1279 # Getting filter tenant list by its id
1280 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'id': self
.__class
__.tenant_id
})
1282 for item
in filter_tenant_list
:
1283 self
.assertEqual(type(item
['id']), str)
1284 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1286 def test_020_get_tenant_list_by_name(self
):
1287 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1288 self
.__class
__.test_index
,
1289 inspect
.currentframe().f_code
.co_name
)
1290 self
.__class
__.test_index
+= 1
1292 # Getting filter tenant list by its name
1293 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant']})
1295 for item
in filter_tenant_list
:
1296 self
.assertEqual(type(item
['name']), str)
1297 self
.assertEqual(item
['name'], test_config
['tenant'])
1299 def test_030_get_tenant_list_by_name_and_id(self
):
1300 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1301 self
.__class
__.test_index
,
1302 inspect
.currentframe().f_code
.co_name
)
1303 self
.__class
__.test_index
+= 1
1305 # Getting filter tenant list by its name and id
1306 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant'],
1307 'id': self
.__class
__.tenant_id
})
1309 for item
in filter_tenant_list
:
1310 self
.assertEqual(type(item
['name']), str)
1311 self
.assertEqual(type(item
['id']), str)
1312 self
.assertEqual(item
['name'], test_config
['tenant'])
1313 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1315 def test_040_get_tenant_list_negative(self
):
1316 non_exist_tenant_name
= "Tenant_123"
1317 non_exist_tenant_id
= "kjhgrt456-45345kjhdfgnbdk-34dsfjdfg"
1318 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1319 self
.__class
__.test_index
,
1320 inspect
.currentframe().f_code
.co_name
)
1321 self
.__class
__.test_index
+= 1
1323 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': non_exist_tenant_name
,
1324 'id': non_exist_tenant_id
})
1326 self
.assertEqual(filter_tenant_list
, [])
1328 class test_vimconn_new_tenant(test_base
):
1331 def test_000_new_tenant(self
):
1332 tenant_name
= _get_random_string(20)
1333 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1334 self
.__class
__.test_index
,
1335 inspect
.currentframe().f_code
.co_name
)
1336 self
.__class
__.test_index
+= 1
1338 self
.__class
__.tenant_id
= test_config
["vim_conn"].new_tenant(tenant_name
)
1341 self
.assertEqual(type(self
.__class
__.tenant_id
), str)
1343 def test_010_new_tenant_negative(self
):
1344 Invalid_tenant_name
= 10121
1345 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1346 self
.__class
__.test_index
,
1347 inspect
.currentframe().f_code
.co_name
)
1348 self
.__class
__.test_index
+= 1
1350 with self
.assertRaises(Exception) as context
:
1351 test_config
["vim_conn"].new_tenant(Invalid_tenant_name
)
1353 self
.assertEqual((context
.exception
).http_code
, 400)
1355 def test_020_delete_tenant(self
):
1356 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1357 self
.__class
__.test_index
,
1358 inspect
.currentframe().f_code
.co_name
)
1359 self
.__class
__.test_index
+= 1
1361 tenant_id
= test_config
["vim_conn"].delete_tenant(self
.__class
__.tenant_id
)
1362 self
.assertEqual(type(tenant_id
), str)
1364 def test_030_delete_tenant_negative(self
):
1365 Non_exist_tenant_name
= 'Test_30_tenant'
1366 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1367 self
.__class
__.test_index
,
1368 inspect
.currentframe().f_code
.co_name
)
1369 self
.__class
__.test_index
+= 1
1371 with self
.assertRaises(Exception) as context
:
1372 test_config
["vim_conn"].delete_tenant(Non_exist_tenant_name
)
1374 self
.assertEqual((context
.exception
).http_code
, 404)
1378 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
1379 scenario based tests.
1381 class descriptor_based_scenario_test(test_base
):
1383 scenario_test_path
= None
1384 scenario_uuid
= None
1385 instance_scenario_uuid
= None
1389 def setUpClass(cls
):
1391 cls
.to_delete_list
= []
1392 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
1393 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
1396 def tearDownClass(cls
):
1397 test_config
["test_number"] += 1
1399 def test_000_load_scenario(self
):
1400 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1401 inspect
.currentframe().f_code
.co_name
,
1402 test_config
["test_folder"])
1403 self
.__class
__.test_index
+= 1
1404 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
1405 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
1406 if len(vnfd_files
) == 0 or len(scenario_file
) > 1:
1407 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
1408 test_config
["test_folder"]))
1411 for vnfd
in vnfd_files
:
1412 with
open(vnfd
, 'r') as stream
:
1413 vnf_descriptor
= yaml
.load(stream
)
1415 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
1416 for vnfc
in vnfc_list
:
1417 vnfc
['image name'] = test_config
["image_name"]
1418 devices
= vnfc
.get('devices',[])
1419 for device
in devices
:
1420 if device
['type'] == 'disk' and 'image name' in device
:
1421 device
['image name'] = test_config
["image_name"]
1423 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
1424 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
)
1426 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
1427 "params": {"uuid": vnf
['vnf']['uuid']}})
1429 #load the scenario definition
1430 with
open(scenario_file
[0], 'r') as stream
:
1431 scenario_descriptor
= yaml
.load(stream
)
1432 networks
= scenario_descriptor
['scenario']['networks']
1433 networks
[test_config
["mgmt_net"]] = networks
.pop('mgmt')
1434 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
1435 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
1436 logger
.debug(scenario
)
1437 self
.__class
__.to_delete_list
.insert(0,{"item": "scenario", "function": test_config
["client"].delete_scenario
,
1438 "params":{"uuid": scenario
['scenario']['uuid']} })
1439 self
.__class
__.scenario_uuid
= scenario
['scenario']['uuid']
1441 def test_010_instantiate_scenario(self
):
1442 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1443 inspect
.currentframe().f_code
.co_name
,
1444 test_config
["test_folder"])
1445 self
.__class
__.test_index
+= 1
1447 instance
= test_config
["client"].create_instance(scenario_id
=self
.__class
__.scenario_uuid
,
1448 name
=self
.__class
__.test_text
)
1449 self
.__class
__.instance_scenario_uuid
= instance
['uuid']
1450 logger
.debug(instance
)
1451 self
.__class
__.to_delete_list
.insert(0, {"item": "instance", "function": test_config
["client"].delete_instance
,
1452 "params": {"uuid": instance
['uuid']}})
1454 def test_020_check_deployent(self
):
1455 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1456 inspect
.currentframe().f_code
.co_name
,
1457 test_config
["test_folder"])
1458 self
.__class
__.test_index
+= 1
1460 if test_config
["manual"]:
1461 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1464 keep_waiting
= test_config
["timeout"]
1465 instance_active
= False
1467 result
= check_instance_scenario_active(self
.__class
__.instance_scenario_uuid
)
1470 elif 'ERROR' in result
[1]:
1471 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
1473 raise Exception(msg
)
1475 if keep_waiting
>= 5:
1478 elif keep_waiting
> 0:
1479 time
.sleep(keep_waiting
)
1482 msg
= 'Timeout reached while waiting instance scenario to get active'
1484 raise Exception(msg
)
1486 def test_030_clean_deployment(self
):
1487 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1488 inspect
.currentframe().f_code
.co_name
,
1489 test_config
["test_folder"])
1490 self
.__class
__.test_index
+= 1
1491 #At the moment if you delete an scenario right after creating it, in openstack datacenters
1492 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1494 for item
in self
.__class
__.to_delete_list
:
1495 response
= item
["function"](**item
["params"])
1496 logger
.debug(response
)
1499 def _get_random_string(maxLength
):
1500 '''generates a string with random characters string.letters and string.digits
1501 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1505 minLength
= min_string
- len(prefix
)
1506 if maxLength
< min_string
: maxLength
= min_string
1507 maxLength
-= len(prefix
)
1508 length
= random
.randint(minLength
,maxLength
)
1509 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
1512 def test_vimconnector(args
):
1514 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1515 test_config
['vimtype'] = args
.vimtype
1516 if args
.vimtype
== "vmware":
1517 import vimconn_vmware
as vim
1519 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1521 tenant_name
= args
.tenant_name
1522 test_config
['tenant'] = tenant_name
1523 config_params
= json
.loads(args
.config_param
)
1524 org_name
= config_params
.get('orgname')
1525 org_user
= config_params
.get('user')
1526 org_passwd
= config_params
.get('passwd')
1527 vim_url
= args
.endpoint_url
1528 test_config
['image_path'] = args
.image_path
1529 test_config
['image_name'] = args
.image_name
1531 # vmware connector obj
1532 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
)
1534 elif args
.vimtype
== "aws":
1535 import vimconn_aws
as vim
1536 elif args
.vimtype
== "openstack":
1537 import vimconn_openstack
as vim
1538 elif args
.vimtype
== "openvim":
1539 import vimconn_openvim
as vim
1541 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
1545 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1546 # If only want to obtain a tests list print it and exit
1549 for cls
in clsmembers
:
1550 if cls
[0].startswith('test_vimconnector'):
1551 tests_names
.append(cls
[0])
1553 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
1558 # Create the list of tests to be run
1559 code_based_tests
= []
1561 for test
in args
.tests
:
1562 for t
in test
.split(','):
1563 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1564 if len(matches_code_based_tests
) > 0:
1565 code_based_tests
.append(matches_code_based_tests
[0][1])
1567 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1569 if not code_based_tests
:
1571 for cls
in clsmembers
:
1572 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1573 if cls
[0].startswith('test_vimconnector'):
1574 code_based_tests
.append(cls
[1])
1576 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1578 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1579 # This is handled in the tests using logging.
1580 stream
= open('/dev/null', 'w')
1582 # Run code based tests
1583 basic_tests_suite
= unittest
.TestSuite()
1584 for test
in code_based_tests
:
1585 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1586 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1587 executed
+= result
.testsRun
1588 failed
+= len(result
.failures
) + len(result
.errors
)
1589 if failfast
and failed
:
1591 if len(result
.failures
) > 0:
1592 logger
.debug("failures : {}".format(result
.failures
))
1593 if len(result
.errors
) > 0:
1594 logger
.debug("errors : {}".format(result
.errors
))
1595 return executed
, failed
1600 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1601 import openmanoclient
1604 test_config
["client"] = openmanoclient
.openmanoclient(
1605 endpoint_url
=args
.endpoint_url
,
1606 tenant_name
=args
.tenant_name
,
1607 datacenter_name
=args
.datacenter
,
1608 debug
=args
.debug
, logger
=test_config
["logger_name"])
1609 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1610 # If only want to obtain a tests list print it and exit
1613 for cls
in clsmembers
:
1614 if cls
[0].startswith('test_VIM'):
1615 tests_names
.append(cls
[0])
1617 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
1618 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1619 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1624 # Create the list of tests to be run
1625 code_based_tests
= []
1627 for test
in args
.tests
:
1628 for t
in test
.split(','):
1629 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1630 if len(matches_code_based_tests
) > 0:
1631 code_based_tests
.append(matches_code_based_tests
[0][1])
1633 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1635 if not code_based_tests
:
1637 for cls
in clsmembers
:
1638 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1639 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
1640 code_based_tests
.append(cls
[1])
1642 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1644 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1645 # This is handled in the tests using logging.
1646 stream
= open('/dev/null', 'w')
1648 # Run code based tests
1649 basic_tests_suite
= unittest
.TestSuite()
1650 for test
in code_based_tests
:
1651 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1652 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1653 executed
+= result
.testsRun
1654 failed
+= len(result
.failures
) + len(result
.errors
)
1655 if failfast
and failed
:
1657 if len(result
.failures
) > 0:
1658 logger
.debug("failures : {}".format(result
.failures
))
1659 if len(result
.errors
) > 0:
1660 logger
.debug("errors : {}".format(result
.errors
))
1661 return executed
, failed
1664 def test_deploy(args
):
1666 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1667 import openmanoclient
1670 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1671 test_config
["image_name"] = args
.image_name
1672 test_config
["mgmt_net"] = args
.mgmt_net
1673 test_config
["manual"] = args
.manual
1674 test_directory_content
= os
.listdir(test_config
["test_directory"])
1675 # If only want to obtain a tests list print it and exit
1677 msg
= "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
1682 descriptor_based_tests
= []
1683 # Create the list of tests to be run
1684 code_based_tests
= []
1686 for test
in args
.tests
:
1687 for t
in test
.split(','):
1688 if t
in test_directory_content
:
1689 descriptor_based_tests
.append(t
)
1691 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1693 if not descriptor_based_tests
:
1695 descriptor_based_tests
= test_directory_content
1697 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1699 # import openmanoclient from relative path
1700 test_config
["client"] = openmanoclient
.openmanoclient(
1701 endpoint_url
=args
.endpoint_url
,
1702 tenant_name
=args
.tenant_name
,
1703 datacenter_name
=args
.datacenter
,
1704 debug
=args
.debug
, logger
=test_config
["logger_name"])
1706 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1707 # This is handled in the tests using logging.
1708 stream
= open('/dev/null', 'w')
1709 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1710 for test
in descriptor_based_tests
:
1711 test_config
["test_folder"] = test
1712 test_suite
= unittest
.TestSuite()
1713 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
1714 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
1715 executed
+= result
.testsRun
1716 failed
+= len(result
.failures
) + len(result
.errors
)
1717 if failfast
and failed
:
1719 if len(result
.failures
) > 0:
1720 logger
.debug("failures : {}".format(result
.failures
))
1721 if len(result
.errors
) > 0:
1722 logger
.debug("errors : {}".format(result
.errors
))
1724 return executed
, failed
1726 if __name__
=="__main__":
1728 parser
= ArgumentParser(description
='Test RO module')
1729 parser
.add_argument('-v','--version', action
='version', help="Show current version",
1730 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
1733 parent_parser
= ArgumentParser(add_help
=False)
1734 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1735 dest
='failfast', action
="store_true", default
=False)
1736 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1737 dest
='failed', action
="store_true", default
=False)
1738 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
1739 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
1741 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
1742 help='Set the logger file. By default '+default_logger_file
)
1743 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
1744 help="Set the openmano tenant to use for the test. By default 'osm'")
1745 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
1746 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1747 dest
='timeout', type=int, default
=300)
1748 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
1750 subparsers
= parser
.add_subparsers(help='test sets')
1752 # Deployment test set
1753 # -------------------
1754 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
1755 help="test deployment using descriptors at RO_test folder ")
1756 deploy_parser
.set_defaults(func
=test_deploy
)
1758 # Mandatory arguments
1759 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
1760 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1761 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
1762 help='Image name available at datacenter used for the tests')
1763 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
1764 help='Set the vim management network to use for tests')
1766 # Optional arguments
1767 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
1768 help='Pause execution once deployed to allow manual checking of the '
1769 'deployed instance scenario')
1770 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1771 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1774 # -------------------
1775 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
1776 vimconn_parser
.set_defaults(func
=test_vimconnector
)
1777 # Mandatory arguments
1778 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1779 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
1780 help='Set the vimconnector type to test')
1781 mandatory_arguments
.add_argument('-c', '--config', dest
='config_param', required
=True,
1782 help='Set the vimconnector specific config parameters in dictionary format')
1783 mandatory_arguments
.add_argument('-u', '--url', dest
='endpoint_url',required
=True, help="Set the vim connector url or Host IP")
1784 # Optional arguments
1785 vimconn_parser
.add_argument('-i', '--image-path', dest
='image_path', help="Provide image path present at RO container")
1786 vimconn_parser
.add_argument('-n', '--image-name', dest
='image_name', help="Provide image name for test")
1787 # TODO add optional arguments for vimconn tests
1788 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1790 # Datacenter test set
1791 # -------------------
1792 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
1793 vimconn_parser
.set_defaults(func
=test_vim
)
1795 # Mandatory arguments
1796 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1797 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1799 # Optional arguments
1800 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1801 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1803 argcomplete
.autocomplete(parser
)
1804 args
= parser
.parse_args()
1808 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1809 logger_level
= 'INFO'
1811 logger_level
= 'DEBUG'
1813 logger_level
= 'WARNING'
1814 logger_name
= os
.path
.basename(__file__
)
1815 test_config
["logger_name"] = logger_name
1816 logger
= logging
.getLogger(logger_name
)
1817 logger
.setLevel(logger_level
)
1818 failfast
= args
.failfast
1820 # Configure a logging handler to store in a logging file
1821 if args
.logger_file
:
1822 fileHandler
= logging
.FileHandler(args
.logger_file
)
1823 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1824 fileHandler
.setFormatter(formatter_fileHandler
)
1825 logger
.addHandler(fileHandler
)
1827 # Configure a handler to print to stdout
1828 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
1829 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1830 consoleHandler
.setFormatter(formatter_consoleHandler
)
1831 logger
.addHandler(consoleHandler
)
1833 logger
.debug('Program started with the following arguments: ' + str(args
))
1835 # set test config parameters
1836 test_config
["timeout"] = args
.timeout
1837 test_config
["test_number"] = 1
1839 executed
, failed
= args
.func(args
)
1842 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
1843 sys
.exit(1 if failed
else 0)