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 argparse
import ArgumentParser
43 __author__
= "Pablo Montes, Alfonso Tierno"
44 __date__
= "$16-Feb-2017 17:08:16$"
46 version_date
= "Oct 2017"
48 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 ('uuid' 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', ""))
283 class test_vimconn_connect(test_base
):
285 def test_000_connect(self
):
286 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
287 self
.__class
__.test_index
,
288 inspect
.currentframe().f_code
.co_name
)
290 self
.__class
__.test_index
+= 1
291 if test_config
['vimtype'] == 'vmware':
292 vca_object
= test_config
["vim_conn"].connect()
293 logger
.debug("{}".format(vca_object
))
294 self
.assertIsNotNone(vca_object
)
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
656 with self
.assertRaises(Exception) as context
:
657 test_config
["vim_conn"].get_network(Non_exist_id
)
659 self
.assertEqual((context
.exception
).http_code
, 404)
661 class test_vimconn_delete_network(test_base
):
664 def test_000_delete_network(self
):
666 self
.__class
__.network_name
= _get_random_string(20)
667 self
.__class
__.net_type
= 'bridge'
668 network
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
669 net_type
=self
.__class
__.net_type
)
670 self
.__class
__.network_id
= network
671 logger
.debug("{}".format(network
))
673 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
674 self
.__class
__.test_index
,
675 inspect
.currentframe().f_code
.co_name
)
676 self
.__class
__.test_index
+= 1
678 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
680 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
682 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
684 # after deleting network we check in network list
685 network_list
= test_config
["vim_conn"].get_network_list({ 'id':self
.__class
__.network_id
})
686 self
.assertEqual(network_list
, [])
688 def test_010_delete_network_negative(self
):
689 Non_exist_id
= str(uuid
.uuid4())
691 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
692 self
.__class
__.test_index
,
693 inspect
.currentframe().f_code
.co_name
)
694 self
.__class
__.test_index
+= 1
696 with self
.assertRaises(Exception) as context
:
697 test_config
["vim_conn"].delete_network(Non_exist_id
)
699 self
.assertEqual((context
.exception
).http_code
, 400)
701 class test_vimconn_get_flavor(test_base
):
703 def test_000_get_flavor(self
):
704 test_directory_content
= os
.listdir(test_config
["test_directory"])
706 for dir_name
in test_directory_content
:
707 if dir_name
== 'simple_linux':
708 self
.__class
__.scenario_test_path
= test_config
["test_directory"] + '/'+ dir_name
709 vnfd_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/vnfd_*.yaml')
712 for vnfd
in vnfd_files
:
713 with
open(vnfd
, 'r') as stream
:
714 vnf_descriptor
= yaml
.load(stream
)
716 vnfc_list
= vnf_descriptor
['vnf']['VNFC']
717 for item
in vnfc_list
:
718 if 'ram' in item
and 'vcpus' in item
and 'disk' in item
:
720 vcpus
= item
['vcpus']
723 flavor_data
= {'ram': ram
,
728 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
729 self
.__class
__.test_index
,
730 inspect
.currentframe().f_code
.co_name
)
731 self
.__class
__.test_index
+= 1
733 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
735 result
= test_config
["vim_conn"].get_flavor(flavor_id
)
736 self
.assertEqual(ram
, result
['ram'])
737 self
.assertEqual(vcpus
, result
['vcpus'])
738 self
.assertEqual(disk
, result
['disk'])
741 result
= test_config
["vim_conn"].delete_flavor(flavor_id
)
743 logger
.info("Flavor id {} sucessfully deleted".format(result
))
745 logger
.info("Failed to delete flavor id {}".format(result
))
747 def test_010_get_flavor_negative(self
):
748 Non_exist_flavor_id
= str(uuid
.uuid4())
750 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
751 self
.__class
__.test_index
,
752 inspect
.currentframe().f_code
.co_name
)
753 self
.__class
__.test_index
+= 1
755 with self
.assertRaises(Exception) as context
:
756 test_config
["vim_conn"].get_flavor(Non_exist_flavor_id
)
758 self
.assertEqual((context
.exception
).http_code
, 404)
760 class test_vimconn_new_flavor(test_base
):
763 def test_000_new_flavor(self
):
764 flavor_data
= {'ram': 1024, 'vpcus': 1, 'disk': 10}
766 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
767 self
.__class
__.test_index
,
768 inspect
.currentframe().f_code
.co_name
)
769 self
.__class
__.test_index
+= 1
772 self
.__class
__.flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
773 self
.assertEqual(type(self
.__class
__.flavor_id
),str)
774 self
.assertIsInstance(uuid
.UUID(self
.__class
__.flavor_id
),uuid
.UUID
)
776 def test_010_delete_flavor(self
):
777 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
778 self
.__class
__.test_index
,
779 inspect
.currentframe().f_code
.co_name
)
780 self
.__class
__.test_index
+= 1
783 result
= test_config
["vim_conn"].delete_flavor(self
.__class
__.flavor_id
)
785 logger
.info("Flavor id {} sucessfully deleted".format(result
))
787 logger
.error("Failed to delete flavor id {}".format(result
))
788 raise Exception ("Failed to delete created flavor")
790 def test_020_new_flavor_negative(self
):
791 Invalid_flavor_data
= {'ram': '1024', 'vcpus': 2.0, 'disk': 2.0}
793 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
794 self
.__class
__.test_index
,
795 inspect
.currentframe().f_code
.co_name
)
796 self
.__class
__.test_index
+= 1
798 with self
.assertRaises(Exception) as context
:
799 test_config
["vim_conn"].new_flavor(Invalid_flavor_data
)
801 self
.assertEqual((context
.exception
).http_code
, 400)
803 def test_030_delete_flavor_negative(self
):
804 Non_exist_flavor_id
= str(uuid
.uuid4())
806 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
807 self
.__class
__.test_index
,
808 inspect
.currentframe().f_code
.co_name
)
809 self
.__class
__.test_index
+= 1
811 with self
.assertRaises(Exception) as context
:
812 test_config
["vim_conn"].delete_flavor(Non_exist_flavor_id
)
814 self
.assertEqual((context
.exception
).http_code
, 404)
816 class test_vimconn_new_image(test_base
):
818 def test_000_new_image(self
):
819 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
820 self
.__class
__.test_index
,
821 inspect
.currentframe().f_code
.co_name
)
822 self
.__class
__.test_index
+= 1
824 image_path
= test_config
['image_path']
826 self
.__class
__.image_id
= test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : image_path
})
828 self
.assertEqual(type(self
.__class
__.image_id
),str)
829 self
.assertIsInstance(uuid
.UUID(self
.__class
__.image_id
),uuid
.UUID
)
831 self
.skipTest("Skipping test as image file not present at RO container")
833 def test_010_new_image_negative(self
):
834 Non_exist_image_path
= '/temp1/cirros.ovf'
836 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
837 self
.__class
__.test_index
,
838 inspect
.currentframe().f_code
.co_name
)
839 self
.__class
__.test_index
+= 1
841 with self
.assertRaises(Exception) as context
:
842 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
844 self
.assertEqual((context
.exception
).http_code
, 400)
846 def test_020_delete_image(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_id
= test_config
["vim_conn"].delete_image(self
.__class
__.image_id
)
853 self
.assertEqual(type(image_id
),str)
855 def test_030_delete_image_negative(self
):
856 Non_exist_image_id
= str(uuid
.uuid4())
858 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
859 self
.__class
__.test_index
,
860 inspect
.currentframe().f_code
.co_name
)
861 self
.__class
__.test_index
+= 1
863 with self
.assertRaises(Exception) as context
:
864 test_config
["vim_conn"].delete_image(Non_exist_image_id
)
866 self
.assertEqual((context
.exception
).http_code
, 404)
868 class test_vimconn_get_image_id_from_path(test_base
):
870 def test_000_get_image_id_from_path(self
):
871 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
872 self
.__class
__.test_index
,
873 inspect
.currentframe().f_code
.co_name
)
874 self
.__class
__.test_index
+= 1
876 image_path
= test_config
['image_path']
878 image_id
= test_config
["vim_conn"].get_image_id_from_path( image_path
)
879 self
.assertEqual(type(image_id
),str)
881 self
.skipTest("Skipping test as image file not present at RO container")
883 def test_010_get_image_id_from_path_negative(self
):
884 Non_exist_image_path
= '/temp1/cirros.ovf'
886 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
887 self
.__class
__.test_index
,
888 inspect
.currentframe().f_code
.co_name
)
889 self
.__class
__.test_index
+= 1
891 with self
.assertRaises(Exception) as context
:
892 test_config
["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path
})
894 self
.assertEqual((context
.exception
).http_code
, 400)
896 class test_vimconn_get_image_list(test_base
):
900 def test_000_get_image_list(self
):
901 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
902 self
.__class
__.test_index
,
903 inspect
.currentframe().f_code
.co_name
)
904 self
.__class
__.test_index
+= 1
905 image_list
= test_config
["vim_conn"].get_image_list()
907 for item
in image_list
:
909 self
.__class
__.image_name
= item
['name']
910 self
.__class
__.image_id
= item
['id']
911 self
.assertEqual(type(self
.__class
__.image_name
),str)
912 self
.assertEqual(type(self
.__class
__.image_id
),str)
914 def test_010_get_image_list_by_name(self
):
915 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
916 self
.__class
__.test_index
,
917 inspect
.currentframe().f_code
.co_name
)
918 self
.__class
__.test_index
+= 1
920 image_list
= test_config
["vim_conn"].get_image_list({'name': self
.__class
__.image_name
})
922 for item
in image_list
:
923 self
.assertEqual(type(item
['id']), str)
924 self
.assertEqual(item
['id'], self
.__class
__.image_id
)
925 self
.assertEqual(type(item
['name']), str)
926 self
.assertEqual(item
['name'], self
.__class
__.image_name
)
928 def test_020_get_image_list_by_id(self
):
929 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
930 self
.__class
__.test_index
,
931 inspect
.currentframe().f_code
.co_name
)
932 self
.__class
__.test_index
+= 1
934 filter_image_list
= test_config
["vim_conn"].get_image_list({'id': self
.__class
__.image_id
})
936 for item1
in filter_image_list
:
937 self
.assertEqual(type(item1
.get('id')), str)
938 self
.assertEqual(item1
.get('id'), self
.__class
__.image_id
)
939 self
.assertEqual(type(item1
.get('name')), str)
940 self
.assertEqual(item1
.get('name'), self
.__class
__.image_name
)
942 def test_030_get_image_list_negative(self
):
943 Non_exist_image_id
= uuid
.uuid4()
944 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
945 self
.__class
__.test_index
,
946 inspect
.currentframe().f_code
.co_name
)
947 self
.__class
__.test_index
+= 1
948 image_list
= test_config
["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id
})
950 self
.assertIsNotNone(image_list
, None)
951 self
.assertEqual(image_list
, [])
953 class test_vimconn_new_vminstance(test_base
):
962 self
.__class
__.network_name
= _get_random_string(20)
963 self
.__class
__.net_type
= 'bridge'
965 self
.__class
__.network_id
= test_config
["vim_conn"].new_network(net_name
=self
.__class
__.network_name
,
966 net_type
=self
.__class
__.net_type
)
969 test_base
.tearDown(self
)
970 # Deleting created network
971 result
= test_config
["vim_conn"].delete_network(self
.__class
__.network_id
)
973 logger
.info("Network id {} sucessfully deleted".format(self
.__class
__.network_id
))
975 logger
.info("Failed to delete network id {}".format(self
.__class
__.network_id
))
977 def test_000_new_vminstance(self
):
978 vpci
= "0000:00:11.0"
981 flavor_data
= {'ram': 1024, 'vcpus': 1, 'disk': 10}
984 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
986 # find image name and image id
987 if test_config
['image_name']:
988 image_list
= test_config
['vim_conn'].get_image_list({'name': test_config
['image_name']})
989 if len(image_list
) == 0:
990 raise Exception("Image {} is not found at VIM".format(test_config
['image_name']))
992 self
.__class
__.image_id
= image_list
[0]['id']
994 image_list
= test_config
['vim_conn'].get_image_list()
995 if len(image_list
) == 0:
996 raise Exception("Not found any image at VIM")
998 self
.__class
__.image_id
= image_list
[0]['id']
1000 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1001 self
.__class
__.test_index
,
1002 inspect
.currentframe().f_code
.co_name
)
1003 self
.__class
__.test_index
+= 1
1005 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
}]
1007 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
)
1009 self
.assertEqual(type(self
.__class
__.instance_id
),str)
1011 def test_010_new_vminstance_by_model(self
):
1012 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1013 model_name
= 'e1000'
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': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'model': model_name
, '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_020_new_vminstance_by_net_use(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': net_use
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', '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_030_new_vminstance_by_net_type(self
):
1060 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1065 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1067 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1068 self
.__class
__.test_index
,
1069 inspect
.currentframe().f_code
.co_name
)
1070 self
.__class
__.test_index
+= 1
1072 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': _type
, 'net_id': self
.__class
__.network_id
}]
1074 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=self
.__class
__.image_id
,
1075 flavor_id
=flavor_id
,
1077 self
.assertEqual(type(instance_id
),str)
1078 # Deleting created vm instance
1079 logger
.info("Deleting created vm intance")
1080 test_config
["vim_conn"].delete_vminstance(instance_id
)
1083 def test_040_new_vminstance_by_cloud_config(self
):
1084 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1086 user_name
= 'test_user'
1088 key_pairs
= ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com']
1090 users_data
= [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com'], 'name': user_name
}]
1092 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
}
1095 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1097 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1098 self
.__class
__.test_index
,
1099 inspect
.currentframe().f_code
.co_name
)
1100 self
.__class
__.test_index
+= 1
1102 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1104 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='Cloud_vm', image_id
=self
.__class
__.image_id
,
1105 flavor_id
=flavor_id
,
1107 cloud_config
=cloud_data
)
1108 self
.assertEqual(type(instance_id
),str)
1109 # Deleting created vm instance
1110 logger
.info("Deleting created vm intance")
1111 test_config
["vim_conn"].delete_vminstance(instance_id
)
1114 def test_050_new_vminstance_by_disk_list(self
):
1115 flavor_data
= {'ram': 1024, 'vcpus': 2, 'disk': 10}
1118 device_data
= [{'image_id': self
.__class
__.image_id
, 'size': '5'}]
1121 flavor_id
= test_config
["vim_conn"].new_flavor(flavor_data
)
1123 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1124 self
.__class
__.test_index
,
1125 inspect
.currentframe().f_code
.co_name
)
1126 self
.__class
__.test_index
+= 1
1128 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1130 instance_id
, _
= test_config
["vim_conn"].new_vminstance(name
='VM_test1', image_id
=self
.__class
__.image_id
,
1131 flavor_id
=flavor_id
,
1133 disk_list
=device_data
)
1134 self
.assertEqual(type(instance_id
),str)
1135 # Deleting created vm instance
1136 logger
.info("Deleting created vm intance")
1137 test_config
["vim_conn"].delete_vminstance(instance_id
)
1140 def test_060_new_vminstance_negative(self
):
1141 unknown_flavor_id
= str(uuid
.uuid4())
1142 unknown_image_id
= str(uuid
.uuid4())
1145 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1146 self
.__class
__.test_index
,
1147 inspect
.currentframe().f_code
.co_name
)
1148 self
.__class
__.test_index
+= 1
1150 net_list
= [{'use': self
.__class
__.net_type
, 'name': name
, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self
.__class
__.network_id
}]
1152 with self
.assertRaises(Exception) as context
:
1153 test_config
["vim_conn"].new_vminstance(name
='Test1_vm', image_id
=unknown_image_id
,
1154 flavor_id
=unknown_flavor_id
,
1156 self
.assertEqual((context
.exception
).http_code
, 404)
1158 def test_070_get_vminstance(self
):
1159 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1160 self
.__class
__.test_index
,
1161 inspect
.currentframe().f_code
.co_name
)
1162 self
.__class
__.test_index
+= 1
1164 # Get instance by its id
1165 vm_info
= test_config
["vim_conn"].get_vminstance(self
.__class
__.instance_id
)
1167 if test_config
['vimtype'] == 'vmware':
1168 for attr
in vm_info
:
1169 if attr
== 'status':
1170 self
.assertEqual(vm_info
[attr
], 'ACTIVE')
1171 if attr
== 'hostId':
1172 self
.assertEqual(type(vm_info
[attr
]), str)
1173 if attr
== 'interfaces':
1174 self
.assertEqual(type(vm_info
[attr
]), list)
1175 self
.assertEqual(vm_info
[attr
][0]['IsConnected'], 'true')
1176 if attr
== 'IsEnabled':
1177 self
.assertEqual(vm_info
[attr
], 'true')
1179 def test_080_get_vminstance_negative(self
):
1180 unknown_instance_id
= str(uuid
.uuid4())
1182 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1183 self
.__class
__.test_index
,
1184 inspect
.currentframe().f_code
.co_name
)
1185 self
.__class
__.test_index
+= 1
1187 with self
.assertRaises(Exception) as context
:
1188 test_config
["vim_conn"].get_vminstance(unknown_instance_id
)
1190 self
.assertEqual((context
.exception
).http_code
, 404)
1192 def test_090_refresh_vms_status(self
):
1193 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1194 self
.__class
__.test_index
,
1195 inspect
.currentframe().f_code
.co_name
)
1196 self
.__class
__.test_index
+= 1
1198 vm_list
.append(self
.__class
__.instance_id
)
1201 vm_info
= test_config
["vim_conn"].refresh_vms_status(vm_list
)
1202 for attr
in vm_info
[self
.__class
__.instance_id
]:
1203 if attr
== 'status':
1204 self
.assertEqual(vm_info
[self
.__class
__.instance_id
][attr
], 'ACTIVE')
1205 if attr
== 'interfaces':
1206 self
.assertEqual(type(vm_info
[self
.__class
__.instance_id
][attr
]), list)
1208 def test_100_refresh_vms_status_negative(self
):
1209 unknown_id
= str(uuid
.uuid4())
1211 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1212 self
.__class
__.test_index
,
1213 inspect
.currentframe().f_code
.co_name
)
1214 self
.__class
__.test_index
+= 1
1216 vm_dict
= test_config
["vim_conn"].refresh_vms_status([unknown_id
])
1217 self
.assertEqual(vm_dict
, {})
1219 def test_110_action_vminstance(self
):
1220 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1221 self
.__class
__.test_index
,
1222 inspect
.currentframe().f_code
.co_name
)
1223 self
.__class
__.test_index
+= 1
1225 action_list
= ['shutdown','start','shutoff','rebuild','pause','resume']
1226 # various action on vminstace
1227 for action
in action_list
:
1228 instance_id
= test_config
["vim_conn"].action_vminstance(self
.__class
__.instance_id
,
1230 self
.assertEqual(instance_id
, self
.__class
__.instance_id
)
1232 def test_120_action_vminstance_negative(self
):
1233 non_exist_id
= str(uuid
.uuid4())
1234 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1235 self
.__class
__.test_index
,
1236 inspect
.currentframe().f_code
.co_name
)
1237 self
.__class
__.test_index
+= 1
1240 with self
.assertRaises(Exception) as context
:
1241 test_config
["vim_conn"].action_vminstance(non_exist_id
, { action
: None})
1243 self
.assertEqual((context
.exception
).http_code
, 400)
1245 def test_130_delete_vminstance(self
):
1246 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1247 self
.__class
__.test_index
,
1248 inspect
.currentframe().f_code
.co_name
)
1249 self
.__class
__.test_index
+= 1
1251 # Deleting created vm instance
1252 logger
.info("Deleting created vm instance")
1253 test_config
["vim_conn"].delete_vminstance(self
.__class
__.instance_id
)
1256 class test_vimconn_get_tenant_list(test_base
):
1259 def test_000_get_tenant_list(self
):
1260 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1261 self
.__class
__.test_index
,
1262 inspect
.currentframe().f_code
.co_name
)
1263 self
.__class
__.test_index
+= 1
1265 # Getting tenant list
1266 tenant_list
= test_config
["vim_conn"].get_tenant_list()
1268 for item
in tenant_list
:
1269 if test_config
['tenant'] == item
['name']:
1270 self
.__class
__.tenant_id
= item
['id']
1271 self
.assertEqual(type(item
['name']), str)
1272 self
.assertEqual(type(item
['id']), str)
1274 def test_010_get_tenant_list_by_id(self
):
1275 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1276 self
.__class
__.test_index
,
1277 inspect
.currentframe().f_code
.co_name
)
1278 self
.__class
__.test_index
+= 1
1280 # Getting filter tenant list by its id
1281 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'id': self
.__class
__.tenant_id
})
1283 for item
in filter_tenant_list
:
1284 self
.assertEqual(type(item
['id']), str)
1285 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1287 def test_020_get_tenant_list_by_name(self
):
1288 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1289 self
.__class
__.test_index
,
1290 inspect
.currentframe().f_code
.co_name
)
1291 self
.__class
__.test_index
+= 1
1293 # Getting filter tenant list by its name
1294 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant']})
1296 for item
in filter_tenant_list
:
1297 self
.assertEqual(type(item
['name']), str)
1298 self
.assertEqual(item
['name'], test_config
['tenant'])
1300 def test_030_get_tenant_list_by_name_and_id(self
):
1301 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1302 self
.__class
__.test_index
,
1303 inspect
.currentframe().f_code
.co_name
)
1304 self
.__class
__.test_index
+= 1
1306 # Getting filter tenant list by its name and id
1307 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': test_config
['tenant'],
1308 'id': self
.__class
__.tenant_id
})
1310 for item
in filter_tenant_list
:
1311 self
.assertEqual(type(item
['name']), str)
1312 self
.assertEqual(type(item
['id']), str)
1313 self
.assertEqual(item
['name'], test_config
['tenant'])
1314 self
.assertEqual(item
['id'], self
.__class
__.tenant_id
)
1316 def test_040_get_tenant_list_negative(self
):
1317 non_exist_tenant_name
= "Tenant_123"
1318 non_exist_tenant_id
= "kjhgrt456-45345kjhdfgnbdk-34dsfjdfg"
1319 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1320 self
.__class
__.test_index
,
1321 inspect
.currentframe().f_code
.co_name
)
1322 self
.__class
__.test_index
+= 1
1324 filter_tenant_list
= test_config
["vim_conn"].get_tenant_list({'name': non_exist_tenant_name
,
1325 'id': non_exist_tenant_id
})
1327 self
.assertEqual(filter_tenant_list
, [])
1329 class test_vimconn_new_tenant(test_base
):
1332 def test_000_new_tenant(self
):
1333 tenant_name
= _get_random_string(20)
1334 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1335 self
.__class
__.test_index
,
1336 inspect
.currentframe().f_code
.co_name
)
1337 self
.__class
__.test_index
+= 1
1339 self
.__class
__.tenant_id
= test_config
["vim_conn"].new_tenant(tenant_name
)
1342 self
.assertEqual(type(self
.__class
__.tenant_id
), str)
1344 def test_010_new_tenant_negative(self
):
1345 Invalid_tenant_name
= 10121
1346 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1347 self
.__class
__.test_index
,
1348 inspect
.currentframe().f_code
.co_name
)
1349 self
.__class
__.test_index
+= 1
1351 with self
.assertRaises(Exception) as context
:
1352 test_config
["vim_conn"].new_tenant(Invalid_tenant_name
)
1354 self
.assertEqual((context
.exception
).http_code
, 400)
1356 def test_020_delete_tenant(self
):
1357 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1358 self
.__class
__.test_index
,
1359 inspect
.currentframe().f_code
.co_name
)
1360 self
.__class
__.test_index
+= 1
1362 tenant_id
= test_config
["vim_conn"].delete_tenant(self
.__class
__.tenant_id
)
1363 self
.assertEqual(type(tenant_id
), str)
1365 def test_030_delete_tenant_negative(self
):
1366 Non_exist_tenant_name
= 'Test_30_tenant'
1367 self
.__class
__.test_text
= "{}.{}. TEST {}".format(test_config
["test_number"],
1368 self
.__class
__.test_index
,
1369 inspect
.currentframe().f_code
.co_name
)
1370 self
.__class
__.test_index
+= 1
1372 with self
.assertRaises(Exception) as context
:
1373 test_config
["vim_conn"].delete_tenant(Non_exist_tenant_name
)
1375 self
.assertEqual((context
.exception
).http_code
, 404)
1380 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
1381 scenario based tests.
1383 class descriptor_based_scenario_test(test_base
):
1385 scenario_test_path
= None
1388 def setUpClass(cls
):
1390 cls
.to_delete_list
= []
1391 cls
.scenario_uuids
= []
1392 cls
.instance_scenario_uuids
= []
1393 cls
.scenario_test_path
= test_config
["test_directory"] + '/' + test_config
["test_folder"]
1394 logger
.info("{}. {} {}".format(test_config
["test_number"], cls
.__name
__, test_config
["test_folder"]))
1397 def tearDownClass(cls
):
1398 test_config
["test_number"] += 1
1400 def test_000_load_scenario(self
):
1401 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1402 inspect
.currentframe().f_code
.co_name
,
1403 test_config
["test_folder"])
1404 self
.__class
__.test_index
+= 1
1406 descriptor_files
= glob
.glob(self
.__class
__.scenario_test_path
+'/*.yaml')
1407 vnf_descriptors
= []
1408 scenario_descriptors
= []
1409 for descriptor_file
in descriptor_files
:
1410 with
open(descriptor_file
, 'r') as stream
:
1411 descriptor
= yaml
.load(stream
)
1412 if "vnf" in descriptor
or "vnfd:vnfd-catalog" in descriptor
or "vnfd-catalog" in descriptor
:
1413 vnf_descriptors
.append(descriptor
)
1415 scenario_descriptors
.append(descriptor
)
1417 scenario_file
= glob
.glob(self
.__class
__.scenario_test_path
+ '/scenario_*.yaml')
1418 if not vnf_descriptors
or not scenario_descriptors
or len(scenario_descriptors
) > 1:
1419 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
1420 test_config
["test_folder"]))
1423 for vnf_descriptor
in vnf_descriptors
:
1424 logger
.debug("VNF descriptor: {}".format(vnf_descriptor
))
1425 vnf
= test_config
["client"].create_vnf(descriptor
=vnf_descriptor
, image_name
=test_config
["image_name"])
1428 vnf_uuid
= vnf
['vnf']['uuid']
1430 vnf_uuid
= vnf
['vnfd'][0]['uuid']
1431 self
.__class
__.to_delete_list
.insert(0, {"item": "vnf", "function": test_config
["client"].delete_vnf
,
1432 "params": {"uuid": vnf_uuid
}})
1434 # load the scenario definition
1435 for scenario_descriptor
in scenario_descriptors
:
1436 # networks = scenario_descriptor['scenario']['networks']
1437 # networks[test_config["mgmt_net"]] = networks.pop('mgmt')
1438 logger
.debug("Scenario descriptor: {}".format(scenario_descriptor
))
1439 scenario
= test_config
["client"].create_scenario(descriptor
=scenario_descriptor
)
1440 logger
.debug(scenario
)
1441 if 'scenario' in scenario
:
1442 scenario_uuid
= scenario
['scenario']['uuid']
1444 scenario_uuid
= scenario
['nsd'][0]['uuid']
1445 self
.__class
__.to_delete_list
.insert(0, {"item": "scenario",
1446 "function": test_config
["client"].delete_scenario
,
1447 "params": {"uuid": scenario_uuid
}})
1448 self
.__class
__.scenario_uuids
.append(scenario_uuid
)
1450 def test_010_instantiate_scenario(self
):
1451 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1452 inspect
.currentframe().f_code
.co_name
,
1453 test_config
["test_folder"])
1454 self
.__class
__.test_index
+= 1
1455 for scenario_uuid
in self
.__class
__.scenario_uuids
:
1456 instance_descriptor
= {
1458 "name": self
.__class
__.test_text
,
1459 "scenario": scenario_uuid
,
1461 "mgmt": {"sites": [ { "netmap-use": test_config
["mgmt_net"]} ]}
1465 instance
= test_config
["client"].create_instance(instance_descriptor
)
1466 self
.__class
__.instance_scenario_uuids
.append(instance
['uuid'])
1467 logger
.debug(instance
)
1468 self
.__class
__.to_delete_list
.insert(0, {"item": "instance",
1469 "function": test_config
["client"].delete_instance
,
1470 "params": {"uuid": instance
['uuid']}})
1472 def test_020_check_deployent(self
):
1473 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1474 inspect
.currentframe().f_code
.co_name
,
1475 test_config
["test_folder"])
1476 self
.__class
__.test_index
+= 1
1478 if test_config
["manual"]:
1479 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1482 keep_waiting
= test_config
["timeout"]
1483 pending_instance_scenario_uuids
= list(self
.__class
__.instance_scenario_uuids
) # make a copy
1484 while pending_instance_scenario_uuids
:
1486 while index
< len(pending_instance_scenario_uuids
):
1487 result
= check_instance_scenario_active(pending_instance_scenario_uuids
[index
])
1489 del pending_instance_scenario_uuids
[index
]
1491 elif 'ERROR' in result
[1]:
1492 msg
= 'Got error while waiting for the instance to get active: '+result
[1]
1494 raise Exception(msg
)
1497 if keep_waiting
>= 5:
1500 elif keep_waiting
> 0:
1501 time
.sleep(keep_waiting
)
1504 msg
= 'Timeout reached while waiting instance scenario to get active'
1506 raise Exception(msg
)
1508 def test_030_clean_deployment(self
):
1509 self
.__class
__.test_text
= "{}.{}. TEST {} {}".format(test_config
["test_number"], self
.__class
__.test_index
,
1510 inspect
.currentframe().f_code
.co_name
,
1511 test_config
["test_folder"])
1512 self
.__class
__.test_index
+= 1
1513 #At the moment if you delete an scenario right after creating it, in openstack datacenters
1514 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1516 for item
in self
.__class
__.to_delete_list
:
1517 response
= item
["function"](**item
["params"])
1518 logger
.debug(response
)
1521 def _get_random_string(maxLength
):
1522 '''generates a string with random characters string.letters and string.digits
1523 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1527 minLength
= min_string
- len(prefix
)
1528 if maxLength
< min_string
: maxLength
= min_string
1529 maxLength
-= len(prefix
)
1530 length
= random
.randint(minLength
,maxLength
)
1531 return 'testing_'+"".join([random
.choice(string
.letters
+string
.digits
) for i
in xrange(length
)])
1534 def test_vimconnector(args
):
1536 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1537 test_config
['vimtype'] = args
.vimtype
1538 if args
.vimtype
== "vmware":
1539 import vimconn_vmware
as vim
1541 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1543 tenant_name
= args
.tenant_name
1544 test_config
['tenant'] = tenant_name
1545 config_params
= json
.loads(args
.config_param
)
1546 org_name
= config_params
.get('orgname')
1547 org_user
= config_params
.get('user')
1548 org_passwd
= config_params
.get('passwd')
1549 vim_url
= args
.endpoint_url
1550 test_config
['image_path'] = args
.image_path
1551 test_config
['image_name'] = args
.image_name
1553 # vmware connector obj
1554 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
)
1556 elif args
.vimtype
== "aws":
1557 import vimconn_aws
as vim
1558 elif args
.vimtype
== "openstack":
1559 import vimconn_openstack
as vim
1560 elif args
.vimtype
== "openvim":
1561 import vimconn_openvim
as vim
1563 logger
.critical("vimtype '{}' not supported".format(args
.vimtype
))
1567 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1568 # If only want to obtain a tests list print it and exit
1571 for cls
in clsmembers
:
1572 if cls
[0].startswith('test_vimconnector'):
1573 tests_names
.append(cls
[0])
1575 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
))
1580 # Create the list of tests to be run
1581 code_based_tests
= []
1583 for test
in args
.tests
:
1584 for t
in test
.split(','):
1585 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1586 if len(matches_code_based_tests
) > 0:
1587 code_based_tests
.append(matches_code_based_tests
[0][1])
1589 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1591 if not code_based_tests
:
1593 for cls
in clsmembers
:
1594 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1595 if cls
[0].startswith('test_vimconnector'):
1596 code_based_tests
.append(cls
[1])
1598 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1600 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1601 # This is handled in the tests using logging.
1602 stream
= open('/dev/null', 'w')
1604 # Run code based tests
1605 basic_tests_suite
= unittest
.TestSuite()
1606 for test
in code_based_tests
:
1607 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1608 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1609 executed
+= result
.testsRun
1610 failed
+= len(result
.failures
) + len(result
.errors
)
1611 if failfast
and failed
:
1613 if len(result
.failures
) > 0:
1614 logger
.debug("failures : {}".format(result
.failures
))
1615 if len(result
.errors
) > 0:
1616 logger
.debug("errors : {}".format(result
.errors
))
1617 return executed
, failed
1622 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1623 import openmanoclient
1626 test_config
["client"] = openmanoclient
.openmanoclient(
1627 endpoint_url
=args
.endpoint_url
,
1628 tenant_name
=args
.tenant_name
,
1629 datacenter_name
=args
.datacenter
,
1630 debug
=args
.debug
, logger
=test_config
["logger_name"])
1631 clsmembers
= inspect
.getmembers(sys
.modules
[__name__
], inspect
.isclass
)
1632 # If only want to obtain a tests list print it and exit
1635 for cls
in clsmembers
:
1636 if cls
[0].startswith('test_VIM'):
1637 tests_names
.append(cls
[0])
1639 msg
= "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names
)) +\
1640 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1641 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1646 # Create the list of tests to be run
1647 code_based_tests
= []
1649 for test
in args
.tests
:
1650 for t
in test
.split(','):
1651 matches_code_based_tests
= [item
for item
in clsmembers
if item
[0] == t
]
1652 if len(matches_code_based_tests
) > 0:
1653 code_based_tests
.append(matches_code_based_tests
[0][1])
1655 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1657 if not code_based_tests
:
1659 for cls
in clsmembers
:
1660 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1661 if cls
[0].startswith('test_VIM') and cls
[0] != 'test_VIM_tenant_operations':
1662 code_based_tests
.append(cls
[1])
1664 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1666 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1667 # This is handled in the tests using logging.
1668 stream
= open('/dev/null', 'w')
1670 # Run code based tests
1671 basic_tests_suite
= unittest
.TestSuite()
1672 for test
in code_based_tests
:
1673 basic_tests_suite
.addTest(unittest
.makeSuite(test
))
1674 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=failfast
).run(basic_tests_suite
)
1675 executed
+= result
.testsRun
1676 failed
+= len(result
.failures
) + len(result
.errors
)
1677 if failfast
and failed
:
1679 if len(result
.failures
) > 0:
1680 logger
.debug("failures : {}".format(result
.failures
))
1681 if len(result
.errors
) > 0:
1682 logger
.debug("errors : {}".format(result
.errors
))
1683 return executed
, failed
1686 def test_deploy(args
):
1688 sys
.path
.append(os
.path
.dirname(os
.path
.dirname(os
.path
.abspath(__file__
))) + "/osm_ro")
1689 import openmanoclient
1692 test_config
["test_directory"] = os
.path
.dirname(__file__
) + "/RO_tests"
1693 test_config
["image_name"] = args
.image_name
1694 test_config
["mgmt_net"] = args
.mgmt_net
1695 test_config
["manual"] = args
.manual
1696 test_directory_content
= os
.listdir(test_config
["test_directory"])
1697 # If only want to obtain a tests list print it and exit
1699 msg
= "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content
))
1704 descriptor_based_tests
= []
1705 # Create the list of tests to be run
1706 code_based_tests
= []
1708 for test
in args
.tests
:
1709 for t
in test
.split(','):
1710 if t
in test_directory_content
:
1711 descriptor_based_tests
.append(t
)
1713 logger
.critical("Test '{}' is not among the possible ones".format(t
))
1715 if not descriptor_based_tests
:
1717 descriptor_based_tests
= test_directory_content
1719 logger
.debug("tests to be executed: {}".format(code_based_tests
))
1721 # import openmanoclient from relative path
1722 test_config
["client"] = openmanoclient
.openmanoclient(
1723 endpoint_url
=args
.endpoint_url
,
1724 tenant_name
=args
.tenant_name
,
1725 datacenter_name
=args
.datacenter
,
1726 debug
=args
.debug
, logger
=test_config
["logger_name"])
1728 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1729 # This is handled in the tests using logging.
1730 stream
= open('/dev/null', 'w')
1731 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1732 for test
in descriptor_based_tests
:
1733 test_config
["test_folder"] = test
1734 test_suite
= unittest
.TestSuite()
1735 test_suite
.addTest(unittest
.makeSuite(descriptor_based_scenario_test
))
1736 result
= unittest
.TextTestRunner(stream
=stream
, failfast
=False).run(test_suite
)
1737 executed
+= result
.testsRun
1738 failed
+= len(result
.failures
) + len(result
.errors
)
1739 if failfast
and failed
:
1741 if len(result
.failures
) > 0:
1742 logger
.debug("failures : {}".format(result
.failures
))
1743 if len(result
.errors
) > 0:
1744 logger
.debug("errors : {}".format(result
.errors
))
1746 return executed
, failed
1748 if __name__
=="__main__":
1750 parser
= ArgumentParser(description
='Test RO module')
1751 parser
.add_argument('-v','--version', action
='version', help="Show current version",
1752 version
='%(prog)s version ' + __version__
+ ' ' + version_date
)
1755 parent_parser
= ArgumentParser(add_help
=False)
1756 parent_parser
.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1757 dest
='failfast', action
="store_true", default
=False)
1758 parent_parser
.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1759 dest
='failed', action
="store_true", default
=False)
1760 default_logger_file
= os
.path
.dirname(__file__
)+'/'+os
.path
.splitext(os
.path
.basename(__file__
))[0]+'.log'
1761 parent_parser
.add_argument('--list-tests', help='List all available tests', dest
='list_tests', action
="store_true",
1763 parent_parser
.add_argument('--logger_file', dest
='logger_file', default
=default_logger_file
,
1764 help='Set the logger file. By default '+default_logger_file
)
1765 parent_parser
.add_argument("-t", '--tenant', dest
='tenant_name', default
="osm",
1766 help="Set the openmano tenant to use for the test. By default 'osm'")
1767 parent_parser
.add_argument('--debug', help='Set logs to debug level', dest
='debug', action
="store_true")
1768 parent_parser
.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1769 dest
='timeout', type=int, default
=300)
1770 parent_parser
.add_argument('--test', '--tests', help='Specify the tests to run', dest
='tests', action
="append")
1772 subparsers
= parser
.add_subparsers(help='test sets')
1774 # Deployment test set
1775 # -------------------
1776 deploy_parser
= subparsers
.add_parser('deploy', parents
=[parent_parser
],
1777 help="test deployment using descriptors at RO_test folder ")
1778 deploy_parser
.set_defaults(func
=test_deploy
)
1780 # Mandatory arguments
1781 mandatory_arguments
= deploy_parser
.add_argument_group('mandatory arguments')
1782 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1783 mandatory_arguments
.add_argument("-i", '--image-name', required
=True, dest
="image_name",
1784 help='Image name available at datacenter used for the tests')
1785 mandatory_arguments
.add_argument("-n", '--mgmt-net-name', required
=True, dest
='mgmt_net',
1786 help='Set the vim management network to use for tests')
1788 # Optional arguments
1789 deploy_parser
.add_argument('-m', '--manual-check', dest
='manual', action
="store_true", default
=False,
1790 help='Pause execution once deployed to allow manual checking of the '
1791 'deployed instance scenario')
1792 deploy_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1793 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1796 # -------------------
1797 vimconn_parser
= subparsers
.add_parser('vimconn', parents
=[parent_parser
], help="test vimconnector plugin")
1798 vimconn_parser
.set_defaults(func
=test_vimconnector
)
1799 # Mandatory arguments
1800 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1801 mandatory_arguments
.add_argument('--vimtype', choices
=['vmware', 'aws', 'openstack', 'openvim'], required
=True,
1802 help='Set the vimconnector type to test')
1803 mandatory_arguments
.add_argument('-c', '--config', dest
='config_param', required
=True,
1804 help='Set the vimconnector specific config parameters in dictionary format')
1805 mandatory_arguments
.add_argument('-u', '--url', dest
='endpoint_url',required
=True, help="Set the vim connector url or Host IP")
1806 # Optional arguments
1807 vimconn_parser
.add_argument('-i', '--image-path', dest
='image_path', help="Provide image path present at RO container")
1808 vimconn_parser
.add_argument('-n', '--image-name', dest
='image_name', help="Provide image name for test")
1809 # TODO add optional arguments for vimconn tests
1810 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1812 # Datacenter test set
1813 # -------------------
1814 vimconn_parser
= subparsers
.add_parser('vim', parents
=[parent_parser
], help="test vim")
1815 vimconn_parser
.set_defaults(func
=test_vim
)
1817 # Mandatory arguments
1818 mandatory_arguments
= vimconn_parser
.add_argument_group('mandatory arguments')
1819 mandatory_arguments
.add_argument('-d', '--datacenter', required
=True, help='Set the datacenter to test')
1821 # Optional arguments
1822 vimconn_parser
.add_argument('-u', '--url', dest
='endpoint_url', default
='http://localhost:9090/openmano',
1823 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1825 argcomplete
.autocomplete(parser
)
1826 args
= parser
.parse_args()
1830 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1831 logger_level
= 'INFO'
1833 logger_level
= 'DEBUG'
1835 logger_level
= 'WARNING'
1836 logger_name
= os
.path
.basename(__file__
)
1837 test_config
["logger_name"] = logger_name
1838 logger
= logging
.getLogger(logger_name
)
1839 logger
.setLevel(logger_level
)
1840 failfast
= args
.failfast
1842 # Configure a logging handler to store in a logging file
1843 if args
.logger_file
:
1844 fileHandler
= logging
.FileHandler(args
.logger_file
)
1845 formatter_fileHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1846 fileHandler
.setFormatter(formatter_fileHandler
)
1847 logger
.addHandler(fileHandler
)
1849 # Configure a handler to print to stdout
1850 consoleHandler
= logging
.StreamHandler(sys
.stdout
)
1851 formatter_consoleHandler
= logging
.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1852 consoleHandler
.setFormatter(formatter_consoleHandler
)
1853 logger
.addHandler(consoleHandler
)
1855 logger
.debug('Program started with the following arguments: ' + str(args
))
1857 # set test config parameters
1858 test_config
["timeout"] = args
.timeout
1859 test_config
["test_number"] = 1
1861 executed
, failed
= args
.func(args
)
1864 logger
.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed
, failed
))
1865 sys
.exit(1 if failed
else 0)