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
})
826 self
.assertEqual(type(image_id
),str)
827 self
.assertIsInstance(uuid
.UUID(image_id
),uuid
.UUID
)
829 self
.skipTest("Skipping test as image file not present at RO container")
831 def test_010_new_image_negative(self
):
832 Non_exist_image_path
= '/temp1/cirros.ovf'
834 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
835 self
.__class
__.test_index
,
836 inspect
.currentframe().f_code
.co_name
)
837 self
.__class
__.test_index
+= 1
839 with self
.assertRaises(Exception) as context
:
840 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
842 self
.assertEqual((context
.exception
).http_code
, 400)
844 class test_vimconn_get_image_id_from_path(test_base
):
846 def test_000_get_image_id_from_path(self
):
847 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
848 self
.__class
__.test_index
,
849 inspect
.currentframe().f_code
.co_name
)
850 self
.__class
__.test_index
+= 1
852 image_path
= test_config
['image_path']
854 image_id
= test_config
["vim_conn"].get_image_id_from_path( image_path
)
855 self
.assertEqual(type(image_id
),str)
857 self
.skipTest("Skipping test as image file not present at RO container")
859 def test_010_get_image_id_from_path_negative(self
):
860 Non_exist_image_path
= '/temp1/cirros.ovf'
862 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
863 self
.__class
__.test_index
,
864 inspect
.currentframe().f_code
.co_name
)
865 self
.__class
__.test_index
+= 1
867 with self
.assertRaises(Exception) as context
:
868 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
870 self
.assertEqual((context
.exception
).http_code
, 400)
872 class test_vimconn_get_image_list(test_base
):
876 def test_000_get_image_list(self
):
877 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
878 self
.__class
__.test_index
,
879 inspect
.currentframe().f_code
.co_name
)
880 self
.__class
__.test_index
+= 1
881 image_list
= test_config
["vim_conn"].get_image_list()
883 for item
in image_list
:
885 self
.__class
__.image_name
= item
['name']
886 self
.__class
__.image_id
= item
['id']
887 self
.assertEqual(type(self
.__class
__.image_name
),str)
888 self
.assertEqual(type(self
.__class
__.image_id
),str)
890 def test_010_get_image_list_by_name(self
):
891 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
892 self
.__class
__.test_index
,
893 inspect
.currentframe().f_code
.co_name
)
894 self
.__class
__.test_index
+= 1
896 image_list
= test_config
["vim_conn"].get_image_list({'name': self
.__class
__.image_name
})
898 for item
in image_list
:
899 self
.assertEqual(type(item
['id']), str)
900 self
.assertEqual(item
['id'], self
.__class
__.image_id
)
901 self
.assertEqual(type(item
['name']), str)
902 self
.assertEqual(item
['name'], self
.__class
__.image_name
)
904 def test_020_get_image_list_by_id(self
):
905 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
906 self
.__class
__.test_index
,
907 inspect
.currentframe().f_code
.co_name
)
908 self
.__class
__.test_index
+= 1
910 filter_image_list
= test_config
["vim_conn"].get_image_list({'id': self
.__class
__.image_id
})
912 for item1
in filter_image_list
:
913 self
.assertEqual(type(item1
.get('id')), str)
914 self
.assertEqual(item1
.get('id'), self
.__class
__.image_id
)
915 self
.assertEqual(type(item1
.get('name')), str)
916 self
.assertEqual(item1
.get('name'), self
.__class
__.image_name
)
918 def test_030_get_image_list_negative(self
):
919 Non_exist_image_id
= uuid
.uuid4()
920 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
921 self
.__class
__.test_index
,
922 inspect
.currentframe().f_code
.co_name
)
923 self
.__class
__.test_index
+= 1
924 image_list
= test_config
["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id
})
926 self
.assertIsNotNone(image_list
, None)
927 self
.assertEqual(image_list
, [])
929 class test_vimconn_new_vminstance(test_base
):
938 self
.__class
__.network_name
= _get_random_string(20)
939 self
.__class
__.net_type
= 'bridge'
941 self
.__class
__.network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
942 net_type
=self
.__class
__.net_type
)
945 test_base
.tearDown(self
)
946 # Deleting created network
947 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
949 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
951 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
953 def test_000_new_vminstance(self
):
954 vpci
= "0000:00:11.0"
957 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
960 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
962 # find image name and image id
963 if test_config
['image_name']:
964 image_list
= test_config
['vim_conn'].get_image_list({'name': test_config
['image_name']})
965 if len(image_list
) == 0:
966 raise Exception("Image {} is not found at VIM".format(test_config
['image_name']))
968 self
.__class
__.image_id
= image_list
[0]['id']
970 image_list
= test_config
['vim_conn'].get_image_list()
971 if len(image_list
) == 0:
972 raise Exception("Not found any image at VIM")
974 self
.__class
__.image_id
= image_list
[0]['id']
976 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
977 self
.__class
__.test_index
,
978 inspect
.currentframe().f_code
.co_name
)
979 self
.__class
__.test_index
+= 1
981 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
}]
983 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
)
985 self
.assertEqual(type(self
.__class
__.instance_id
),str)
987 def test_010_new_vminstance_by_model(self
):
988 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
993 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
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 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
}]
1002 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1003 flavor_id
=flavor_id
,
1005 self
.assertEqual(type(instance_id
),str)
1006 # Deleting created vm instance
1007 logger
.info("Deleting created vm intance")
1008 test_config
["vim_conn"].delete_vminstance(instance_id
)
1011 def test_020_new_vminstance_by_net_use(self
):
1012 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1017 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1019 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1020 self
.__class
__.test_index
,
1021 inspect
.currentframe().f_code
.co_name
)
1022 self
.__class
__.test_index
+= 1
1024 net_list
= [{'use': net_use
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1026 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1027 flavor_id
=flavor_id
,
1029 self
.assertEqual(type(instance_id
),str)
1030 # Deleting created vm instance
1031 logger
.info("Deleting created vm intance")
1032 test_config
["vim_conn"].delete_vminstance(instance_id
)
1035 def test_030_new_vminstance_by_net_type(self
):
1036 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1041 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1043 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1044 self
.__class
__.test_index
,
1045 inspect
.currentframe().f_code
.co_name
)
1046 self
.__class
__.test_index
+= 1
1048 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': _type
, 'net_id': self
.__class
__.network_id
}]
1050 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1051 flavor_id
=flavor_id
,
1053 self
.assertEqual(type(instance_id
),str)
1054 # Deleting created vm instance
1055 logger
.info("Deleting created vm intance")
1056 test_config
["vim_conn"].delete_vminstance(instance_id
)
1059 def test_040_new_vminstance_by_cloud_config(self
):
1060 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1062 user_name
= 'test_user'
1064 key_pairs
= ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com']
1066 users_data
= [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com'], 'name': user_name
}]
1068 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
}
1071 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1073 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1074 self
.__class
__.test_index
,
1075 inspect
.currentframe().f_code
.co_name
)
1076 self
.__class
__.test_index
+= 1
1078 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1080 instance_id
= test_config
["vim_conn"].new_vminstance(name
='Cloud_vm', image_id
=self
.__class
__.image_id
,
1081 flavor_id
=flavor_id
,
1083 cloud_config
=cloud_data
)
1084 self
.assertEqual(type(instance_id
),str)
1085 # Deleting created vm instance
1086 logger
.info("Deleting created vm intance")
1087 test_config
["vim_conn"].delete_vminstance(instance_id
)
1090 def test_050_new_vminstance_by_disk_list(self
):
1091 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1094 device_data
= [{'image_id': self
.__class
__.image_id
, 'size': '5'}]
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, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1106 instance_id
= test_config
["vim_conn"].new_vminstance(name
='VM_test1', image_id
=self
.__class
__.image_id
,
1107 flavor_id
=flavor_id
,
1109 disk_list
=device_data
)
1110 self
.assertEqual(type(instance_id
),str)
1111 # Deleting created vm instance
1112 logger
.info("Deleting created vm intance")
1113 test_config
["vim_conn"].delete_vminstance(instance_id
)
1116 def test_060_new_vminstance_negative(self
):
1117 unknown_flavor_id
= str(uuid
.uuid4())
1118 unknown_image_id
= str(uuid
.uuid4())
1121 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1122 self
.__class
__.test_index
,
1123 inspect
.currentframe().f_code
.co_name
)
1124 self
.__class
__.test_index
+= 1
1126 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1128 with self
.assertRaises(Exception) as context
:
1129 test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=unknown_image_id
,
1130 flavor_id
=unknown_flavor_id
,
1132 self
.assertEqual((context
.exception
).http_code
, 404)
1134 def test_070_get_vminstance(self
):
1135 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1136 self
.__class
__.test_index
,
1137 inspect
.currentframe().f_code
.co_name
)
1138 self
.__class
__.test_index
+= 1
1140 # Get instance by its id
1141 vm_info
= test_config
["vim_conn"].get_vminstance(self
.__class
__.instance_id
)
1143 if test_config
['vimtype'] == 'vmware':
1144 for attr
in vm_info
:
1145 if attr
== 'status':
1146 self
.assertEqual(vm_info
[attr
], 'ACTIVE')
1147 if attr
== 'hostId':
1148 self
.assertEqual(type(vm_info
[attr
]), str)
1149 if attr
== 'interfaces':
1150 self
.assertEqual(type(vm_info
[attr
]), list)
1151 self
.assertEqual(vm_info
[attr
][0]['IsConnected'], 'true')
1152 if attr
== 'IsEnabled':
1153 self
.assertEqual(vm_info
[attr
], 'true')
1155 def test_080_get_vminstance_negative(self
):
1156 unknown_instance_id
= str(uuid
.uuid4())
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 with self
.assertRaises(Exception) as context
:
1164 test_config
["vim_conn"].get_vminstance(unknown_instance_id
)
1166 self
.assertEqual((context
.exception
).http_code
, 404)
1168 def test_090_refresh_vms_status(self
):
1169 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1170 self
.__class
__.test_index
,
1171 inspect
.currentframe().f_code
.co_name
)
1172 self
.__class
__.test_index
+= 1
1174 vm_list
.append(self
.__class
__.instance_id
)
1177 vm_info
= test_config
["vim_conn"].refresh_vms_status(vm_list
)
1178 for attr
in vm_info
[self
.__class
__.instance_id
]:
1179 if attr
== 'status':
1180 self
.assertEqual(vm_info
[self
.__class
__.instance_id
][attr
], 'ACTIVE')
1181 if attr
== 'interfaces':
1182 self
.assertEqual(type(vm_info
[self
.__class
__.instance_id
][attr
]), list)
1184 def test_100_refresh_vms_status_negative(self
):
1185 unknown_id
= str(uuid
.uuid4())
1187 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1188 self
.__class
__.test_index
,
1189 inspect
.currentframe().f_code
.co_name
)
1190 self
.__class
__.test_index
+= 1
1192 vm_dict
= test_config
["vim_conn"].refresh_vms_status([unknown_id
])
1193 self
.assertEqual(vm_dict
, {})
1195 def test_110_action_vminstance(self
):
1196 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1197 self
.__class
__.test_index
,
1198 inspect
.currentframe().f_code
.co_name
)
1199 self
.__class
__.test_index
+= 1
1201 action_list
= ['shutdown','start','shutoff','rebuild','pause','resume']
1202 # various action on vminstace
1203 for action
in action_list
:
1204 instance_id
= test_config
["vim_conn"].action_vminstance(self
.__class
__.instance_id
,
1206 self
.assertEqual(instance_id
, self
.__class
__.instance_id
)
1208 def test_120_action_vminstance_negative(self
):
1209 non_exist_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
1216 with self
.assertRaises(Exception) as context
:
1217 test_config
["vim_conn"].action_vminstance(non_exist_id
, { action
: None})
1219 self
.assertEqual((context
.exception
).http_code
, 400)
1221 def test_130_delete_vminstance(self
):
1222 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1223 self
.__class
__.test_index
,
1224 inspect
.currentframe().f_code
.co_name
)
1225 self
.__class
__.test_index
+= 1
1227 # Deleting created vm instance
1228 logger
.info("Deleting created vm instance")
1229 test_config
["vim_conn"].delete_vminstance(self
.__class
__.instance_id
)
1232 class test_vimconn_get_tenant_list(test_base
):
1235 def test_000_get_tenant_list(self
):
1236 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1237 self
.__class
__.test_index
,
1238 inspect
.currentframe().f_code
.co_name
)
1239 self
.__class
__.test_index
+= 1
1241 # Getting tenant list
1242 tenant_list
= test_config
["vim_conn"].get_tenant_list()
1244 for item
in tenant_list
:
1245 if test_config
['tenant'] == item
['name']:
1246 self
.__class
__.tenant_id
= item
['id']
1247 self
.assertEqual(type(item
['name']), str)
1248 self
.assertEqual(type(item
['id']), str)
1250 def test_010_get_tenant_list_by_id(self
):
1251 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1252 self
.__class
__.test_index
,
1253 inspect
.currentframe().f_code
.co_name
)
1254 self
.__class
__.test_index
+= 1
1256 # Getting filter tenant list by its id
1257 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'id': self
.__class
__.tenant_id
})
1259 for item
in filter_tenant_list
:
1260 self
.assertEqual(type(item
['id']), str)
1261 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1263 def test_020_get_tenant_list_by_name(self
):
1264 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1265 self
.__class
__.test_index
,
1266 inspect
.currentframe().f_code
.co_name
)
1267 self
.__class
__.test_index
+= 1
1269 # Getting filter tenant list by its name
1270 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant']})
1272 for item
in filter_tenant_list
:
1273 self
.assertEqual(type(item
['name']), str)
1274 self
.assertEqual(item
['name'], test_config
['tenant'])
1276 def test_030_get_tenant_list_by_name_and_id(self
):
1277 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1278 self
.__class
__.test_index
,
1279 inspect
.currentframe().f_code
.co_name
)
1280 self
.__class
__.test_index
+= 1
1282 # Getting filter tenant list by its name and id
1283 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant'],
1284 'id': self
.__class
__.tenant_id
})
1286 for item
in filter_tenant_list
:
1287 self
.assertEqual(type(item
['name']), str)
1288 self
.assertEqual(type(item
['id']), str)
1289 self
.assertEqual(item
['name'], test_config
['tenant'])
1290 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1292 def test_040_get_tenant_list_negative(self
):
1293 non_exist_tenant_name
= "Tenant_123"
1294 non_exist_tenant_id
= "kjhgrt456-45345kjhdfgnbdk-34dsfjdfg"
1295 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1296 self
.__class
__.test_index
,
1297 inspect
.currentframe().f_code
.co_name
)
1298 self
.__class
__.test_index
+= 1
1300 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': non_exist_tenant_name
,
1301 'id': non_exist_tenant_id
})
1303 self
.assertEqual(filter_tenant_list
, [])
1305 class test_vimconn_new_tenant(test_base
):
1308 def test_000_new_tenant(self
):
1309 tenant_name
= _get_random_string(20)
1310 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1311 self
.__class
__.test_index
,
1312 inspect
.currentframe().f_code
.co_name
)
1313 self
.__class
__.test_index
+= 1
1315 self
.__class
__.tenant_id
= test_config
["vim_conn"].new_tenant(tenant_name
)
1318 self
.assertEqual(type(self
.__class
__.tenant_id
), str)
1320 def test_010_new_tenant_negative(self
):
1321 Invalid_tenant_name
= 10121
1322 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1323 self
.__class
__.test_index
,
1324 inspect
.currentframe().f_code
.co_name
)
1325 self
.__class
__.test_index
+= 1
1327 with self
.assertRaises(Exception) as context
:
1328 test_config
["vim_conn"].new_tenant(Invalid_tenant_name
)
1330 self
.assertEqual((context
.exception
).http_code
, 400)
1332 def test_020_delete_tenant(self
):
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 tenant_id
= test_config
["vim_conn"].delete_tenant(self
.__class
__.tenant_id
)
1339 self
.assertEqual(type(tenant_id
), str)
1341 def test_030_delete_tenant_negative(self
):
1342 Non_exist_tenant_name
= 'Test_30_tenant'
1343 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1344 self
.__class
__.test_index
,
1345 inspect
.currentframe().f_code
.co_name
)
1346 self
.__class
__.test_index
+= 1
1348 with self
.assertRaises(Exception) as context
:
1349 test_config
["vim_conn"].delete_tenant(Non_exist_tenant_name
)
1351 self
.assertEqual((context
.exception
).http_code
, 404)
1355 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
1356 scenario based tests.
1358 class descriptor_based_scenario_test(test_base
):
1360 scenario_test_path
= None
1361 scenario_uuid
= None
1362 instance_scenario_uuid
= None
1366 def setUpClass(cls
):
1368 cls
.to_delete_list
= []
1369 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
1370 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
1373 def tearDownClass(cls
):
1374 test_config
["test_number"] += 1
1376 def test_000_load_scenario(self
):
1377 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1378 inspect
.currentframe().f_code
.co_name
,
1379 test_config
["test_folder"])
1380 self
.__class
__.test_index
+= 1
1381 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
1382 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
1383 if len(vnfd_files
) == 0 or len(scenario_file
) > 1:
1384 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
1385 test_config
["test_folder"]))
1388 for vnfd
in vnfd_files
:
1389 with
open(vnfd
, 'r') as stream
:
1390 vnf_descriptor
= yaml
.load(stream
)
1392 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
1393 for vnfc
in vnfc_list
:
1394 vnfc
['image name'] = test_config
["image_name"]
1395 devices
= vnfc
.get('devices',[])
1396 for device
in devices
:
1397 if device
['type'] == 'disk' and 'image name' in device
:
1398 device
['image name'] = test_config
["image_name"]
1400 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
1401 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
)
1403 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
1404 "params": {"uuid": vnf
['vnf']['uuid']}})
1406 #load the scenario definition
1407 with
open(scenario_file
[0], 'r') as stream
:
1408 scenario_descriptor
= yaml
.load(stream
)
1409 networks
= scenario_descriptor
['scenario']['networks']
1410 networks
[test_config
["mgmt_net"]] = networks
.pop('mgmt')
1411 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
1412 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
1413 logger
.debug(scenario
)
1414 self
.__class
__.to_delete_list
.insert(0,{"item": "scenario", "function": test_config
["client"].delete_scenario
,
1415 "params":{"uuid": scenario
['scenario']['uuid']} })
1416 self
.__class
__.scenario_uuid
= scenario
['scenario']['uuid']
1418 def test_010_instantiate_scenario(self
):
1419 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1420 inspect
.currentframe().f_code
.co_name
,
1421 test_config
["test_folder"])
1422 self
.__class
__.test_index
+= 1
1424 instance
= test_config
["client"].create_instance(scenario_id
=self
.__class
__.scenario_uuid
,
1425 name
=self
.__class
__.test_text
)
1426 self
.__class
__.instance_scenario_uuid
= instance
['uuid']
1427 logger
.debug(instance
)
1428 self
.__class
__.to_delete_list
.insert(0, {"item": "instance", "function": test_config
["client"].delete_instance
,
1429 "params": {"uuid": instance
['uuid']}})
1431 def test_020_check_deployent(self
):
1432 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1433 inspect
.currentframe().f_code
.co_name
,
1434 test_config
["test_folder"])
1435 self
.__class
__.test_index
+= 1
1437 if test_config
["manual"]:
1438 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1441 keep_waiting
= test_config
["timeout"]
1442 instance_active
= False
1444 result
= check_instance_scenario_active(self
.__class
__.instance_scenario_uuid
)
1447 elif 'ERROR' in result
[1]:
1448 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
1450 raise Exception(msg
)
1452 if keep_waiting
>= 5:
1455 elif keep_waiting
> 0:
1456 time
.sleep(keep_waiting
)
1459 msg
= 'Timeout reached while waiting instance scenario to get active'
1461 raise Exception(msg
)
1463 def test_030_clean_deployment(self
):
1464 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1465 inspect
.currentframe().f_code
.co_name
,
1466 test_config
["test_folder"])
1467 self
.__class
__.test_index
+= 1
1468 #At the moment if you delete an scenario right after creating it, in openstack datacenters
1469 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1471 for item
in self
.__class
__.to_delete_list
:
1472 response
= item
["function"](**item
["params"])
1473 logger
.debug(response
)
1476 def _get_random_string(maxLength
):
1477 '''generates a string with random characters string.letters and string.digits
1478 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1482 minLength
= min_string
- len(prefix
)
1483 if maxLength
< min_string
: maxLength
= min_string
1484 maxLength
-= len(prefix
)
1485 length
= random
.randint(minLength
,maxLength
)
1486 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
1489 def test_vimconnector(args
):
1491 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1492 test_config
['vimtype'] = args
.vimtype
1493 if args
.vimtype
== "vmware":
1494 import vimconn_vmware
as vim
1496 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1498 tenant_name
= args
.tenant_name
1499 test_config
['tenant'] = tenant_name
1500 config_params
= json
.loads(args
.config_param
)
1501 org_name
= config_params
.get('orgname')
1502 org_user
= config_params
.get('user')
1503 org_passwd
= config_params
.get('passwd')
1504 vim_url
= args
.endpoint_url
1505 test_config
['image_path'] = args
.image_path
1506 test_config
['image_name'] = args
.image_name
1508 # vmware connector obj
1509 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
)
1511 elif args
.vimtype
== "aws":
1512 import vimconn_aws
as vim
1513 elif args
.vimtype
== "openstack":
1514 import vimconn_openstack
as vim
1515 elif args
.vimtype
== "openvim":
1516 import vimconn_openvim
as vim
1518 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
1522 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1523 # If only want to obtain a tests list print it and exit
1526 for cls
in clsmembers
:
1527 if cls
[0].startswith('test_vimconnector'):
1528 tests_names
.append(cls
[0])
1530 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
1535 # Create the list of tests to be run
1536 code_based_tests
= []
1538 for test
in args
.tests
:
1539 for t
in test
.split(','):
1540 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1541 if len(matches_code_based_tests
) > 0:
1542 code_based_tests
.append(matches_code_based_tests
[0][1])
1544 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1546 if not code_based_tests
:
1548 for cls
in clsmembers
:
1549 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1550 if cls
[0].startswith('test_vimconnector'):
1551 code_based_tests
.append(cls
[1])
1553 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1555 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1556 # This is handled in the tests using logging.
1557 stream
= open('/dev/null', 'w')
1559 # Run code based tests
1560 basic_tests_suite
= unittest
.TestSuite()
1561 for test
in code_based_tests
:
1562 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1563 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1564 executed
+= result
.testsRun
1565 failed
+= len(result
.failures
) + len(result
.errors
)
1566 if failfast
and failed
:
1568 if len(result
.failures
) > 0:
1569 logger
.debug("failures : {}".format(result
.failures
))
1570 if len(result
.errors
) > 0:
1571 logger
.debug("errors : {}".format(result
.errors
))
1572 return executed
, failed
1577 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1578 import openmanoclient
1581 test_config
["client"] = openmanoclient
.openmanoclient(
1582 endpoint_url
=args
.endpoint_url
,
1583 tenant_name
=args
.tenant_name
,
1584 datacenter_name
=args
.datacenter
,
1585 debug
=args
.debug
, logger
=test_config
["logger_name"])
1586 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1587 # If only want to obtain a tests list print it and exit
1590 for cls
in clsmembers
:
1591 if cls
[0].startswith('test_VIM'):
1592 tests_names
.append(cls
[0])
1594 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
1595 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1596 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1601 # Create the list of tests to be run
1602 code_based_tests
= []
1604 for test
in args
.tests
:
1605 for t
in test
.split(','):
1606 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1607 if len(matches_code_based_tests
) > 0:
1608 code_based_tests
.append(matches_code_based_tests
[0][1])
1610 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1612 if not code_based_tests
:
1614 for cls
in clsmembers
:
1615 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1616 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
1617 code_based_tests
.append(cls
[1])
1619 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1621 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1622 # This is handled in the tests using logging.
1623 stream
= open('/dev/null', 'w')
1625 # Run code based tests
1626 basic_tests_suite
= unittest
.TestSuite()
1627 for test
in code_based_tests
:
1628 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1629 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1630 executed
+= result
.testsRun
1631 failed
+= len(result
.failures
) + len(result
.errors
)
1632 if failfast
and failed
:
1634 if len(result
.failures
) > 0:
1635 logger
.debug("failures : {}".format(result
.failures
))
1636 if len(result
.errors
) > 0:
1637 logger
.debug("errors : {}".format(result
.errors
))
1638 return executed
, failed
1641 def test_deploy(args
):
1643 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1644 import openmanoclient
1647 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1648 test_config
["image_name"] = args
.image_name
1649 test_config
["mgmt_net"] = args
.mgmt_net
1650 test_config
["manual"] = args
.manual
1651 test_directory_content
= os
.listdir(test_config
["test_directory"])
1652 # If only want to obtain a tests list print it and exit
1654 msg
= "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
1659 descriptor_based_tests
= []
1660 # Create the list of tests to be run
1661 code_based_tests
= []
1663 for test
in args
.tests
:
1664 for t
in test
.split(','):
1665 if t
in test_directory_content
:
1666 descriptor_based_tests
.append(t
)
1668 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1670 if not descriptor_based_tests
:
1672 descriptor_based_tests
= test_directory_content
1674 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1676 # import openmanoclient from relative path
1677 test_config
["client"] = openmanoclient
.openmanoclient(
1678 endpoint_url
=args
.endpoint_url
,
1679 tenant_name
=args
.tenant_name
,
1680 datacenter_name
=args
.datacenter
,
1681 debug
=args
.debug
, logger
=test_config
["logger_name"])
1683 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1684 # This is handled in the tests using logging.
1685 stream
= open('/dev/null', 'w')
1686 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1687 for test
in descriptor_based_tests
:
1688 test_config
["test_folder"] = test
1689 test_suite
= unittest
.TestSuite()
1690 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
1691 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
1692 executed
+= result
.testsRun
1693 failed
+= len(result
.failures
) + len(result
.errors
)
1694 if failfast
and failed
:
1696 if len(result
.failures
) > 0:
1697 logger
.debug("failures : {}".format(result
.failures
))
1698 if len(result
.errors
) > 0:
1699 logger
.debug("errors : {}".format(result
.errors
))
1701 return executed
, failed
1703 if __name__
=="__main__":
1705 parser
= ArgumentParser(description
='Test RO module')
1706 parser
.add_argument('-v','--version', action
='version', help="Show current version",
1707 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
1710 parent_parser
= ArgumentParser(add_help
=False)
1711 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1712 dest
='failfast', action
="store_true", default
=False)
1713 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1714 dest
='failed', action
="store_true", default
=False)
1715 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
1716 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
1718 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
1719 help='Set the logger file. By default '+default_logger_file
)
1720 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
1721 help="Set the openmano tenant to use for the test. By default 'osm'")
1722 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
1723 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1724 dest
='timeout', type=int, default
=300)
1725 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
1727 subparsers
= parser
.add_subparsers(help='test sets')
1729 # Deployment test set
1730 # -------------------
1731 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
1732 help="test deployment using descriptors at RO_test folder ")
1733 deploy_parser
.set_defaults(func
=test_deploy
)
1735 # Mandatory arguments
1736 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
1737 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1738 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
1739 help='Image name available at datacenter used for the tests')
1740 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
1741 help='Set the vim management network to use for tests')
1743 # Optional arguments
1744 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
1745 help='Pause execution once deployed to allow manual checking of the '
1746 'deployed instance scenario')
1747 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1748 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1751 # -------------------
1752 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
1753 vimconn_parser
.set_defaults(func
=test_vimconnector
)
1754 # Mandatory arguments
1755 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1756 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
1757 help='Set the vimconnector type to test')
1758 mandatory_arguments
.add_argument('-c', '--config', dest
='config_param', required
=True,
1759 help='Set the vimconnector specific config parameters in dictionary format')
1760 mandatory_arguments
.add_argument('-u', '--url', dest
='endpoint_url',required
=True, help="Set the vim connector url or Host IP")
1761 # Optional arguments
1762 vimconn_parser
.add_argument('-i', '--image-path', dest
='image_path', help="Provide image path present at RO container")
1763 vimconn_parser
.add_argument('-n', '--image-name', dest
='image_name', help="Provide image name for test")
1764 # TODO add optional arguments for vimconn tests
1765 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1767 # Datacenter test set
1768 # -------------------
1769 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
1770 vimconn_parser
.set_defaults(func
=test_vim
)
1772 # Mandatory arguments
1773 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1774 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1776 # Optional arguments
1777 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1778 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1780 argcomplete
.autocomplete(parser
)
1781 args
= parser
.parse_args()
1785 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1786 logger_level
= 'INFO'
1788 logger_level
= 'DEBUG'
1790 logger_level
= 'WARNING'
1791 logger_name
= os
.path
.basename(__file__
)
1792 test_config
["logger_name"] = logger_name
1793 logger
= logging
.getLogger(logger_name
)
1794 logger
.setLevel(logger_level
)
1795 failfast
= args
.failfast
1797 # Configure a logging handler to store in a logging file
1798 if args
.logger_file
:
1799 fileHandler
= logging
.FileHandler(args
.logger_file
)
1800 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1801 fileHandler
.setFormatter(formatter_fileHandler
)
1802 logger
.addHandler(fileHandler
)
1804 # Configure a handler to print to stdout
1805 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
1806 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1807 consoleHandler
.setFormatter(formatter_consoleHandler
)
1808 logger
.addHandler(consoleHandler
)
1810 logger
.debug('Program started with the following arguments: ' + str(args
))
1812 # set test config parameters
1813 test_config
["timeout"] = args
.timeout
1814 test_config
["test_number"] = 1
1816 executed
, failed
= args
.func(args
)
1819 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
1820 sys
.exit(1 if failed
else 0)