2 # -*- coding: utf-8 -*-
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
25 from time
import sleep
28 __author__
= "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
29 __date__
= "$2018-03-01$"
31 version_date
= "Oct 2018"
35 print("Usage: ", sys
.argv
[0], "[options]")
36 print(" Performs system tests over running NBI. It can be used for real OSM test using option '--test-osm'")
37 print(" If this is the case env variables 'OSMNBITEST_VIM_NAME' must be suplied to create a VIM if not exist "
38 "where deployment is done")
40 print(" -h|--help: shows this help")
41 print(" --insecure: Allows non trusted https NBI server")
42 print(" --list: list available tests")
43 print(" --manual-check: Deployment tests stop after deployed to allow manual inspection. Only make sense with "
45 print(" -p|--password PASSWORD: NBI access password. 'admin' by default")
46 print(" ---project PROJECT: NBI access project. 'admin' by default")
47 print(" --test TEST[,...]: Execute only a test or a comma separated list of tests")
48 print(" --params key=val: params to the previous test. key can be vnfd-files, nsd-file, ns-name, ns-config")
49 print(" --test-osm: If missing this test is intended for NBI only, no other OSM components are expected. Use "
50 "this flag to test the system. LCM and RO components are expected to be up and running")
51 print(" --timeout TIMEOUT: General NBI timeout, by default {}s".format(timeout
))
52 print(" --timeout-deploy TIMEOUT: Timeout used for getting NS deployed, by default {}s".format(timeout_deploy
))
53 print(" --timeout-configure TIMEOUT: Timeout used for getting NS deployed and configured,"
54 " by default {}s".format(timeout_configure
))
55 print(" -u|--user USERNAME: NBI access username. 'admin' by default")
56 print(" --url URL: complete NBI server URL. 'https//localhost:9999/osm' by default")
57 print(" -v|--verbose print debug information, can be used several times")
58 print(" --no-verbose remove verbosity")
59 print(" --version: prints current version")
60 print("ENV variables used for real deployment tests with option osm-test.")
61 print(" export OSMNBITEST_VIM_NAME=vim-name")
62 print(" export OSMNBITEST_VIM_URL=vim-url")
63 print(" export OSMNBITEST_VIM_TYPE=vim-type")
64 print(" export OSMNBITEST_VIM_TENANT=vim-tenant")
65 print(" export OSMNBITEST_VIM_USER=vim-user")
66 print(" export OSMNBITEST_VIM_PASSWORD=vim-password")
67 print(" export OSMNBITEST_VIM_CONFIG=\"vim-config\"")
68 print(" export OSMNBITEST_NS_NAME=\"vim-config\"")
72 r_header_json
= {"Content-type": "application/json"}
74 "Content-type": "application/json",
75 "Accept": "application/json",
77 r_header_yaml
= {"Content-type": "application/yaml"}
79 "Content-type": "application/yaml",
80 "Accept": "application/yaml",
82 r_header_text
= {"Content-type": "text/plain"}
83 r_header_octect
= {"Content-type": "application/octet-stream"}
85 "Accept": "text/plain",
87 r_header_zip
= {"Content-type": "application/zip"}
89 "Accept": "application/zip",
92 "Accept": "application/yaml", "Content-type": "application/zip"
96 # test ones authorized
97 test_authorized_list
= (
98 ("AU1", "Invalid vnfd id", "GET", "/vnfpkgm/v1/vnf_packages/non-existing-id",
99 headers_json
, None, 404, r_header_json
, "json"),
100 ("AU2", "Invalid nsd id", "GET", "/nsd/v1/ns_descriptors/non-existing-id",
101 headers_yaml
, None, 404, r_header_yaml
, "yaml"),
102 ("AU3", "Invalid nsd id", "DELETE", "/nsd/v1/ns_descriptors_content/non-existing-id",
103 headers_yaml
, None, 404, r_header_yaml
, "yaml"),
105 timeout
= 120 # general timeout
106 timeout_deploy
= 60*10 # timeout for NS deploying without charms
107 timeout_configure
= 60*20 # timeout for NS deploying and configuring
110 class TestException(Exception):
115 def __init__(self
, url_base
, header_base
=None, verify
=False, user
="admin", password
="admin", project
="admin"):
116 self
.url_base
= url_base
117 if header_base
is None:
118 self
.header_base
= {}
120 self
.header_base
= header_base
.copy()
121 self
.s
= requests
.session()
122 self
.s
.headers
= self
.header_base
126 self
.password
= password
127 self
.project
= project
129 # contains ID of tests obtained from Location response header. "" key contains last obtained id
131 self
.old_test_description
= ""
132 self
.test_name
= None
135 def set_header(self
, header
):
136 self
.s
.headers
.update(header
)
138 def set_tet_name(self
, test_name
):
139 self
.test_name
= test_name
141 def unset_header(self
, key
):
142 if key
in self
.s
.headers
:
143 del self
.s
.headers
[key
]
145 def test(self
, name
, description
, method
, url
, headers
, payload
, expected_codes
, expected_headers
,
146 expected_payload
, store_file
=None):
148 Performs an http request and check http code response. Exit if different than allowed. It get the returned id
149 that can be used by following test in the URL with {name} where name is the name of the test
150 :param name: short name of the test
151 :param description: description of the test
152 :param method: HTTP method: GET,PUT,POST,DELETE,...
153 :param url: complete URL or relative URL
154 :param headers: request headers to add to the base headers
155 :param payload: Can be a dict, transformed to json, a text or a file if starts with '@'
156 :param expected_codes: expected response codes, can be int, int tuple or int range
157 :param expected_headers: expected response headers, dict with key values
158 :param expected_payload: expected payload, 0 if empty, 'yaml', 'json', 'text', 'zip', 'octet-stream'
159 :param store_file: filename to store content
160 :return: requests response
165 self
.s
= requests
.session()
169 elif not url
.startswith("http"):
170 url
= self
.url_base
+ url
172 var_start
= url
.find("<") + 1
174 var_end
= url
.find(">", var_start
)
177 var_name
= url
[var_start
:var_end
]
178 if var_name
in self
.test_ids
:
179 url
= url
[:var_start
-1] + self
.test_ids
[var_name
] + url
[var_end
+1:]
180 var_start
+= len(self
.test_ids
[var_name
])
181 var_start
= url
.find("<", var_start
) + 1
183 if isinstance(payload
, str):
184 if payload
.startswith("@"):
186 file_name
= payload
[1:]
187 if payload
.startswith("@b"):
189 file_name
= payload
[2:]
190 with
open(file_name
, mode
) as f
:
192 elif isinstance(payload
, dict):
193 payload
= json
.dumps(payload
)
195 test_description
= "Test {} {} {} {}".format(name
, description
, method
, url
)
196 if self
.old_test_description
!= test_description
:
197 self
.old_test_description
= test_description
198 logger
.warning(test_description
)
200 if expected_payload
in ("zip", "octet-string") or store_file
:
202 r
= getattr(self
.s
, method
.lower())(url
, data
=payload
, headers
=headers
, verify
=self
.verify
, stream
=stream
)
203 if expected_payload
in ("zip", "octet-string") or store_file
:
204 logger
.debug("RX {}".format(r
.status_code
))
206 logger
.debug("RX {}: {}".format(r
.status_code
, r
.text
))
210 if isinstance(expected_codes
, int):
211 expected_codes
= (expected_codes
,)
212 if r
.status_code
not in expected_codes
:
214 "Got status {}. Expected {}. {}".format(r
.status_code
, expected_codes
, r
.text
))
217 for header_key
, header_val
in expected_headers
.items():
218 if header_key
.lower() not in r
.headers
:
219 raise TestException("Header {} not present".format(header_key
))
220 if header_val
and header_val
.lower() not in r
.headers
[header_key
]:
221 raise TestException("Header {} does not contain {} but {}".format(header_key
, header_val
,
222 r
.headers
[header_key
]))
224 if expected_payload
is not None:
225 if expected_payload
== 0 and len(r
.content
) > 0:
226 raise TestException("Expected empty payload")
227 elif expected_payload
== "json":
230 except Exception as e
:
231 raise TestException("Expected json response payload, but got Exception {}".format(e
))
232 elif expected_payload
== "yaml":
234 yaml
.safe_load(r
.text
)
235 except Exception as e
:
236 raise TestException("Expected yaml response payload, but got Exception {}".format(e
))
237 elif expected_payload
in ("zip", "octet-string"):
238 if len(r
.content
) == 0:
239 raise TestException("Expected some response payload, but got empty")
241 # tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
242 # for tarinfo in tar:
243 # tarname = tarinfo.name
245 # except Exception as e:
246 # raise TestException("Expected zip response payload, but got Exception {}".format(e))
247 elif expected_payload
== "text":
248 if len(r
.content
) == 0:
249 raise TestException("Expected some response payload, but got empty")
252 with
open(store_file
, 'wb') as fd
:
253 for chunk
in r
.iter_content(chunk_size
=128):
256 location
= r
.headers
.get("Location")
258 _id
= location
[location
.rfind("/") + 1:]
260 self
.test_ids
[name
] = str(_id
)
261 self
.test_ids
["last_id"] = str(_id
) # last id
262 self
.test_ids
[""] = str(_id
) # last id
264 except TestException
as e
:
268 r_status_code
= r
.status_code
270 logger
.error("{} \nRX code{}: {}".format(e
, r_status_code
, r_text
))
273 logger
.error("Cannot open file {}".format(e
))
276 def get_autorization(self
): # user=None, password=None, project=None):
277 if self
.token
: # and self.user == user and self.password == password and self.project == project:
280 # self.password = password
281 # self.project = project
282 r
= self
.test("TOKEN", "Obtain token", "POST", "/admin/v1/tokens", headers_json
,
283 {"username": self
.user
, "password": self
.password
, "project_id": self
.project
},
284 (200, 201), r_header_json
, "json")
286 self
.token
= response
["id"]
287 self
.set_header({"Authorization": "Bearer {}".format(self
.token
)})
289 def remove_authorization(self
):
291 self
.test("TOKEN_DEL", "Delete token", "DELETE", "/admin/v1/tokens/{}".format(self
.token
), headers_json
,
292 None, (200, 201, 204), None, None)
294 self
.unset_header("Authorization")
296 def get_create_vim(self
, test_osm
):
299 self
.get_autorization()
301 vim_name
= os
.environ
.get("OSMNBITEST_VIM_NAME")
304 "Needed to define OSMNBITEST_VIM_XXX variables to create a real VIM for deployment")
308 r
= self
.test("_VIMGET1", "Get VIM ID", "GET", "/admin/v1/vim_accounts?name={}".format(vim_name
), headers_json
,
309 None, 200, r_header_json
, "json")
312 return vims
[0]["_id"]
315 # check needed environ parameters:
316 if not os
.environ
.get("OSMNBITEST_VIM_URL") or not os
.environ
.get("OSMNBITEST_VIM_TENANT"):
317 raise TestException("Env OSMNBITEST_VIM_URL and OSMNBITEST_VIM_TENANT are needed for create a real VIM"
318 " to deploy on whit the --test-osm option")
319 vim_data
= "{{schema_version: '1.0', name: '{}', vim_type: {}, vim_url: '{}', vim_tenant_name: '{}', "\
320 "vim_user: {}, vim_password: {}".format(vim_name
,
321 os
.environ
.get("OSMNBITEST_VIM_TYPE", "openstack"),
322 os
.environ
.get("OSMNBITEST_VIM_URL"),
323 os
.environ
.get("OSMNBITEST_VIM_TENANT"),
324 os
.environ
.get("OSMNBITEST_VIM_USER"),
325 os
.environ
.get("OSMNBITEST_VIM_PASSWORD"))
326 if os
.environ
.get("OSMNBITEST_VIM_CONFIG"):
327 vim_data
+= " ,config: {}".format(os
.environ
.get("OSMNBITEST_VIM_CONFIG"))
330 vim_data
= "{schema_version: '1.0', name: fakeVim, vim_type: openstack, vim_url: 'http://10.11.12.13/fake'"\
331 ", vim_tenant_name: 'vimtenant', vim_user: vimuser, vim_password: vimpassword}"
332 r
= self
.test("_VIMGET2", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_yaml
, vim_data
,
333 (201), {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/yaml"}, "yaml")
334 location
= r
.headers
.get("Location")
335 return location
[location
.rfind("/") + 1:]
338 class TestNonAuthorized
:
339 description
= "test invalid URLs. methods and no authorization"
342 def run(engine
, test_osm
, manual_check
, test_params
=None):
343 engine
.remove_authorization()
344 test_not_authorized_list
= (
345 ("NA1", "Invalid token", "GET", "/admin/v1/users", headers_json
, None, 401, r_header_json
, "json"),
346 ("NA2", "Invalid URL", "POST", "/admin/v1/nonexist", headers_yaml
, None, 405, r_header_yaml
, "yaml"),
347 ("NA3", "Invalid version", "DELETE", "/admin/v2/users", headers_yaml
, None, 405, r_header_yaml
, "yaml"),
349 for t
in test_not_authorized_list
:
353 class TestUsersProjects
:
354 description
= "test project and user creation"
357 def run(engine
, test_osm
, manual_check
, test_params
=None):
358 engine
.get_autorization()
359 engine
.test("PU1", "Create project non admin", "POST", "/admin/v1/projects", headers_json
, {"name": "P1"},
360 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
361 engine
.test("PU2", "Create project admin", "POST", "/admin/v1/projects", headers_json
,
362 {"name": "Padmin", "admin": True}, (201, 204),
363 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
364 engine
.test("PU3", "Create project bad format", "POST", "/admin/v1/projects", headers_json
, {"name": 1}, 422,
365 r_header_json
, "json")
366 engine
.test("PU4", "Create user with bad project", "POST", "/admin/v1/users", headers_json
,
367 {"username": "U1", "projects": ["P1", "P2", "Padmin"], "password": "pw1"}, 409,
368 r_header_json
, "json")
369 engine
.test("PU5", "Create user with bad project and force", "POST", "/admin/v1/users?FORCE=True", headers_json
,
370 {"username": "U1", "projects": ["P1", "P2", "Padmin"], "password": "pw1"}, 201,
371 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
372 engine
.test("PU6", "Create user 2", "POST", "/admin/v1/users", headers_json
,
373 {"username": "U2", "projects": ["P1"], "password": "pw2"}, 201,
374 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
376 engine
.test("PU7", "Edit user U1, delete P2 project", "PATCH", "/admin/v1/users/U1", headers_json
,
377 {"projects": {"$'P2'": None}}, 204, None, None)
378 res
= engine
.test("PU1", "Check user U1, contains the right projects", "GET", "/admin/v1/users/U1",
379 headers_json
, None, 200, None, json
)
382 expected_projects
= ["P1", "Padmin"]
383 if u1
["projects"] != expected_projects
:
384 raise TestException("User content projects '{}' different than expected '{}'. Edition has not done"
385 " properly".format(u1
["projects"], expected_projects
))
387 engine
.test("PU8", "Edit user U1, set Padmin as default project", "PUT", "/admin/v1/users/U1", headers_json
,
388 {"projects": {"$'Padmin'": None, "$+[0]": "Padmin"}}, 204, None, None)
389 res
= engine
.test("PU1", "Check user U1, contains the right projects", "GET", "/admin/v1/users/U1",
390 headers_json
, None, 200, None, json
)
393 expected_projects
= ["Padmin", "P1"]
394 if u1
["projects"] != expected_projects
:
395 raise TestException("User content projects '{}' different than expected '{}'. Edition has not done"
396 " properly".format(u1
["projects"], expected_projects
))
398 engine
.test("PU9", "Edit user U1, change password", "PATCH", "/admin/v1/users/U1", headers_json
,
399 {"password": "pw1_new"}, 204, None, None)
401 engine
.test("PU10", "Change to project P1 non existing", "POST", "/admin/v1/tokens/", headers_json
,
402 {"project_id": "P1"}, 401, r_header_json
, "json")
404 res
= engine
.test("PU1", "Change to user U1 project P1", "POST", "/admin/v1/tokens", headers_json
,
405 {"username": "U1", "password": "pw1_new", "project_id": "P1"}, (200, 201),
406 r_header_json
, "json")
407 response
= res
.json()
408 engine
.set_header({"Authorization": "Bearer {}".format(response
["id"])})
410 engine
.test("PU11", "Edit user projects non admin", "PUT", "/admin/v1/users/U1", headers_json
,
411 {"projects": {"$'P1'": None}}, 401, r_header_json
, "json")
412 engine
.test("PU12", "Add new project non admin", "POST", "/admin/v1/projects", headers_json
,
413 {"name": "P2"}, 401, r_header_json
, "json")
414 engine
.test("PU13", "Add new user non admin", "POST", "/admin/v1/users", headers_json
,
415 {"username": "U3", "projects": ["P1"], "password": "pw3"}, 401,
416 r_header_json
, "json")
418 res
= engine
.test("PU14", "Change to user U1 project Padmin", "POST", "/admin/v1/tokens", headers_json
,
419 {"project_id": "Padmin"}, (200, 201), r_header_json
, "json")
420 response
= res
.json()
421 engine
.set_header({"Authorization": "Bearer {}".format(response
["id"])})
423 engine
.test("PU15", "Add new project admin", "POST", "/admin/v1/projects", headers_json
, {"name": "P2"},
424 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
425 engine
.test("PU16", "Add new user U3 admin", "POST", "/admin/v1/users",
426 headers_json
, {"username": "U3", "projects": ["P2"], "password": "pw3"}, (201, 204),
427 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
428 engine
.test("PU17", "Edit user projects admin", "PUT", "/admin/v1/users/U3", headers_json
,
429 {"projects": ["P2"]}, 204, None, None)
431 engine
.test("PU18", "Delete project P2 conflict", "DELETE", "/admin/v1/projects/P2", headers_json
, None, 409,
432 r_header_json
, "json")
433 engine
.test("PU19", "Delete project P2 forcing", "DELETE", "/admin/v1/projects/P2?FORCE=True", headers_json
,
434 None, 204, None, None)
436 engine
.test("PU20", "Delete user U1. Conflict deleting own user", "DELETE", "/admin/v1/users/U1", headers_json
,
437 None, 409, r_header_json
, "json")
438 engine
.test("PU21", "Delete user U2", "DELETE", "/admin/v1/users/U2", headers_json
, None, 204, None, None)
439 engine
.test("PU22", "Delete user U3", "DELETE", "/admin/v1/users/U3", headers_json
, None, 204, None, None)
441 engine
.remove_authorization() # To force get authorization
442 engine
.get_autorization()
443 engine
.test("PU23", "Delete user U1", "DELETE", "/admin/v1/users/U1", headers_json
, None, 204, None, None)
444 engine
.test("PU24", "Delete project P1", "DELETE", "/admin/v1/projects/P1", headers_json
, None, 204, None, None)
445 engine
.test("PU25", "Delete project Padmin", "DELETE", "/admin/v1/projects/Padmin", headers_json
, None, 204,
450 description
= "Creates/edit/delete fake VIMs and SDN controllers"
454 "schema_version": "1.0",
455 "schema_type": "No idea",
457 "description": "Descriptor name",
458 "vim_type": "openstack",
459 "vim_url": "http://localhost:/vim",
460 "vim_tenant_name": "vimTenant",
462 "vim_password": "password",
463 "config": {"config_param": 1}
467 "description": "sdn-description",
468 "dpid": "50:50:52:54:00:94:21:21",
469 "ip": "192.168.15.17",
471 "type": "opendaylight",
476 self
.port_mapping
= [
477 {"compute_node": "compute node 1",
478 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/1", "switch_mac": "52:54:00:94:21:21"},
479 {"pci": "0000:81:00.1", "switch_port": "port-2/2", "switch_mac": "52:54:00:94:21:22"}
481 {"compute_node": "compute node 2",
482 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/3", "switch_mac": "52:54:00:94:21:23"},
483 {"pci": "0000:81:00.1", "switch_port": "port-2/4", "switch_mac": "52:54:00:94:21:24"}
487 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
489 vim_bad
= self
.vim
.copy()
492 engine
.get_autorization()
493 engine
.test("FVIM1", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
, (201, 204),
494 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json")
495 engine
.test("FVIM2", "Create VIM without name, bad schema", "POST", "/admin/v1/vim_accounts", headers_json
,
496 vim_bad
, 422, None, headers_json
)
497 engine
.test("FVIM3", "Create VIM name repeated", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
,
498 409, None, headers_json
)
499 engine
.test("FVIM4", "Show VIMs", "GET", "/admin/v1/vim_accounts", headers_yaml
, None, 200, r_header_yaml
,
501 engine
.test("FVIM5", "Show VIM", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
, None, 200,
502 r_header_yaml
, "yaml")
505 engine
.test("FVIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/<FVIM1>?FORCE=True", headers_yaml
,
507 engine
.test("FVIM7", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
, None,
508 404, r_header_yaml
, "yaml")
510 # delete and wait until is really deleted
511 engine
.test("FVIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
, None, 202,
515 r
= engine
.test("FVIM7", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<FVIM1>", headers_yaml
,
516 None, None, r_header_yaml
, "yaml")
517 if r
.status_code
== 404:
519 elif r
.status_code
== 200:
523 raise TestException("Vim created at 'FVIM1' is not delete after {} seconds".format(timeout
))
526 class TestVIMSDN(TestFakeVim
):
527 description
= "Creates VIM with SDN editing SDN controllers and port_mapping"
530 TestFakeVim
.__init
__(self
)
532 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
533 engine
.get_autorization()
535 engine
.test("VIMSDN1", "Create SDN", "POST", "/admin/v1/sdns", headers_json
, self
.sdn
, (201, 204),
536 {"Location": "/admin/v1/sdns/", "Content-Type": "application/json"}, "json")
539 engine
.test("VIMSDN2", "Edit SDN", "PATCH", "/admin/v1/sdns/<VIMSDN1>", headers_json
, {"name": "new_sdn_name"},
543 self
.vim
["config"]["sdn-controller"] = engine
.test_ids
["VIMSDN1"]
544 self
.vim
["config"]["sdn-port-mapping"] = self
.port_mapping
545 engine
.test("VIMSDN3", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
, (200, 204, 201),
546 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json"),
548 self
.port_mapping
[0]["compute_node"] = "compute node XX"
549 engine
.test("VIMSDN4", "Edit VIM change port-mapping", "PUT", "/admin/v1/vim_accounts/<VIMSDN3>", headers_json
,
550 {"config": {"sdn-port-mapping": self
.port_mapping
}}, 204, None, None)
551 engine
.test("VIMSDN5", "Edit VIM remove port-mapping", "PUT", "/admin/v1/vim_accounts/<VIMSDN3>", headers_json
,
552 {"config": {"sdn-port-mapping": None}}, 204, None, None)
556 engine
.test("VIMSDN6", "Delete VIM remove port-mapping", "DELETE",
557 "/admin/v1/vim_accounts/<VIMSDN3>?FORCE=True", headers_json
, None, 202, None, 0)
558 engine
.test("VIMSDN7", "Delete SDNC", "DELETE", "/admin/v1/sdns/<VIMSDN1>?FORCE=True", headers_json
, None,
561 engine
.test("VIMSDN8", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<VIMSDN3>", headers_yaml
,
562 None, 404, r_header_yaml
, "yaml")
563 engine
.test("VIMSDN9", "Check SDN is deleted", "GET", "/admin/v1/sdns/<VIMSDN1>", headers_yaml
, None,
564 404, r_header_yaml
, "yaml")
566 # delete and wait until is really deleted
567 engine
.test("VIMSDN6", "Delete VIM remove port-mapping", "DELETE", "/admin/v1/vim_accounts/<VIMSDN3>",
568 headers_json
, None, (202, 201, 204), None, 0)
569 engine
.test("VIMSDN7", "Delete SDN", "DELETE", "/admin/v1/sdns/<VIMSDN1>", headers_json
, None,
570 (202, 201, 204), None, 0)
573 r
= engine
.test("VIMSDN8", "Check VIM is deleted", "GET", "/admin/v1/vim_accounts/<VIMSDN3>",
574 headers_yaml
, None, None, r_header_yaml
, "yaml")
575 if r
.status_code
== 404:
577 elif r
.status_code
== 200:
581 raise TestException("Vim created at 'VIMSDN3' is not delete after {} seconds".format(timeout
))
583 r
= engine
.test("VIMSDN9", "Check SDNC is deleted", "GET", "/admin/v1/sdns/<VIMSDN1>",
584 headers_yaml
, None, None, r_header_yaml
, "yaml")
585 if r
.status_code
== 404:
587 elif r
.status_code
== 200:
591 raise TestException("SDNC created at 'VIMSDN1' is not delete after {} seconds".format(timeout
))
595 description
= "Base class for downloading descriptors from ETSI, onboard and deploy in real VIM"
606 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
607 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
608 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
609 self
.descriptor_edit
= None
610 self
.uses_configuration
= False
616 self
.passed_tests
= 0
620 def create_descriptors(self
, engine
):
621 temp_dir
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/temp/"
622 if not os
.path
.exists(temp_dir
):
623 os
.makedirs(temp_dir
)
624 for vnfd_index
, vnfd_filename
in enumerate(self
.vnfd_filenames
):
625 if "/" in vnfd_filename
:
626 vnfd_filename_path
= vnfd_filename
627 if not os
.path
.exists(vnfd_filename_path
):
628 raise TestException("File '{}' does not exist".format(vnfd_filename_path
))
630 vnfd_filename_path
= temp_dir
+ vnfd_filename
631 if not os
.path
.exists(vnfd_filename_path
):
632 with
open(vnfd_filename_path
, "wb") as file:
633 response
= requests
.get(self
.descriptor_url
+ vnfd_filename
)
634 if response
.status_code
>= 300:
635 raise TestException("Error downloading descriptor from '{}': {}".format(
636 self
.descriptor_url
+ vnfd_filename
, response
.status_code
))
637 file.write(response
.content
)
638 if vnfd_filename_path
.endswith(".yaml"):
639 headers
= headers_yaml
641 headers
= headers_zip_yaml
642 if self
.step
% 2 == 0:
643 # vnfd CREATE AND UPLOAD in one step:
644 test_name
= "DEPLOY{}".format(self
.step
)
645 engine
.test(test_name
, "Onboard VNFD in one step", "POST",
646 "/vnfpkgm/v1/vnf_packages_content" + self
.qforce
, headers
, "@b" + vnfd_filename_path
, 201,
647 {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"},
649 self
.vnfds_test
.append(test_name
)
650 self
.vnfds_id
.append(engine
.test_ids
["last_id"])
653 # vnfd CREATE AND UPLOAD ZIP
654 test_name
= "DEPLOY{}".format(self
.step
)
655 engine
.test(test_name
, "Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages",
656 headers_json
, None, 201,
657 {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
658 self
.vnfds_test
.append(test_name
)
659 self
.vnfds_id
.append(engine
.test_ids
["last_id"])
661 # location = r.headers["Location"]
662 # vnfd_id = location[location.rfind("/")+1:]
663 engine
.test("DEPLOY{}".format(self
.step
), "Onboard VNFD step 2 as ZIP", "PUT",
664 "/vnfpkgm/v1/vnf_packages/<>/package_content" + self
.qforce
,
665 headers
, "@b" + vnfd_filename_path
, 204, None, 0)
668 if self
.descriptor_edit
:
669 if "vnfd{}".format(vnfd_index
) in self
.descriptor_edit
:
671 engine
.test("DEPLOY{}".format(self
.step
), "Edit VNFD ", "PATCH",
672 "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfds_id
[-1]),
673 headers_yaml
, self
.descriptor_edit
["vnfd{}".format(vnfd_index
)], 204, None, None)
676 if "/" in self
.nsd_filename
:
677 nsd_filename_path
= self
.nsd_filename
678 if not os
.path
.exists(nsd_filename_path
):
679 raise TestException("File '{}' does not exist".format(nsd_filename_path
))
681 nsd_filename_path
= temp_dir
+ self
.nsd_filename
682 if not os
.path
.exists(nsd_filename_path
):
683 with
open(nsd_filename_path
, "wb") as file:
684 response
= requests
.get(self
.descriptor_url
+ self
.nsd_filename
)
685 if response
.status_code
>= 300:
686 raise TestException("Error downloading descriptor from '{}': {}".format(
687 self
.descriptor_url
+ self
.nsd_filename
, response
.status_code
))
688 file.write(response
.content
)
689 if nsd_filename_path
.endswith(".yaml"):
690 headers
= headers_yaml
692 headers
= headers_zip_yaml
694 self
.nsd_test
= "DEPLOY" + str(self
.step
)
695 if self
.step
% 2 == 0:
696 # nsd CREATE AND UPLOAD in one step:
697 engine
.test("DEPLOY{}".format(self
.step
), "Onboard NSD in one step", "POST",
698 "/nsd/v1/ns_descriptors_content" + self
.qforce
, headers
, "@b" + nsd_filename_path
, 201,
699 {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}, yaml
)
701 self
.nsd_id
= engine
.test_ids
["last_id"]
703 # nsd CREATE AND UPLOAD ZIP
704 engine
.test("DEPLOY{}".format(self
.step
), "Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors",
705 headers_json
, None, 201,
706 {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
708 self
.nsd_id
= engine
.test_ids
["last_id"]
709 # location = r.headers["Location"]
710 # vnfd_id = location[location.rfind("/")+1:]
711 engine
.test("DEPLOY{}".format(self
.step
), "Onboard NSD step 2 as ZIP", "PUT",
712 "/nsd/v1/ns_descriptors/<>/nsd_content" + self
.qforce
,
713 headers
, "@b" + nsd_filename_path
, 204, None, 0)
716 if self
.descriptor_edit
and "nsd" in self
.descriptor_edit
:
718 engine
.test("DEPLOY{}".format(self
.step
), "Edit NSD ", "PATCH",
719 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
),
720 headers_yaml
, self
.descriptor_edit
["nsd"], 204, None, None)
723 def delete_descriptors(self
, engine
):
725 engine
.test("DEPLOY{}".format(self
.step
), "Delete NSSD SOL005", "DELETE",
726 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
),
727 headers_yaml
, None, 204, None, 0)
729 for vnfd_id
in self
.vnfds_id
:
730 engine
.test("DEPLOY{}".format(self
.step
), "Delete VNFD SOL005", "DELETE",
731 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id
), headers_yaml
, None, 204, None, 0)
734 def instantiate(self
, engine
, ns_data
):
735 ns_data_text
= yaml
.safe_dump(ns_data
, default_flow_style
=True, width
=256)
736 # create NS Two steps
737 r
= engine
.test("DEPLOY{}".format(self
.step
), "Create NS step 1", "POST", "/nslcm/v1/ns_instances",
738 headers_yaml
, ns_data_text
, 201,
739 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"}, "yaml")
740 self
.ns_test
= "DEPLOY{}".format(self
.step
)
741 self
.ns_id
= engine
.test_ids
["last_id"]
743 r
= engine
.test("DEPLOY{}".format(self
.step
), "Instantiate NS step 2", "POST",
744 "/nslcm/v1/ns_instances/<{}>/instantiate".format(self
.ns_test
), headers_yaml
, ns_data_text
,
745 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
746 nslcmop_test
= "DEPLOY{}".format(self
.step
)
750 # Wait until status is Ok
751 wait
= timeout_configure
if self
.uses_configuration
else timeout_deploy
753 r
= engine
.test("DEPLOY{}".format(self
.step
), "Wait until NS is deployed and configured", "GET",
754 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test
), headers_json
, None,
755 200, r_header_json
, "json")
757 if "COMPLETED" in nslcmop
["operationState"]:
759 elif "FAILED" in nslcmop
["operationState"]:
760 raise TestException("NS instantiate has failed: {}".format(nslcmop
["detailed-status"]))
764 raise TestException("NS instantiate is not done after {} seconds".format(timeout_deploy
))
767 def _wait_nslcmop_ready(self
, engine
, nslcmop_test
, timeout_deploy
, expected_fail
=False):
770 r
= engine
.test("DEPLOY{}".format(self
.step
), "Wait to ns lcm operation complete", "GET",
771 "/nslcm/v1/ns_lcm_op_occs/<{}>".format(nslcmop_test
), headers_json
, None,
772 200, r_header_json
, "json")
774 if "COMPLETED" in nslcmop
["operationState"]:
776 raise TestException("NS terminate has success, expecting failing: {}".format(
777 nslcmop
["detailed-status"]))
779 elif "FAILED" in nslcmop
["operationState"]:
780 if not expected_fail
:
781 raise TestException("NS terminate has failed: {}".format(nslcmop
["detailed-status"]))
786 raise TestException("NS instantiate is not terminate after {} seconds".format(timeout
))
788 def terminate(self
, engine
):
791 r
= engine
.test("DEPLOY{}".format(self
.step
), "Terminate NS", "POST",
792 "/nslcm/v1/ns_instances/<{}>/terminate".format(self
.ns_test
), headers_yaml
, None,
793 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
794 nslcmop2_test
= "DEPLOY{}".format(self
.step
)
796 # Wait until status is Ok
797 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_test
, timeout_deploy
)
799 r
= engine
.test("DEPLOY{}".format(self
.step
), "Delete NS", "DELETE",
800 "/nslcm/v1/ns_instances/<{}>".format(self
.ns_test
), headers_yaml
, None,
804 r
= engine
.test("DEPLOY{}".format(self
.step
), "Delete NS with FORCE", "DELETE",
805 "/nslcm/v1/ns_instances/<{}>?FORCE=True".format(self
.ns_test
), headers_yaml
, None,
809 # check all it is deleted
810 r
= engine
.test("DEPLOY{}".format(self
.step
), "Check NS is deleted", "GET",
811 "/nslcm/v1/ns_instances/<{}>".format(self
.ns_test
), headers_yaml
, None,
814 r
= engine
.test("DEPLOY{}".format(self
.step
), "Check NSLCMOPs are deleted", "GET",
815 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId=<{}>".format(self
.ns_test
), headers_json
, None,
818 if not isinstance(nslcmops
, list) or nslcmops
:
819 raise TestException("NS {} deleted but with ns_lcm_op_occ active: {}".format(self
.ns_test
, nslcmops
))
821 def test_ns(self
, engine
, test_osm
, commands
=None, users
=None, passwds
=None, keys
=None, timeout
=0):
824 r
= engine
.test("TEST_NS{}".format(n
), "GET VNFR_IDs", "GET",
825 "/nslcm/v1/ns_instances/{}".format(self
.ns_id
), headers_json
, None,
826 200, r_header_json
, "json")
830 vnfr_list
= ns_data
['constituent-vnfr-ref']
833 for vnfr_id
in vnfr_list
:
834 self
.total_tests
+= 1
835 r
= engine
.test("TEST_NS{}".format(n
), "GET IP_ADDRESS OF VNFR", "GET",
836 "/nslcm/v1/vnfrs/{}".format(vnfr_id
), headers_json
, None,
837 200, r_header_json
, "json")
841 if vnfr_data
.get("ip-address"):
842 name
= "TEST_NS{}".format(n
)
843 description
= "Run tests in VNFR with IP {}".format(vnfr_data
['ip-address'])
845 test_description
= "Test {} {}".format(name
, description
)
846 logger
.warning(test_description
)
847 vnf_index
= str(vnfr_data
["member-vnf-index-ref"])
848 while timeout
>= time
:
849 result
, message
= self
.do_checks([vnfr_data
["ip-address"]],
850 vnf_index
=vnfr_data
["member-vnf-index-ref"],
851 commands
=commands
.get(vnf_index
), user
=users
.get(vnf_index
),
852 passwd
=passwds
.get(vnf_index
), key
=keys
.get(vnf_index
))
854 logger
.warning(message
)
860 logger
.critical(message
)
864 logger
.critical(message
)
866 logger
.critical("VNFR {} has not mgmt address. Check failed".format(vnfr_id
))
868 def do_checks(self
, ip
, vnf_index
, commands
=[], user
=None, passwd
=None, key
=None):
871 from pssh
.clients
import ParallelSSHClient
872 from pssh
.utils
import load_private_key
873 from ssh2
import exceptions
as ssh2Exception
874 except ImportError as e
:
875 logger
.critical("Package <pssh> or/and <urllib3> is not installed. Please add them with 'pip3 install "
876 "parallel-ssh urllib3': {}".format(e
))
877 urllib3
.disable_warnings(urllib3
.exceptions
.InsecureRequestWarning
)
879 p_host
= os
.environ
.get("PROXY_HOST")
880 p_user
= os
.environ
.get("PROXY_USER")
881 p_password
= os
.environ
.get("PROXY_PASSWD")
884 pkey
= load_private_key(key
)
888 client
= ParallelSSHClient(ip
, user
=user
, password
=passwd
, pkey
=pkey
, proxy_host
=p_host
,
889 proxy_user
=p_user
, proxy_password
=p_password
, timeout
=10, num_retries
=0)
891 output
= client
.run_command(cmd
)
893 if output
[ip
[0]].exit_code
:
894 return -1, " VNFR {} could not be checked: {}".format(ip
[0], output
[ip
[0]].stderr
)
896 self
.passed_tests
+= 1
897 return 1, " Test successful"
898 except (ssh2Exception
.ChannelFailure
, ssh2Exception
.SocketDisconnectError
, ssh2Exception
.SocketTimeout
,
899 ssh2Exception
.SocketRecvError
) as e
:
900 return 0, "Timeout accessing the VNFR {}: {}".format(ip
[0], str(e
))
901 except Exception as e
:
902 return -1, "ERROR checking the VNFR {}: {}".format(ip
[0], str(e
))
904 def aditional_operations(self
, engine
, test_osm
, manual_check
):
907 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
908 engine
.get_autorization()
909 nsname
= os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
911 if "vnfd-files" in test_params
:
912 self
.vnfd_filenames
= test_params
["vnfd-files"].split(",")
913 if "nsd-file" in test_params
:
914 self
.nsd_filename
= test_params
["nsd-file"]
915 if test_params
.get("ns-name"):
916 nsname
= test_params
["ns-name"]
917 self
.create_descriptors(engine
)
919 # create real VIM if not exist
920 self
.vim_id
= engine
.get_create_vim(test_osm
)
921 ns_data
= {"nsDescription": "default description", "nsName": nsname
, "nsdId": self
.nsd_id
,
922 "vimAccountId": self
.vim_id
}
923 if test_params
and test_params
.get("ns-config"):
924 if isinstance(test_params
["ns-config"], str):
925 ns_data
.update(yaml
.load(test_params
["ns-config"]))
927 ns_data
.update(test_params
["ns-config"])
928 self
.instantiate(engine
, ns_data
)
931 input('NS has been deployed. Perform manual check and press enter to resume')
933 self
.test_ns(engine
, test_osm
, self
.cmds
, self
.uss
, self
.pss
, self
.keys
, self
.timeout
)
934 self
.aditional_operations(engine
, test_osm
, manual_check
)
935 self
.terminate(engine
)
936 self
.delete_descriptors(engine
)
939 def print_results(self
):
940 print("\n\n\n--------------------------------------------")
941 print("TEST RESULTS:\n PASSED TESTS: {} - TOTAL TESTS: {}".format(self
.total_tests
, self
.passed_tests
))
942 print("--------------------------------------------")
945 class TestDeployHackfestCirros(TestDeploy
):
946 description
= "Load and deploy Hackfest cirros_2vnf_ns example"
950 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
951 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
952 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
953 self
.uss
= {'1': "cirros", '2': "cirros"}
954 self
.pss
= {'1': "cubswin:)", '2': "cubswin:)"}
957 class TestDeployHackfest1(TestDeploy
):
958 description
= "Load and deploy Hackfest_1_vnfd example"
962 self
.vnfd_filenames
= ("hackfest_1_vnfd.tar.gz",)
963 self
.nsd_filename
= "hackfest_1_nsd.tar.gz"
964 # self.cmds = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
965 # self.uss = {'1': "cirros", '2': "cirros"}
966 # self.pss = {'1': "cubswin:)", '2': "cubswin:)"}
969 class TestDeployHackfestCirrosScaling(TestDeploy
):
970 description
= "Load and deploy Hackfest cirros_2vnf_ns example with scaling modifications"
974 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
975 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
977 def create_descriptors(self
, engine
):
978 super().create_descriptors(engine
)
979 # Modify VNFD to add scaling and count=2
982 "$id: 'cirros_vnfd-VM'":
984 scaling-group-descriptor:
985 - name: "scale_cirros"
986 max-instance-count: 2
988 - vdu-id-ref: cirros_vnfd-VM
991 engine
.test("DEPLOY{}".format(self
.step
), "Edit VNFD ", "PATCH",
992 "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfds_id
[0]),
993 headers_yaml
, payload
, 204, None, None)
996 def aditional_operations(self
, engine
, test_osm
, manual_check
):
999 # 2 perform scale out twice
1000 payload
= '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, 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 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_out
= "DEPLOY{}".format(self
.step
)
1007 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_scale_out
, timeout_deploy
)
1009 input('NS scale out done. Check that two more vdus are there')
1010 # TODO check automatic
1012 # 2 perform scale in
1013 payload
= '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1014 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
1015 for i
in range(0, 2):
1016 engine
.test("DEPLOY{}".format(self
.step
), "Execute scale IN action over NS", "POST",
1017 "/nslcm/v1/ns_instances/<{}>/scale".format(self
.ns_test
), headers_yaml
, payload
,
1018 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1019 nslcmop2_scale_in
= "DEPLOY{}".format(self
.step
)
1020 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_scale_in
, timeout_deploy
)
1022 input('NS scale in done. Check that two less vdus are there')
1023 # TODO check automatic
1025 # perform scale in that must fail as reached limit
1026 engine
.test("DEPLOY{}".format(self
.step
), "Execute scale IN out of limit action over NS", "POST",
1027 "/nslcm/v1/ns_instances/<{}>/scale".format(self
.ns_test
), headers_yaml
, payload
,
1028 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1029 nslcmop2_scale_in
= "DEPLOY{}".format(self
.step
)
1030 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_scale_in
, timeout_deploy
, expected_fail
=True)
1033 class TestDeployIpMac(TestDeploy
):
1034 description
= "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
1038 self
.vnfd_filenames
= ("vnfd_2vdu_set_ip_mac2.yaml", "vnfd_2vdu_set_ip_mac.yaml")
1039 self
.nsd_filename
= "scenario_2vdu_set_ip_mac.yaml"
1040 self
.descriptor_url
= \
1041 "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
1042 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1043 self
.uss
= {'1': "osm", '2': "osm"}
1044 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1047 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1048 # super().run(engine, test_osm, manual_check, test_params)
1049 # run again setting IPs with instantiate parameters
1050 instantiation_params
= {
1053 "member-vnf-index": "1",
1056 "name": "internal_vld1", # net_internal
1058 "ip-version": "ipv4",
1059 "subnet-address": "10.9.8.0/24",
1060 "dhcp-params": {"count": 100, "start-address": "10.9.8.100"}
1062 "internal-connection-point": [
1065 "ip-address": "10.9.8.2",
1069 "ip-address": "10.9.8.3",
1080 # "name": "iface11",
1081 # "floating-ip-required": True,
1085 "mac-address": "52:33:44:55:66:13"
1094 "ip-address": "10.31.31.22",
1095 "mac-address": "52:33:44:55:66:21"
1104 super().run(engine
, test_osm
, manual_check
, test_params
={"ns-config": instantiation_params
})
1107 class TestDeployHackfest4(TestDeploy
):
1108 description
= "Load and deploy Hackfest 4 example."
1112 self
.vnfd_filenames
= ("hackfest_4_vnfd.tar.gz",)
1113 self
.nsd_filename
= "hackfest_4_nsd.tar.gz"
1114 self
.uses_configuration
= True
1115 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1116 self
.uss
= {'1': "ubuntu", '2': "ubuntu"}
1117 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1119 def create_descriptors(self
, engine
):
1120 super().create_descriptors(engine
)
1121 # Modify VNFD to add scaling
1123 scaling-group-descriptor:
1124 - name: "scale_dataVM"
1125 max-instance-count: 10
1127 - name: "auto_cpu_util_above_threshold"
1128 scaling-type: "automatic"
1132 - name: "cpu_util_above_threshold"
1133 scale-in-threshold: 15
1134 scale-in-relational-operation: "LE"
1135 scale-out-threshold: 60
1136 scale-out-relational-operation: "GE"
1137 vnf-monitoring-param-ref: "all_aaa_cpu_util"
1139 - vdu-id-ref: dataVM
1141 scaling-config-action:
1142 - trigger: post-scale-out
1143 vnf-config-primitive-name-ref: touch
1144 - trigger: pre-scale-in
1145 vnf-config-primitive-name-ref: touch
1152 default-value: '/home/ubuntu/touched'
1154 engine
.test("DEPLOY{}".format(self
.step
), "Edit VNFD ", "PATCH",
1155 "/vnfpkgm/v1/vnf_packages/<{}>".format(self
.vnfds_test
[0]), headers_yaml
, payload
, 204, None, None)
1159 class TestDeployHackfest3Charmed(TestDeploy
):
1160 description
= "Load and deploy Hackfest 3charmed_ns example. Modifies it for adding scaling and performs " \
1161 "primitive actions and scaling"
1165 self
.vnfd_filenames
= ("hackfest_3charmed_vnfd.tar.gz",)
1166 self
.nsd_filename
= "hackfest_3charmed_nsd.tar.gz"
1167 self
.uses_configuration
= True
1168 self
.cmds
= {'1': [''], '2': ['ls -lrt /home/ubuntu/first-touch', ]}
1169 self
.uss
= {'1': "ubuntu", '2': "ubuntu"}
1170 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1172 # def create_descriptors(self, engine):
1173 # super().create_descriptors(engine)
1174 # # Modify VNFD to add scaling
1176 # scaling-group-descriptor:
1177 # - name: "scale_dataVM"
1178 # max-instance-count: 10
1180 # - name: "auto_cpu_util_above_threshold"
1181 # scaling-type: "automatic"
1185 # - name: "cpu_util_above_threshold"
1186 # scale-in-threshold: 15
1187 # scale-in-relational-operation: "LE"
1188 # scale-out-threshold: 60
1189 # scale-out-relational-operation: "GE"
1190 # vnf-monitoring-param-ref: "all_aaa_cpu_util"
1192 # - vdu-id-ref: dataVM
1194 # scaling-config-action:
1195 # - trigger: post-scale-out
1196 # vnf-config-primitive-name-ref: touch
1197 # - trigger: pre-scale-in
1198 # vnf-config-primitive-name-ref: touch
1199 # vnf-configuration:
1205 # default-value: '/home/ubuntu/touched'
1207 # engine.test("DEPLOY{}".format(self.step), "Edit VNFD ", "PATCH",
1208 # "/vnfpkgm/v1/vnf_packages/<{}>".format(self.vnfds_test[0]),
1209 # headers_yaml, payload, 200,
1210 # r_header_yaml, yaml)
1211 # self.vnfds_test.append("DEPLOY" + str(self.step))
1214 def aditional_operations(self
, engine
, test_osm
, manual_check
):
1218 payload
= '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
1219 engine
.test("DEPLOY{}".format(self
.step
), "Executer service primitive over NS", "POST",
1220 "/nslcm/v1/ns_instances/<{}>/action".format(self
.ns_test
), headers_yaml
, payload
,
1221 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1222 nslcmop2_action
= "DEPLOY{}".format(self
.step
)
1224 # Wait until status is Ok
1225 self
._wait
_nslcmop
_ready
(engine
, nslcmop2_action
, timeout_deploy
)
1227 input('NS service primitive has been executed. Check that file /home/ubuntu/OSMTESTNBI is present at '
1230 cmds
= {'1': [''], '2': ['ls -lrt /home/ubuntu/OSMTESTNBI', ]}
1231 uss
= {'1': "ubuntu", '2': "ubuntu"}
1232 pss
= {'1': "osm4u", '2': "osm4u"}
1233 self
.test_ns(engine
, test_osm
, cmds
, uss
, pss
, self
.keys
, self
.timeout
)
1235 # # 2 perform scale out
1236 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1237 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1238 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1239 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1240 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1241 # nslcmop2_scale_out = "DEPLOY{}".format(self.step)
1242 # self._wait_nslcmop_ready(engine, nslcmop2_scale_out, timeout_deploy)
1244 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
1245 # # TODO check automatic
1247 # # 2 perform scale in
1248 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1249 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1250 # engine.test("DEPLOY{}".format(self.step), "Execute scale action over NS", "POST",
1251 # "/nslcm/v1/ns_instances/<{}>/scale".format(self.ns_test), headers_yaml, payload,
1252 # 201, {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}, "yaml")
1253 # nslcmop2_scale_in = "DEPLOY{}".format(self.step)
1254 # self._wait_nslcmop_ready(engine, nslcmop2_scale_in, timeout_deploy)
1256 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
1257 # # TODO check automatic
1260 class TestDeploySingleVdu(TestDeployHackfest3Charmed
):
1261 description
= "Generate a single VDU base on editing Hackfest3Charmed descriptors and deploy"
1265 self
.qforce
= "?FORCE=True"
1266 self
.descriptor_edit
= {
1267 # Modify VNFD to remove one VDU
1271 "interface": {"$[0]": {"external-connection-point-ref": "pdu-mgmt"}}
1275 "vnf-configuration": None,
1276 "connection-point": {
1280 "short-name": "pdu-mgmt"
1284 "mgmt-interface": {"cp": "pdu-mgmt"},
1285 "description": "A vnf single vdu to be used as PDU",
1289 "id": "pdu_internal",
1290 "name": "pdu_internal",
1291 "internal-connection-point": {"$[1]": None},
1292 "short-name": "pdu_internal",
1298 # Modify NSD accordingly
1300 "constituent-vnfd": {
1301 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
1304 "description": "A nsd to deploy the vnf to act as as PDU",
1306 "name": "nsd-as-pdu",
1307 "short-name": "nsd-as-pdu",
1312 "short-name": "mgmt_pdu",
1313 "vnfd-connection-point-ref": {
1315 "vnfd-connection-point-ref": "pdu-mgmt",
1316 "vnfd-id-ref": "vdu-as-pdu",
1328 class TestDeployHnfd(TestDeployHackfest3Charmed
):
1329 description
= "Generate a HNFD base on editing Hackfest3Charmed descriptors and deploy"
1333 self
.pduDeploy
= TestDeploySingleVdu()
1334 self
.pdu_interface_0
= {}
1335 self
.pdu_interface_1
= {}
1338 # self.vnf_to_pdu = """
1341 # pdu-type: PDU-TYPE-1
1346 # name: pdu-iface-internal
1348 # description: HFND, one PDU + One VDU
1354 self
.pdu_descriptor
= {
1356 "type": "PDU-TYPE-1",
1357 "vim_accounts": "to-override",
1360 "name": "mgmt-iface",
1363 "ip-address": "to override",
1364 "mac-address": "mac_address",
1365 "vim-network-name": "mgmt",
1368 "name": "pdu-iface-internal",
1371 "ip-address": "to override",
1372 "mac-address": "mac_address",
1373 "vim-network-name": "pdu_internal", # OSMNBITEST-PDU-pdu_internal
1377 self
.vnfd_filenames
= ("hackfest_3charmed_vnfd.tar.gz", "hackfest_3charmed_vnfd.tar.gz")
1379 self
.descriptor_edit
= {
1383 "short-name": "hfn1",
1386 "pdu-type": "PDU-TYPE-1",
1388 "$[0]": {"name": "mgmt-iface"},
1389 "$[1]": {"name": "pdu-iface-internal"},
1395 "constituent-vnfd": {
1396 "$[1]": {"vnfd-id-ref": "hfnd1"}
1401 def create_descriptors(self
, engine
):
1402 super().create_descriptors(engine
)
1405 self
.pdu_descriptor
["interfaces"][0].update(self
.pdu_interface_0
)
1406 self
.pdu_descriptor
["interfaces"][1].update(self
.pdu_interface_1
)
1407 self
.pdu_descriptor
["vim_accounts"] = [self
.vim_id
]
1408 # TODO get vim-network-name from vnfr.vld.name
1409 self
.pdu_descriptor
["interfaces"][1]["vim-network-name"] = "{}-{}-{}".format(
1410 os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST"),
1411 "PDU", self
.pdu_descriptor
["interfaces"][1]["vim-network-name"])
1412 test_name
= "DEPLOY{}".format(self
.step
)
1413 engine
.test(test_name
, "Onboard PDU descriptor", "POST", "/pdu/v1/pdu_descriptors",
1414 {"Location": "/pdu/v1/pdu_descriptors/", "Content-Type": "application/yaml"}, self
.pdu_descriptor
,
1415 201, r_header_yaml
, "yaml")
1416 self
.pdu_id
= engine
.test_ids
["last_id"]
1419 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1420 engine
.get_autorization()
1421 nsname
= os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
1423 # create real VIM if not exist
1424 self
.vim_id
= engine
.get_create_vim(test_osm
)
1426 self
.pduDeploy
.create_descriptors(engine
)
1427 self
.pduDeploy
.instantiate(engine
, {"nsDescription": "to be used as PDU", "nsName": nsname
+ "-PDU",
1428 "nsdId": self
.pduDeploy
.nsd_id
, "vimAccountId": self
.vim_id
})
1430 input('VNF to be used as PDU has been deployed. Perform manual check and press enter to resume')
1432 self
.pduDeploy
.test_ns(engine
, test_osm
, self
.pduDeploy
.cmds
, self
.pduDeploy
.uss
, self
.pduDeploy
.pss
,
1433 self
.pduDeploy
.keys
, self
.pduDeploy
.timeout
)
1436 r
= engine
.test("DEPLOY{}".format(self
.step
), "GET IP_ADDRESS OF VNFR", "GET",
1437 "/nslcm/v1/vnfrs?nsr-id-ref={}".format(self
.pduDeploy
.ns_id
), headers_json
, None,
1438 200, r_header_json
, "json")
1440 vnfr_data
= r
.json()
1443 self
.pdu_interface_0
["ip-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][0].get("ip-address")
1444 self
.pdu_interface_1
["ip-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][1].get("ip-address")
1445 self
.pdu_interface_0
["mac-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][0].get("mac-address")
1446 self
.pdu_interface_1
["mac-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][1].get("mac-address")
1447 if not self
.pdu_interface_0
["ip-address"]:
1448 raise TestException("Vnfr has not managment ip address")
1450 self
.pdu_interface_0
["ip-address"] = "192.168.10.10"
1451 self
.pdu_interface_1
["ip-address"] = "192.168.11.10"
1452 self
.pdu_interface_0
["mac-address"] = "52:33:44:55:66:13"
1453 self
.pdu_interface_1
["mac-address"] = "52:33:44:55:66:14"
1455 self
.create_descriptors(engine
)
1457 ns_data
= {"nsDescription": "default description", "nsName": nsname
, "nsdId": self
.nsd_id
,
1458 "vimAccountId": self
.vim_id
}
1459 if test_params
and test_params
.get("ns-config"):
1460 if isinstance(test_params
["ns-config"], str):
1461 ns_data
.update(yaml
.load(test_params
["ns-config"]))
1463 ns_data
.update(test_params
["ns-config"])
1465 self
.instantiate(engine
, ns_data
)
1467 input('NS has been deployed. Perform manual check and press enter to resume')
1469 self
.test_ns(engine
, test_osm
, self
.cmds
, self
.uss
, self
.pss
, self
.keys
, self
.timeout
)
1470 self
.aditional_operations(engine
, test_osm
, manual_check
)
1471 self
.terminate(engine
)
1472 self
.pduDeploy
.terminate(engine
)
1473 self
.delete_descriptors(engine
)
1474 self
.pduDeploy
.delete_descriptors(engine
)
1478 self
.print_results()
1480 def delete_descriptors(self
, engine
):
1481 super().delete_descriptors(engine
)
1483 engine
.test("DEPLOY{}".format(self
.step
), "Delete PDU SOL005", "DELETE",
1484 "/pdu/v1/pdu_descriptors/{}".format(self
.pdu_id
),
1485 headers_yaml
, None, 204, None, 0)
1488 class TestDescriptors
:
1489 description
= "Test VNFD, NSD, PDU descriptors CRUD and dependencies"
1493 self
.vnfd_filename
= "hackfest_3charmed_vnfd.tar.gz"
1494 self
.nsd_filename
= "hackfest_3charmed_nsd.tar.gz"
1495 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
1499 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1500 engine
.get_autorization()
1501 temp_dir
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/temp/"
1502 if not os
.path
.exists(temp_dir
):
1503 os
.makedirs(temp_dir
)
1506 for filename
in (self
.vnfd_filename
, self
.nsd_filename
):
1507 filename_path
= temp_dir
+ filename
1508 if not os
.path
.exists(filename_path
):
1509 with
open(filename_path
, "wb") as file:
1510 response
= requests
.get(self
.descriptor_url
+ filename
)
1511 if response
.status_code
>= 300:
1512 raise TestException("Error downloading descriptor from '{}': {}".format(
1513 self
.descriptor_url
+ filename
, response
.status_code
))
1514 file.write(response
.content
)
1516 vnfd_filename_path
= temp_dir
+ self
.vnfd_filename
1517 nsd_filename_path
= temp_dir
+ self
.nsd_filename
1519 # vnfd CREATE AND UPLOAD in one step:
1520 test_name
= "DESCRIPTOR{}".format(self
.step
)
1521 engine
.test(test_name
, "Onboard VNFD in one step", "POST",
1522 "/vnfpkgm/v1/vnf_packages_content", headers_zip_yaml
, "@b" + vnfd_filename_path
, 201,
1523 {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"}, "yaml")
1524 self
.vnfd_id
= engine
.test_ids
["last_id"]
1527 # get vnfd descriptor
1528 engine
.test("DESCRIPTOR" + str(self
.step
), "Get VNFD descriptor", "GET",
1529 "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_id
), headers_yaml
, None, 200, r_header_yaml
, "yaml")
1532 # get vnfd file descriptor
1533 engine
.test("DESCRIPTOR" + str(self
.step
), "Get VNFD file descriptor", "GET",
1534 "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(self
.vnfd_id
), headers_text
, None, 200,
1535 r_header_text
, "text", temp_dir
+"vnfd-yaml")
1537 # TODO compare files: diff vnfd-yaml hackfest_3charmed_vnfd/hackfest_3charmed_vnfd.yaml
1539 # get vnfd zip file package
1540 engine
.test("DESCRIPTOR" + str(self
.step
), "Get VNFD zip package", "GET",
1541 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self
.vnfd_id
), headers_zip
, None, 200,
1542 r_header_zip
, "zip", temp_dir
+"vnfd-zip")
1544 # TODO compare files: diff vnfd-zip hackfest_3charmed_vnfd.tar.gz
1547 engine
.test("DESCRIPTOR" + str(self
.step
), "Get VNFD artifact package", "GET",
1548 "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/osm.png".format(self
.vnfd_id
), headers_zip
, None, 200,
1549 r_header_octect
, "octet-string", temp_dir
+"vnfd-icon")
1551 # TODO compare files: diff vnfd-icon hackfest_3charmed_vnfd/icons/osm.png
1553 # nsd CREATE AND UPLOAD in one step:
1554 test_name
= "DESCRIPTOR{}".format(self
.step
)
1555 engine
.test(test_name
, "Onboard NSD in one step", "POST",
1556 "/nsd/v1/ns_descriptors_content", headers_zip_yaml
, "@b" + nsd_filename_path
, 201,
1557 {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}, "yaml")
1558 self
.nsd_id
= engine
.test_ids
["last_id"]
1561 # get nsd descriptor
1562 engine
.test("DESCRIPTOR" + str(self
.step
), "Get NSD descriptor", "GET",
1563 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
), headers_yaml
, None, 200, r_header_yaml
, "yaml")
1566 # get nsd file descriptor
1567 engine
.test("DESCRIPTOR" + str(self
.step
), "Get NSD file descriptor", "GET",
1568 "/nsd/v1/ns_descriptors/{}/nsd".format(self
.nsd_id
), headers_text
, None, 200,
1569 r_header_text
, "text", temp_dir
+"nsd-yaml")
1571 # TODO compare files: diff nsd-yaml hackfest_3charmed_nsd/hackfest_3charmed_nsd.yaml
1573 # get nsd zip file package
1574 engine
.test("DESCRIPTOR" + str(self
.step
), "Get NSD zip package", "GET",
1575 "/nsd/v1/ns_descriptors/{}/nsd_content".format(self
.nsd_id
), headers_zip
, None, 200,
1576 r_header_zip
, "zip", temp_dir
+"nsd-zip")
1578 # TODO compare files: diff nsd-zip hackfest_3charmed_nsd.tar.gz
1581 engine
.test("DESCRIPTOR" + str(self
.step
), "Get NSD artifact package", "GET",
1582 "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm.png".format(self
.nsd_id
), headers_zip
, None, 200,
1583 r_header_octect
, "octet-string", temp_dir
+"nsd-icon")
1585 # TODO compare files: diff nsd-icon hackfest_3charmed_nsd/icons/osm.png
1588 test_rest
.test("DESCRIPTOR" + str(self
.step
), "Delete VNFD conflict", "DELETE",
1589 "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_id
), headers_yaml
, None, 409, None, None)
1592 test_rest
.test("DESCRIPTOR" + str(self
.step
), "Delete VNFD force", "DELETE",
1593 "/vnfpkgm/v1/vnf_packages/{}?FORCE=TRUE".format(self
.vnfd_id
), headers_yaml
, None, 204, None, 0)
1597 test_rest
.test("DESCRIPTOR" + str(self
.step
), "Delete NSD", "DELETE",
1598 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
), headers_yaml
, None, 204, None, 0)
1602 class TestNetSliceTemplates
:
1603 description
= "Upload a NST to OSM"
1606 self
.nst_filenames
= ("@./cirros_slice/cirros_slice.yaml")
1608 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1610 engine
.get_autorization()
1611 r
= engine
.test("NST1", "Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml
,
1613 201, {"Location": "/nst/v1/netslice_templates_content", "Content-Type": "application/yaml"},
1615 location
= r
.headers
["Location"]
1616 nst_id
= location
[location
.rfind("/")+1:]
1618 # nstd SHOW OSM format
1619 r
= engine
.test("NST2", "Show NSTD OSM format", "GET",
1620 "/nst/v1/netslice_templates/{}".format(nst_id
), headers_json
, None,
1621 200, r_header_json
, "json")
1624 r
= engine
.test("NST3", "Delete NSTD", "DELETE",
1625 "/nst/v1/netslice_templates/{}".format(nst_id
), headers_json
, None,
1629 class TestNetSliceInstances
:
1630 description
= "Upload a NST to OSM"
1633 self
.nst_filenames
= ("@./cirros_slice/cirros_slice.yaml")
1635 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1637 engine
.get_autorization()
1638 r
= engine
.test("NST1", "Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml
,
1639 self
.nst_filenames
, 201,
1640 {"Location": "/nst/v1/netslice_templates_content", "Content-Type": "application/yaml"}, "yaml")
1641 location
= r
.headers
["Location"]
1642 nst_id
= location
[location
.rfind("/")+1:]
1646 self
.vim_id
= engine
.get_create_vim(test_osm
)
1648 r
= engine
.test("VIM1", "Get available VIM", "GET", "/admin/v1/vim_accounts",
1649 headers_json
, None, None, r_header_json
, "json")
1650 r_json
= json
.loads(r
.text
)
1652 self
.vim_id
= vim
["_id"]
1654 ns_data
= {"nsiDescription": "default description", "nsiName": "my_slice", "nstId": nst_id
,
1655 "vimAccountId": self
.vim_id
}
1656 ns_data_text
= yaml
.safe_dump(ns_data
, default_flow_style
=True, width
=256)
1658 r
= engine
.test("NSI1", "Onboard NSI", "POST", "/nsilcm/v1/netslice_instances_content", headers_yaml
,
1660 {"Location": "/nsilcm/v1/netslice_instances_content", "Content-Type":
1661 "application/yaml"}, "yaml")
1662 location
= r
.headers
["Location"]
1663 nsi_id
= location
[location
.rfind("/")+1:]
1665 # TODO: Improve the wait with a polling if NSI was deployed
1670 r
= engine
.test("NSI2", "Wait until NSI is deployed", "GET",
1671 "/nsilcm/v1/netslice_instances_content/{}".format(nsi_id
), headers_json
, None,
1672 200, r_header_json
, "json")
1675 r
= engine
.test("NSI3", "Delete NSI", "DELETE",
1676 "/nsilcm/v1/netslice_instances_content/{}".format(nsi_id
), headers_json
, None,
1677 202, r_header_json
, "json")
1682 r
= engine
.test("NST2", "Delete NSTD", "DELETE",
1683 "/nst/v1/netslice_templates/{}".format(nst_id
), headers_json
, None,
1687 if __name__
== "__main__":
1691 # Disable warnings from self-signed certificates.
1692 requests
.packages
.urllib3
.disable_warnings()
1694 logging
.basicConfig(format
="%(levelname)s %(message)s", level
=logging
.ERROR
)
1695 logger
= logging
.getLogger('NBI')
1696 # load parameters and configuration
1697 opts
, args
= getopt
.getopt(sys
.argv
[1:], "hvu:p:",
1698 ["url=", "user=", "password=", "help", "version", "verbose", "no-verbose",
1699 "project=", "insecure", "timeout", "timeout-deploy", "timeout-configure",
1700 "test=", "list", "test-osm", "manual-check", "params="])
1701 url
= "https://localhost:9999/osm"
1702 user
= password
= project
= "admin"
1704 manual_check
= False
1708 "NonAuthorized": TestNonAuthorized
,
1709 "FakeVIM": TestFakeVim
,
1710 "TestUsersProjects": TestUsersProjects
,
1711 "VIM-SDN": TestVIMSDN
,
1712 "Deploy-Custom": TestDeploy
,
1713 "Deploy-Hackfest-Cirros": TestDeployHackfestCirros
,
1714 "Deploy-Hackfest-Cirros-Scaling": TestDeployHackfestCirrosScaling
,
1715 "Deploy-Hackfest-3Charmed": TestDeployHackfest3Charmed
,
1716 "Deploy-Hackfest-4": TestDeployHackfest4
,
1717 "Deploy-CirrosMacIp": TestDeployIpMac
,
1718 "TestDescriptors": TestDescriptors
,
1719 "TestDeployHackfest1": TestDeployHackfest1
,
1720 # "Deploy-MultiVIM": TestDeployMultiVIM,
1721 "DeploySingleVdu": TestDeploySingleVdu
,
1722 "DeployHnfd": TestDeployHnfd
,
1723 "Upload-Slice-Template": TestNetSliceTemplates
,
1724 "Deploy-Slice-Instance": TestNetSliceInstances
,
1730 # print("parameter:", o, a)
1731 if o
== "--version":
1732 print("test version " + __version__
+ ' ' + version_date
)
1735 for test
, test_class
in test_classes
.items():
1736 print("{:20} {}".format(test
+ ":", test_class
.description
))
1738 elif o
in ("-v", "--verbose"):
1740 elif o
== "no-verbose":
1742 elif o
in ("-h", "--help"):
1745 elif o
== "--test-osm":
1747 elif o
== "--manual-check":
1751 elif o
in ("-u", "--user"):
1753 elif o
in ("-p", "--password"):
1755 elif o
== "--project":
1758 # print("asdfadf", o, a, a.split(","))
1759 for _test
in a
.split(","):
1760 if _test
not in test_classes
:
1761 print("Invalid test name '{}'. Use option '--list' to show available tests".format(_test
),
1764 test_to_do
.append(_test
)
1765 elif o
== "--params":
1766 param_key
, _
, param_value
= a
.partition("=")
1767 text_index
= len(test_to_do
)
1768 if text_index
not in test_params
:
1769 test_params
[text_index
] = {}
1770 test_params
[text_index
][param_key
] = param_value
1771 elif o
== "--insecure":
1773 elif o
== "--timeout":
1775 elif o
== "--timeout-deploy":
1776 timeout_deploy
= int(a
)
1777 elif o
== "--timeout-configure":
1778 timeout_configure
= int(a
)
1780 assert False, "Unhandled option"
1782 logger
.setLevel(logging
.WARNING
)
1784 logger
.setLevel(logging
.DEBUG
)
1786 logger
.setLevel(logging
.ERROR
)
1788 test_rest
= TestRest(url
, user
=user
, password
=password
, project
=project
)
1789 # print("tests to do:", test_to_do)
1792 for test
in test_to_do
:
1794 test_class
= test_classes
[test
]
1795 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(text_index
))
1797 for test
, test_class
in test_classes
.items():
1798 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(0))
1804 except TestException
as e
:
1805 logger
.error(test
+ "Test {} Exception: {}".format(test
, str(e
)))
1807 except getopt
.GetoptError
as e
:
1809 print(e
, file=sys
.stderr
)
1811 except Exception as e
:
1812 logger
.critical(test
+ " Exception: " + str(e
), exc_info
=True)