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
26 from random
import randint
28 from sys
import stderr
30 __author__
= "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
31 __date__
= "$2018-03-01$"
33 version_date
= "Oct 2018"
37 print("Usage: ", sys
.argv
[0], "[options]")
38 print(" Performs system tests over running NBI. It can be used for real OSM test using option '--test-osm'")
39 print(" If this is the case env variables 'OSMNBITEST_VIM_NAME' must be supplied to create a VIM if not exist "
40 "where deployment is done")
42 print(" -h|--help: shows this help")
43 print(" --insecure: Allows non trusted https NBI server")
44 print(" --list: list available tests")
45 print(" --manual-check: Deployment tests stop after deployed to allow manual inspection. Only make sense with "
47 print(" -p|--password PASSWORD: NBI access password. 'admin' by default")
48 print(" ---project PROJECT: NBI access project. 'admin' by default")
49 print(" --test TEST[,...]: Execute only a test or a comma separated list of tests")
50 print(" --params key=val: params to the previous test. key can be vnfd-files, nsd-file, ns-name, ns-config")
51 print(" --test-osm: If missing this test is intended for NBI only, no other OSM components are expected. Use "
52 "this flag to test the system. LCM and RO components are expected to be up and running")
53 print(" --timeout TIMEOUT: General NBI timeout, by default {}s".format(timeout
))
54 print(" --timeout-deploy TIMEOUT: Timeout used for getting NS deployed, by default {}s".format(timeout_deploy
))
55 print(" --timeout-configure TIMEOUT: Timeout used for getting NS deployed and configured,"
56 " by default {}s".format(timeout_configure
))
57 print(" -u|--user USERNAME: NBI access username. 'admin' by default")
58 print(" --url URL: complete NBI server URL. 'https//localhost:9999/osm' by default")
59 print(" -v|--verbose print debug information, can be used several times")
60 print(" --no-verbose remove verbosity")
61 print(" --version: prints current version")
62 print("ENV variables used for real deployment tests with option osm-test.")
63 print(" export OSMNBITEST_VIM_NAME=vim-name")
64 print(" export OSMNBITEST_VIM_URL=vim-url")
65 print(" export OSMNBITEST_VIM_TYPE=vim-type")
66 print(" export OSMNBITEST_VIM_TENANT=vim-tenant")
67 print(" export OSMNBITEST_VIM_USER=vim-user")
68 print(" export OSMNBITEST_VIM_PASSWORD=vim-password")
69 print(" export OSMNBITEST_VIM_CONFIG=\"vim-config\"")
70 print(" export OSMNBITEST_NS_NAME=\"vim-config\"")
74 r_header_json
= {"Content-type": "application/json"}
75 headers_json
= {"Content-type": "application/json", "Accept": "application/json"}
76 r_header_yaml
= {"Content-type": "application/yaml"}
77 headers_yaml
= {"Content-type": "application/yaml", "Accept": "application/yaml"}
78 r_header_text
= {"Content-type": "text/plain"}
79 r_header_octect
= {"Content-type": "application/octet-stream"}
80 headers_text
= {"Accept": "text/plain,application/yaml"}
81 r_header_zip
= {"Content-type": "application/zip"}
82 headers_zip
= {"Accept": "application/zip,application/yaml"}
83 headers_zip_yaml
= {"Accept": "application/yaml", "Content-type": "application/zip"}
84 r_headers_yaml_location_vnfd
= {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"}
85 r_headers_yaml_location_nsd
= {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}
86 r_headers_yaml_location_nst
= {"Location": "/nst/v1/netslice_templates_content", "Content-Type": "application/yaml"}
87 r_headers_yaml_location_nslcmop
= {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}
89 # test ones authorized
90 test_authorized_list
= (
91 ("AU1", "Invalid vnfd id", "GET", "/vnfpkgm/v1/vnf_packages/non-existing-id",
92 headers_json
, None, 404, r_header_json
, "json"),
93 ("AU2", "Invalid nsd id", "GET", "/nsd/v1/ns_descriptors/non-existing-id",
94 headers_yaml
, None, 404, r_header_yaml
, "yaml"),
95 ("AU3", "Invalid nsd id", "DELETE", "/nsd/v1/ns_descriptors_content/non-existing-id",
96 headers_yaml
, None, 404, r_header_yaml
, "yaml"),
98 timeout
= 120 # general timeout
99 timeout_deploy
= 60*10 # timeout for NS deploying without charms
100 timeout_configure
= 60*20 # timeout for NS deploying and configuring
103 class TestException(Exception):
108 def __init__(self
, url_base
, header_base
=None, verify
=False, user
="admin", password
="admin", project
="admin"):
109 self
.url_base
= url_base
110 if header_base
is None:
111 self
.header_base
= {}
113 self
.header_base
= header_base
.copy()
114 self
.s
= requests
.session()
115 self
.s
.headers
= self
.header_base
119 self
.password
= password
120 self
.project
= project
122 # contains ID of tests obtained from Location response header. "" key contains last obtained id
124 self
.test_name
= None
125 self
.step
= 0 # number of subtest under test
126 self
.passed_tests
= 0
127 self
.failed_tests
= 0
129 def set_test_name(self
, test_name
):
130 self
.test_name
= test_name
134 def set_header(self
, header
):
135 self
.s
.headers
.update(header
)
137 def set_tet_name(self
, test_name
):
138 self
.test_name
= test_name
140 def unset_header(self
, key
):
141 if key
in self
.s
.headers
:
142 del self
.s
.headers
[key
]
144 def test(self
, description
, method
, url
, headers
, payload
, expected_codes
, expected_headers
,
145 expected_payload
, store_file
=None, pooling
=False):
147 Performs an http request and check http code response. Exit if different than allowed. It get the returned id
148 that can be used by following test in the URL with {name} where name is the name of the test
149 :param description: description of the test
150 :param method: HTTP method: GET,PUT,POST,DELETE,...
151 :param url: complete URL or relative URL
152 :param headers: request headers to add to the base headers
153 :param payload: Can be a dict, transformed to json, a text or a file if starts with '@'
154 :param expected_codes: expected response codes, can be int, int tuple or int range
155 :param expected_headers: expected response headers, dict with key values
156 :param expected_payload: expected payload, 0 if empty, 'yaml', 'json', 'text', 'zip', 'octet-stream'
157 :param store_file: filename to store content
158 :param pooling: if True do not count neither log this test. Because a pooling is done with many equal requests
159 :return: requests response
164 self
.s
= requests
.session()
168 elif not url
.startswith("http"):
169 url
= self
.url_base
+ url
171 # replace url <> with the last ID
172 url
= url
.replace("<>", self
.last_id
)
174 if isinstance(payload
, str):
175 if payload
.startswith("@"):
177 file_name
= payload
[1:]
178 if payload
.startswith("@b"):
180 file_name
= payload
[2:]
181 with
open(file_name
, mode
) as f
:
183 elif isinstance(payload
, dict):
184 payload
= json
.dumps(payload
)
187 test_description
= "Test {}{} {} {} {}".format(self
.test_name
, self
.step
, description
, method
, url
)
188 logger
.warning(test_description
)
191 if expected_payload
in ("zip", "octet-string") or store_file
:
193 r
= getattr(self
.s
, method
.lower())(url
, data
=payload
, headers
=headers
, verify
=self
.verify
, stream
=stream
)
194 if expected_payload
in ("zip", "octet-string") or store_file
:
195 logger
.debug("RX {}".format(r
.status_code
))
197 logger
.debug("RX {}: {}".format(r
.status_code
, r
.text
))
201 if isinstance(expected_codes
, int):
202 expected_codes
= (expected_codes
,)
203 if r
.status_code
not in expected_codes
:
205 "Got status {}. Expected {}. {}".format(r
.status_code
, expected_codes
, r
.text
))
208 for header_key
, header_val
in expected_headers
.items():
209 if header_key
.lower() not in r
.headers
:
210 raise TestException("Header {} not present".format(header_key
))
211 if header_val
and header_val
.lower() not in r
.headers
[header_key
]:
212 raise TestException("Header {} does not contain {} but {}".format(header_key
, header_val
,
213 r
.headers
[header_key
]))
215 if expected_payload
is not None:
216 if expected_payload
== 0 and len(r
.content
) > 0:
217 raise TestException("Expected empty payload")
218 elif expected_payload
== "json":
221 except Exception as e
:
222 raise TestException("Expected json response payload, but got Exception {}".format(e
))
223 elif expected_payload
== "yaml":
225 yaml
.safe_load(r
.text
)
226 except Exception as e
:
227 raise TestException("Expected yaml response payload, but got Exception {}".format(e
))
228 elif expected_payload
in ("zip", "octet-string"):
229 if len(r
.content
) == 0:
230 raise TestException("Expected some response payload, but got empty")
232 # tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
233 # for tarinfo in tar:
234 # tarname = tarinfo.name
236 # except Exception as e:
237 # raise TestException("Expected zip response payload, but got Exception {}".format(e))
238 elif expected_payload
== "text":
239 if len(r
.content
) == 0:
240 raise TestException("Expected some response payload, but got empty")
243 with
open(store_file
, 'wb') as fd
:
244 for chunk
in r
.iter_content(chunk_size
=128):
247 location
= r
.headers
.get("Location")
249 _id
= location
[location
.rfind("/") + 1:]
251 self
.last_id
= str(_id
)
253 self
.passed_tests
+= 1
255 except TestException
as e
:
256 self
.failed_tests
+= 1
260 r_status_code
= r
.status_code
262 logger
.error("{} \nRX code{}: {}".format(e
, r_status_code
, r_text
))
267 logger
.error("Cannot open file {}: {}".format(store_file
, e
))
269 logger
.error("Exception: {}".format(e
), exc_info
=True)
270 self
.failed_tests
+= 1
274 def get_autorization(self
): # user=None, password=None, project=None):
275 if self
.token
: # and self.user == user and self.password == password and self.project == project:
278 # self.password = password
279 # self.project = project
280 r
= self
.test("Obtain token", "POST", "/admin/v1/tokens", headers_json
,
281 {"username": self
.user
, "password": self
.password
, "project_id": self
.project
},
282 (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("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("Get VIM ID", "GET", "/admin/v1/vim_accounts?name={}".format(vim_name
), headers_json
,
309 None, 200, r_header_json
, "json")
314 return vims
[0]["_id"]
317 # check needed environ parameters:
318 if not os
.environ
.get("OSMNBITEST_VIM_URL") or not os
.environ
.get("OSMNBITEST_VIM_TENANT"):
319 raise TestException("Env OSMNBITEST_VIM_URL and OSMNBITEST_VIM_TENANT are needed for create a real VIM"
320 " to deploy on whit the --test-osm option")
321 vim_data
= "{{schema_version: '1.0', name: '{}', vim_type: {}, vim_url: '{}', vim_tenant_name: '{}', "\
322 "vim_user: {}, vim_password: {}".format(vim_name
,
323 os
.environ
.get("OSMNBITEST_VIM_TYPE", "openstack"),
324 os
.environ
.get("OSMNBITEST_VIM_URL"),
325 os
.environ
.get("OSMNBITEST_VIM_TENANT"),
326 os
.environ
.get("OSMNBITEST_VIM_USER"),
327 os
.environ
.get("OSMNBITEST_VIM_PASSWORD"))
328 if os
.environ
.get("OSMNBITEST_VIM_CONFIG"):
329 vim_data
+= " ,config: {}".format(os
.environ
.get("OSMNBITEST_VIM_CONFIG"))
332 vim_data
= "{schema_version: '1.0', name: fakeVim, vim_type: openstack, vim_url: 'http://10.11.12.13/fake'"\
333 ", vim_tenant_name: 'vimtenant', vim_user: vimuser, vim_password: vimpassword}"
334 self
.test("Create VIM", "POST", "/admin/v1/vim_accounts", headers_yaml
, vim_data
,
335 (201), {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/yaml"}, "yaml")
338 def print_results(self
):
339 print("\n\n\n--------------------------------------------")
340 print("TEST RESULTS: Total: {}, Passed: {}, Failed: {}".format(self
.passed_tests
+ self
.failed_tests
,
341 self
.passed_tests
, self
.failed_tests
))
342 print("--------------------------------------------")
344 def wait_until_delete(self
, url_op
, timeout_delete
):
346 Make a pooling until topic is not present, because of deleted
348 :param timeout_delete:
351 description
= "Wait to topic being deleted"
352 test_description
= "Test {}{} {} {} {}".format(self
.test_name
, self
.step
, description
, "GET", url_op
)
353 logger
.warning(test_description
)
356 wait
= timeout_delete
358 r
= self
.test(description
, "GET", url
, headers_yaml
, (200, 404), None, r_header_yaml
, "yaml", pooling
=True)
361 if r
.status_code
== 404:
362 self
.passed_tests
+= 1
364 elif r
.status_code
== 200:
368 raise TestException("Topic is not deleted after {} seconds".format(timeout_delete
))
369 self
.failed_tests
+= 1
371 def wait_operation_ready(self
, ns_nsi
, opp_id
, timeout
, expected_fail
=False):
373 Wait until nslcmop or nsilcmop finished
374 :param ns_nsi: "ns" o "nsi"
375 :param opp_id: Id o fthe operation
377 :param expected_fail:
378 :return: None. Updates passed/failed_tests
381 url_op
= "/nslcm/v1/ns_lcm_op_occs/{}".format(opp_id
)
383 url_op
= "/nsilcm/v1/nsi_lcm_op_occs/{}".format(opp_id
)
384 description
= "Wait to {} lcm operation complete".format(ns_nsi
)
385 test_description
= "Test {}{} {} {} {}".format(self
.test_name
, self
.step
, description
, "GET", url_op
)
386 logger
.warning(test_description
)
390 r
= self
.test(description
, "GET", url_op
, headers_json
, None,
391 200, r_header_json
, "json", pooling
=True)
395 if "COMPLETED" in nslcmop
["operationState"]:
397 logger
.error("NS terminate has success, expecting failing: {}".format(
398 nslcmop
["detailed-status"]))
399 self
.failed_tests
+= 1
401 self
.passed_tests
+= 1
403 elif "FAILED" in nslcmop
["operationState"]:
404 if not expected_fail
:
405 logger
.error("NS terminate has failed: {}".format(nslcmop
["detailed-status"]))
407 self
.passed_tests
+= 1
410 print(".", end
="", file=stderr
)
414 self
.failed_tests
+= 1
415 logger
.error("NS instantiate is not terminate after {} seconds".format(timeout
))
417 print("", file=stderr
)
420 class TestNonAuthorized
:
421 description
= "test invalid URLs. methods and no authorization"
424 def run(engine
, test_osm
, manual_check
, test_params
=None):
425 engine
.set_test_name("NonAuth")
426 engine
.remove_authorization()
427 test_not_authorized_list
= (
428 ("Invalid token", "GET", "/admin/v1/users", headers_json
, None, 401, r_header_json
, "json"),
429 ("Invalid URL", "POST", "/admin/v1/nonexist", headers_yaml
, None, 405, r_header_yaml
, "yaml"),
430 ("Invalid version", "DELETE", "/admin/v2/users", headers_yaml
, None, 405, r_header_yaml
, "yaml"),
432 for t
in test_not_authorized_list
:
436 class TestUsersProjects
:
437 description
= "test project and user creation"
440 def run(engine
, test_osm
, manual_check
, test_params
=None):
441 engine
.set_test_name("UserProject")
442 engine
.get_autorization()
443 engine
.test("Create project non admin", "POST", "/admin/v1/projects", headers_json
, {"name": "P1"},
444 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
445 engine
.test("Create project admin", "POST", "/admin/v1/projects", headers_json
,
446 {"name": "Padmin", "admin": True}, (201, 204),
447 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
448 engine
.test("Create project bad format", "POST", "/admin/v1/projects", headers_json
, {"name": 1}, 422,
449 r_header_json
, "json")
450 engine
.test("Create user with bad project", "POST", "/admin/v1/users", headers_json
,
451 {"username": "U1", "projects": ["P1", "P2", "Padmin"], "password": "pw1"}, 409,
452 r_header_json
, "json")
453 engine
.test("Create user with bad project and force", "POST", "/admin/v1/users?FORCE=True", headers_json
,
454 {"username": "U1", "projects": ["P1", "P2", "Padmin"], "password": "pw1"}, 201,
455 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
456 engine
.test("Create user 2", "POST", "/admin/v1/users", headers_json
,
457 {"username": "U2", "projects": ["P1"], "password": "pw2"}, 201,
458 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
459 engine
.test("Edit user U1, delete P2 project", "PATCH", "/admin/v1/users/U1", headers_json
,
460 {"projects": {"$'P2'": None}}, 204, None, None)
461 res
= engine
.test("Check user U1, contains the right projects", "GET", "/admin/v1/users/U1",
462 headers_json
, None, 200, None, json
)
466 expected_projects
= ["P1", "Padmin"]
467 if u1
["projects"] != expected_projects
:
468 logger
.error("User content projects '{}' different than expected '{}'. Edition has not done"
469 " properly".format(u1
["projects"], expected_projects
))
470 engine
.failed_tests
+= 1
472 engine
.test("Edit user U1, set Padmin as default project", "PUT", "/admin/v1/users/U1", headers_json
,
473 {"projects": {"$'Padmin'": None, "$+[0]": "Padmin"}}, 204, None, None)
474 res
= engine
.test("Check user U1, contains the right projects", "GET", "/admin/v1/users/U1",
475 headers_json
, None, 200, None, json
)
479 expected_projects
= ["Padmin", "P1"]
480 if u1
["projects"] != expected_projects
:
481 logger
.error("User content projects '{}' different than expected '{}'. Edition has not done"
482 " properly".format(u1
["projects"], expected_projects
))
483 engine
.failed_tests
+= 1
485 engine
.test("Edit user U1, change password", "PATCH", "/admin/v1/users/U1", headers_json
,
486 {"password": "pw1_new"}, 204, None, None)
488 engine
.test("Change to project P1 non existing", "POST", "/admin/v1/tokens/", headers_json
,
489 {"project_id": "P1"}, 401, r_header_json
, "json")
491 res
= engine
.test("Change to user U1 project P1", "POST", "/admin/v1/tokens", headers_json
,
492 {"username": "U1", "password": "pw1_new", "project_id": "P1"}, (200, 201),
493 r_header_json
, "json")
495 response
= res
.json()
496 engine
.set_header({"Authorization": "Bearer {}".format(response
["id"])})
498 engine
.test("Edit user projects non admin", "PUT", "/admin/v1/users/U1", headers_json
,
499 {"projects": {"$'P1'": None}}, 401, r_header_json
, "json")
500 engine
.test("Add new project non admin", "POST", "/admin/v1/projects", headers_json
,
501 {"name": "P2"}, 401, r_header_json
, "json")
502 engine
.test("Add new user non admin", "POST", "/admin/v1/users", headers_json
,
503 {"username": "U3", "projects": ["P1"], "password": "pw3"}, 401,
504 r_header_json
, "json")
506 res
= engine
.test("Change to user U1 project Padmin", "POST", "/admin/v1/tokens", headers_json
,
507 {"project_id": "Padmin"}, (200, 201), r_header_json
, "json")
509 response
= res
.json()
510 engine
.set_header({"Authorization": "Bearer {}".format(response
["id"])})
512 engine
.test("Add new project admin", "POST", "/admin/v1/projects", headers_json
, {"name": "P2"},
513 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
514 engine
.test("Add new user U3 admin", "POST", "/admin/v1/users",
515 headers_json
, {"username": "U3", "projects": ["P2"], "password": "pw3"}, (201, 204),
516 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
517 engine
.test("Edit user projects admin", "PUT", "/admin/v1/users/U3", headers_json
,
518 {"projects": ["P2"]}, 204, None, None)
520 engine
.test("Delete project P2 conflict", "DELETE", "/admin/v1/projects/P2", headers_json
, None, 409,
521 r_header_json
, "json")
522 engine
.test("Delete project P2 forcing", "DELETE", "/admin/v1/projects/P2?FORCE=True", headers_json
,
523 None, 204, None, None)
525 engine
.test("Delete user U1. Conflict deleting own user", "DELETE", "/admin/v1/users/U1", headers_json
,
526 None, 409, r_header_json
, "json")
527 engine
.test("Delete user U2", "DELETE", "/admin/v1/users/U2", headers_json
, None, 204, None, None)
528 engine
.test("Delete user U3", "DELETE", "/admin/v1/users/U3", headers_json
, None, 204, None, None)
530 engine
.remove_authorization() # To force get authorization
531 engine
.get_autorization()
532 engine
.test("Delete user U1", "DELETE", "/admin/v1/users/U1", headers_json
, None, 204, None, None)
533 engine
.test("Delete project P1", "DELETE", "/admin/v1/projects/P1", headers_json
, None, 204, None, None)
534 engine
.test("Delete project Padmin", "DELETE", "/admin/v1/projects/Padmin", headers_json
, None, 204,
539 description
= "Creates/edit/delete fake VIMs and SDN controllers"
543 "schema_version": "1.0",
544 "schema_type": "No idea",
546 "description": "Descriptor name",
547 "vim_type": "openstack",
548 "vim_url": "http://localhost:/vim",
549 "vim_tenant_name": "vimTenant",
551 "vim_password": "password",
552 "config": {"config_param": 1}
556 "description": "sdn-description",
557 "dpid": "50:50:52:54:00:94:21:21",
558 "ip": "192.168.15.17",
560 "type": "opendaylight",
565 self
.port_mapping
= [
566 {"compute_node": "compute node 1",
567 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/1", "switch_mac": "52:54:00:94:21:21"},
568 {"pci": "0000:81:00.1", "switch_port": "port-2/2", "switch_mac": "52:54:00:94:21:22"}
570 {"compute_node": "compute node 2",
571 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/3", "switch_mac": "52:54:00:94:21:23"},
572 {"pci": "0000:81:00.1", "switch_port": "port-2/4", "switch_mac": "52:54:00:94:21:24"}
576 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
578 vim_bad
= self
.vim
.copy()
581 engine
.set_test_name("FakeVim")
582 engine
.get_autorization()
583 engine
.test("Create VIM", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
, (201, 204),
584 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json")
585 vim_id
= engine
.last_id
586 engine
.test("Create VIM without name, bad schema", "POST", "/admin/v1/vim_accounts", headers_json
,
587 vim_bad
, 422, None, headers_json
)
588 engine
.test("Create VIM name repeated", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
,
589 409, None, headers_json
)
590 engine
.test("Show VIMs", "GET", "/admin/v1/vim_accounts", headers_yaml
, None, 200, r_header_yaml
,
592 engine
.test("Show VIM", "GET", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_yaml
, None, 200,
593 r_header_yaml
, "yaml")
596 engine
.test("Delete VIM", "DELETE", "/admin/v1/vim_accounts/{}?FORCE=True".format(vim_id
), headers_yaml
,
598 engine
.test("Check VIM is deleted", "GET", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_yaml
, None,
599 404, r_header_yaml
, "yaml")
601 # delete and wait until is really deleted
602 engine
.test("Delete VIM", "DELETE", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_yaml
, None, 202,
604 engine
.wait_until_delete("/admin/v1/vim_accounts/{}".format(vim_id
), timeout
)
607 class TestVIMSDN(TestFakeVim
):
608 description
= "Creates VIM with SDN editing SDN controllers and port_mapping"
611 TestFakeVim
.__init
__(self
)
613 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
614 engine
.set_test_name("VimSdn")
615 engine
.get_autorization()
617 engine
.test("Create SDN", "POST", "/admin/v1/sdns", headers_json
, self
.sdn
, (201, 204),
618 {"Location": "/admin/v1/sdns/", "Content-Type": "application/json"}, "json")
619 sdnc_id
= engine
.last_id
622 engine
.test("Edit SDN", "PATCH", "/admin/v1/sdns/{}".format(sdnc_id
), headers_json
, {"name": "new_sdn_name"},
626 self
.vim
["config"]["sdn-controller"] = sdnc_id
627 self
.vim
["config"]["sdn-port-mapping"] = self
.port_mapping
628 engine
.test("Create VIM", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
, (200, 204, 201),
629 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json"),
631 vim_id
= engine
.last_id
632 self
.port_mapping
[0]["compute_node"] = "compute node XX"
633 engine
.test("Edit VIM change port-mapping", "PUT", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_json
,
634 {"config": {"sdn-port-mapping": self
.port_mapping
}}, 204, None, None)
635 engine
.test("Edit VIM remove port-mapping", "PUT", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_json
,
636 {"config": {"sdn-port-mapping": None}}, 204, None, None)
640 engine
.test("Delete VIM remove port-mapping", "DELETE",
641 "/admin/v1/vim_accounts/{}?FORCE=True".format(vim_id
), headers_json
, None, 202, None, 0)
642 engine
.test("Delete SDNC", "DELETE", "/admin/v1/sdns/{}?FORCE=True".format(sdnc_id
), headers_json
, None,
645 engine
.test("Check VIM is deleted", "GET", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_yaml
,
646 None, 404, r_header_yaml
, "yaml")
647 engine
.test("Check SDN is deleted", "GET", "/admin/v1/sdns/{}".format(sdnc_id
), headers_yaml
, None,
648 404, r_header_yaml
, "yaml")
650 # delete and wait until is really deleted
651 engine
.test("Delete VIM remove port-mapping", "DELETE", "/admin/v1/vim_accounts/{}".format(vim_id
),
652 headers_json
, None, (202, 201, 204), None, 0)
653 engine
.test("Delete SDN", "DELETE", "/admin/v1/sdns/{}".format(sdnc_id
), headers_json
, None,
654 (202, 201, 204), None, 0)
655 engine
.wait_until_delete("/admin/v1/vim_accounts/{}".format(vim_id
), timeout
)
656 engine
.wait_until_delete("/admin/v1/sdns/{}".format(sdnc_id
), timeout
)
660 description
= "Base class for downloading descriptors from ETSI, onboard and deploy in real VIM"
663 self
.test_name
= "DEPLOY"
668 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
669 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
670 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
671 self
.descriptor_edit
= None
672 self
.uses_configuration
= False
680 def create_descriptors(self
, engine
):
681 temp_dir
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/temp/"
682 if not os
.path
.exists(temp_dir
):
683 os
.makedirs(temp_dir
)
684 for vnfd_index
, vnfd_filename
in enumerate(self
.vnfd_filenames
):
685 if "/" in vnfd_filename
:
686 vnfd_filename_path
= vnfd_filename
687 if not os
.path
.exists(vnfd_filename_path
):
688 raise TestException("File '{}' does not exist".format(vnfd_filename_path
))
690 vnfd_filename_path
= temp_dir
+ vnfd_filename
691 if not os
.path
.exists(vnfd_filename_path
):
692 with
open(vnfd_filename_path
, "wb") as file:
693 response
= requests
.get(self
.descriptor_url
+ vnfd_filename
)
694 if response
.status_code
>= 300:
695 raise TestException("Error downloading descriptor from '{}': {}".format(
696 self
.descriptor_url
+ vnfd_filename
, response
.status_code
))
697 file.write(response
.content
)
698 if vnfd_filename_path
.endswith(".yaml"):
699 headers
= headers_yaml
701 headers
= headers_zip_yaml
702 if randint(0, 1) == 0:
703 # vnfd CREATE AND UPLOAD in one step:
704 engine
.test("Onboard VNFD in one step", "POST",
705 "/vnfpkgm/v1/vnf_packages_content" + self
.qforce
, headers
, "@b" + vnfd_filename_path
, 201,
706 r_headers_yaml_location_vnfd
,
708 self
.vnfds_id
.append(engine
.last_id
)
710 # vnfd CREATE AND UPLOAD ZIP
711 engine
.test("Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages",
712 headers_json
, None, 201,
713 {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
714 self
.vnfds_id
.append(engine
.last_id
)
715 engine
.test("Onboard VNFD step 2 as ZIP", "PUT",
716 "/vnfpkgm/v1/vnf_packages/<>/package_content" + self
.qforce
,
717 headers
, "@b" + vnfd_filename_path
, 204, None, 0)
719 if self
.descriptor_edit
:
720 if "vnfd{}".format(vnfd_index
) in self
.descriptor_edit
:
722 engine
.test("Edit VNFD ", "PATCH",
723 "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfds_id
[-1]),
724 headers_yaml
, self
.descriptor_edit
["vnfd{}".format(vnfd_index
)], 204, None, None)
726 if "/" in self
.nsd_filename
:
727 nsd_filename_path
= self
.nsd_filename
728 if not os
.path
.exists(nsd_filename_path
):
729 raise TestException("File '{}' does not exist".format(nsd_filename_path
))
731 nsd_filename_path
= temp_dir
+ self
.nsd_filename
732 if not os
.path
.exists(nsd_filename_path
):
733 with
open(nsd_filename_path
, "wb") as file:
734 response
= requests
.get(self
.descriptor_url
+ self
.nsd_filename
)
735 if response
.status_code
>= 300:
736 raise TestException("Error downloading descriptor from '{}': {}".format(
737 self
.descriptor_url
+ self
.nsd_filename
, response
.status_code
))
738 file.write(response
.content
)
739 if nsd_filename_path
.endswith(".yaml"):
740 headers
= headers_yaml
742 headers
= headers_zip_yaml
744 if randint(0, 1) == 0:
745 # nsd CREATE AND UPLOAD in one step:
746 engine
.test("Onboard NSD in one step", "POST",
747 "/nsd/v1/ns_descriptors_content" + self
.qforce
, headers
, "@b" + nsd_filename_path
, 201,
748 r_headers_yaml_location_nsd
, yaml
)
749 self
.nsd_id
= engine
.last_id
751 # nsd CREATE AND UPLOAD ZIP
752 engine
.test("Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors",
753 headers_json
, None, 201,
754 {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
755 self
.nsd_id
= engine
.last_id
756 engine
.test("Onboard NSD step 2 as ZIP", "PUT",
757 "/nsd/v1/ns_descriptors/<>/nsd_content" + self
.qforce
,
758 headers
, "@b" + nsd_filename_path
, 204, None, 0)
760 if self
.descriptor_edit
and "nsd" in self
.descriptor_edit
:
762 engine
.test("Edit NSD ", "PATCH",
763 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
),
764 headers_yaml
, self
.descriptor_edit
["nsd"], 204, None, None)
766 def delete_descriptors(self
, engine
):
768 engine
.test("Delete NSSD SOL005", "DELETE",
769 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
),
770 headers_yaml
, None, 204, None, 0)
771 for vnfd_id
in self
.vnfds_id
:
772 engine
.test("Delete VNFD SOL005", "DELETE",
773 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id
), headers_yaml
, None, 204, None, 0)
775 def instantiate(self
, engine
, ns_data
):
776 ns_data_text
= yaml
.safe_dump(ns_data
, default_flow_style
=True, width
=256)
777 # create NS Two steps
778 r
= engine
.test("Create NS step 1", "POST", "/nslcm/v1/ns_instances",
779 headers_yaml
, ns_data_text
, 201,
780 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"}, "yaml")
783 self
.ns_id
= engine
.last_id
784 engine
.test("Instantiate NS step 2", "POST",
785 "/nslcm/v1/ns_instances/{}/instantiate".format(self
.ns_id
), headers_yaml
, ns_data_text
,
786 201, r_headers_yaml_location_nslcmop
, "yaml")
787 nslcmop_id
= engine
.last_id
790 # Wait until status is Ok
791 timeout
= timeout_configure
if self
.uses_configuration
else timeout_deploy
792 engine
.wait_operation_ready("ns", nslcmop_id
, timeout
)
794 def terminate(self
, engine
):
797 engine
.test("Terminate NS", "POST", "/nslcm/v1/ns_instances/{}/terminate".format(self
.ns_id
), headers_yaml
,
798 None, 201, r_headers_yaml_location_nslcmop
, "yaml")
799 nslcmop2_id
= engine
.last_id
800 # Wait until status is Ok
801 engine
.wait_operation_ready("ns", nslcmop2_id
, timeout_deploy
)
803 engine
.test("Delete NS", "DELETE", "/nslcm/v1/ns_instances/{}".format(self
.ns_id
), headers_yaml
, None,
806 engine
.test("Delete NS with FORCE", "DELETE", "/nslcm/v1/ns_instances/{}?FORCE=True".format(self
.ns_id
),
807 headers_yaml
, None, 204, None, 0)
809 # check all it is deleted
810 engine
.test("Check NS is deleted", "GET", "/nslcm/v1/ns_instances/{}".format(self
.ns_id
), headers_yaml
, None,
812 r
= engine
.test("Check NSLCMOPs are deleted", "GET",
813 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId={}".format(self
.ns_id
), 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_id
, nslcmops
))
821 def test_ns(self
, engine
, test_osm
, commands
=None, users
=None, passwds
=None, keys
=None, timeout
=0):
823 r
= engine
.test("GET VNFR IDs", "GET",
824 "/nslcm/v1/ns_instances/{}".format(self
.ns_id
), headers_json
, None,
825 200, r_header_json
, "json")
830 vnfr_list
= ns_data
['constituent-vnfr-ref']
833 for vnfr_id
in vnfr_list
:
834 r
= engine
.test("Get VNFR to get IP_ADDRESS", "GET",
835 "/nslcm/v1/vnfrs/{}".format(vnfr_id
), headers_json
, None,
836 200, r_header_json
, "json")
841 vnf_index
= str(vnfr_data
["member-vnf-index-ref"])
842 if not commands
.get(vnf_index
):
844 if vnfr_data
.get("ip-address"):
845 description
= "Exec command='{}' at VNFR={} IP={}".format(commands
.get(vnf_index
)[0], vnf_index
,
846 vnfr_data
['ip-address'])
848 test_description
= "{}{} {}".format(engine
.test_name
, engine
.step
, description
)
849 logger
.warning(test_description
)
850 while timeout
>= time
:
851 result
, message
= self
.do_checks([vnfr_data
["ip-address"]],
852 vnf_index
=vnfr_data
["member-vnf-index-ref"],
853 commands
=commands
.get(vnf_index
), user
=users
.get(vnf_index
),
854 passwd
=passwds
.get(vnf_index
), key
=keys
.get(vnf_index
))
856 engine
.passed_tests
+= 1
857 logger
.debug(message
)
863 engine
.failed_tests
+= 1
864 logger
.error(message
)
868 engine
.failed_tests
+= 1
869 logger
.error(message
)
871 engine
.failed_tests
+= 1
872 logger
.error("VNFR {} has not mgmt address. Check failed".format(vnfr_id
))
874 def do_checks(self
, ip
, vnf_index
, commands
=[], user
=None, passwd
=None, key
=None):
877 from pssh
.clients
import ParallelSSHClient
878 from pssh
.utils
import load_private_key
879 from ssh2
import exceptions
as ssh2Exception
880 except ImportError as e
:
881 logger
.critical("Package <pssh> or/and <urllib3> is not installed. Please add them with 'pip3 install "
882 "parallel-ssh urllib3': {}".format(e
))
883 return -1, "install needed packages 'pip3 install parallel-ssh urllib3'"
884 urllib3
.disable_warnings(urllib3
.exceptions
.InsecureRequestWarning
)
886 p_host
= os
.environ
.get("PROXY_HOST")
887 p_user
= os
.environ
.get("PROXY_USER")
888 p_password
= os
.environ
.get("PROXY_PASSWD")
891 pkey
= load_private_key(key
)
895 client
= ParallelSSHClient(ip
, user
=user
, password
=passwd
, pkey
=pkey
, proxy_host
=p_host
,
896 proxy_user
=p_user
, proxy_password
=p_password
, timeout
=10, num_retries
=0)
898 output
= client
.run_command(cmd
)
900 if output
[ip
[0]].exit_code
:
901 return -1, " VNFR {} command '{}' returns error: {}".format(ip
[0], cmd
, output
[ip
[0]].stderr
)
903 return 1, " VNFR {} command '{}' successful".format(ip
[0], cmd
)
904 except (ssh2Exception
.ChannelFailure
, ssh2Exception
.SocketDisconnectError
, ssh2Exception
.SocketTimeout
,
905 ssh2Exception
.SocketRecvError
) as e
:
906 return 0, "Timeout accessing the VNFR {}: {}".format(ip
[0], str(e
))
907 except Exception as e
:
908 return -1, "ERROR checking the VNFR {}: {}".format(ip
[0], str(e
))
910 def aditional_operations(self
, engine
, test_osm
, manual_check
):
913 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
914 engine
.set_test_name(self
.test_name
)
915 engine
.get_autorization()
916 nsname
= os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
918 if "vnfd-files" in test_params
:
919 self
.vnfd_filenames
= test_params
["vnfd-files"].split(",")
920 if "nsd-file" in test_params
:
921 self
.nsd_filename
= test_params
["nsd-file"]
922 if test_params
.get("ns-name"):
923 nsname
= test_params
["ns-name"]
924 self
.create_descriptors(engine
)
926 # create real VIM if not exist
927 self
.vim_id
= engine
.get_create_vim(test_osm
)
928 ns_data
= {"nsDescription": "default description", "nsName": nsname
, "nsdId": self
.nsd_id
,
929 "vimAccountId": self
.vim_id
}
930 if test_params
and test_params
.get("ns-config"):
931 if isinstance(test_params
["ns-config"], str):
932 ns_data
.update(yaml
.load(test_params
["ns-config"]))
934 ns_data
.update(test_params
["ns-config"])
935 self
.instantiate(engine
, ns_data
)
938 input('NS has been deployed. Perform manual check and press enter to resume')
939 if test_osm
and self
.cmds
:
940 self
.test_ns(engine
, test_osm
, self
.cmds
, self
.uss
, self
.pss
, self
.keys
, self
.timeout
)
941 self
.aditional_operations(engine
, test_osm
, manual_check
)
942 self
.terminate(engine
)
943 self
.delete_descriptors(engine
)
946 class TestDeployHackfestCirros(TestDeploy
):
947 description
= "Load and deploy Hackfest cirros_2vnf_ns example"
951 self
.test_name
= "CIRROS"
952 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
953 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
954 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
955 self
.uss
= {'1': "cirros", '2': "cirros"}
956 self
.pss
= {'1': "cubswin:)", '2': "cubswin:)"}
959 class TestDeployHackfest1(TestDeploy
):
960 description
= "Load and deploy Hackfest_1_vnfd example"
964 self
.test_name
= "HACKFEST1-"
965 self
.vnfd_filenames
= ("hackfest_1_vnfd.tar.gz",)
966 self
.nsd_filename
= "hackfest_1_nsd.tar.gz"
967 # self.cmds = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
968 # self.uss = {'1': "cirros", '2': "cirros"}
969 # self.pss = {'1': "cubswin:)", '2': "cubswin:)"}
972 class TestDeployHackfestCirrosScaling(TestDeploy
):
973 description
= "Load and deploy Hackfest cirros_2vnf_ns example with scaling modifications"
977 self
.test_name
= "CIRROS-SCALE"
978 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
979 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
981 def create_descriptors(self
, engine
):
982 super().create_descriptors(engine
)
983 # Modify VNFD to add scaling and count=2
984 self
.descriptor_edit
= {
987 "$id: 'cirros_vnfd-VM'": {"count": 2}
989 "scaling-group-descriptor": [{
990 "name": "scale_cirros",
991 "max-instance-count": 2,
993 "vdu-id-ref": "cirros_vnfd-VM",
1000 def aditional_operations(self
, engine
, test_osm
, manual_check
):
1003 # 2 perform scale out twice
1004 payload
= '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1005 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
1006 for i
in range(0, 2):
1007 engine
.test("Execute scale action over NS", "POST",
1008 "/nslcm/v1/ns_instances/{}/scale".format(self
.ns_id
), headers_yaml
, payload
,
1009 201, r_headers_yaml_location_nslcmop
, "yaml")
1010 nslcmop2_scale_out
= engine
.last_id
1011 engine
.wait_operation_ready("ns", nslcmop2_scale_out
, timeout_deploy
)
1013 input('NS scale out done. Check that two more vdus are there')
1014 # TODO check automatic
1016 # 2 perform scale in
1017 payload
= '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1018 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
1019 for i
in range(0, 2):
1020 engine
.test("Execute scale IN action over NS", "POST",
1021 "/nslcm/v1/ns_instances/{}/scale".format(self
.ns_id
), headers_yaml
, payload
,
1022 201, r_headers_yaml_location_nslcmop
, "yaml")
1023 nslcmop2_scale_in
= engine
.last_id
1024 engine
.wait_operation_ready("ns", nslcmop2_scale_in
, timeout_deploy
)
1026 input('NS scale in done. Check that two less vdus are there')
1027 # TODO check automatic
1029 # perform scale in that must fail as reached limit
1030 engine
.test("Execute scale IN out of limit action over NS", "POST",
1031 "/nslcm/v1/ns_instances/{}/scale".format(self
.ns_id
), headers_yaml
, payload
,
1032 201, r_headers_yaml_location_nslcmop
, "yaml")
1033 nslcmop2_scale_in
= engine
.last_id
1034 engine
.wait_operation_ready("ns", nslcmop2_scale_in
, timeout_deploy
, expected_fail
=True)
1037 class TestDeployIpMac(TestDeploy
):
1038 description
= "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
1042 self
.test_name
= "SetIpMac"
1043 self
.vnfd_filenames
= ("vnfd_2vdu_set_ip_mac2.yaml", "vnfd_2vdu_set_ip_mac.yaml")
1044 self
.nsd_filename
= "scenario_2vdu_set_ip_mac.yaml"
1045 self
.descriptor_url
= \
1046 "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
1047 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1048 self
.uss
= {'1': "osm", '2': "osm"}
1049 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1052 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1053 # super().run(engine, test_osm, manual_check, test_params)
1054 # run again setting IPs with instantiate parameters
1055 instantiation_params
= {
1058 "member-vnf-index": "1",
1061 "name": "internal_vld1", # net_internal
1063 "ip-version": "ipv4",
1064 "subnet-address": "10.9.8.0/24",
1065 "dhcp-params": {"count": 100, "start-address": "10.9.8.100"}
1067 "internal-connection-point": [
1070 "ip-address": "10.9.8.2",
1074 "ip-address": "10.9.8.3",
1085 # "name": "iface11",
1086 # "floating-ip-required": True,
1090 "mac-address": "52:33:44:55:66:13"
1099 "ip-address": "10.31.31.22",
1100 "mac-address": "52:33:44:55:66:21"
1109 super().run(engine
, test_osm
, manual_check
, test_params
={"ns-config": instantiation_params
})
1112 class TestDeployHackfest4(TestDeploy
):
1113 description
= "Load and deploy Hackfest 4 example."
1117 self
.test_name
= "HACKFEST4-"
1118 self
.vnfd_filenames
= ("hackfest_4_vnfd.tar.gz",)
1119 self
.nsd_filename
= "hackfest_4_nsd.tar.gz"
1120 self
.uses_configuration
= True
1121 self
.cmds
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1122 self
.uss
= {'1': "ubuntu", '2': "ubuntu"}
1123 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1125 def create_descriptors(self
, engine
):
1126 super().create_descriptors(engine
)
1127 # Modify VNFD to add scaling
1128 self
.descriptor_edit
= {
1130 'vnf-configuration': {
1131 'config-primitive': [{
1135 'data-type': 'STRING',
1136 'default-value': '/home/ubuntu/touched'
1140 'scaling-group-descriptor': [{
1141 'name': 'scale_dataVM',
1142 'scaling-policy': [{
1143 'threshold-time': 0,
1144 'name': 'auto_cpu_util_above_threshold',
1145 'scaling-type': 'automatic',
1146 'scaling-criteria': [{
1147 'name': 'cpu_util_above_threshold',
1148 'vnf-monitoring-param-ref': 'all_aaa_cpu_util',
1149 'scale-out-relational-operation': 'GE',
1150 'scale-in-threshold': 15,
1151 'scale-out-threshold': 60,
1152 'scale-in-relational-operation': 'LE'
1156 'max-instance-count': 10,
1157 'scaling-config-action': [
1158 {'vnf-config-primitive-name-ref': 'touch',
1159 'trigger': 'post-scale-out'},
1160 {'vnf-config-primitive-name-ref': 'touch',
1161 'trigger': 'pre-scale-in'}
1164 'vdu-id-ref': 'dataVM',
1172 class TestDeployHackfest3Charmed(TestDeploy
):
1173 description
= "Load and deploy Hackfest 3charmed_ns example. Modifies it for adding scaling and performs " \
1174 "primitive actions and scaling"
1178 self
.test_name
= "HACKFEST3-"
1179 self
.vnfd_filenames
= ("hackfest_3charmed_vnfd.tar.gz",)
1180 self
.nsd_filename
= "hackfest_3charmed_nsd.tar.gz"
1181 self
.uses_configuration
= True
1182 self
.cmds
= {'1': [''], '2': ['ls -lrt /home/ubuntu/first-touch', ]}
1183 self
.uss
= {'1': "ubuntu", '2': "ubuntu"}
1184 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1185 # self.descriptor_edit = {
1186 # "vnfd0": yaml.load("""
1187 # scaling-group-descriptor:
1188 # - name: "scale_dataVM"
1189 # max-instance-count: 10
1191 # - name: "auto_cpu_util_above_threshold"
1192 # scaling-type: "automatic"
1196 # - name: "cpu_util_above_threshold"
1197 # scale-in-threshold: 15
1198 # scale-in-relational-operation: "LE"
1199 # scale-out-threshold: 60
1200 # scale-out-relational-operation: "GE"
1201 # vnf-monitoring-param-ref: "all_aaa_cpu_util"
1203 # - vdu-id-ref: dataVM
1205 # scaling-config-action:
1206 # - trigger: post-scale-out
1207 # vnf-config-primitive-name-ref: touch
1208 # - trigger: pre-scale-in
1209 # vnf-config-primitive-name-ref: touch
1210 # vnf-configuration:
1216 # default-value: '/home/ubuntu/touched'
1220 def aditional_operations(self
, engine
, test_osm
, manual_check
):
1224 payload
= '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
1225 engine
.test("Exec service primitive over NS", "POST",
1226 "/nslcm/v1/ns_instances/{}/action".format(self
.ns_id
), headers_yaml
, payload
,
1227 201, r_headers_yaml_location_nslcmop
, "yaml")
1228 nslcmop2_action
= engine
.last_id
1229 # Wait until status is Ok
1230 engine
.wait_operation_ready("ns", nslcmop2_action
, timeout_deploy
)
1232 input('NS service primitive has been executed. Check that file /home/ubuntu/OSMTESTNBI is present at '
1235 cmds
= {'1': [''], '2': ['ls -lrt /home/ubuntu/OSMTESTNBI', ]}
1236 uss
= {'1': "ubuntu", '2': "ubuntu"}
1237 pss
= {'1': "osm4u", '2': "osm4u"}
1238 self
.test_ns(engine
, test_osm
, cmds
, uss
, pss
, self
.keys
, self
.timeout
)
1240 # # 2 perform scale out
1241 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1242 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1243 # engine.test("Execute scale action over NS", "POST",
1244 # "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1245 # 201, r_headers_yaml_location_nslcmop, "yaml")
1246 # nslcmop2_scale_out = engine.last_id
1247 # engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
1249 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
1250 # # TODO check automatic
1252 # # 2 perform scale in
1253 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1254 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1255 # engine.test("Execute scale action over NS", "POST",
1256 # "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1257 # 201, r_headers_yaml_location_nslcmop, "yaml")
1258 # nslcmop2_scale_in = engine.last_id
1259 # engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
1261 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
1262 # # TODO check automatic
1265 class TestDeploySimpleCharm(TestDeploy
):
1266 description
= "Deploy hackfest-4 hackfest_simplecharm example"
1270 self
.test_name
= "HACKFEST-SIMPLE"
1271 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-4.0-four/4th-hackfest/packages/"
1272 self
.vnfd_filenames
= ("hackfest_simplecharm_vnf.tar.gz",)
1273 self
.nsd_filename
= "hackfest_simplecharm_ns.tar.gz"
1274 self
.uses_configuration
= True
1275 self
.cmds
= {'1': [''], '2': ['ls -lrt /home/ubuntu/first-touch', ]}
1276 self
.uss
= {'1': "ubuntu", '2': "ubuntu"}
1277 self
.pss
= {'1': "osm4u", '2': "osm4u"}
1280 class TestDeploySimpleCharm2(TestDeploySimpleCharm
):
1281 description
= "Deploy hackfest-4 hackfest_simplecharm example changing naming to contain dots on ids and " \
1286 self
.test_name
= "HACKFEST-SIMPLE2-"
1287 self
.qforce
= "?FORCE=True"
1288 self
.descriptor_edit
= {
1290 "id": "hackfest.simplecharm.vnf"
1294 "id": "hackfest.simplecharm.ns",
1295 "constituent-vnfd": {
1296 "$[0]": {"vnfd-id-ref": "hackfest.simplecharm.vnf", "member-vnf-index": "$1"},
1297 "$[1]": {"vnfd-id-ref": "hackfest.simplecharm.vnf", "member-vnf-index": "$2"},
1301 "vnfd-connection-point-ref": {"$[0]": {"member-vnf-index-ref": "$1",
1302 "vnfd-id-ref": "hackfest.simplecharm.vnf"},
1303 "$[1]": {"member-vnf-index-ref": "$2",
1304 "vnfd-id-ref": "hackfest.simplecharm.vnf"}},
1307 "vnfd-connection-point-ref": {"$[0]": {"member-vnf-index-ref": "$1",
1308 "vnfd-id-ref": "hackfest.simplecharm.vnf"},
1309 "$[1]": {"member-vnf-index-ref": "$2",
1310 "vnfd-id-ref": "hackfest.simplecharm.vnf"}},
1317 class TestDeployHackfest3Charmed2(TestDeployHackfest3Charmed
):
1318 description
= "Load and deploy Hackfest 3charmed_ns example modified version of descriptors to have dots in " \
1319 "ids and member-vnf-index"
1323 self
.test_name
= "HACKFEST3bis"
1324 self
.qforce
= "?FORCE=True"
1325 self
.descriptor_edit
= {
1329 "interface": {"$[0]": {"external-connection-point-ref": "pdu-mgmt"}}
1333 "vnf-configuration": None,
1334 "connection-point": {
1338 "short-name": "pdu-mgmt"
1342 "mgmt-interface": {"cp": "pdu-mgmt"},
1343 "description": "A vnf single vdu to be used as PDU",
1347 "id": "pdu_internal",
1348 "name": "pdu_internal",
1349 "internal-connection-point": {"$[1]": None},
1350 "short-name": "pdu_internal",
1356 # Modify NSD accordingly
1358 "constituent-vnfd": {
1359 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
1362 "description": "A nsd to deploy the vnf to act as as PDU",
1364 "name": "nsd-as-pdu",
1365 "short-name": "nsd-as-pdu",
1370 "short-name": "mgmt_pdu",
1371 "vnfd-connection-point-ref": {
1373 "vnfd-connection-point-ref": "pdu-mgmt",
1374 "vnfd-id-ref": "vdu-as-pdu",
1386 class TestDeploySingleVdu(TestDeployHackfest3Charmed
):
1387 description
= "Generate a single VDU base on editing Hackfest3Charmed descriptors and deploy"
1391 self
.test_name
= "SingleVDU"
1392 self
.qforce
= "?FORCE=True"
1393 self
.descriptor_edit
= {
1394 # Modify VNFD to remove one VDU
1398 "interface": {"$[0]": {"external-connection-point-ref": "pdu-mgmt"}}
1402 "vnf-configuration": None,
1403 "connection-point": {
1407 "short-name": "pdu-mgmt"
1411 "mgmt-interface": {"cp": "pdu-mgmt"},
1412 "description": "A vnf single vdu to be used as PDU",
1416 "id": "pdu_internal",
1417 "name": "pdu_internal",
1418 "internal-connection-point": {"$[1]": None},
1419 "short-name": "pdu_internal",
1425 # Modify NSD accordingly
1427 "constituent-vnfd": {
1428 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
1431 "description": "A nsd to deploy the vnf to act as as PDU",
1433 "name": "nsd-as-pdu",
1434 "short-name": "nsd-as-pdu",
1439 "short-name": "mgmt_pdu",
1440 "vnfd-connection-point-ref": {
1442 "vnfd-connection-point-ref": "pdu-mgmt",
1443 "vnfd-id-ref": "vdu-as-pdu",
1455 class TestDeployHnfd(TestDeployHackfest3Charmed
):
1456 description
= "Generate a HNFD base on editing Hackfest3Charmed descriptors and deploy"
1460 self
.test_name
= "HNFD"
1461 self
.pduDeploy
= TestDeploySingleVdu()
1462 self
.pdu_interface_0
= {}
1463 self
.pdu_interface_1
= {}
1466 # self.vnf_to_pdu = """
1469 # pdu-type: PDU-TYPE-1
1474 # name: pdu-iface-internal
1476 # description: HFND, one PDU + One VDU
1482 self
.pdu_descriptor
= {
1484 "type": "PDU-TYPE-1",
1485 "vim_accounts": "to-override",
1488 "name": "mgmt-iface",
1491 "ip-address": "to override",
1492 "mac-address": "mac_address",
1493 "vim-network-name": "mgmt",
1496 "name": "pdu-iface-internal",
1499 "ip-address": "to override",
1500 "mac-address": "mac_address",
1501 "vim-network-name": "pdu_internal", # OSMNBITEST-PDU-pdu_internal
1505 self
.vnfd_filenames
= ("hackfest_3charmed_vnfd.tar.gz", "hackfest_3charmed_vnfd.tar.gz")
1507 self
.descriptor_edit
= {
1511 "short-name": "hfn1",
1514 "pdu-type": "PDU-TYPE-1",
1516 "$[0]": {"name": "mgmt-iface"},
1517 "$[1]": {"name": "pdu-iface-internal"},
1523 "constituent-vnfd": {
1524 "$[1]": {"vnfd-id-ref": "hfnd1"}
1527 "$[0]": {"vnfd-connection-point-ref": {"$[1]": {"vnfd-id-ref": "hfnd1"}}},
1528 "$[1]": {"vnfd-connection-point-ref": {"$[1]": {"vnfd-id-ref": "hfnd1"}}}
1533 def create_descriptors(self
, engine
):
1534 super().create_descriptors(engine
)
1537 self
.pdu_descriptor
["interfaces"][0].update(self
.pdu_interface_0
)
1538 self
.pdu_descriptor
["interfaces"][1].update(self
.pdu_interface_1
)
1539 self
.pdu_descriptor
["vim_accounts"] = [self
.vim_id
]
1540 # TODO get vim-network-name from vnfr.vld.name
1541 self
.pdu_descriptor
["interfaces"][1]["vim-network-name"] = "{}-{}-{}".format(
1542 os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST"),
1543 "PDU", self
.pdu_descriptor
["interfaces"][1]["vim-network-name"])
1544 engine
.test("Onboard PDU descriptor", "POST", "/pdu/v1/pdu_descriptors",
1545 {"Location": "/pdu/v1/pdu_descriptors/", "Content-Type": "application/yaml"}, self
.pdu_descriptor
,
1546 201, r_header_yaml
, "yaml")
1547 self
.pdu_id
= engine
.last_id
1549 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1550 engine
.get_autorization()
1551 engine
.set_test_name(self
.test_name
)
1552 nsname
= os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
1554 # create real VIM if not exist
1555 self
.vim_id
= engine
.get_create_vim(test_osm
)
1557 self
.pduDeploy
.create_descriptors(engine
)
1558 self
.pduDeploy
.instantiate(engine
, {"nsDescription": "to be used as PDU", "nsName": nsname
+ "-PDU",
1559 "nsdId": self
.pduDeploy
.nsd_id
, "vimAccountId": self
.vim_id
})
1561 input('VNF to be used as PDU has been deployed. Perform manual check and press enter to resume')
1563 self
.pduDeploy
.test_ns(engine
, test_osm
, self
.pduDeploy
.cmds
, self
.pduDeploy
.uss
, self
.pduDeploy
.pss
,
1564 self
.pduDeploy
.keys
, self
.pduDeploy
.timeout
)
1567 r
= engine
.test("Get VNFR to obtain IP_ADDRESS", "GET",
1568 "/nslcm/v1/vnfrs?nsr-id-ref={}".format(self
.pduDeploy
.ns_id
), headers_json
, None,
1569 200, r_header_json
, "json")
1572 vnfr_data
= r
.json()
1575 self
.pdu_interface_0
["ip-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][0].get("ip-address")
1576 self
.pdu_interface_1
["ip-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][1].get("ip-address")
1577 self
.pdu_interface_0
["mac-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][0].get("mac-address")
1578 self
.pdu_interface_1
["mac-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][1].get("mac-address")
1579 if not self
.pdu_interface_0
["ip-address"]:
1580 raise TestException("Vnfr has not managment ip address")
1582 self
.pdu_interface_0
["ip-address"] = "192.168.10.10"
1583 self
.pdu_interface_1
["ip-address"] = "192.168.11.10"
1584 self
.pdu_interface_0
["mac-address"] = "52:33:44:55:66:13"
1585 self
.pdu_interface_1
["mac-address"] = "52:33:44:55:66:14"
1587 self
.create_descriptors(engine
)
1589 ns_data
= {"nsDescription": "default description", "nsName": nsname
, "nsdId": self
.nsd_id
,
1590 "vimAccountId": self
.vim_id
}
1591 if test_params
and test_params
.get("ns-config"):
1592 if isinstance(test_params
["ns-config"], str):
1593 ns_data
.update(yaml
.load(test_params
["ns-config"]))
1595 ns_data
.update(test_params
["ns-config"])
1597 self
.instantiate(engine
, ns_data
)
1599 input('NS has been deployed. Perform manual check and press enter to resume')
1601 self
.test_ns(engine
, test_osm
, self
.cmds
, self
.uss
, self
.pss
, self
.keys
, self
.timeout
)
1602 self
.aditional_operations(engine
, test_osm
, manual_check
)
1603 self
.terminate(engine
)
1604 self
.pduDeploy
.terminate(engine
)
1605 self
.delete_descriptors(engine
)
1606 self
.pduDeploy
.delete_descriptors(engine
)
1608 def delete_descriptors(self
, engine
):
1609 super().delete_descriptors(engine
)
1611 engine
.test("Delete PDU SOL005", "DELETE",
1612 "/pdu/v1/pdu_descriptors/{}".format(self
.pdu_id
),
1613 headers_yaml
, None, 204, None, 0)
1616 class TestDescriptors
:
1617 description
= "Test VNFD, NSD, PDU descriptors CRUD and dependencies"
1620 self
.vnfd_filename
= "hackfest_3charmed_vnfd.tar.gz"
1621 self
.nsd_filename
= "hackfest_3charmed_nsd.tar.gz"
1622 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
1625 self
.vnfd_empty
= """vnfd:vnfd-catalog:
1631 self
.vnfd_prova
= """vnfd:vnfd-catalog:
1643 - external-connection-point-ref: cp_0h8m
1651 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1652 engine
.set_test_name("Descriptors")
1653 engine
.get_autorization()
1654 temp_dir
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/temp/"
1655 if not os
.path
.exists(temp_dir
):
1656 os
.makedirs(temp_dir
)
1659 for filename
in (self
.vnfd_filename
, self
.nsd_filename
):
1660 filename_path
= temp_dir
+ filename
1661 if not os
.path
.exists(filename_path
):
1662 with
open(filename_path
, "wb") as file:
1663 response
= requests
.get(self
.descriptor_url
+ filename
)
1664 if response
.status_code
>= 300:
1665 raise TestException("Error downloading descriptor from '{}': {}".format(
1666 self
.descriptor_url
+ filename
, response
.status_code
))
1667 file.write(response
.content
)
1669 vnfd_filename_path
= temp_dir
+ self
.vnfd_filename
1670 nsd_filename_path
= temp_dir
+ self
.nsd_filename
1672 engine
.test("Onboard empty VNFD in one step", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml
,
1673 self
.vnfd_empty
, 201, r_headers_yaml_location_vnfd
, "yaml")
1674 self
.vnfd_id
= engine
.last_id
1677 engine
.test("Upload invalid VNFD ", "PUT", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self
.vnfd_id
),
1678 headers_yaml
, self
.vnfd_prova
, 422, r_header_yaml
, "yaml")
1680 engine
.test("Upload VNFD {}".format(self
.vnfd_filename
), "PUT",
1681 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self
.vnfd_id
), headers_zip_yaml
,
1682 "@b" + vnfd_filename_path
, 204, None, 0)
1685 engine
.test("Upload invalid VNFD ", "PUT", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self
.vnfd_id
),
1686 headers_yaml
, self
.vnfd_prova
, 422, r_header_yaml
, "yaml")
1688 # get vnfd descriptor
1689 engine
.test("Get VNFD descriptor", "GET", "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_id
),
1690 headers_yaml
, None, 200, r_header_yaml
, "yaml")
1692 # get vnfd file descriptor
1693 engine
.test("Get VNFD file descriptor", "GET", "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(self
.vnfd_id
),
1694 headers_text
, None, 200, r_header_text
, "text", temp_dir
+"vnfd-yaml")
1695 # TODO compare files: diff vnfd-yaml hackfest_3charmed_vnfd/hackfest_3charmed_vnfd.yaml
1697 # get vnfd zip file package
1698 engine
.test("Get VNFD zip package", "GET",
1699 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self
.vnfd_id
), headers_zip
, None, 200,
1700 r_header_zip
, "zip", temp_dir
+"vnfd-zip")
1701 # TODO compare files: diff vnfd-zip hackfest_3charmed_vnfd.tar.gz
1704 engine
.test("Get VNFD artifact package", "GET",
1705 "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/osm.png".format(self
.vnfd_id
), headers_zip
, None, 200,
1706 r_header_octect
, "octet-string", temp_dir
+"vnfd-icon")
1707 # TODO compare files: diff vnfd-icon hackfest_3charmed_vnfd/icons/osm.png
1709 # nsd CREATE AND UPLOAD in one step:
1710 engine
.test("Onboard NSD in one step", "POST", "/nsd/v1/ns_descriptors_content", headers_zip_yaml
,
1711 "@b" + nsd_filename_path
, 201, r_headers_yaml_location_nsd
, "yaml")
1712 self
.nsd_id
= engine
.last_id
1714 # get nsd descriptor
1715 engine
.test("Get NSD descriptor", "GET", "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
), headers_yaml
,
1716 None, 200, r_header_yaml
, "yaml")
1718 # get nsd file descriptor
1719 engine
.test("Get NSD file descriptor", "GET", "/nsd/v1/ns_descriptors/{}/nsd".format(self
.nsd_id
), headers_text
,
1720 None, 200, r_header_text
, "text", temp_dir
+"nsd-yaml")
1721 # TODO compare files: diff nsd-yaml hackfest_3charmed_nsd/hackfest_3charmed_nsd.yaml
1723 # get nsd zip file package
1724 engine
.test("Get NSD zip package", "GET", "/nsd/v1/ns_descriptors/{}/nsd_content".format(self
.nsd_id
),
1725 headers_zip
, None, 200, r_header_zip
, "zip", temp_dir
+"nsd-zip")
1726 # TODO compare files: diff nsd-zip hackfest_3charmed_nsd.tar.gz
1729 engine
.test("Get NSD artifact package", "GET",
1730 "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm.png".format(self
.nsd_id
), headers_zip
, None, 200,
1731 r_header_octect
, "octet-string", temp_dir
+"nsd-icon")
1732 # TODO compare files: diff nsd-icon hackfest_3charmed_nsd/icons/osm.png
1735 test_rest
.test("Delete VNFD conflict", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_id
),
1736 headers_yaml
, None, 409, None, None)
1738 test_rest
.test("Delete VNFD force", "DELETE", "/vnfpkgm/v1/vnf_packages/{}?FORCE=TRUE".format(self
.vnfd_id
),
1739 headers_yaml
, None, 204, None, 0)
1742 test_rest
.test("Delete NSD", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
), headers_yaml
, None, 204,
1746 class TestNetSliceTemplates
:
1747 description
= "Upload a NST to OSM"
1750 self
.nst_filenames
= ("@./cirros_slice/cirros_slice_vld.yaml")
1752 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1754 engine
.set_test_name("NST")
1755 engine
.get_autorization()
1756 engine
.test("Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml
, self
.nst_filenames
,
1757 201, r_headers_yaml_location_nst
, "yaml")
1758 nst_id
= engine
.last_id
1760 # nstd SHOW OSM format
1761 engine
.test("Show NSTD OSM format", "GET", "/nst/v1/netslice_templates/{}".format(nst_id
), headers_json
, None,
1762 200, r_header_json
, "json")
1765 engine
.test("Delete NSTD", "DELETE", "/nst/v1/netslice_templates/{}".format(nst_id
), headers_json
, None,
1769 class TestNetSliceInstances
:
1770 description
= "Upload a NST to OSM"
1774 self
.nst_filenames
= ("@./cirros_slice/cirros_slice.yaml")
1776 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1778 engine
.set_test_name("NSI")
1779 engine
.get_autorization()
1780 engine
.test("Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml
, self
.nst_filenames
, 201,
1781 r_headers_yaml_location_nst
, "yaml")
1782 nst_id
= engine
.last_id
1785 self
.vim_id
= engine
.get_create_vim(test_osm
)
1787 ns_data
= {"nsiDescription": "default description", "nsiName": "my_slice", "nstId": nst_id
,
1788 "vimAccountId": self
.vim_id
}
1789 ns_data_text
= yaml
.safe_dump(ns_data
, default_flow_style
=True, width
=256)
1791 engine
.test("Onboard NSI", "POST", "/nsilcm/v1/netslice_instances_content", headers_yaml
, ns_data_text
, 201,
1792 r_headers_yaml_location_nst
, "yaml")
1793 nsi_id
= engine
.last_id
1795 # TODO: Improve the wait with a polling if NSI was deployed
1800 engine
.test("Wait until NSI is deployed", "GET", "/nsilcm/v1/netslice_instances_content/{}".format(nsi_id
),
1801 headers_json
, None, 200, r_header_json
, "json")
1804 engine
.test("Delete NSI", "DELETE", "/nsilcm/v1/netslice_instances_content/{}".format(nsi_id
), headers_json
,
1805 None, 202, r_header_json
, "json")
1810 engine
.test("Delete NSTD", "DELETE", "/nst/v1/netslice_templates/{}".format(nst_id
), headers_json
, None,
1814 if __name__
== "__main__":
1818 # Disable warnings from self-signed certificates.
1819 requests
.packages
.urllib3
.disable_warnings()
1821 logging
.basicConfig(format
="%(levelname)s %(message)s", level
=logging
.ERROR
)
1822 logger
= logging
.getLogger('NBI')
1823 # load parameters and configuration
1824 opts
, args
= getopt
.getopt(sys
.argv
[1:], "hvu:p:",
1825 ["url=", "user=", "password=", "help", "version", "verbose", "no-verbose",
1826 "project=", "insecure", "timeout", "timeout-deploy", "timeout-configure",
1827 "test=", "list", "test-osm", "manual-check", "params=", 'fail-fast'])
1828 url
= "https://localhost:9999/osm"
1829 user
= password
= project
= "admin"
1831 manual_check
= False
1836 "NonAuthorized": TestNonAuthorized
,
1837 "FakeVIM": TestFakeVim
,
1838 "TestUsersProjects": TestUsersProjects
,
1839 "VIM-SDN": TestVIMSDN
,
1840 "Deploy-Custom": TestDeploy
,
1841 "Deploy-Hackfest-Cirros": TestDeployHackfestCirros
,
1842 "Deploy-Hackfest-Cirros-Scaling": TestDeployHackfestCirrosScaling
,
1843 "Deploy-Hackfest-3Charmed": TestDeployHackfest3Charmed
,
1844 "Deploy-Hackfest-4": TestDeployHackfest4
,
1845 "Deploy-CirrosMacIp": TestDeployIpMac
,
1846 "TestDescriptors": TestDescriptors
,
1847 "TestDeployHackfest1": TestDeployHackfest1
,
1848 # "Deploy-MultiVIM": TestDeployMultiVIM,
1849 "DeploySingleVdu": TestDeploySingleVdu
,
1850 "DeployHnfd": TestDeployHnfd
,
1851 # "Upload-Slice-Template": TestNetSliceTemplates,
1852 # "Deploy-Slice-Instance": TestNetSliceInstances,
1853 "TestDeploySimpleCharm": TestDeploySimpleCharm
,
1854 "TestDeploySimpleCharm2": TestDeploySimpleCharm2
,
1860 # print("parameter:", o, a)
1861 if o
== "--version":
1862 print("test version " + __version__
+ ' ' + version_date
)
1865 for test
, test_class
in test_classes
.items():
1866 print("{:20} {}".format(test
+ ":", test_class
.description
))
1868 elif o
in ("-v", "--verbose"):
1870 elif o
== "no-verbose":
1872 elif o
in ("-h", "--help"):
1875 elif o
== "--test-osm":
1877 elif o
== "--manual-check":
1881 elif o
in ("-u", "--user"):
1883 elif o
in ("-p", "--password"):
1885 elif o
== "--project":
1887 elif o
== "--fail-fast":
1890 # print("asdfadf", o, a, a.split(","))
1891 for _test
in a
.split(","):
1892 if _test
not in test_classes
:
1893 print("Invalid test name '{}'. Use option '--list' to show available tests".format(_test
),
1896 test_to_do
.append(_test
)
1897 elif o
== "--params":
1898 param_key
, _
, param_value
= a
.partition("=")
1899 text_index
= len(test_to_do
)
1900 if text_index
not in test_params
:
1901 test_params
[text_index
] = {}
1902 test_params
[text_index
][param_key
] = param_value
1903 elif o
== "--insecure":
1905 elif o
== "--timeout":
1907 elif o
== "--timeout-deploy":
1908 timeout_deploy
= int(a
)
1909 elif o
== "--timeout-configure":
1910 timeout_configure
= int(a
)
1912 assert False, "Unhandled option"
1914 logger
.setLevel(logging
.WARNING
)
1916 logger
.setLevel(logging
.DEBUG
)
1918 logger
.setLevel(logging
.ERROR
)
1920 test_rest
= TestRest(url
, user
=user
, password
=password
, project
=project
)
1921 # print("tests to do:", test_to_do)
1924 for test
in test_to_do
:
1925 if fail_fast
and test_rest
.failed_tests
:
1928 test_class
= test_classes
[test
]
1929 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(text_index
))
1931 for test
, test_class
in test_classes
.items():
1932 if fail_fast
and test_rest
.failed_tests
:
1934 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(0))
1935 test_rest
.print_results()
1936 exit(1 if test_rest
.failed_tests
else 0)
1938 except TestException
as e
:
1939 logger
.error(test
+ "Test {} Exception: {}".format(test
, str(e
)))
1941 except getopt
.GetoptError
as e
:
1943 print(e
, file=sys
.stderr
)
1945 except Exception as e
:
1946 logger
.critical(test
+ " Exception: " + str(e
), exc_info
=True)