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
41 from pyvcloud
.vcloudair
import VCA
42 from argparse
import ArgumentParser
44 __author__
= "Pablo Montes, Alfonso Tierno"
45 __date__
= "$16-Feb-2017 17:08:16$"
47 version_date
= "Oct 2017"
49 test_config
= {} # used for global variables with the test configuration
52 class test_base(unittest
.TestCase
):
58 logger
.info("{}. {}".format(test_config
["test_number"], cls
.__name
__))
61 def tearDownClass(cls
):
62 test_config
["test_number"] += 1
65 exec_info
= sys
.exc_info()
66 if exec_info
== (None, None, None):
67 logger
.info(self
.__class
__.test_text
+" -> TEST OK")
69 logger
.warning(self
.__class
__.test_text
+" -> TEST NOK")
70 logger
.critical("Traceback error",exc_info
=True)
73 def check_instance_scenario_active(uuid
):
74 instance
= test_config
["client"].get_instance(uuid
=uuid
)
76 for net
in instance
['nets']:
77 status
= net
['status']
78 if status
!= 'ACTIVE':
79 return (False, status
)
81 for vnf
in instance
['vnfs']:
84 if status
!= 'ACTIVE':
85 return (False, status
)
92 All unittest classes for code based tests must have prefix 'test_' in order to be taken into account for tests
94 class test_VIM_datacenter_tenant_operations(test_base
):
97 def test_000_create_RO_tenant(self
):
98 self
.__class
__.tenant_name
= _get_random_string(20)
99 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
100 inspect
.currentframe().f_code
.co_name
)
101 self
.__class
__.test_index
+= 1
102 tenant
= test_config
["client"].create_tenant(name
=self
.__class
__.tenant_name
,
103 description
=self
.__class
__.tenant_name
)
104 logger
.debug("{}".format(tenant
))
105 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.tenant_name
)
107 def test_010_list_RO_tenant(self
):
108 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
109 inspect
.currentframe().f_code
.co_name
)
110 self
.__class
__.test_index
+= 1
111 tenant
= test_config
["client"].get_tenant(name
=self
.__class
__.tenant_name
)
112 logger
.debug("{}".format(tenant
))
113 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.tenant_name
)
115 def test_020_delete_RO_tenant(self
):
116 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
117 inspect
.currentframe().f_code
.co_name
)
118 self
.__class
__.test_index
+= 1
119 tenant
= test_config
["client"].delete_tenant(name
=self
.__class
__.tenant_name
)
120 logger
.debug("{}".format(tenant
))
121 assert('deleted' in tenant
.get('result',""))
124 class test_VIM_datacenter_operations(test_base
):
125 datacenter_name
= None
127 def test_000_create_datacenter(self
):
128 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
129 inspect
.currentframe().f_code
.co_name
)
130 self
.__class
__.datacenter_name
= _get_random_string(20)
131 self
.__class
__.test_index
+= 1
132 self
.datacenter
= test_config
["client"].create_datacenter(name
=self
.__class
__.datacenter_name
,
133 vim_url
="http://fakeurl/fake")
134 logger
.debug("{}".format(self
.datacenter
))
135 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name',''), self
.__class
__.datacenter_name
)
137 def test_010_list_datacenter(self
):
138 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
139 inspect
.currentframe().f_code
.co_name
)
141 self
.__class
__.test_index
+= 1
142 self
.datacenter
= test_config
["client"].get_datacenter(all_tenants
=True, name
=self
.__class
__.datacenter_name
)
143 logger
.debug("{}".format(self
.datacenter
))
144 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name', ''), self
.__class
__.datacenter_name
)
146 def test_020_attach_datacenter(self
):
147 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
148 inspect
.currentframe().f_code
.co_name
)
150 self
.__class
__.test_index
+= 1
151 self
.datacenter
= test_config
["client"].attach_datacenter(name
=self
.__class
__.datacenter_name
,
152 vim_tenant_name
='fake')
153 logger
.debug("{}".format(self
.datacenter
))
154 assert ('vim_tenants' in self
.datacenter
.get('datacenter', {}))
156 def test_030_list_attached_datacenter(self
):
157 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
158 inspect
.currentframe().f_code
.co_name
)
160 self
.__class
__.test_index
+= 1
161 self
.datacenter
= test_config
["client"].get_datacenter(all_tenants
=False, name
=self
.__class
__.datacenter_name
)
162 logger
.debug("{}".format(self
.datacenter
))
163 self
.assertEqual (self
.datacenter
.get('datacenter', {}).get('name', ''), self
.__class
__.datacenter_name
)
165 def test_040_detach_datacenter(self
):
166 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
167 inspect
.currentframe().f_code
.co_name
)
169 self
.__class
__.test_index
+= 1
170 self
.datacenter
= test_config
["client"].detach_datacenter(name
=self
.__class
__.datacenter_name
)
171 logger
.debug("{}".format(self
.datacenter
))
172 assert ('detached' in self
.datacenter
.get('result', ""))
174 def test_050_delete_datacenter(self
):
175 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
176 inspect
.currentframe().f_code
.co_name
)
178 self
.__class
__.test_index
+= 1
179 self
.datacenter
= test_config
["client"].delete_datacenter(name
=self
.__class
__.datacenter_name
)
180 logger
.debug("{}".format(self
.datacenter
))
181 assert('deleted' in self
.datacenter
.get('result',""))
184 class test_VIM_network_operations(test_base
):
185 vim_network_name
= None
186 vim_network_uuid
= None
188 def test_000_create_VIM_network(self
):
189 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
190 inspect
.currentframe().f_code
.co_name
)
191 self
.__class
__.vim_network_name
= _get_random_string(20)
192 self
.__class
__.test_index
+= 1
193 network
= test_config
["client"].vim_action("create", "networks", name
=self
.__class
__.vim_network_name
)
194 logger
.debug("{}".format(network
))
195 self
.__class
__.vim_network_uuid
= network
["network"]["id"]
196 self
.assertEqual(network
.get('network', {}).get('name', ''), self
.__class
__.vim_network_name
)
198 def test_010_list_VIM_networks(self
):
199 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
200 inspect
.currentframe().f_code
.co_name
)
201 self
.__class
__.test_index
+= 1
202 networks
= test_config
["client"].vim_action("list", "networks")
203 logger
.debug("{}".format(networks
))
205 def test_020_get_VIM_network_by_uuid(self
):
206 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
207 inspect
.currentframe().f_code
.co_name
)
209 self
.__class
__.test_index
+= 1
210 network
= test_config
["client"].vim_action("show", "networks", uuid
=self
.__class
__.vim_network_uuid
)
211 logger
.debug("{}".format(network
))
212 self
.assertEqual(network
.get('network', {}).get('name', ''), self
.__class
__.vim_network_name
)
214 def test_030_delete_VIM_network_by_uuid(self
):
215 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
216 inspect
.currentframe().f_code
.co_name
)
218 self
.__class
__.test_index
+= 1
219 network
= test_config
["client"].vim_action("delete", "networks", uuid
=self
.__class
__.vim_network_uuid
)
220 logger
.debug("{}".format(network
))
221 assert ('deleted' in network
.get('result', ""))
224 class test_VIM_image_operations(test_base
):
226 def test_000_list_VIM_images(self
):
227 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
228 inspect
.currentframe().f_code
.co_name
)
229 self
.__class
__.test_index
+= 1
230 images
= test_config
["client"].vim_action("list", "images")
231 logger
.debug("{}".format(images
))
234 The following is a non critical test that will fail most of the times.
235 In case of OpenStack datacenter these tests will only success if RO has access to the admin endpoint
236 This test will only be executed in case it is specifically requested by the user
238 class test_VIM_tenant_operations(test_base
):
239 vim_tenant_name
= None
240 vim_tenant_uuid
= None
244 test_base
.setUpClass(cls
)
245 logger
.warning("In case of OpenStack datacenter these tests will only success "
246 "if RO has access to the admin endpoint")
248 def test_000_create_VIM_tenant(self
):
249 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
250 inspect
.currentframe().f_code
.co_name
)
251 self
.__class
__.vim_tenant_name
= _get_random_string(20)
252 self
.__class
__.test_index
+= 1
253 tenant
= test_config
["client"].vim_action("create", "tenants", name
=self
.__class
__.vim_tenant_name
)
254 logger
.debug("{}".format(tenant
))
255 self
.__class
__.vim_tenant_uuid
= tenant
["tenant"]["id"]
256 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.vim_tenant_name
)
258 def test_010_list_VIM_tenants(self
):
259 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
260 inspect
.currentframe().f_code
.co_name
)
261 self
.__class
__.test_index
+= 1
262 tenants
= test_config
["client"].vim_action("list", "tenants")
263 logger
.debug("{}".format(tenants
))
265 def test_020_get_VIM_tenant_by_uuid(self
):
266 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
267 inspect
.currentframe().f_code
.co_name
)
269 self
.__class
__.test_index
+= 1
270 tenant
= test_config
["client"].vim_action("show", "tenants", uuid
=self
.__class
__.vim_tenant_uuid
)
271 logger
.debug("{}".format(tenant
))
272 self
.assertEqual(tenant
.get('tenant', {}).get('name', ''), self
.__class
__.vim_tenant_name
)
274 def test_030_delete_VIM_tenant_by_uuid(self
):
275 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"], self
.__class
__.test_index
,
276 inspect
.currentframe().f_code
.co_name
)
278 self
.__class
__.test_index
+= 1
279 tenant
= test_config
["client"].vim_action("delete", "tenants", uuid
=self
.__class
__.vim_tenant_uuid
)
280 logger
.debug("{}".format(tenant
))
281 assert ('deleted' in tenant
.get('result', ""))
284 class test_vimconn_connect(test_base
):
286 def test_000_connect(self
):
287 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
288 self
.__class
__.test_index
,
289 inspect
.currentframe().f_code
.co_name
)
291 self
.__class
__.test_index
+= 1
292 if test_config
['vimtype'] == 'vmware':
293 vca_object
= test_config
["vim_conn"].connect()
294 logger
.debug("{}".format(vca_object
))
295 self
.assertIsInstance(vca_object
, VCA
)
298 class test_vimconn_new_network(test_base
):
301 def test_000_new_network(self
):
302 self
.__class
__.network_name
= _get_random_string(20)
303 network_type
= 'bridge'
305 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
306 self
.__class
__.test_index
, inspect
.currentframe().f_code
.co_name
)
307 self
.__class
__.test_index
+= 1
309 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
310 net_type
=network_type
)
311 self
.__class
__.network_id
= network
312 logger
.debug("{}".format(network
))
314 network_list
= test_config
["vim_conn"].get_network_list()
315 for net
in network_list
:
316 if self
.__class
__.network_name
in net
.get('name'):
317 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
318 self
.assertEqual(net
.get('type'), network_type
)
320 # Deleting created network
321 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
323 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
325 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
327 def test_010_new_network_by_types(self
):
329 network_types
= ['data','bridge','mgmt']
330 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
331 self
.__class
__.test_index
,
332 inspect
.currentframe().f_code
.co_name
)
333 self
.__class
__.test_index
+= 1
334 for net_type
in network_types
:
335 self
.__class
__.network_name
= _get_random_string(20)
336 network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
339 delete_net_ids
.append(network_id
)
340 logger
.debug("{}".format(network_id
))
342 network_list
= test_config
["vim_conn"].get_network_list()
343 for net
in network_list
:
344 if self
.__class
__.network_name
in net
.get('name'):
345 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
346 if net_type
in net
.get('type'):
347 self
.assertEqual(net
.get('type'), net_type
)
349 self
.assertNotEqual(net
.get('type'), net_type
)
351 # Deleting created network
352 for net_id
in delete_net_ids
:
353 result
= test_config
["vim_conn"].delete_network(net_id
)
355 logger
.info("Network id {} sucessfully deleted".format(net_id
))
357 logger
.info("Failed to delete network id {}".format(net_id
))
359 def test_020_new_network_by_ipprofile(self
):
360 test_directory_content
= os
.listdir(test_config
["test_directory"])
362 for dir_name
in test_directory_content
:
363 if dir_name
== 'simple_multi_vnfc':
364 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
365 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
368 for vnfd
in vnfd_files
:
369 with
open(vnfd
, 'r') as stream
:
370 vnf_descriptor
= yaml
.load(stream
)
372 internal_connections_list
= vnf_descriptor
['vnf']['internal-connections']
373 for item
in internal_connections_list
:
374 if 'ip-profile' in item
:
375 version
= item
['ip-profile']['ip-version']
376 dhcp_count
= item
['ip-profile']['dhcp']['count']
377 dhcp_enabled
= item
['ip-profile']['dhcp']['enabled']
379 self
.__class
__.network_name
= _get_random_string(20)
380 ip_profile
= {'dhcp_count': dhcp_count
,
381 'dhcp_enabled': dhcp_enabled
,
382 'ip_version': version
384 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
385 self
.__class
__.test_index
,
386 inspect
.currentframe().f_code
.co_name
)
387 self
.__class
__.test_index
+= 1
388 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
390 ip_profile
=ip_profile
)
391 self
.__class
__.network_id
= network
392 logger
.debug("{}".format(network
))
394 network_list
= test_config
["vim_conn"].get_network_list()
395 for net
in network_list
:
396 if self
.__class
__.network_name
in net
.get('name'):
397 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
399 # Deleting created network
400 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
402 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
404 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
406 def test_030_new_network_by_isshared(self
):
407 self
.__class
__.network_name
= _get_random_string(20)
409 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
410 self
.__class
__.test_index
,
411 inspect
.currentframe().f_code
.co_name
)
412 self
.__class
__.test_index
+= 1
413 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
416 self
.__class
__.network_id
= network
417 logger
.debug("{}".format(network
))
419 network_list
= test_config
["vim_conn"].get_network_list()
420 for net
in network_list
:
421 if self
.__class
__.network_name
in net
.get('name'):
422 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
423 self
.assertEqual(net
.get('shared'), shared
)
425 # Deleting created network
426 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
428 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
430 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
432 def test_040_new_network_by_negative(self
):
433 self
.__class
__.network_name
= _get_random_string(20)
434 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
435 self
.__class
__.test_index
,
436 inspect
.currentframe().f_code
.co_name
)
437 self
.__class
__.test_index
+= 1
438 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
439 net_type
='unknowntype')
440 self
.__class
__.network_id
= network
441 logger
.debug("{}".format(network
))
442 network_list
= test_config
["vim_conn"].get_network_list()
443 for net
in network_list
:
444 if self
.__class
__.network_name
in net
.get('name'):
445 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
447 # Deleting created network
448 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
450 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
452 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
454 def test_050_refresh_nets_status(self
):
455 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
456 self
.__class
__.test_index
,
457 inspect
.currentframe().f_code
.co_name
)
458 self
.__class
__.test_index
+= 1
459 # creating new network
460 network_name
= _get_random_string(20)
462 network_id
= test_config
["vim_conn"].new_network(net_name
=network_name
,
465 net_dict
= test_config
["vim_conn"].refresh_nets_status([network_id
])
466 for attr
in net_dict
[network_id
]:
468 self
.assertEqual(net_dict
[network_id
][attr
], 'ACTIVE')
470 # Deleting created network
471 result
= test_config
["vim_conn"].delete_network(network_id
)
473 logger
.info("Network id {} sucessfully deleted".format(network_id
))
475 logger
.info("Failed to delete network id {}".format(network_id
))
477 def test_060_refresh_nets_status_negative(self
):
478 unknown_net_id
= str(uuid
.uuid4())
479 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
480 self
.__class
__.test_index
,
481 inspect
.currentframe().f_code
.co_name
)
482 self
.__class
__.test_index
+= 1
485 net_dict
= test_config
["vim_conn"].refresh_nets_status([unknown_net_id
])
486 self
.assertEqual(net_dict
, {})
488 class test_vimconn_get_network_list(test_base
):
492 # creating new network
493 self
.__class
__.network_name
= _get_random_string(20)
494 self
.__class
__.net_type
= 'bridge'
495 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
496 net_type
=self
.__class
__.net_type
)
497 self
.__class
__.network_id
= network
498 logger
.debug("{}".format(network
))
501 test_base
.tearDown(self
)
503 # Deleting created network
504 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
506 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
508 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
510 def test_000_get_network_list(self
):
511 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
512 self
.__class
__.test_index
,
513 inspect
.currentframe().f_code
.co_name
)
514 self
.__class
__.test_index
+= 1
516 network_list
= test_config
["vim_conn"].get_network_list()
517 for net
in network_list
:
518 if self
.__class
__.network_name
in net
.get('name'):
519 self
.assertIn(self
.__class
__.network_name
, net
.get('name'))
520 self
.assertEqual(net
.get('type'), self
.__class
__.net_type
)
521 self
.assertEqual(net
.get('status'), 'ACTIVE')
522 self
.assertEqual(net
.get('shared'), False)
524 def test_010_get_network_list_by_name(self
):
525 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
526 self
.__class
__.test_index
,
527 inspect
.currentframe().f_code
.co_name
)
528 self
.__class
__.test_index
+= 1
530 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
532 # find network from list by it's name
533 new_network_list
= test_config
["vim_conn"].get_network_list({'name': network_name
})
534 for list_item
in new_network_list
:
535 if self
.__class
__.network_name
in list_item
.get('name'):
536 self
.assertEqual(network_name
, list_item
.get('name'))
537 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
538 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
540 def test_020_get_network_list_by_id(self
):
541 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
542 self
.__class
__.test_index
,
543 inspect
.currentframe().f_code
.co_name
)
544 self
.__class
__.test_index
+= 1
546 # find network from list by it's id
547 new_network_list
= test_config
["vim_conn"].get_network_list({'id':self
.__class
__.network_id
})
548 for list_item
in new_network_list
:
549 if self
.__class
__.network_id
in list_item
.get('id'):
550 self
.assertEqual(self
.__class
__.network_id
, list_item
.get('id'))
551 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
552 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
554 def test_030_get_network_list_by_shared(self
):
556 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
557 self
.__class
__.test_index
,
558 inspect
.currentframe().f_code
.co_name
)
559 self
.__class
__.test_index
+= 1
561 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
562 # find network from list by it's shared value
563 new_network_list
= test_config
["vim_conn"].get_network_list({'shared':Shared
,
564 'name':network_name
})
565 for list_item
in new_network_list
:
566 if list_item
.get('shared') == Shared
:
567 self
.assertEqual(list_item
.get('shared'), Shared
)
568 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
569 self
.assertEqual(network_name
, list_item
.get('name'))
571 def test_040_get_network_list_by_tenant_id(self
):
572 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
573 self
.__class
__.test_index
,
574 inspect
.currentframe().f_code
.co_name
)
575 self
.__class
__.test_index
+= 1
577 tenant_list
= test_config
["vim_conn"].get_tenant_list()
578 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
580 for tenant_item
in tenant_list
:
581 if test_config
['tenant'] == tenant_item
.get('name'):
582 # find network from list by it's tenant id
583 tenant_id
= tenant_item
.get('id')
584 new_network_list
= test_config
["vim_conn"].get_network_list({'tenant_id':tenant_id
,
585 'name':network_name
})
586 for list_item
in new_network_list
:
587 self
.assertEqual(tenant_id
, list_item
.get('tenant_id'))
588 self
.assertEqual(network_name
, list_item
.get('name'))
589 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
590 self
.assertEqual(list_item
.get('status'), 'ACTIVE')
592 def test_050_get_network_list_by_status(self
):
593 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
594 self
.__class
__.test_index
,
595 inspect
.currentframe().f_code
.co_name
)
596 self
.__class
__.test_index
+= 1
599 network_name
= test_config
['vim_conn'].get_network_name_by_id(self
.__class
__.network_id
)
601 # find network from list by it's status
602 new_network_list
= test_config
["vim_conn"].get_network_list({'status':status
,
603 'name': network_name
})
604 for list_item
in new_network_list
:
605 self
.assertIn(self
.__class
__.network_name
, list_item
.get('name'))
606 self
.assertEqual(list_item
.get('type'), self
.__class
__.net_type
)
607 self
.assertEqual(list_item
.get('status'), status
)
609 def test_060_get_network_list_by_negative(self
):
610 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
611 self
.__class
__.test_index
,
612 inspect
.currentframe().f_code
.co_name
)
613 self
.__class
__.test_index
+= 1
615 network_list
= test_config
["vim_conn"].get_network_list({'name': 'unknown_name'})
616 self
.assertEqual(network_list
, [])
618 class test_vimconn_get_network(test_base
):
622 # creating new network
623 self
.__class
__.network_name
= _get_random_string(20)
624 self
.__class
__.net_type
= 'bridge'
625 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
626 net_type
=self
.__class
__.net_type
)
627 self
.__class
__.network_id
= network
628 logger
.debug("{}".format(network
))
631 test_base
.tearDown(self
)
633 # Deleting created network
634 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
636 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
638 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
640 def test_000_get_network(self
):
641 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
642 self
.__class
__.test_index
,
643 inspect
.currentframe().f_code
.co_name
)
644 self
.__class
__.test_index
+= 1
646 network_info
= test_config
["vim_conn"].get_network(self
.__class
__.network_id
)
647 self
.assertEqual(network_info
.get('status'), 'ACTIVE')
648 self
.assertIn(self
.__class
__.network_name
, network_info
.get('name'))
649 self
.assertEqual(network_info
.get('type'), self
.__class
__.net_type
)
650 self
.assertEqual(network_info
.get('id'), self
.__class
__.network_id
)
652 def test_010_get_network_negative(self
):
653 Non_exist_id
= str(uuid
.uuid4())
654 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
655 self
.__class
__.test_index
,
656 inspect
.currentframe().f_code
.co_name
)
657 self
.__class
__.test_index
+= 1
659 network_info
= test_config
["vim_conn"].get_network(Non_exist_id
)
660 self
.assertEqual(network_info
, {})
662 class test_vimconn_delete_network(test_base
):
665 def test_000_delete_network(self
):
667 self
.__class
__.network_name
= _get_random_string(20)
668 self
.__class
__.net_type
= 'bridge'
669 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
670 net_type
=self
.__class
__.net_type
)
671 self
.__class
__.network_id
= network
672 logger
.debug("{}".format(network
))
674 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
675 self
.__class
__.test_index
,
676 inspect
.currentframe().f_code
.co_name
)
677 self
.__class
__.test_index
+= 1
679 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
681 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
683 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
685 # after deleting network we check in network list
686 network_list
= test_config
["vim_conn"].get_network_list({ 'id':self
.__class
__.network_id
})
687 self
.assertEqual(network_list
, [])
689 def test_010_delete_network_negative(self
):
690 Non_exist_id
= str(uuid
.uuid4())
692 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
693 self
.__class
__.test_index
,
694 inspect
.currentframe().f_code
.co_name
)
695 self
.__class
__.test_index
+= 1
697 with self
.assertRaises(Exception) as context
:
698 test_config
["vim_conn"].delete_network(Non_exist_id
)
700 self
.assertEqual((context
.exception
).http_code
, 400)
702 class test_vimconn_get_flavor(test_base
):
704 def test_000_get_flavor(self
):
705 test_directory_content
= os
.listdir(test_config
["test_directory"])
707 for dir_name
in test_directory_content
:
708 if dir_name
== 'simple_linux':
709 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
710 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
713 for vnfd
in vnfd_files
:
714 with
open(vnfd
, 'r') as stream
:
715 vnf_descriptor
= yaml
.load(stream
)
717 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
718 for item
in vnfc_list
:
719 if 'ram' in item
and 'vcpus' in item
and 'disk' in item
:
721 vcpus
= item
['vcpus']
724 flavor_data
= {'ram': ram
,
729 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
730 self
.__class
__.test_index
,
731 inspect
.currentframe().f_code
.co_name
)
732 self
.__class
__.test_index
+= 1
734 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
736 result
= test_config
["vim_conn"].get_flavor(flavor_id
)
737 self
.assertEqual(ram
, result
['ram'])
738 self
.assertEqual(vcpus
, result
['vcpus'])
739 self
.assertEqual(disk
, result
['disk'])
742 result
= test_config
["vim_conn"].delete_flavor(flavor_id
)
744 logger
.info("Flavor id {} sucessfully deleted".format(result
))
746 logger
.info("Failed to delete flavor id {}".format(result
))
748 def test_010_get_flavor_negative(self
):
749 Non_exist_flavor_id
= str(uuid
.uuid4())
751 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
752 self
.__class
__.test_index
,
753 inspect
.currentframe().f_code
.co_name
)
754 self
.__class
__.test_index
+= 1
756 with self
.assertRaises(Exception) as context
:
757 test_config
["vim_conn"].get_flavor(Non_exist_flavor_id
)
759 self
.assertEqual((context
.exception
).http_code
, 404)
761 class test_vimconn_new_flavor(test_base
):
764 def test_000_new_flavor(self
):
765 flavor_data
= {'ram': 1024, 'vpcus': 1, 'disk': 10}
767 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
768 self
.__class
__.test_index
,
769 inspect
.currentframe().f_code
.co_name
)
770 self
.__class
__.test_index
+= 1
773 self
.__class
__.flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
774 self
.assertEqual(type(self
.__class
__.flavor_id
),str)
775 self
.assertIsInstance(uuid
.UUID(self
.__class
__.flavor_id
),uuid
.UUID
)
777 def test_010_delete_flavor(self
):
778 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
779 self
.__class
__.test_index
,
780 inspect
.currentframe().f_code
.co_name
)
781 self
.__class
__.test_index
+= 1
784 result
= test_config
["vim_conn"].delete_flavor(self
.__class
__.flavor_id
)
786 logger
.info("Flavor id {} sucessfully deleted".format(result
))
788 logger
.error("Failed to delete flavor id {}".format(result
))
789 raise Exception ("Failed to delete created flavor")
791 def test_020_new_flavor_negative(self
):
792 Invalid_flavor_data
= {'ram': '1024', 'vcpus': 2.0, 'disk': 2.0}
794 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
795 self
.__class
__.test_index
,
796 inspect
.currentframe().f_code
.co_name
)
797 self
.__class
__.test_index
+= 1
799 with self
.assertRaises(Exception) as context
:
800 test_config
["vim_conn"].new_flavor(Invalid_flavor_data
)
802 self
.assertEqual((context
.exception
).http_code
, 400)
804 def test_030_delete_flavor_negative(self
):
805 Non_exist_flavor_id
= str(uuid
.uuid4())
807 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
808 self
.__class
__.test_index
,
809 inspect
.currentframe().f_code
.co_name
)
810 self
.__class
__.test_index
+= 1
812 with self
.assertRaises(Exception) as context
:
813 test_config
["vim_conn"].delete_flavor(Non_exist_flavor_id
)
815 self
.assertEqual((context
.exception
).http_code
, 404)
817 class test_vimconn_new_image(test_base
):
819 def test_000_new_image(self
):
820 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
821 self
.__class
__.test_index
,
822 inspect
.currentframe().f_code
.co_name
)
823 self
.__class
__.test_index
+= 1
825 image_path
= test_config
['image_path']
827 image_id
= test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : image_path
})
829 self
.assertEqual(type(image_id
),str)
830 self
.assertIsInstance(uuid
.UUID(image_id
),uuid
.UUID
)
832 self
.skipTest("Skipping test as image file not present at RO container")
834 def test_010_new_image_negative(self
):
835 Non_exist_image_path
= '/temp1/cirros.ovf'
837 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
838 self
.__class
__.test_index
,
839 inspect
.currentframe().f_code
.co_name
)
840 self
.__class
__.test_index
+= 1
842 with self
.assertRaises(Exception) as context
:
843 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
845 self
.assertEqual((context
.exception
).http_code
, 400)
847 def test_020_delete_image(self
):
848 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
849 self
.__class
__.test_index
,
850 inspect
.currentframe().f_code
.co_name
)
851 self
.__class
__.test_index
+= 1
853 image_id
= test_config
["vim_conn"].delete_image(self
.__class
__.image_id
)
854 self
.assertEqual(type(image_id
),str)
856 def test_030_delete_image_negative(self
):
857 Non_exist_image_id
= str(uuid
.uuid4())
859 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
860 self
.__class
__.test_index
,
861 inspect
.currentframe().f_code
.co_name
)
862 self
.__class
__.test_index
+= 1
864 with self
.assertRaises(Exception) as context
:
865 test_config
["vim_conn"].delete_image(Non_exist_image_id
)
867 self
.assertEqual((context
.exception
).http_code
, 404)
869 class test_vimconn_get_image_id_from_path(test_base
):
871 def test_000_get_image_id_from_path(self
):
872 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
873 self
.__class
__.test_index
,
874 inspect
.currentframe().f_code
.co_name
)
875 self
.__class
__.test_index
+= 1
877 image_path
= test_config
['image_path']
879 image_id
= test_config
["vim_conn"].get_image_id_from_path( image_path
)
880 self
.assertEqual(type(image_id
),str)
882 self
.skipTest("Skipping test as image file not present at RO container")
884 def test_010_get_image_id_from_path_negative(self
):
885 Non_exist_image_path
= '/temp1/cirros.ovf'
887 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
888 self
.__class
__.test_index
,
889 inspect
.currentframe().f_code
.co_name
)
890 self
.__class
__.test_index
+= 1
892 with self
.assertRaises(Exception) as context
:
893 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
895 self
.assertEqual((context
.exception
).http_code
, 400)
897 class test_vimconn_get_image_list(test_base
):
901 def test_000_get_image_list(self
):
902 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
903 self
.__class
__.test_index
,
904 inspect
.currentframe().f_code
.co_name
)
905 self
.__class
__.test_index
+= 1
906 image_list
= test_config
["vim_conn"].get_image_list()
908 for item
in image_list
:
910 self
.__class
__.image_name
= item
['name']
911 self
.__class
__.image_id
= item
['id']
912 self
.assertEqual(type(self
.__class
__.image_name
),str)
913 self
.assertEqual(type(self
.__class
__.image_id
),str)
915 def test_010_get_image_list_by_name(self
):
916 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
917 self
.__class
__.test_index
,
918 inspect
.currentframe().f_code
.co_name
)
919 self
.__class
__.test_index
+= 1
921 image_list
= test_config
["vim_conn"].get_image_list({'name': self
.__class
__.image_name
})
923 for item
in image_list
:
924 self
.assertEqual(type(item
['id']), str)
925 self
.assertEqual(item
['id'], self
.__class
__.image_id
)
926 self
.assertEqual(type(item
['name']), str)
927 self
.assertEqual(item
['name'], self
.__class
__.image_name
)
929 def test_020_get_image_list_by_id(self
):
930 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
931 self
.__class
__.test_index
,
932 inspect
.currentframe().f_code
.co_name
)
933 self
.__class
__.test_index
+= 1
935 filter_image_list
= test_config
["vim_conn"].get_image_list({'id': self
.__class
__.image_id
})
937 for item1
in filter_image_list
:
938 self
.assertEqual(type(item1
.get('id')), str)
939 self
.assertEqual(item1
.get('id'), self
.__class
__.image_id
)
940 self
.assertEqual(type(item1
.get('name')), str)
941 self
.assertEqual(item1
.get('name'), self
.__class
__.image_name
)
943 def test_030_get_image_list_negative(self
):
944 Non_exist_image_id
= uuid
.uuid4()
945 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
946 self
.__class
__.test_index
,
947 inspect
.currentframe().f_code
.co_name
)
948 self
.__class
__.test_index
+= 1
949 image_list
= test_config
["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id
})
951 self
.assertIsNotNone(image_list
, None)
952 self
.assertEqual(image_list
, [])
954 class test_vimconn_new_vminstance(test_base
):
963 self
.__class
__.network_name
= _get_random_string(20)
964 self
.__class
__.net_type
= 'bridge'
966 self
.__class
__.network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
967 net_type
=self
.__class
__.net_type
)
970 test_base
.tearDown(self
)
971 # Deleting created network
972 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
974 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
976 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
978 def test_000_new_vminstance(self
):
979 vpci
= "0000:00:11.0"
982 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
985 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
987 # find image name and image id
988 if test_config
['image_name']:
989 image_list
= test_config
['vim_conn'].get_image_list({'name': test_config
['image_name']})
990 if len(image_list
) == 0:
991 raise Exception("Image {} is not found at VIM".format(test_config
['image_name']))
993 self
.__class
__.image_id
= image_list
[0]['id']
995 image_list
= test_config
['vim_conn'].get_image_list()
996 if len(image_list
) == 0:
997 raise Exception("Not found any image at VIM")
999 self
.__class
__.image_id
= image_list
[0]['id']
1001 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1002 self
.__class
__.test_index
,
1003 inspect
.currentframe().f_code
.co_name
)
1004 self
.__class
__.test_index
+= 1
1006 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
}]
1008 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
)
1010 self
.assertEqual(type(self
.__class
__.instance_id
),str)
1012 def test_010_new_vminstance_by_model(self
):
1013 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1014 model_name
= 'e1000'
1018 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1020 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1021 self
.__class
__.test_index
,
1022 inspect
.currentframe().f_code
.co_name
)
1023 self
.__class
__.test_index
+= 1
1025 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
}]
1027 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1028 flavor_id
=flavor_id
,
1030 self
.assertEqual(type(instance_id
),str)
1031 # Deleting created vm instance
1032 logger
.info("Deleting created vm intance")
1033 test_config
["vim_conn"].delete_vminstance(instance_id
)
1036 def test_020_new_vminstance_by_net_use(self
):
1037 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1042 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1044 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1045 self
.__class
__.test_index
,
1046 inspect
.currentframe().f_code
.co_name
)
1047 self
.__class
__.test_index
+= 1
1049 net_list
= [{'use': net_use
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1051 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1052 flavor_id
=flavor_id
,
1054 self
.assertEqual(type(instance_id
),str)
1055 # Deleting created vm instance
1056 logger
.info("Deleting created vm intance")
1057 test_config
["vim_conn"].delete_vminstance(instance_id
)
1060 def test_030_new_vminstance_by_net_type(self
):
1061 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1066 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1068 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1069 self
.__class
__.test_index
,
1070 inspect
.currentframe().f_code
.co_name
)
1071 self
.__class
__.test_index
+= 1
1073 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': _type
, 'net_id': self
.__class
__.network_id
}]
1075 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1076 flavor_id
=flavor_id
,
1078 self
.assertEqual(type(instance_id
),str)
1079 # Deleting created vm instance
1080 logger
.info("Deleting created vm intance")
1081 test_config
["vim_conn"].delete_vminstance(instance_id
)
1084 def test_040_new_vminstance_by_cloud_config(self
):
1085 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1087 user_name
= 'test_user'
1089 key_pairs
= ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com']
1091 users_data
= [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com'], 'name': user_name
}]
1093 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
}
1096 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1098 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1099 self
.__class
__.test_index
,
1100 inspect
.currentframe().f_code
.co_name
)
1101 self
.__class
__.test_index
+= 1
1103 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1105 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Cloud_vm', image_id
=self
.__class
__.image_id
,
1106 flavor_id
=flavor_id
,
1108 cloud_config
=cloud_data
)
1109 self
.assertEqual(type(instance_id
),str)
1110 # Deleting created vm instance
1111 logger
.info("Deleting created vm intance")
1112 test_config
["vim_conn"].delete_vminstance(instance_id
)
1115 def test_050_new_vminstance_by_disk_list(self
):
1116 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1119 device_data
= [{'image_id': self
.__class
__.image_id
, 'size': '5'}]
1122 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1124 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1125 self
.__class
__.test_index
,
1126 inspect
.currentframe().f_code
.co_name
)
1127 self
.__class
__.test_index
+= 1
1129 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1131 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='VM_test1', image_id
=self
.__class
__.image_id
,
1132 flavor_id
=flavor_id
,
1134 disk_list
=device_data
)
1135 self
.assertEqual(type(instance_id
),str)
1136 # Deleting created vm instance
1137 logger
.info("Deleting created vm intance")
1138 test_config
["vim_conn"].delete_vminstance(instance_id
)
1141 def test_060_new_vminstance_negative(self
):
1142 unknown_flavor_id
= str(uuid
.uuid4())
1143 unknown_image_id
= str(uuid
.uuid4())
1146 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1147 self
.__class
__.test_index
,
1148 inspect
.currentframe().f_code
.co_name
)
1149 self
.__class
__.test_index
+= 1
1151 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1153 with self
.assertRaises(Exception) as context
:
1154 test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=unknown_image_id
,
1155 flavor_id
=unknown_flavor_id
,
1157 self
.assertEqual((context
.exception
).http_code
, 404)
1159 def test_070_get_vminstance(self
):
1160 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1161 self
.__class
__.test_index
,
1162 inspect
.currentframe().f_code
.co_name
)
1163 self
.__class
__.test_index
+= 1
1165 # Get instance by its id
1166 vm_info
= test_config
["vim_conn"].get_vminstance(self
.__class
__.instance_id
)
1168 if test_config
['vimtype'] == 'vmware':
1169 for attr
in vm_info
:
1170 if attr
== 'status':
1171 self
.assertEqual(vm_info
[attr
], 'ACTIVE')
1172 if attr
== 'hostId':
1173 self
.assertEqual(type(vm_info
[attr
]), str)
1174 if attr
== 'interfaces':
1175 self
.assertEqual(type(vm_info
[attr
]), list)
1176 self
.assertEqual(vm_info
[attr
][0]['IsConnected'], 'true')
1177 if attr
== 'IsEnabled':
1178 self
.assertEqual(vm_info
[attr
], 'true')
1180 def test_080_get_vminstance_negative(self
):
1181 unknown_instance_id
= str(uuid
.uuid4())
1183 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1184 self
.__class
__.test_index
,
1185 inspect
.currentframe().f_code
.co_name
)
1186 self
.__class
__.test_index
+= 1
1188 with self
.assertRaises(Exception) as context
:
1189 test_config
["vim_conn"].get_vminstance(unknown_instance_id
)
1191 self
.assertEqual((context
.exception
).http_code
, 404)
1193 def test_090_refresh_vms_status(self
):
1194 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1195 self
.__class
__.test_index
,
1196 inspect
.currentframe().f_code
.co_name
)
1197 self
.__class
__.test_index
+= 1
1199 vm_list
.append(self
.__class
__.instance_id
)
1202 vm_info
= test_config
["vim_conn"].refresh_vms_status(vm_list
)
1203 for attr
in vm_info
[self
.__class
__.instance_id
]:
1204 if attr
== 'status':
1205 self
.assertEqual(vm_info
[self
.__class
__.instance_id
][attr
], 'ACTIVE')
1206 if attr
== 'interfaces':
1207 self
.assertEqual(type(vm_info
[self
.__class
__.instance_id
][attr
]), list)
1209 def test_100_refresh_vms_status_negative(self
):
1210 unknown_id
= str(uuid
.uuid4())
1212 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1213 self
.__class
__.test_index
,
1214 inspect
.currentframe().f_code
.co_name
)
1215 self
.__class
__.test_index
+= 1
1217 vm_dict
= test_config
["vim_conn"].refresh_vms_status([unknown_id
])
1218 self
.assertEqual(vm_dict
, {})
1220 def test_110_action_vminstance(self
):
1221 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1222 self
.__class
__.test_index
,
1223 inspect
.currentframe().f_code
.co_name
)
1224 self
.__class
__.test_index
+= 1
1226 action_list
= ['shutdown','start','shutoff','rebuild','pause','resume']
1227 # various action on vminstace
1228 for action
in action_list
:
1229 instance_id
= test_config
["vim_conn"].action_vminstance(self
.__class
__.instance_id
,
1231 self
.assertEqual(instance_id
, self
.__class
__.instance_id
)
1233 def test_120_action_vminstance_negative(self
):
1234 non_exist_id
= str(uuid
.uuid4())
1235 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1236 self
.__class
__.test_index
,
1237 inspect
.currentframe().f_code
.co_name
)
1238 self
.__class
__.test_index
+= 1
1241 with self
.assertRaises(Exception) as context
:
1242 test_config
["vim_conn"].action_vminstance(non_exist_id
, { action
: None})
1244 self
.assertEqual((context
.exception
).http_code
, 400)
1246 def test_130_delete_vminstance(self
):
1247 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1248 self
.__class
__.test_index
,
1249 inspect
.currentframe().f_code
.co_name
)
1250 self
.__class
__.test_index
+= 1
1252 # Deleting created vm instance
1253 logger
.info("Deleting created vm instance")
1254 test_config
["vim_conn"].delete_vminstance(self
.__class
__.instance_id
)
1257 class test_vimconn_get_tenant_list(test_base
):
1260 def test_000_get_tenant_list(self
):
1261 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1262 self
.__class
__.test_index
,
1263 inspect
.currentframe().f_code
.co_name
)
1264 self
.__class
__.test_index
+= 1
1266 # Getting tenant list
1267 tenant_list
= test_config
["vim_conn"].get_tenant_list()
1269 for item
in tenant_list
:
1270 if test_config
['tenant'] == item
['name']:
1271 self
.__class
__.tenant_id
= item
['id']
1272 self
.assertEqual(type(item
['name']), str)
1273 self
.assertEqual(type(item
['id']), str)
1275 def test_010_get_tenant_list_by_id(self
):
1276 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1277 self
.__class
__.test_index
,
1278 inspect
.currentframe().f_code
.co_name
)
1279 self
.__class
__.test_index
+= 1
1281 # Getting filter tenant list by its id
1282 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'id': self
.__class
__.tenant_id
})
1284 for item
in filter_tenant_list
:
1285 self
.assertEqual(type(item
['id']), str)
1286 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1288 def test_020_get_tenant_list_by_name(self
):
1289 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1290 self
.__class
__.test_index
,
1291 inspect
.currentframe().f_code
.co_name
)
1292 self
.__class
__.test_index
+= 1
1294 # Getting filter tenant list by its name
1295 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant']})
1297 for item
in filter_tenant_list
:
1298 self
.assertEqual(type(item
['name']), str)
1299 self
.assertEqual(item
['name'], test_config
['tenant'])
1301 def test_030_get_tenant_list_by_name_and_id(self
):
1302 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1303 self
.__class
__.test_index
,
1304 inspect
.currentframe().f_code
.co_name
)
1305 self
.__class
__.test_index
+= 1
1307 # Getting filter tenant list by its name and id
1308 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant'],
1309 'id': self
.__class
__.tenant_id
})
1311 for item
in filter_tenant_list
:
1312 self
.assertEqual(type(item
['name']), str)
1313 self
.assertEqual(type(item
['id']), str)
1314 self
.assertEqual(item
['name'], test_config
['tenant'])
1315 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1317 def test_040_get_tenant_list_negative(self
):
1318 non_exist_tenant_name
= "Tenant_123"
1319 non_exist_tenant_id
= "kjhgrt456-45345kjhdfgnbdk-34dsfjdfg"
1320 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1321 self
.__class
__.test_index
,
1322 inspect
.currentframe().f_code
.co_name
)
1323 self
.__class
__.test_index
+= 1
1325 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': non_exist_tenant_name
,
1326 'id': non_exist_tenant_id
})
1328 self
.assertEqual(filter_tenant_list
, [])
1330 class test_vimconn_new_tenant(test_base
):
1333 def test_000_new_tenant(self
):
1334 tenant_name
= _get_random_string(20)
1335 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1336 self
.__class
__.test_index
,
1337 inspect
.currentframe().f_code
.co_name
)
1338 self
.__class
__.test_index
+= 1
1340 self
.__class
__.tenant_id
= test_config
["vim_conn"].new_tenant(tenant_name
)
1343 self
.assertEqual(type(self
.__class
__.tenant_id
), str)
1345 def test_010_new_tenant_negative(self
):
1346 Invalid_tenant_name
= 10121
1347 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1348 self
.__class
__.test_index
,
1349 inspect
.currentframe().f_code
.co_name
)
1350 self
.__class
__.test_index
+= 1
1352 with self
.assertRaises(Exception) as context
:
1353 test_config
["vim_conn"].new_tenant(Invalid_tenant_name
)
1355 self
.assertEqual((context
.exception
).http_code
, 400)
1357 def test_020_delete_tenant(self
):
1358 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1359 self
.__class
__.test_index
,
1360 inspect
.currentframe().f_code
.co_name
)
1361 self
.__class
__.test_index
+= 1
1363 tenant_id
= test_config
["vim_conn"].delete_tenant(self
.__class
__.tenant_id
)
1364 self
.assertEqual(type(tenant_id
), str)
1366 def test_030_delete_tenant_negative(self
):
1367 Non_exist_tenant_name
= 'Test_30_tenant'
1368 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1369 self
.__class
__.test_index
,
1370 inspect
.currentframe().f_code
.co_name
)
1371 self
.__class
__.test_index
+= 1
1373 with self
.assertRaises(Exception) as context
:
1374 test_config
["vim_conn"].delete_tenant(Non_exist_tenant_name
)
1376 self
.assertEqual((context
.exception
).http_code
, 404)
1381 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
1382 scenario based tests.
1384 class descriptor_based_scenario_test(test_base
):
1386 scenario_test_path
= None
1389 def setUpClass(cls
):
1391 cls
.to_delete_list
= []
1392 cls
.scenario_uuids
= []
1393 cls
.instance_scenario_uuids
= []
1394 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
1395 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
1398 def tearDownClass(cls
):
1399 test_config
["test_number"] += 1
1401 def test_000_load_scenario(self
):
1402 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1403 inspect
.currentframe().f_code
.co_name
,
1404 test_config
["test_folder"])
1405 self
.__class
__.test_index
+= 1
1407 descriptor_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/*.yaml')
1408 vnf_descriptors
= []
1409 scenario_descriptors
= []
1410 for descriptor_file
in descriptor_files
:
1411 with
open(descriptor_file
, 'r') as stream
:
1412 descriptor
= yaml
.load(stream
)
1413 if "vnf" in descriptor
or "vnfd:vnfd-catalog" in descriptor
or "vnfd-catalog" in descriptor
:
1414 vnf_descriptors
.append(descriptor
)
1416 scenario_descriptors
.append(descriptor
)
1418 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
1419 if not vnf_descriptors
or not scenario_descriptors
or len(scenario_descriptors
) > 1:
1420 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
1421 test_config
["test_folder"]))
1424 for vnf_descriptor
in vnf_descriptors
:
1425 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
1426 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
, image_name
=test_config
["image_name"])
1429 vnf_uuid
= vnf
['vnf']['uuid']
1431 vnf_uuid
= vnf
['vnfd'][0]['uuid']
1432 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
1433 "params": {"uuid": vnf_uuid
}})
1435 # load the scenario definition
1436 for scenario_descriptor
in scenario_descriptors
:
1437 # networks = scenario_descriptor['scenario']['networks']
1438 # networks[test_config["mgmt_net"]] = networks.pop('mgmt')
1439 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
1440 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
1441 logger
.debug(scenario
)
1442 if 'scenario' in scenario
:
1443 scenario_uuid
= scenario
['scenario']['uuid']
1445 scenario_uuid
= scenario
['nsd'][0]['uuid']
1446 self
.__class
__.to_delete_list
.insert(0, {"item": "scenario",
1447 "function": test_config
["client"].delete_scenario
,
1448 "params": {"uuid": scenario_uuid
}})
1449 self
.__class
__.scenario_uuids
.append(scenario_uuid
)
1451 def test_010_instantiate_scenario(self
):
1452 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1453 inspect
.currentframe().f_code
.co_name
,
1454 test_config
["test_folder"])
1455 self
.__class
__.test_index
+= 1
1456 for scenario_uuid
in self
.__class
__.scenario_uuids
:
1457 instance_descriptor
= {
1459 "name": self
.__class
__.test_text
,
1460 "scenario": scenario_uuid
,
1462 "mgmt": {"sites": [ { "netmap-use": test_config
["mgmt_net"]} ]}
1466 instance
= test_config
["client"].create_instance(instance_descriptor
)
1467 self
.__class
__.instance_scenario_uuids
.append(instance
['uuid'])
1468 logger
.debug(instance
)
1469 self
.__class
__.to_delete_list
.insert(0, {"item": "instance",
1470 "function": test_config
["client"].delete_instance
,
1471 "params": {"uuid": instance
['uuid']}})
1473 def test_020_check_deployent(self
):
1474 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1475 inspect
.currentframe().f_code
.co_name
,
1476 test_config
["test_folder"])
1477 self
.__class
__.test_index
+= 1
1479 if test_config
["manual"]:
1480 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1483 keep_waiting
= test_config
["timeout"]
1484 pending_instance_scenario_uuids
= list(self
.__class
__.instance_scenario_uuids
) # make a copy
1485 while pending_instance_scenario_uuids
:
1487 while index
< len(pending_instance_scenario_uuids
):
1488 result
= check_instance_scenario_active(pending_instance_scenario_uuids
[index
])
1490 del pending_instance_scenario_uuids
[index
]
1492 elif 'ERROR' in result
[1]:
1493 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
1495 raise Exception(msg
)
1498 if keep_waiting
>= 5:
1501 elif keep_waiting
> 0:
1502 time
.sleep(keep_waiting
)
1505 msg
= 'Timeout reached while waiting instance scenario to get active'
1507 raise Exception(msg
)
1509 def test_030_clean_deployment(self
):
1510 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1511 inspect
.currentframe().f_code
.co_name
,
1512 test_config
["test_folder"])
1513 self
.__class
__.test_index
+= 1
1514 #At the moment if you delete an scenario right after creating it, in openstack datacenters
1515 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1517 for item
in self
.__class
__.to_delete_list
:
1518 response
= item
["function"](**item
["params"])
1519 logger
.debug(response
)
1522 def _get_random_string(maxLength
):
1523 '''generates a string with random characters string.letters and string.digits
1524 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1528 minLength
= min_string
- len(prefix
)
1529 if maxLength
< min_string
: maxLength
= min_string
1530 maxLength
-= len(prefix
)
1531 length
= random
.randint(minLength
,maxLength
)
1532 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
1535 def test_vimconnector(args
):
1537 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1538 test_config
['vimtype'] = args
.vimtype
1539 if args
.vimtype
== "vmware":
1540 import vimconn_vmware
as vim
1542 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1544 tenant_name
= args
.tenant_name
1545 test_config
['tenant'] = tenant_name
1546 config_params
= json
.loads(args
.config_param
)
1547 org_name
= config_params
.get('orgname')
1548 org_user
= config_params
.get('user')
1549 org_passwd
= config_params
.get('passwd')
1550 vim_url
= args
.endpoint_url
1551 test_config
['image_path'] = args
.image_path
1552 test_config
['image_name'] = args
.image_name
1554 # vmware connector obj
1555 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
)
1557 elif args
.vimtype
== "aws":
1558 import vimconn_aws
as vim
1559 elif args
.vimtype
== "openstack":
1560 import vimconn_openstack
as vim
1561 elif args
.vimtype
== "openvim":
1562 import vimconn_openvim
as vim
1564 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
1568 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1569 # If only want to obtain a tests list print it and exit
1572 for cls
in clsmembers
:
1573 if cls
[0].startswith('test_vimconnector'):
1574 tests_names
.append(cls
[0])
1576 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
1581 # Create the list of tests to be run
1582 code_based_tests
= []
1584 for test
in args
.tests
:
1585 for t
in test
.split(','):
1586 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1587 if len(matches_code_based_tests
) > 0:
1588 code_based_tests
.append(matches_code_based_tests
[0][1])
1590 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1592 if not code_based_tests
:
1594 for cls
in clsmembers
:
1595 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1596 if cls
[0].startswith('test_vimconnector'):
1597 code_based_tests
.append(cls
[1])
1599 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1601 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1602 # This is handled in the tests using logging.
1603 stream
= open('/dev/null', 'w')
1605 # Run code based tests
1606 basic_tests_suite
= unittest
.TestSuite()
1607 for test
in code_based_tests
:
1608 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1609 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1610 executed
+= result
.testsRun
1611 failed
+= len(result
.failures
) + len(result
.errors
)
1612 if failfast
and failed
:
1614 if len(result
.failures
) > 0:
1615 logger
.debug("failures : {}".format(result
.failures
))
1616 if len(result
.errors
) > 0:
1617 logger
.debug("errors : {}".format(result
.errors
))
1618 return executed
, failed
1623 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1624 import openmanoclient
1627 test_config
["client"] = openmanoclient
.openmanoclient(
1628 endpoint_url
=args
.endpoint_url
,
1629 tenant_name
=args
.tenant_name
,
1630 datacenter_name
=args
.datacenter
,
1631 debug
=args
.debug
, logger
=test_config
["logger_name"])
1632 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1633 # If only want to obtain a tests list print it and exit
1636 for cls
in clsmembers
:
1637 if cls
[0].startswith('test_VIM'):
1638 tests_names
.append(cls
[0])
1640 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
1641 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1642 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1647 # Create the list of tests to be run
1648 code_based_tests
= []
1650 for test
in args
.tests
:
1651 for t
in test
.split(','):
1652 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1653 if len(matches_code_based_tests
) > 0:
1654 code_based_tests
.append(matches_code_based_tests
[0][1])
1656 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1658 if not code_based_tests
:
1660 for cls
in clsmembers
:
1661 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1662 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
1663 code_based_tests
.append(cls
[1])
1665 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1667 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1668 # This is handled in the tests using logging.
1669 stream
= open('/dev/null', 'w')
1671 # Run code based tests
1672 basic_tests_suite
= unittest
.TestSuite()
1673 for test
in code_based_tests
:
1674 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1675 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1676 executed
+= result
.testsRun
1677 failed
+= len(result
.failures
) + len(result
.errors
)
1678 if failfast
and failed
:
1680 if len(result
.failures
) > 0:
1681 logger
.debug("failures : {}".format(result
.failures
))
1682 if len(result
.errors
) > 0:
1683 logger
.debug("errors : {}".format(result
.errors
))
1684 return executed
, failed
1687 def test_deploy(args
):
1689 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1690 import openmanoclient
1693 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1694 test_config
["image_name"] = args
.image_name
1695 test_config
["mgmt_net"] = args
.mgmt_net
1696 test_config
["manual"] = args
.manual
1697 test_directory_content
= os
.listdir(test_config
["test_directory"])
1698 # If only want to obtain a tests list print it and exit
1700 msg
= "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
1705 descriptor_based_tests
= []
1706 # Create the list of tests to be run
1707 code_based_tests
= []
1709 for test
in args
.tests
:
1710 for t
in test
.split(','):
1711 if t
in test_directory_content
:
1712 descriptor_based_tests
.append(t
)
1714 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1716 if not descriptor_based_tests
:
1718 descriptor_based_tests
= test_directory_content
1720 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1722 # import openmanoclient from relative path
1723 test_config
["client"] = openmanoclient
.openmanoclient(
1724 endpoint_url
=args
.endpoint_url
,
1725 tenant_name
=args
.tenant_name
,
1726 datacenter_name
=args
.datacenter
,
1727 debug
=args
.debug
, logger
=test_config
["logger_name"])
1729 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1730 # This is handled in the tests using logging.
1731 stream
= open('/dev/null', 'w')
1732 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1733 for test
in descriptor_based_tests
:
1734 test_config
["test_folder"] = test
1735 test_suite
= unittest
.TestSuite()
1736 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
1737 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
1738 executed
+= result
.testsRun
1739 failed
+= len(result
.failures
) + len(result
.errors
)
1740 if failfast
and failed
:
1742 if len(result
.failures
) > 0:
1743 logger
.debug("failures : {}".format(result
.failures
))
1744 if len(result
.errors
) > 0:
1745 logger
.debug("errors : {}".format(result
.errors
))
1747 return executed
, failed
1749 if __name__
=="__main__":
1751 parser
= ArgumentParser(description
='Test RO module')
1752 parser
.add_argument('-v','--version', action
='version', help="Show current version",
1753 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
1756 parent_parser
= ArgumentParser(add_help
=False)
1757 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1758 dest
='failfast', action
="store_true", default
=False)
1759 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1760 dest
='failed', action
="store_true", default
=False)
1761 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
1762 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
1764 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
1765 help='Set the logger file. By default '+default_logger_file
)
1766 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
1767 help="Set the openmano tenant to use for the test. By default 'osm'")
1768 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
1769 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1770 dest
='timeout', type=int, default
=300)
1771 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
1773 subparsers
= parser
.add_subparsers(help='test sets')
1775 # Deployment test set
1776 # -------------------
1777 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
1778 help="test deployment using descriptors at RO_test folder ")
1779 deploy_parser
.set_defaults(func
=test_deploy
)
1781 # Mandatory arguments
1782 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
1783 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1784 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
1785 help='Image name available at datacenter used for the tests')
1786 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
1787 help='Set the vim management network to use for tests')
1789 # Optional arguments
1790 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
1791 help='Pause execution once deployed to allow manual checking of the '
1792 'deployed instance scenario')
1793 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1794 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1797 # -------------------
1798 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
1799 vimconn_parser
.set_defaults(func
=test_vimconnector
)
1800 # Mandatory arguments
1801 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1802 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
1803 help='Set the vimconnector type to test')
1804 mandatory_arguments
.add_argument('-c', '--config', dest
='config_param', required
=True,
1805 help='Set the vimconnector specific config parameters in dictionary format')
1806 mandatory_arguments
.add_argument('-u', '--url', dest
='endpoint_url',required
=True, help="Set the vim connector url or Host IP")
1807 # Optional arguments
1808 vimconn_parser
.add_argument('-i', '--image-path', dest
='image_path', help="Provide image path present at RO container")
1809 vimconn_parser
.add_argument('-n', '--image-name', dest
='image_name', help="Provide image name for test")
1810 # TODO add optional arguments for vimconn tests
1811 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1813 # Datacenter test set
1814 # -------------------
1815 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
1816 vimconn_parser
.set_defaults(func
=test_vim
)
1818 # Mandatory arguments
1819 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1820 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1822 # Optional arguments
1823 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1824 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1826 argcomplete
.autocomplete(parser
)
1827 args
= parser
.parse_args()
1831 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1832 logger_level
= 'INFO'
1834 logger_level
= 'DEBUG'
1836 logger_level
= 'WARNING'
1837 logger_name
= os
.path
.basename(__file__
)
1838 test_config
["logger_name"] = logger_name
1839 logger
= logging
.getLogger(logger_name
)
1840 logger
.setLevel(logger_level
)
1841 failfast
= args
.failfast
1843 # Configure a logging handler to store in a logging file
1844 if args
.logger_file
:
1845 fileHandler
= logging
.FileHandler(args
.logger_file
)
1846 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1847 fileHandler
.setFormatter(formatter_fileHandler
)
1848 logger
.addHandler(fileHandler
)
1850 # Configure a handler to print to stdout
1851 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
1852 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1853 consoleHandler
.setFormatter(formatter_consoleHandler
)
1854 logger
.addHandler(consoleHandler
)
1856 logger
.debug('Program started with the following arguments: ' + str(args
))
1858 # set test config parameters
1859 test_config
["timeout"] = args
.timeout
1860 test_config
["test_number"] = 1
1862 executed
, failed
= args
.func(args
)
1865 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
1866 sys
.exit(1 if failed
else 0)