2 # -*- coding: utf-8 -*-
12 from time
import sleep
15 __author__
= "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
16 __date__
= "$2018-03-01$"
18 version_date
= "Oct 2018"
22 print("Usage: ", sys
.argv
[0], "[options]")
23 print(" Performs system tests over running NBI. It can be used for real OSM test using option '--test-osm'")
24 print(" If this is the case env variables 'OSMNBITEST_VIM_NAME' must be suplied to create a VIM if not exist "
25 "where deployment is done")
27 print(" -h|--help: shows this help")
28 print(" --insecure: Allows non trusted https NBI server")
29 print(" --list: list available tests")
30 print(" --manual-check: Deployment tests stop after deployed to allow manual inspection. Only make sense with "
32 print(" -p|--password PASSWORD: NBI access password. 'admin' by default")
33 print(" ---project PROJECT: NBI access project. 'admin' by default")
34 print(" --test TEST[,...]: Execute only a test or a comma separated list of tests")
35 print(" --params key=val: params to the previous test. key can be vnfd-files, nsd-file, ns-name, ns-config")
36 print(" --test-osm: If missing this test is intended for NBI only, no other OSM components are expected. Use "
37 "this flag to test the system. LCM and RO components are expected to be up and running")
38 print(" --timeout TIMEOUT: General NBI timeout, by default {}s".format(timeout
))
39 print(" --timeout-deploy TIMEOUT: Timeout used for getting NS deployed, by default {}s".format(timeout_deploy
))
40 print(" --timeout-configure TIMEOUT: Timeout used for getting NS deployed and configured,"
41 " by default {}s".format(timeout_configure
))
42 print(" -u|--user USERNAME: NBI access username. 'admin' by default")
43 print(" --url URL: complete NBI server URL. 'https//localhost:9999/osm' by default")
44 print(" -v|--verbose print debug information, can be used several times")
45 print(" --no-verbose remove verbosity")
46 print(" --version: prints current version")
47 print("ENV variables used for real deployment tests with option osm-test.")
48 print(" export OSMNBITEST_VIM_NAME=vim-name")
49 print(" export OSMNBITEST_VIM_URL=vim-url")
50 print(" export OSMNBITEST_VIM_TYPE=vim-type")
51 print(" export OSMNBITEST_VIM_TENANT=vim-tenant")
52 print(" export OSMNBITEST_VIM_USER=vim-user")
53 print(" export OSMNBITEST_VIM_PASSWORD=vim-password")
54 print(" export OSMNBITEST_VIM_CONFIG=\"vim-config\"")
55 print(" export OSMNBITEST_NS_NAME=\"vim-config\"")
59 r_header_json
= {"Content-type": "application/json"}
61 "Content-type": "application/json",
62 "Accept": "application/json",
64 r_header_yaml
= {"Content-type": "application/yaml"}
66 "Content-type": "application/yaml",
67 "Accept": "application/yaml",
69 r_header_text
= {"Content-type": "text/plain"}
70 r_header_octect
= {"Content-type": "application/octet-stream"}
72 "Accept": "text/plain",
74 r_header_zip
= {"Content-type": "application/zip"}
76 "Accept": "application/zip",
79 "Accept": "application/yaml", "Content-type": "application/zip"
83 # test ones authorized
84 test_authorized_list
= (
85 ("AU1", "Invalid vnfd id", "GET", "/vnfpkgm/v1/vnf_packages/non-existing-id",
86 headers_json
, None, 404, r_header_json
, "json"),
87 ("AU2", "Invalid nsd id", "GET", "/nsd/v1/ns_descriptors/non-existing-id",
88 headers_yaml
, None, 404, r_header_yaml
, "yaml"),
89 ("AU3", "Invalid nsd id", "DELETE", "/nsd/v1/ns_descriptors_content/non-existing-id",
90 headers_yaml
, None, 404, r_header_yaml
, "yaml"),
92 timeout
= 120 # general timeout
93 timeout_deploy
= 60*10 # timeout for NS deploying without charms
94 timeout_configure
= 60*20 # timeout for NS deploying and configuring
97 class TestException(Exception):
102 def __init__(self
, url_base
, header_base
=None, verify
=False, user
="admin", password
="admin", project
="admin"):
103 self
.url_base
= url_base
104 if header_base
is None:
105 self
.header_base
= {}
107 self
.header_base
= header_base
.copy()
108 self
.s
= requests
.session()
109 self
.s
.headers
= self
.header_base
113 self
.password
= password
114 self
.project
= project
116 # contains ID of tests obtained from Location response header. "" key contains last obtained id
118 self
.old_test_description
= ""
119 self
.test_name
= None
122 def set_header(self
, header
):
123 self
.s
.headers
.update(header
)
125 def set_tet_name(self
, test_name
):
126 self
.test_name
= test_name
128 def unset_header(self
, key
):
129 if key
in self
.s
.headers
:
130 del self
.s
.headers
[key
]
132 def test(self
, name
, description
, method
, url
, headers
, payload
, expected_codes
, expected_headers
,
133 expected_payload
, store_file
=None):
135 Performs an http request and check http code response. Exit if different than allowed. It get the returned id
136 that can be used by following test in the URL with {name} where name is the name of the test
137 :param name: short name of the test
138 :param description: description of the test
139 :param method: HTTP method: GET,PUT,POST,DELETE,...
140 :param url: complete URL or relative URL
141 :param headers: request headers to add to the base headers
142 :param payload: Can be a dict, transformed to json, a text or a file if starts with '@'
143 :param expected_codes: expected response codes, can be int, int tuple or int range
144 :param expected_headers: expected response headers, dict with key values
145 :param expected_payload: expected payload, 0 if empty, 'yaml', 'json', 'text', 'zip', 'octet-stream'
146 :param store_file: filename to store content
147 :return: requests response
152 self
.s
= requests
.session()
156 elif not url
.startswith("http"):
157 url
= self
.url_base
+ url
159 var_start
= url
.find("<") + 1
161 var_end
= url
.find(">", var_start
)
164 var_name
= url
[var_start
:var_end
]
165 if var_name
in self
.test_ids
:
166 url
= url
[:var_start
-1] + self
.test_ids
[var_name
] + url
[var_end
+1:]
167 var_start
+= len(self
.test_ids
[var_name
])
168 var_start
= url
.find("<", var_start
) + 1
170 if isinstance(payload
, str):
171 if payload
.startswith("@"):
173 file_name
= payload
[1:]
174 if payload
.startswith("@b"):
176 file_name
= payload
[2:]
177 with
open(file_name
, mode
) as f
:
179 elif isinstance(payload
, dict):
180 payload
= json
.dumps(payload
)
182 test_description
= "Test {} {} {} {}".format(name
, description
, method
, url
)
183 if self
.old_test_description
!= test_description
:
184 self
.old_test_description
= test_description
185 logger
.warning(test_description
)
187 if expected_payload
in ("zip", "octet-string") or store_file
:
189 r
= getattr(self
.s
, method
.lower())(url
, data
=payload
, headers
=headers
, verify
=self
.verify
, stream
=stream
)
190 if expected_payload
in ("zip", "octet-string") or store_file
:
191 logger
.debug("RX {}".format(r
.status_code
))
193 logger
.debug("RX {}: {}".format(r
.status_code
, r
.text
))
197 if isinstance(expected_codes
, int):
198 expected_codes
= (expected_codes
,)
199 if r
.status_code
not in expected_codes
:
201 "Got status {}. Expected {}. {}".format(r
.status_code
, expected_codes
, r
.text
))
204 for header_key
, header_val
in expected_headers
.items():
205 if header_key
.lower() not in r
.headers
:
206 raise TestException("Header {} not present".format(header_key
))
207 if header_val
and header_val
.lower() not in r
.headers
[header_key
]:
208 raise TestException("Header {} does not contain {} but {}".format(header_key
, header_val
,
209 r
.headers
[header_key
]))
211 if expected_payload
is not None:
212 if expected_payload
== 0 and len(r
.content
) > 0:
213 raise TestException("Expected empty payload")
214 elif expected_payload
== "json":
217 except Exception as e
:
218 raise TestException("Expected json response payload, but got Exception {}".format(e
))
219 elif expected_payload
== "yaml":
221 yaml
.safe_load(r
.text
)
222 except Exception as e
:
223 raise TestException("Expected yaml response payload, but got Exception {}".format(e
))
224 elif expected_payload
in ("zip", "octet-string"):
225 if len(r
.content
) == 0:
226 raise TestException("Expected some response payload, but got empty")
228 # tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
229 # for tarinfo in tar:
230 # tarname = tarinfo.name
232 # except Exception as e:
233 # raise TestException("Expected zip response payload, but got Exception {}".format(e))
234 elif expected_payload
== "text":
235 if len(r
.content
) == 0:
236 raise TestException("Expected some response payload, but got empty")
239 with
open(store_file
, 'wb') as fd
:
240 for chunk
in r
.iter_content(chunk_size
=128):
243 location
= r
.headers
.get("Location")
245 _id
= location
[location
.rfind("/") + 1:]
247 self
.test_ids
[name
] = str(_id
)
248 self
.test_ids
["last_id"] = str(_id
) # last id
249 self
.test_ids
[""] = str(_id
) # last id
251 except TestException
as e
:
255 r_status_code
= r
.status_code
257 logger
.error("{} \nRX code{}: {}".format(e
, r_status_code
, r_text
))
260 logger
.error("Cannot open file {}".format(e
))
263 def get_autorization(self
): # user=None, password=None, project=None):
264 if self
.token
: # and self.user == user and self.password == password and self.project == project:
267 # self.password = password
268 # self.project = project
269 r
= self
.test("TOKEN", "Obtain token", "POST", "/admin/v1/tokens", headers_json
,
270 {"username": self
.user
, "password": self
.password
, "project_id": self
.project
},
271 (200, 201), r_header_json
, "json")
273 self
.token
= response
["id"]
274 self
.set_header({"Authorization": "Bearer {}".format(self
.token
)})
276 def remove_authorization(self
):
278 self
.test("TOKEN_DEL", "Delete token", "DELETE", "/admin/v1/tokens/{}".format(self
.token
), headers_json
,
279 None, (200, 201, 204), None, None)
281 self
.unset_header("Authorization")
283 def get_create_vim(self
, test_osm
):
286 self
.get_autorization()
288 vim_name
= os
.environ
.get("OSMNBITEST_VIM_NAME")
291 "Needed to define OSMNBITEST_VIM_XXX variables to create a real VIM for deployment")
295 r
= self
.test("_VIMGET1", "Get VIM ID", "GET", "/admin/v1/vim_accounts?name={}".format(vim_name
), headers_json
,
296 None, 200, r_header_json
, "json")
299 return vims
[0]["_id"]
302 # check needed environ parameters:
303 if not os
.environ
.get("OSMNBITEST_VIM_URL") or not os
.environ
.get("OSMNBITEST_VIM_TENANT"):
304 raise TestException("Env OSMNBITEST_VIM_URL and OSMNBITEST_VIM_TENANT are needed for create a real VIM"
305 " to deploy on whit the --test-osm option")
306 vim_data
= "{{schema_version: '1.0', name: '{}', vim_type: {}, vim_url: '{}', vim_tenant_name: '{}', "\
307 "vim_user: {}, vim_password: {}".format(vim_name
,
308 os
.environ
.get("OSMNBITEST_VIM_TYPE", "openstack"),
309 os
.environ
.get("OSMNBITEST_VIM_URL"),
310 os
.environ
.get("OSMNBITEST_VIM_TENANT"),
311 os
.environ
.get("OSMNBITEST_VIM_USER"),
312 os
.environ
.get("OSMNBITEST_VIM_PASSWORD"))
313 if os
.environ
.get("OSMNBITEST_VIM_CONFIG"):
314 vim_data
+= " ,config: {}".format(os
.environ
.get("OSMNBITEST_VIM_CONFIG"))
317 vim_data
= "{schema_version: '1.0', name: fakeVim, vim_type: openstack, vim_url: 'http://10.11.12.13/fake'"\
318 ", vim_tenant_name: 'vimtenant', vim_user: vimuser, vim_password: vimpassword}"
319 r
= self
.test("_VIMGET2", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_yaml
, vim_data
,
320 (201), {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/yaml"}, "yaml")
321 location
= r
.headers
.get("Location")
322 return location
[location
.rfind("/") + 1:]
325 class TestNonAuthorized
:
326 description
= "test invalid URLs. methods and no authorization"
329 def run(engine
, test_osm
, manual_check
, test_params
=None):
330 engine
.remove_authorization()
331 test_not_authorized_list
= (
332 ("NA1", "Invalid token", "GET", "/admin/v1/users", headers_json
, None, 401, r_header_json
, "json"),
333 ("NA2", "Invalid URL", "POST", "/admin/v1/nonexist", headers_yaml
, None, 405, r_header_yaml
, "yaml"),
334 ("NA3", "Invalid version", "DELETE", "/admin/v2/users", headers_yaml
, None, 405, r_header_yaml
, "yaml"),
336 for t
in test_not_authorized_list
:
340 class TestUsersProjects
:
341 description
= "test project and user creation"
344 def run(engine
, test_osm
, manual_check
, test_params
=None):
345 engine
.get_autorization()
346 engine
.test("PU1", "Create project non admin", "POST", "/admin/v1/projects", headers_json
, {"name": "P1"},
347 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
348 engine
.test("PU2", "Create project admin", "POST", "/admin/v1/projects", headers_json
,
349 {"name": "Padmin", "admin": True}, (201, 204),
350 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
351 engine
.test("PU3", "Create project bad format", "POST", "/admin/v1/projects", headers_json
, {"name": 1}, 422,
352 r_header_json
, "json")
353 engine
.test("PU4", "Create user with bad project", "POST", "/admin/v1/users", headers_json
,
354 {"username": "U1", "projects": ["P1", "P2", "Padmin"], "password": "pw1"}, 409,
355 r_header_json
, "json")
356 engine
.test("PU5", "Create user with bad project and force", "POST", "/admin/v1/users?FORCE=True", headers_json
,
357 {"username": "U1", "projects": ["P1", "P2", "Padmin"], "password": "pw1"}, 201,
358 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
359 engine
.test("PU6", "Create user 2", "POST", "/admin/v1/users", headers_json
,
360 {"username": "U2", "projects": ["P1"], "password": "pw2"}, 201,
361 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
363 engine
.test("PU7", "Edit user U1, delete P2 project", "PATCH", "/admin/v1/users/U1", headers_json
,
364 {"projects": {"$'P2'": None}}, 204, None, None)
365 res
= engine
.test("PU1", "Check user U1, contains the right projects", "GET", "/admin/v1/users/U1",
366 headers_json
, None, 200, None, json
)
369 expected_projects
= ["P1", "Padmin"]
370 if u1
["projects"] != expected_projects
:
371 raise TestException("User content projects '{}' different than expected '{}'. Edition has not done"
372 " properly".format(u1
["projects"], expected_projects
))
374 engine
.test("PU8", "Edit user U1, set Padmin as default project", "PUT", "/admin/v1/users/U1", headers_json
,
375 {"projects": {"$'Padmin'": None, "$+[0]": "Padmin"}}, 204, None, None)
376 res
= engine
.test("PU1", "Check user U1, contains the right projects", "GET", "/admin/v1/users/U1",
377 headers_json
, None, 200, None, json
)
380 expected_projects
= ["Padmin", "P1"]
381 if u1
["projects"] != expected_projects
:
382 raise TestException("User content projects '{}' different than expected '{}'. Edition has not done"
383 " properly".format(u1
["projects"], expected_projects
))
385 engine
.test("PU9", "Edit user U1, change password", "PATCH", "/admin/v1/users/U1", headers_json
,
386 {"password": "pw1_new"}, 204, None, None)
388 engine
.test("PU10", "Change to project P1 non existing", "POST", "/admin/v1/tokens/", headers_json
,
389 {"project_id": "P1"}, 401, r_header_json
, "json")
391 res
= engine
.test("PU1", "Change to user U1 project P1", "POST", "/admin/v1/tokens", headers_json
,
392 {"username": "U1", "password": "pw1_new", "project_id": "P1"}, (200, 201),
393 r_header_json
, "json")
394 response
= res
.json()
395 engine
.set_header({"Authorization": "Bearer {}".format(response
["id"])})
397 engine
.test("PU11", "Edit user projects non admin", "PUT", "/admin/v1/users/U1", headers_json
,
398 {"projects": {"$'P1'": None}}, 401, r_header_json
, "json")
399 engine
.test("PU12", "Add new project non admin", "POST", "/admin/v1/projects", headers_json
,
400 {"name": "P2"}, 401, r_header_json
, "json")
401 engine
.test("PU13", "Add new user non admin", "POST", "/admin/v1/users", headers_json
,
402 {"username": "U3", "projects": ["P1"], "password": "pw3"}, 401,
403 r_header_json
, "json")
405 res
= engine
.test("PU14", "Change to user U1 project Padmin", "POST", "/admin/v1/tokens", headers_json
,
406 {"project_id": "Padmin"}, (200, 201), r_header_json
, "json")
407 response
= res
.json()
408 engine
.set_header({"Authorization": "Bearer {}".format(response
["id"])})
410 engine
.test("PU15", "Add new project admin", "POST", "/admin/v1/projects", headers_json
, {"name": "P2"},
411 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
412 engine
.test("PU16", "Add new user U3 admin", "POST", "/admin/v1/users",
413 headers_json
, {"username": "U3", "projects": ["P2"], "password": "pw3"}, (201, 204),
414 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
415 engine
.test("PU17", "Edit user projects admin", "PUT", "/admin/v1/users/U3", headers_json
,
416 {"projects": ["P2"]}, 204, None, None)
418 engine
.test("PU18", "Delete project P2 conflict", "DELETE", "/admin/v1/projects/P2", headers_json
, None, 409,
419 r_header_json
, "json")
420 engine
.test("PU19", "Delete project P2 forcing", "DELETE", "/admin/v1/projects/P2?FORCE=True", headers_json
,
421 None, 204, None, None)
423 engine
.test("PU20", "Delete user U1. Conflict deleting own user", "DELETE", "/admin/v1/users/U1", headers_json
,
424 None, 409, r_header_json
, "json")
425 engine
.test("PU21", "Delete user U2", "DELETE", "/admin/v1/users/U2", headers_json
, None, 204, None, None)
426 engine
.test("PU22", "Delete user U3", "DELETE", "/admin/v1/users/U3", headers_json
, None, 204, None, None)
428 engine
.remove_authorization() # To force get authorization
429 engine
.get_autorization()
430 engine
.test("PU23", "Delete user U1", "DELETE", "/admin/v1/users/U1", headers_json
, None, 204, None, None)
431 engine
.test("PU24", "Delete project P1", "DELETE", "/admin/v1/projects/P1", headers_json
, None, 204, None, None)
432 engine
.test("PU25", "Delete project Padmin", "DELETE", "/admin/v1/projects/Padmin", headers_json
, None, 204,
437 description
= "Creates/edit/delete fake VIMs and SDN controllers"
441 "schema_version": "1.0",
442 "schema_type": "No idea",
444 "description": "Descriptor name",
445 "vim_type": "openstack",
446 "vim_url": "http://localhost:/vim",
447 "vim_tenant_name": "vimTenant",
449 "vim_password": "password",
450 "config": {"config_param": 1}
454 "description": "sdn-description",
455 "dpid": "50:50:52:54:00:94:21:21",
456 "ip": "192.168.15.17",
458 "type": "opendaylight",
463 self
.port_mapping
= [
464 {"compute_node": "compute node 1",
465 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/1", "switch_mac": "52:54:00:94:21:21"},
466 {"pci": "0000:81:00.1", "switch_port": "port-2/2", "switch_mac": "52:54:00:94:21:22"}
468 {"compute_node": "compute node 2",
469 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/3", "switch_mac": "52:54:00:94:21:23"},
470 {"pci": "0000:81:00.1", "switch_port": "port-2/4", "switch_mac": "52:54:00:94:21:24"}
474 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
476 vim_bad
= self
.vim
.copy()
479 engine
.get_autorization()
480 engine
.test("FVIM1", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
, (201, 204),
481 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json")
482 engine
.test("FVIM2", "Create VIM without name, bad schema", "POST", "/admin/v1/vim_accounts", headers_json
,
483 vim_bad
, 422, None, headers_json
)
484 engine
.test("FVIM3", "Create VIM name repeated", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
,
485 409, None, headers_json
)
486 engine
.test("FVIM4", "Show VIMs", "GET", "/admin/v1/vim_accounts", headers_yaml
, None, 200, r_header_yaml
,
488 engine
.test("FVIM5", "Show VIM", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
, None, 200,
489 r_header_yaml
, "yaml")
492 engine
.test("FVIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/<FVIM1>?FORCE=True", headers_yaml
,
494 engine
.test("FVIM7", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
, None,
495 404, r_header_yaml
, "yaml")
497 # delete and wait until is really deleted
498 engine
.test("FVIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
, None, 202,
502 r
= engine
.test("FVIM7", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
,
503 None, None, r_header_yaml
, "yaml")
504 if r
.status_code
== 404:
506 elif r
.status_code
== 200:
510 raise TestException("Vim created at 'FVIM1' is not delete after {} seconds".format(timeout
))
513 class TestVIMSDN(TestFakeVim
):
514 description
= "Creates VIM with SDN editing SDN controllers and port_mapping"
517 TestFakeVim
.__init
__(self
)
519 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
520 engine
.get_autorization()
522 engine
.test("VIMSDN1", "Create SDN", "POST", "/admin/v1/sdns", headers_json
, self
.sdn
, (201, 204),
523 {"Location": "/admin/v1/sdns/", "Content-Type": "application/json"}, "json")
526 engine
.test("VIMSDN2", "Edit SDN", "PATCH", "/admin/v1/sdns/<VIMSDN1>", headers_json
, {"name": "new_sdn_name"},
530 self
.vim
["config"]["sdn-controller"] = engine
.test_ids
["VIMSDN1"]
531 self
.vim
["config"]["sdn-port-mapping"] = self
.port_mapping
532 engine
.test("VIMSDN3", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
, (200, 204, 201),
533 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json"),
535 self
.port_mapping
[0]["compute_node"] = "compute node XX"
536 engine
.test("VIMSDN4", "Edit VIM change port-mapping", "PUT", "/admin/v1/vim_accounts/<VIMSDN3>", headers_json
,
537 {"config": {"sdn-port-mapping": self
.port_mapping
}}, 204, None, None)
538 engine
.test("VIMSDN5", "Edit VIM remove port-mapping", "PUT", "/admin/v1/vim_accounts/<VIMSDN3>", headers_json
,
539 {"config": {"sdn-port-mapping": None}}, 204, None, None)
543 engine
.test("VIMSDN6", "Delete VIM remove port-mapping", "DELETE",
544 "/admin/v1/vim_accounts/<VIMSDN3>?FORCE=True", headers_json
, None, 202, None, 0)
545 engine
.test("VIMSDN7", "Delete SDNC", "DELETE", "/admin/v1/sdns/<VIMSDN1>?FORCE=True", headers_json
, None,
548 engine
.test("VIMSDN8", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<VIMSDN3>", headers_yaml
,
549 None, 404, r_header_yaml
, "yaml")
550 engine
.test("VIMSDN9", "Check SDN is deleted", "GET", "/admin/v1/sdns/<VIMSDN1>", headers_yaml
, None,
551 404, r_header_yaml
, "yaml")
553 # delete and wait until is really deleted
554 engine
.test("VIMSDN6", "Delete VIM remove port-mapping", "DELETE", "/admin/v1/vim_accounts/<VIMSDN3>",
555 headers_json
, None, (202, 201, 204), None, 0)
556 engine
.test("VIMSDN7", "Delete SDN", "DELETE", "/admin/v1/sdns/<VIMSDN1>", headers_json
, None,
557 (202, 201, 204), None, 0)
560 r
= engine
.test("VIMSDN8", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<VIMSDN3>",
561 headers_yaml
, None, None, r_header_yaml
, "yaml")
562 if r
.status_code
== 404:
564 elif r
.status_code
== 200:
568 raise TestException("Vim created at 'VIMSDN3' is not delete after {} seconds".format(timeout
))
570 r
= engine
.test("VIMSDN9", "Check SDNC is deleted", "GET", "/admin/v1/sdns/<VIMSDN1>",
571 headers_yaml
, None, None, r_header_yaml
, "yaml")
572 if r
.status_code
== 404:
574 elif r
.status_code
== 200:
578 raise TestException("SDNC created at 'VIMSDN1' is not delete after {} seconds".format(timeout
))
582 description
= "Base class for downloading descriptors from ETSI, onboard and deploy in real VIM"
593 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
594 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
595 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
596 self
.descriptor_edit
= None
597 self
.uses_configuration
= False
603 self
.passed_tests
= 0
607 def create_descriptors(self
, engine
):
608 temp_dir
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/temp/"
609 if not os
.path
.exists(temp_dir
):
610 os
.makedirs(temp_dir
)
611 for vnfd_index
, vnfd_filename
in enumerate(self
.vnfd_filenames
):
612 if "/" in vnfd_filename
:
613 vnfd_filename_path
= vnfd_filename
614 if not os
.path
.exists(vnfd_filename_path
):
615 raise TestException("File '{}' does not exist".format(vnfd_filename_path
))
617 vnfd_filename_path
= temp_dir
+ vnfd_filename
618 if not os
.path
.exists(vnfd_filename_path
):
619 with
open(vnfd_filename_path
, "wb") as file:
620 response
= requests
.get(self
.descriptor_url
+ vnfd_filename
)
621 if response
.status_code
>= 300:
622 raise TestException("Error downloading descriptor from '{}': {}".format(
623 self
.descriptor_url
+ vnfd_filename
, response
.status_code
))
624 file.write(response
.content
)
625 if vnfd_filename_path
.endswith(".yaml"):
626 headers
= headers_yaml
628 headers
= headers_zip_yaml
629 if self
.step
% 2 == 0:
630 # vnfd CREATE AND UPLOAD in one step:
631 test_name
= "DEPLOY{}".format(self
.step
)
632 engine
.test(test_name
, "Onboard VNFD in one step", "POST",
633 "/vnfpkgm/v1/vnf_packages_content" + self
.qforce
, headers
, "@b" + vnfd_filename_path
, 201,
634 {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"},
636 self
.vnfds_test
.append(test_name
)
637 self
.vnfds_id
.append(engine
.test_ids
["last_id"])
640 # vnfd CREATE AND UPLOAD ZIP
641 test_name
= "DEPLOY{}".format(self
.step
)
642 engine
.test(test_name
, "Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages",
643 headers_json
, None, 201,
644 {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
645 self
.vnfds_test
.append(test_name
)
646 self
.vnfds_id
.append(engine
.test_ids
["last_id"])
648 # location = r.headers["Location"]
649 # vnfd_id = location[location.rfind("/")+1:]
650 engine
.test("DEPLOY{}".format(self
.step
), "Onboard VNFD step 2 as ZIP", "PUT",
651 "/vnfpkgm/v1/vnf_packages/<>/package_content" + self
.qforce
,
652 headers
, "@b" + vnfd_filename_path
, 204, None, 0)
655 if self
.descriptor_edit
:
656 if "vnfd{}".format(vnfd_index
) in self
.descriptor_edit
:
658 engine
.test("DEPLOY{}".format(self
.step
), "Edit VNFD ", "PATCH",
659 "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfds_id
[-1]),
660 headers_yaml
, self
.descriptor_edit
["vnfd{}".format(vnfd_index
)], 204, None, None)
663 if "/" in self
.nsd_filename
:
664 nsd_filename_path
= self
.nsd_filename
665 if not os
.path
.exists(nsd_filename_path
):
666 raise TestException("File '{}' does not exist".format(nsd_filename_path
))
668 nsd_filename_path
= temp_dir
+ self
.nsd_filename
669 if not os
.path
.exists(nsd_filename_path
):
670 with
open(nsd_filename_path
, "wb") as file:
671 response
= requests
.get(self
.descriptor_url
+ self
.nsd_filename
)
672 if response
.status_code
>= 300:
673 raise TestException("Error downloading descriptor from '{}': {}".format(
674 self
.descriptor_url
+ self
.nsd_filename
, response
.status_code
))
675 file.write(response
.content
)
676 if nsd_filename_path
.endswith(".yaml"):
677 headers
= headers_yaml
679 headers
= headers_zip_yaml
681 self
.nsd_test
= "DEPLOY" + str(self
.step
)
682 if self
.step
% 2 == 0:
683 # nsd CREATE AND UPLOAD in one step:
684 engine
.test("DEPLOY{}".format(self
.step
), "Onboard NSD in one step", "POST",
685 "/nsd/v1/ns_descriptors_content" + self
.qforce
, headers
, "@b" + nsd_filename_path
, 201,
686 {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}, yaml
)
688 self
.nsd_id
= engine
.test_ids
["last_id"]
690 # nsd CREATE AND UPLOAD ZIP
691 engine
.test("DEPLOY{}".format(self
.step
), "Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors",
692 headers_json
, None, 201,
693 {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
695 self
.nsd_id
= engine
.test_ids
["last_id"]
696 # location = r.headers["Location"]
697 # vnfd_id = location[location.rfind("/")+1:]
698 engine
.test("DEPLOY{}".format(self
.step
), "Onboard NSD step 2 as ZIP", "PUT",
699 "/nsd/v1/ns_descriptors/<>/nsd_content" + self
.qforce
,
700 headers
, "@b" + nsd_filename_path
, 204, None, 0)
703 if self
.descriptor_edit
and "nsd" in self
.descriptor_edit
:
705 engine
.test("DEPLOY{}".format(self
.step
), "Edit NSD ", "PATCH",
706 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
),
707 headers_yaml
, self
.descriptor_edit
["nsd"], 204, None, None)
710 def delete_descriptors(self
, engine
):
712 engine
.test("DEPLOY{}".format(self
.step
), "Delete NSSD SOL005", "DELETE",
713 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
),
714 headers_yaml
, None, 204, None, 0)
716 for vnfd_id
in self
.vnfds_id
:
717 engine
.test("DEPLOY{}".format(self
.step
), "Delete VNFD SOL005", "DELETE",
718 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id
), headers_yaml
, None, 204, None, 0)
721 def instantiate(self
, engine
, ns_data
):
722 ns_data_text
= yaml
.safe_dump(ns_data
, default_flow_style
=True, width
=256)
723 # create NS Two steps
724 r
= engine
.test("DEPLOY{}".format(self
.step
), "Create NS step 1", "POST", "/nslcm/v1/ns_instances",
725 headers_yaml
, ns_data_text
, 201,
726 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"}, "yaml")
727 self
.ns_test
= "DEPLOY{}".format(self
.step
)
728 self
.ns_id
= engine
.test_ids
["last_id"]
730 r
= engine
.test("DEPLOY{}".format(self
.step
), "Instantiate NS step 2", "POST",
731 "/nslcm/v1/ns_instances/<{}>/instantiate".format(self
.ns_test
), headers_yaml
, ns_data_text
,
732 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
733 nslcmop_test
= "DEPLOY{}".format(self
.step
)
737 # Wait until status is Ok
738 wait
= timeout_configure
if self
.uses_configuration
else timeout_deploy
740 r
= engine
.test("DEPLOY{}".format(self
.step
), "Wait until NS is deployed and configured", "GET",
741 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test
), headers_json
, None,
742 200, r_header_json
, "json")
744 if "COMPLETED" in nslcmop
["operationState"]:
746 elif "FAILED" in nslcmop
["operationState"]:
747 raise TestException("NS instantiate has failed: {}".format(nslcmop
["detailed-status"]))
751 raise TestException("NS instantiate is not done after {} seconds".format(timeout_deploy
))
754 def _wait_nslcmop_ready(self
, engine
, nslcmop_test
, timeout_deploy
, expected_fail
=False):
757 r
= engine
.test("DEPLOY{}".format(self
.step
), "Wait to ns lcm operation complete", "GET",
758 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test
), headers_json
, None,
759 200, r_header_json
, "json")
761 if "COMPLETED" in nslcmop
["operationState"]:
763 raise TestException("NS terminate has success, expecting failing: {}".format(
764 nslcmop
["detailed-status"]))
766 elif "FAILED" in nslcmop
["operationState"]:
767 if not expected_fail
:
768 raise TestException("NS terminate has failed: {}".format(nslcmop
["detailed-status"]))
773 raise TestException("NS instantiate is not terminate after {} seconds".format(timeout
))
775 def terminate(self
, engine
):
778 r
= engine
.test("DEPLOY{}".format(self
.step
), "Terminate NS", "POST",
779 "/nslcm/v1/ns_instances/<{}>/terminate".format(self
.ns_test
), headers_yaml
, None,
780 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
781 nslcmop2_test
= "DEPLOY{}".format(self
.step
)
783 # Wait until status is Ok
784 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_test
, timeout_deploy
)
786 r
= engine
.test("DEPLOY{}".format(self
.step
), "Delete NS", "DELETE",
787 "/nslcm/v1/ns_instances/<{}>".format(self
.ns_test
), headers_yaml
, None,
791 r
= engine
.test("DEPLOY{}".format(self
.step
), "Delete NS with FORCE", "DELETE",
792 "/nslcm/v1/ns_instances/<{}>?FORCE=True".format(self
.ns_test
), headers_yaml
, None,
796 # check all it is deleted
797 r
= engine
.test("DEPLOY{}".format(self
.step
), "Check NS is deleted", "GET",
798 "/nslcm/v1/ns_instances/<{}>".format(self
.ns_test
), headers_yaml
, None,
801 r
= engine
.test("DEPLOY{}".format(self
.step
), "Check NSLCMOPs are deleted", "GET",
802 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId=<{}>".format(self
.ns_test
), headers_json
, None,
805 if not isinstance(nslcmops
, list) or nslcmops
:
806 raise TestException("NS {} deleted but with ns_lcm_op_occ active: {}".format(self
.ns_test
, nslcmops
))
808 def test_ns(self
, engine
, test_osm
, commands
=None, users
=None, passwds
=None, keys
=None, timeout
=0):
811 r
= engine
.test("TEST_NS{}".format(n
), "GET VNFR_IDs", "GET",
812 "/nslcm/v1/ns_instances/{}".format(self
.ns_id
), headers_json
, None,
813 200, r_header_json
, "json")
817 vnfr_list
= ns_data
['constituent-vnfr-ref']
820 for vnfr_id
in vnfr_list
:
821 self
.total_tests
+= 1
822 r
= engine
.test("TEST_NS{}".format(n
), "GET IP_ADDRESS OF VNFR", "GET",
823 "/nslcm/v1/vnfrs/{}".format(vnfr_id
), headers_json
, None,
824 200, r_header_json
, "json")
828 if vnfr_data
.get("ip-address"):
829 name
= "TEST_NS{}".format(n
)
830 description
= "Run tests in VNFR with IP {}".format(vnfr_data
['ip-address'])
832 test_description
= "Test {} {}".format(name
, description
)
833 logger
.warning(test_description
)
834 vnf_index
= str(vnfr_data
["member-vnf-index-ref"])
835 while timeout
>= time
:
836 result
, message
= self
.do_checks([vnfr_data
["ip-address"]],
837 vnf_index
=vnfr_data
["member-vnf-index-ref"],
838 commands
=commands
.get(vnf_index
), user
=users
.get(vnf_index
),
839 passwd
=passwds
.get(vnf_index
), key
=keys
.get(vnf_index
))
841 logger
.warning(message
)
847 logger
.critical(message
)
851 logger
.critical(message
)
853 logger
.critical("VNFR {} has not mgmt address. Check failed".format(vnfr_id
))
855 def do_checks(self
, ip
, vnf_index
, commands
=[], user
=None, passwd
=None, key
=None):
858 from pssh
.clients
import ParallelSSHClient
859 from pssh
.utils
import load_private_key
860 from ssh2
import exceptions
as ssh2Exception
861 except ImportError as e
:
862 logger
.critical("Package <pssh> or/and <urllib3> is not installed. Please add them with 'pip3 install "
863 "parallel-ssh urllib3': {}".format(e
))
864 urllib3
.disable_warnings(urllib3
.exceptions
.InsecureRequestWarning
)
866 p_host
= os
.environ
.get("PROXY_HOST")
867 p_user
= os
.environ
.get("PROXY_USER")
868 p_password
= os
.environ
.get("PROXY_PASSWD")
871 pkey
= load_private_key(key
)
875 client
= ParallelSSHClient(ip
, user
=user
, password
=passwd
, pkey
=pkey
, proxy_host
=p_host
,
876 proxy_user
=p_user
, proxy_password
=p_password
, timeout
=10, num_retries
=0)
878 output
= client
.run_command(cmd
)
880 if output
[ip
[0]].exit_code
:
881 return -1, " VNFR {} could not be checked: {}".format(ip
[0], output
[ip
[0]].stderr
)
883 self
.passed_tests
+= 1
884 return 1, " Test successful"
885 except (ssh2Exception
.ChannelFailure
, ssh2Exception
.SocketDisconnectError
, ssh2Exception
.SocketTimeout
,
886 ssh2Exception
.SocketRecvError
) as e
:
887 return 0, "Timeout accessing the VNFR {}: {}".format(ip
[0], str(e
))
888 except Exception as e
:
889 return -1, "ERROR checking the VNFR {}: {}".format(ip
[0], str(e
))
891 def aditional_operations(self
, engine
, test_osm
, manual_check
):
894 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
895 engine
.get_autorization()
896 nsname
= os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
898 if "vnfd-files" in test_params
:
899 self
.vnfd_filenames
= test_params
["vnfd-files"].split(",")
900 if "nsd-file" in test_params
:
901 self
.nsd_filename
= test_params
["nsd-file"]
902 if test_params
.get("ns-name"):
903 nsname
= test_params
["ns-name"]
904 self
.create_descriptors(engine
)
906 # create real VIM if not exist
907 self
.vim_id
= engine
.get_create_vim(test_osm
)
908 ns_data
= {"nsDescription": "default description", "nsName": nsname
, "nsdId": self
.nsd_id
,
909 "vimAccountId": self
.vim_id
}
910 if test_params
and test_params
.get("ns-config"):
911 if isinstance(test_params
["ns-config"], str):
912 ns_data
.update(yaml
.load(test_params
["ns-config"]))
914 ns_data
.update(test_params
["ns-config"])
915 self
.instantiate(engine
, ns_data
)
918 input('NS has been deployed. Perform manual check and press enter to resume')
920 self
.test_ns(engine
, test_osm
, self
.cmds
, self
.uss
, self
.pss
, self
.keys
, self
.timeout
)
921 self
.aditional_operations(engine
, test_osm
, manual_check
)
922 self
.terminate(engine
)
923 self
.delete_descriptors(engine
)
926 def print_results(self
):
927 print("\n\n\n--------------------------------------------")
928 print("TEST RESULTS:\n PASSED TESTS: {} - TOTAL TESTS: {}".format(self
.total_tests
, self
.passed_tests
))
929 print("--------------------------------------------")
932 class TestDeployHackfestCirros(TestDeploy
):
933 description
= "Load and deploy Hackfest cirros_2vnf_ns example"
937 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
938 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
939 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
940 self
.uss
= {'1': "cirros", '2': "cirros"}
941 self
.pss
= {'1': "cubswin:)", '2': "cubswin:)"}
944 class TestDeployHackfest1(TestDeploy
):
945 description
= "Load and deploy Hackfest_1_vnfd example"
949 self
.vnfd_filenames
= ("hackfest_1_vnfd.tar.gz",)
950 self
.nsd_filename
= "hackfest_1_nsd.tar.gz"
951 # self.cmds = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
952 # self.uss = {'1': "cirros", '2': "cirros"}
953 # self.pss = {'1': "cubswin:)", '2': "cubswin:)"}
956 class TestDeployHackfestCirrosScaling(TestDeploy
):
957 description
= "Load and deploy Hackfest cirros_2vnf_ns example with scaling modifications"
961 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
962 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
964 def create_descriptors(self
, engine
):
965 super().create_descriptors(engine
)
966 # Modify VNFD to add scaling and count=2
969 "$id: 'cirros_vnfd-VM'":
971 scaling-group-descriptor:
972 - name: "scale_cirros"
973 max-instance-count: 2
975 - vdu-id-ref: cirros_vnfd-VM
978 engine
.test("DEPLOY{}".format(self
.step
), "Edit VNFD ", "PATCH",
979 "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfds_id
[0]),
980 headers_yaml
, payload
, 204, None, None)
983 def aditional_operations(self
, engine
, test_osm
, manual_check
):
986 # 2 perform scale out twice
987 payload
= '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
988 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
989 for i
in range(0, 2):
990 engine
.test("DEPLOY{}".format(self
.step
), "Execute scale action over NS", "POST",
991 "/nslcm/v1/ns_instances/<{}>/scale".format(self
.ns_test
), headers_yaml
, payload
,
992 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
993 nslcmop2_scale_out
= "DEPLOY{}".format(self
.step
)
994 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_scale_out
, timeout_deploy
)
996 input('NS scale out done. Check that two more vdus are there')
997 # TODO check automatic
1000 payload
= '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1001 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
1002 for i
in range(0, 2):
1003 engine
.test("DEPLOY{}".format(self
.step
), "Execute scale IN action over NS", "POST",
1004 "/nslcm/v1/ns_instances/<{}>/scale".format(self
.ns_test
), headers_yaml
, payload
,
1005 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1006 nslcmop2_scale_in
= "DEPLOY{}".format(self
.step
)
1007 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_scale_in
, timeout_deploy
)
1009 input('NS scale in done. Check that two less vdus are there')
1010 # TODO check automatic
1012 # perform scale in that must fail as reached limit
1013 engine
.test("DEPLOY{}".format(self
.step
), "Execute scale IN out of limit action over NS", "POST",
1014 "/nslcm/v1/ns_instances/<{}>/scale".format(self
.ns_test
), headers_yaml
, payload
,
1015 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1016 nslcmop2_scale_in
= "DEPLOY{}".format(self
.step
)
1017 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_scale_in
, timeout_deploy
, expected_fail
=True)
1020 class TestDeployIpMac(TestDeploy
):
1021 description
= "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
1025 self
.vnfd_filenames
= ("vnfd_2vdu_set_ip_mac2.yaml", "vnfd_2vdu_set_ip_mac.yaml")
1026 self
.nsd_filename
= "scenario_2vdu_set_ip_mac.yaml"
1027 self
.descriptor_url
= \
1028 "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
1029 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1030 self
.uss
= {'1': "osm", '2': "osm"}
1031 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1034 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1035 # super().run(engine, test_osm, manual_check, test_params)
1036 # run again setting IPs with instantiate parameters
1037 instantiation_params
= {
1040 "member-vnf-index": "1",
1043 "name": "internal_vld1", # net_internal
1045 "ip-version": "ipv4",
1046 "subnet-address": "10.9.8.0/24",
1047 "dhcp-params": {"count": 100, "start-address": "10.9.8.100"}
1049 "internal-connection-point": [
1052 "ip-address": "10.9.8.2",
1056 "ip-address": "10.9.8.3",
1067 # "name": "iface11",
1068 # "floating-ip-required": True,
1072 "mac-address": "52:33:44:55:66:13"
1081 "ip-address": "10.31.31.22",
1082 "mac-address": "52:33:44:55:66:21"
1091 super().run(engine
, test_osm
, manual_check
, test_params
={"ns-config": instantiation_params
})
1094 class TestDeployHackfest4(TestDeploy
):
1095 description
= "Load and deploy Hackfest 4 example."
1099 self
.vnfd_filenames
= ("hackfest_4_vnfd.tar.gz",)
1100 self
.nsd_filename
= "hackfest_4_nsd.tar.gz"
1101 self
.uses_configuration
= True
1102 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1103 self
.uss
= {'1': "ubuntu", '2': "ubuntu"}
1104 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1106 def create_descriptors(self
, engine
):
1107 super().create_descriptors(engine
)
1108 # Modify VNFD to add scaling
1110 scaling-group-descriptor:
1111 - name: "scale_dataVM"
1112 max-instance-count: 10
1114 - name: "auto_cpu_util_above_threshold"
1115 scaling-type: "automatic"
1119 - name: "cpu_util_above_threshold"
1120 scale-in-threshold: 15
1121 scale-in-relational-operation: "LE"
1122 scale-out-threshold: 60
1123 scale-out-relational-operation: "GE"
1124 vnf-monitoring-param-ref: "all_aaa_cpu_util"
1126 - vdu-id-ref: dataVM
1128 scaling-config-action:
1129 - trigger: post-scale-out
1130 vnf-config-primitive-name-ref: touch
1131 - trigger: pre-scale-in
1132 vnf-config-primitive-name-ref: touch
1139 default-value: '/home/ubuntu/touched'
1141 engine
.test("DEPLOY{}".format(self
.step
), "Edit VNFD ", "PATCH",
1142 "/vnfpkgm/v1/vnf_packages/<{}>".format(self
.vnfds_test
[0]), headers_yaml
, payload
, 204, None, None)
1146 class TestDeployHackfest3Charmed(TestDeploy
):
1147 description
= "Load and deploy Hackfest 3charmed_ns example. Modifies it for adding scaling and performs " \
1148 "primitive actions and scaling"
1152 self
.vnfd_filenames
= ("hackfest_3charmed_vnfd.tar.gz",)
1153 self
.nsd_filename
= "hackfest_3charmed_nsd.tar.gz"
1154 self
.uses_configuration
= True
1155 self
.cmds
= {'1': [''], '2': ['ls -lrt /home/ubuntu/first-touch', ]}
1156 self
.uss
= {'1': "ubuntu", '2': "ubuntu"}
1157 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1159 # def create_descriptors(self, engine):
1160 # super().create_descriptors(engine)
1161 # # Modify VNFD to add scaling
1163 # scaling-group-descriptor:
1164 # - name: "scale_dataVM"
1165 # max-instance-count: 10
1167 # - name: "auto_cpu_util_above_threshold"
1168 # scaling-type: "automatic"
1172 # - name: "cpu_util_above_threshold"
1173 # scale-in-threshold: 15
1174 # scale-in-relational-operation: "LE"
1175 # scale-out-threshold: 60
1176 # scale-out-relational-operation: "GE"
1177 # vnf-monitoring-param-ref: "all_aaa_cpu_util"
1179 # - vdu-id-ref: dataVM
1181 # scaling-config-action:
1182 # - trigger: post-scale-out
1183 # vnf-config-primitive-name-ref: touch
1184 # - trigger: pre-scale-in
1185 # vnf-config-primitive-name-ref: touch
1186 # vnf-configuration:
1192 # default-value: '/home/ubuntu/touched'
1194 # engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
1195 # "/vnfpkgm/v1/vnf_packages/<{}>".format(self.vnfds_test[0]),
1196 # headers_yaml, payload, 200,
1197 # r_header_yaml, yaml)
1198 # self.vnfds_test.append("DEPLOY" + str(self.step))
1201 def aditional_operations(self
, engine
, test_osm
, manual_check
):
1205 payload
= '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
1206 engine
.test("DEPLOY{}".format(self
.step
), "Executer service primitive over NS", "POST",
1207 "/nslcm/v1/ns_instances/<{}>/action".format(self
.ns_test
), headers_yaml
, payload
,
1208 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1209 nslcmop2_action
= "DEPLOY{}".format(self
.step
)
1211 # Wait until status is Ok
1212 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_action
, timeout_deploy
)
1214 input('NS service primitive has been executed. Check that file /home/ubuntu/OSMTESTNBI is present at '
1217 cmds
= {'1': [''], '2': ['ls -lrt /home/ubuntu/OSMTESTNBI', ]}
1218 uss
= {'1': "ubuntu", '2': "ubuntu"}
1219 pss
= {'1': "osm4u", '2': "osm4u"}
1220 self
.test_ns(engine
, test_osm
, cmds
, uss
, pss
, self
.keys
, self
.timeout
)
1222 # # 2 perform scale out
1223 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1224 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1225 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1226 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1227 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1228 # nslcmop2_scale_out = "DEPLOY{}".format(self.step)
1229 # self._wait_nslcmop_ready(engine, nslcmop2_scale_out, timeout_deploy)
1231 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
1232 # # TODO check automatic
1234 # # 2 perform scale in
1235 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1236 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1237 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1238 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1239 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1240 # nslcmop2_scale_in = "DEPLOY{}".format(self.step)
1241 # self._wait_nslcmop_ready(engine, nslcmop2_scale_in, timeout_deploy)
1243 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
1244 # # TODO check automatic
1247 class TestDeploySingleVdu(TestDeployHackfest3Charmed
):
1248 description
= "Generate a single VDU base on editing Hackfest3Charmed descriptors and deploy"
1252 self
.qforce
= "?FORCE=True"
1253 self
.descriptor_edit
= {
1254 # Modify VNFD to remove one VDU
1258 "interface": {"$[0]": {"external-connection-point-ref": "pdu-mgmt"}}
1262 "vnf-configuration": None,
1263 "connection-point": {
1267 "short-name": "pdu-mgmt"
1271 "mgmt-interface": {"cp": "pdu-mgmt"},
1272 "description": "A vnf single vdu to be used as PDU",
1276 "id": "pdu_internal",
1277 "name": "pdu_internal",
1278 "internal-connection-point": {"$[1]": None},
1279 "short-name": "pdu_internal",
1285 # Modify NSD accordingly
1287 "constituent-vnfd": {
1288 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
1291 "description": "A nsd to deploy the vnf to act as as PDU",
1293 "name": "nsd-as-pdu",
1294 "short-name": "nsd-as-pdu",
1299 "short-name": "mgmt_pdu",
1300 "vnfd-connection-point-ref": {
1302 "vnfd-connection-point-ref": "pdu-mgmt",
1303 "vnfd-id-ref": "vdu-as-pdu",
1315 class TestDeployHnfd(TestDeployHackfest3Charmed
):
1316 description
= "Generate a HNFD base on editing Hackfest3Charmed descriptors and deploy"
1320 self
.pduDeploy
= TestDeploySingleVdu()
1321 self
.pdu_interface_0
= {}
1322 self
.pdu_interface_1
= {}
1325 # self.vnf_to_pdu = """
1328 # pdu-type: PDU-TYPE-1
1333 # name: pdu-iface-internal
1335 # description: HFND, one PDU + One VDU
1341 self
.pdu_descriptor
= {
1343 "type": "PDU-TYPE-1",
1344 "vim_accounts": "to-override",
1347 "name": "mgmt-iface",
1350 "ip-address": "to override",
1351 "mac-address": "mac_address",
1352 "vim-network-name": "mgmt",
1355 "name": "pdu-iface-internal",
1358 "ip-address": "to override",
1359 "mac-address": "mac_address",
1360 "vim-network-name": "pdu_internal", # OSMNBITEST-PDU-pdu_internal
1364 self
.vnfd_filenames
= ("hackfest_3charmed_vnfd.tar.gz", "hackfest_3charmed_vnfd.tar.gz")
1366 self
.descriptor_edit
= {
1370 "short-name": "hfn1",
1373 "pdu-type": "PDU-TYPE-1",
1375 "$[0]": {"name": "mgmt-iface"},
1376 "$[1]": {"name": "pdu-iface-internal"},
1382 "constituent-vnfd": {
1383 "$[1]": {"vnfd-id-ref": "hfnd1"}
1388 def create_descriptors(self
, engine
):
1389 super().create_descriptors(engine
)
1392 self
.pdu_descriptor
["interfaces"][0].update(self
.pdu_interface_0
)
1393 self
.pdu_descriptor
["interfaces"][1].update(self
.pdu_interface_1
)
1394 self
.pdu_descriptor
["vim_accounts"] = [self
.vim_id
]
1395 # TODO get vim-network-name from vnfr.vld.name
1396 self
.pdu_descriptor
["interfaces"][1]["vim-network-name"] = "{}-{}-{}".format(
1397 os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST"),
1398 "PDU", self
.pdu_descriptor
["interfaces"][1]["vim-network-name"])
1399 test_name
= "DEPLOY{}".format(self
.step
)
1400 engine
.test(test_name
, "Onboard PDU descriptor", "POST", "/pdu/v1/pdu_descriptors",
1401 {"Location": "/pdu/v1/pdu_descriptors/", "Content-Type": "application/yaml"}, self
.pdu_descriptor
,
1402 201, r_header_yaml
, "yaml")
1403 self
.pdu_id
= engine
.test_ids
["last_id"]
1406 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1407 engine
.get_autorization()
1408 nsname
= os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
1410 # create real VIM if not exist
1411 self
.vim_id
= engine
.get_create_vim(test_osm
)
1413 self
.pduDeploy
.create_descriptors(engine
)
1414 self
.pduDeploy
.instantiate(engine
, {"nsDescription": "to be used as PDU", "nsName": nsname
+ "-PDU",
1415 "nsdId": self
.pduDeploy
.nsd_id
, "vimAccountId": self
.vim_id
})
1417 input('VNF to be used as PDU has been deployed. Perform manual check and press enter to resume')
1419 self
.pduDeploy
.test_ns(engine
, test_osm
, self
.pduDeploy
.cmds
, self
.pduDeploy
.uss
, self
.pduDeploy
.pss
,
1420 self
.pduDeploy
.keys
, self
.pduDeploy
.timeout
)
1423 r
= engine
.test("DEPLOY{}".format(self
.step
), "GET IP_ADDRESS OF VNFR", "GET",
1424 "/nslcm/v1/vnfrs?nsr-id-ref={}".format(self
.pduDeploy
.ns_id
), headers_json
, None,
1425 200, r_header_json
, "json")
1427 vnfr_data
= r
.json()
1430 self
.pdu_interface_0
["ip-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][0].get("ip-address")
1431 self
.pdu_interface_1
["ip-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][1].get("ip-address")
1432 self
.pdu_interface_0
["mac-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][0].get("mac-address")
1433 self
.pdu_interface_1
["mac-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][1].get("mac-address")
1434 if not self
.pdu_interface_0
["ip-address"]:
1435 raise TestException("Vnfr has not managment ip address")
1437 self
.pdu_interface_0
["ip-address"] = "192.168.10.10"
1438 self
.pdu_interface_1
["ip-address"] = "192.168.11.10"
1439 self
.pdu_interface_0
["mac-address"] = "52:33:44:55:66:13"
1440 self
.pdu_interface_1
["mac-address"] = "52:33:44:55:66:14"
1442 self
.create_descriptors(engine
)
1444 ns_data
= {"nsDescription": "default description", "nsName": nsname
, "nsdId": self
.nsd_id
,
1445 "vimAccountId": self
.vim_id
}
1446 if test_params
and test_params
.get("ns-config"):
1447 if isinstance(test_params
["ns-config"], str):
1448 ns_data
.update(yaml
.load(test_params
["ns-config"]))
1450 ns_data
.update(test_params
["ns-config"])
1452 self
.instantiate(engine
, ns_data
)
1454 input('NS has been deployed. Perform manual check and press enter to resume')
1456 self
.test_ns(engine
, test_osm
, self
.cmds
, self
.uss
, self
.pss
, self
.keys
, self
.timeout
)
1457 self
.aditional_operations(engine
, test_osm
, manual_check
)
1458 self
.terminate(engine
)
1459 self
.pduDeploy
.terminate(engine
)
1460 self
.delete_descriptors(engine
)
1461 self
.pduDeploy
.delete_descriptors(engine
)
1465 self
.print_results()
1467 def delete_descriptors(self
, engine
):
1468 super().delete_descriptors(engine
)
1470 engine
.test("DEPLOY{}".format(self
.step
), "Delete PDU SOL005", "DELETE",
1471 "/pdu/v1/pdu_descriptors/{}".format(self
.pdu_id
),
1472 headers_yaml
, None, 204, None, 0)
1475 class TestDescriptors
:
1476 description
= "Test VNFD, NSD, PDU descriptors CRUD and dependencies"
1480 self
.vnfd_filename
= "hackfest_3charmed_vnfd.tar.gz"
1481 self
.nsd_filename
= "hackfest_3charmed_nsd.tar.gz"
1482 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
1486 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1487 engine
.get_autorization()
1488 temp_dir
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/temp/"
1489 if not os
.path
.exists(temp_dir
):
1490 os
.makedirs(temp_dir
)
1493 for filename
in (self
.vnfd_filename
, self
.nsd_filename
):
1494 filename_path
= temp_dir
+ filename
1495 if not os
.path
.exists(filename_path
):
1496 with
open(filename_path
, "wb") as file:
1497 response
= requests
.get(self
.descriptor_url
+ filename
)
1498 if response
.status_code
>= 300:
1499 raise TestException("Error downloading descriptor from '{}': {}".format(
1500 self
.descriptor_url
+ filename
, response
.status_code
))
1501 file.write(response
.content
)
1503 vnfd_filename_path
= temp_dir
+ self
.vnfd_filename
1504 nsd_filename_path
= temp_dir
+ self
.nsd_filename
1506 # vnfd CREATE AND UPLOAD in one step:
1507 test_name
= "DESCRIPTOR{}".format(self
.step
)
1508 engine
.test(test_name
, "Onboard VNFD in one step", "POST",
1509 "/vnfpkgm/v1/vnf_packages_content", headers_zip_yaml
, "@b" + vnfd_filename_path
, 201,
1510 {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"}, "yaml")
1511 self
.vnfd_id
= engine
.test_ids
["last_id"]
1514 # get vnfd descriptor
1515 engine
.test("DESCRIPTOR" + str(self
.step
), "Get VNFD descriptor", "GET",
1516 "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_id
), headers_yaml
, None, 200, r_header_yaml
, "yaml")
1519 # get vnfd file descriptor
1520 engine
.test("DESCRIPTOR" + str(self
.step
), "Get VNFD file descriptor", "GET",
1521 "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(self
.vnfd_id
), headers_text
, None, 200,
1522 r_header_text
, "text", temp_dir
+"vnfd-yaml")
1524 # TODO compare files: diff vnfd-yaml hackfest_3charmed_vnfd/hackfest_3charmed_vnfd.yaml
1526 # get vnfd zip file package
1527 engine
.test("DESCRIPTOR" + str(self
.step
), "Get VNFD zip package", "GET",
1528 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self
.vnfd_id
), headers_zip
, None, 200,
1529 r_header_zip
, "zip", temp_dir
+"vnfd-zip")
1531 # TODO compare files: diff vnfd-zip hackfest_3charmed_vnfd.tar.gz
1534 engine
.test("DESCRIPTOR" + str(self
.step
), "Get VNFD artifact package", "GET",
1535 "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/osm.png".format(self
.vnfd_id
), headers_zip
, None, 200,
1536 r_header_octect
, "octet-string", temp_dir
+"vnfd-icon")
1538 # TODO compare files: diff vnfd-icon hackfest_3charmed_vnfd/icons/osm.png
1540 # nsd CREATE AND UPLOAD in one step:
1541 test_name
= "DESCRIPTOR{}".format(self
.step
)
1542 engine
.test(test_name
, "Onboard NSD in one step", "POST",
1543 "/nsd/v1/ns_descriptors_content", headers_zip_yaml
, "@b" + nsd_filename_path
, 201,
1544 {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}, "yaml")
1545 self
.nsd_id
= engine
.test_ids
["last_id"]
1548 # get nsd descriptor
1549 engine
.test("DESCRIPTOR" + str(self
.step
), "Get NSD descriptor", "GET",
1550 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
), headers_yaml
, None, 200, r_header_yaml
, "yaml")
1553 # get nsd file descriptor
1554 engine
.test("DESCRIPTOR" + str(self
.step
), "Get NSD file descriptor", "GET",
1555 "/nsd/v1/ns_descriptors/{}/nsd".format(self
.nsd_id
), headers_text
, None, 200,
1556 r_header_text
, "text", temp_dir
+"nsd-yaml")
1558 # TODO compare files: diff nsd-yaml hackfest_3charmed_nsd/hackfest_3charmed_nsd.yaml
1560 # get nsd zip file package
1561 engine
.test("DESCRIPTOR" + str(self
.step
), "Get NSD zip package", "GET",
1562 "/nsd/v1/ns_descriptors/{}/nsd_content".format(self
.nsd_id
), headers_zip
, None, 200,
1563 r_header_zip
, "zip", temp_dir
+"nsd-zip")
1565 # TODO compare files: diff nsd-zip hackfest_3charmed_nsd.tar.gz
1568 engine
.test("DESCRIPTOR" + str(self
.step
), "Get NSD artifact package", "GET",
1569 "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm.png".format(self
.nsd_id
), headers_zip
, None, 200,
1570 r_header_octect
, "octet-string", temp_dir
+"nsd-icon")
1572 # TODO compare files: diff nsd-icon hackfest_3charmed_nsd/icons/osm.png
1575 test_rest
.test("DESCRIPTOR" + str(self
.step
), "Delete VNFD conflict", "DELETE",
1576 "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_id
), headers_yaml
, None, 409, None, None)
1579 test_rest
.test("DESCRIPTOR" + str(self
.step
), "Delete VNFD force", "DELETE",
1580 "/vnfpkgm/v1/vnf_packages/{}?FORCE=TRUE".format(self
.vnfd_id
), headers_yaml
, None, 204, None, 0)
1584 test_rest
.test("DESCRIPTOR" + str(self
.step
), "Delete NSD", "DELETE",
1585 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
), headers_yaml
, None, 204, None, 0)
1589 if __name__
== "__main__":
1593 # Disable warnings from self-signed certificates.
1594 requests
.packages
.urllib3
.disable_warnings()
1596 logging
.basicConfig(format
="%(levelname)s %(message)s", level
=logging
.ERROR
)
1597 logger
= logging
.getLogger('NBI')
1598 # load parameters and configuration
1599 opts
, args
= getopt
.getopt(sys
.argv
[1:], "hvu:p:",
1600 ["url=", "user=", "password=", "help", "version", "verbose", "no-verbose",
1601 "project=", "insecure", "timeout", "timeout-deploy", "timeout-configure",
1602 "test=", "list", "test-osm", "manual-check", "params="])
1603 url
= "https://localhost:9999/osm"
1604 user
= password
= project
= "admin"
1606 manual_check
= False
1610 "NonAuthorized": TestNonAuthorized
,
1611 "FakeVIM": TestFakeVim
,
1612 "TestUsersProjects": TestUsersProjects
,
1613 "VIM-SDN": TestVIMSDN
,
1614 "Deploy-Custom": TestDeploy
,
1615 "Deploy-Hackfest-Cirros": TestDeployHackfestCirros
,
1616 "Deploy-Hackfest-Cirros-Scaling": TestDeployHackfestCirrosScaling
,
1617 "Deploy-Hackfest-3Charmed": TestDeployHackfest3Charmed
,
1618 "Deploy-Hackfest-4": TestDeployHackfest4
,
1619 "Deploy-CirrosMacIp": TestDeployIpMac
,
1620 "TestDescriptors": TestDescriptors
,
1621 "TestDeployHackfest1": TestDeployHackfest1
,
1622 # "Deploy-MultiVIM": TestDeployMultiVIM,
1623 "DeploySingleVdu": TestDeploySingleVdu
,
1624 "DeployHnfd": TestDeployHnfd
,
1630 # print("parameter:", o, a)
1631 if o
== "--version":
1632 print("test version " + __version__
+ ' ' + version_date
)
1635 for test
, test_class
in test_classes
.items():
1636 print("{:20} {}".format(test
+ ":", test_class
.description
))
1638 elif o
in ("-v", "--verbose"):
1640 elif o
== "no-verbose":
1642 elif o
in ("-h", "--help"):
1645 elif o
== "--test-osm":
1647 elif o
== "--manual-check":
1651 elif o
in ("-u", "--user"):
1653 elif o
in ("-p", "--password"):
1655 elif o
== "--project":
1658 # print("asdfadf", o, a, a.split(","))
1659 for _test
in a
.split(","):
1660 if _test
not in test_classes
:
1661 print("Invalid test name '{}'. Use option '--list' to show available tests".format(_test
),
1664 test_to_do
.append(_test
)
1665 elif o
== "--params":
1666 param_key
, _
, param_value
= a
.partition("=")
1667 text_index
= len(test_to_do
)
1668 if text_index
not in test_params
:
1669 test_params
[text_index
] = {}
1670 test_params
[text_index
][param_key
] = param_value
1671 elif o
== "--insecure":
1673 elif o
== "--timeout":
1675 elif o
== "--timeout-deploy":
1676 timeout_deploy
= int(a
)
1677 elif o
== "--timeout-configure":
1678 timeout_configure
= int(a
)
1680 assert False, "Unhandled option"
1682 logger
.setLevel(logging
.WARNING
)
1684 logger
.setLevel(logging
.DEBUG
)
1686 logger
.setLevel(logging
.ERROR
)
1688 test_rest
= TestRest(url
, user
=user
, password
=password
, project
=project
)
1689 # print("tests to do:", test_to_do)
1692 for test
in test_to_do
:
1694 test_class
= test_classes
[test
]
1695 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(text_index
))
1697 for test
, test_class
in test_classes
.items():
1698 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(0))
1704 except TestException
as e
:
1705 logger
.error(test
+ "Test {} Exception: {}".format(test
, str(e
)))
1707 except getopt
.GetoptError
as e
:
1709 print(e
, file=sys
.stderr
)
1711 except Exception as e
:
1712 logger
.critical(test
+ " Exception: " + str(e
), exc_info
=True)