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
29 from uuid
import uuid4
32 __author__
= "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
33 __date__
= "$2018-03-01$"
35 version_date
= "Oct 2018"
39 print("Usage: ", sys
.argv
[0], "[options]")
40 print(" Performs system tests over running NBI. It can be used for real OSM test using option '--test-osm'")
41 print(" If this is the case env variables 'OSMNBITEST_VIM_NAME' must be supplied to create a VIM if not exist "
42 "where deployment is done")
44 print(" -h|--help: shows this help")
45 print(" --insecure: Allows non trusted https NBI server")
46 print(" --list: list available tests")
47 print(" --manual-check: Deployment tests stop after deployed to allow manual inspection. Only make sense with "
49 print(" -p|--password PASSWORD: NBI access password. 'admin' by default")
50 print(" ---project PROJECT: NBI access project. 'admin' by default")
51 print(" --test TEST[,...]: Execute only a test or a comma separated list of tests")
52 print(" --params key=val: params to the previous test. key can be vnfd-files, nsd-file, ns-name, ns-config")
53 print(" --test-osm: If missing this test is intended for NBI only, no other OSM components are expected. Use "
54 "this flag to test the system. LCM and RO components are expected to be up and running")
55 print(" --timeout TIMEOUT: General NBI timeout, by default {}s".format(timeout
))
56 print(" --timeout-deploy TIMEOUT: Timeout used for getting NS deployed, by default {}s".format(timeout_deploy
))
57 print(" --timeout-configure TIMEOUT: Timeout used for getting NS deployed and configured,"
58 " by default {}s".format(timeout_configure
))
59 print(" -u|--user USERNAME: NBI access username. 'admin' by default")
60 print(" --url URL: complete NBI server URL. 'https//localhost:9999/osm' by default")
61 print(" -v|--verbose print debug information, can be used several times")
62 print(" --no-verbose remove verbosity")
63 print(" --version: prints current version")
64 print("ENV variables used for real deployment tests with option osm-test.")
65 print(" export OSMNBITEST_VIM_NAME=vim-name")
66 print(" export OSMNBITEST_VIM_URL=vim-url")
67 print(" export OSMNBITEST_VIM_TYPE=vim-type")
68 print(" export OSMNBITEST_VIM_TENANT=vim-tenant")
69 print(" export OSMNBITEST_VIM_USER=vim-user")
70 print(" export OSMNBITEST_VIM_PASSWORD=vim-password")
71 print(" export OSMNBITEST_VIM_CONFIG=\"vim-config\"")
72 print(" export OSMNBITEST_NS_NAME=\"vim-config\"")
76 r_header_json
= {"Content-type": "application/json"}
77 headers_json
= {"Content-type": "application/json", "Accept": "application/json"}
78 r_header_yaml
= {"Content-type": "application/yaml"}
79 headers_yaml
= {"Content-type": "application/yaml", "Accept": "application/yaml"}
80 r_header_text
= {"Content-type": "text/plain"}
81 r_header_octect
= {"Content-type": "application/octet-stream"}
82 headers_text
= {"Accept": "text/plain,application/yaml"}
83 r_header_zip
= {"Content-type": "application/zip"}
84 headers_zip
= {"Accept": "application/zip,application/yaml"}
85 headers_zip_yaml
= {"Accept": "application/yaml", "Content-type": "application/zip"}
86 headers_zip_json
= {"Accept": "application/json", "Content-type": "application/zip"}
87 headers_txt_json
= {"Accept": "application/json", "Content-type": "text/plain"}
88 r_headers_yaml_location_vnfd
= {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"}
89 r_headers_yaml_location_nsd
= {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}
90 r_headers_yaml_location_nst
= {"Location": "/nst/v1/netslice_templates_content", "Content-Type": "application/yaml"}
91 r_headers_yaml_location_nslcmop
= {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}
92 r_headers_yaml_location_nsilcmop
= {"Location": "/osm/nsilcm/v1/nsi_lcm_op_occs/", "Content-Type": "application/yaml"}
94 # test ones authorized
95 test_authorized_list
= (
96 ("AU1", "Invalid vnfd id", "GET", "/vnfpkgm/v1/vnf_packages/non-existing-id",
97 headers_json
, None, 404, r_header_json
, "json"),
98 ("AU2", "Invalid nsd id", "GET", "/nsd/v1/ns_descriptors/non-existing-id",
99 headers_yaml
, None, 404, r_header_yaml
, "yaml"),
100 ("AU3", "Invalid nsd id", "DELETE", "/nsd/v1/ns_descriptors_content/non-existing-id",
101 headers_yaml
, None, 404, r_header_yaml
, "yaml"),
103 timeout
= 120 # general timeout
104 timeout_deploy
= 60*10 # timeout for NS deploying without charms
105 timeout_configure
= 60*20 # timeout for NS deploying and configuring
108 class TestException(Exception):
113 def __init__(self
, url_base
, header_base
=None, verify
=False, user
="admin", password
="admin", project
="admin"):
114 self
.url_base
= url_base
115 if header_base
is None:
116 self
.header_base
= {}
118 self
.header_base
= header_base
.copy()
119 self
.s
= requests
.session()
120 self
.s
.headers
= self
.header_base
124 self
.password
= password
125 self
.project
= project
127 # contains ID of tests obtained from Location response header. "" key contains last obtained id
129 self
.test_name
= None
130 self
.step
= 0 # number of subtest under test
131 self
.passed_tests
= 0
132 self
.failed_tests
= 0
134 def set_test_name(self
, test_name
):
135 self
.test_name
= test_name
139 def set_header(self
, header
):
140 self
.s
.headers
.update(header
)
142 def set_tet_name(self
, test_name
):
143 self
.test_name
= test_name
145 def unset_header(self
, key
):
146 if key
in self
.s
.headers
:
147 del self
.s
.headers
[key
]
149 def test(self
, description
, method
, url
, headers
, payload
, expected_codes
, expected_headers
,
150 expected_payload
, store_file
=None, pooling
=False):
152 Performs an http request and check http code response. Exit if different than allowed. It get the returned id
153 that can be used by following test in the URL with {name} where name is the name of the test
154 :param description: description of the test
155 :param method: HTTP method: GET,PUT,POST,DELETE,...
156 :param url: complete URL or relative URL
157 :param headers: request headers to add to the base headers
158 :param payload: Can be a dict, transformed to json, a text or a file if starts with '@'
159 :param expected_codes: expected response codes, can be int, int tuple or int range
160 :param expected_headers: expected response headers, dict with key values
161 :param expected_payload: expected payload, 0 if empty, 'yaml', 'json', 'text', 'zip', 'octet-stream'
162 :param store_file: filename to store content
163 :param pooling: if True do not count neither log this test. Because a pooling is done with many equal requests
164 :return: requests response
169 self
.s
= requests
.session()
173 elif not url
.startswith("http"):
174 url
= self
.url_base
+ url
176 # replace url <> with the last ID
177 url
= url
.replace("<>", self
.last_id
)
179 if isinstance(payload
, str):
180 if payload
.startswith("@"):
182 file_name
= payload
[1:]
183 if payload
.startswith("@b"):
185 file_name
= payload
[2:]
186 with
open(file_name
, mode
) as f
:
188 elif isinstance(payload
, dict):
189 payload
= json
.dumps(payload
)
192 test_description
= "Test {}{} {} {} {}".format(self
.test_name
, self
.step
, description
, method
, url
)
193 logger
.warning(test_description
)
196 if expected_payload
in ("zip", "octet-string") or store_file
:
201 r
= getattr(self
.s
, method
.lower())(url
, data
=payload
, headers
=headers
, verify
=self
.verify
,
204 except requests
.exceptions
.ConnectionError
as e
:
207 logger
.error("Exception {}. Retrying".format(e
))
210 if expected_payload
in ("zip", "octet-string") or store_file
:
211 logger
.debug("RX {}".format(r
.status_code
))
213 logger
.debug("RX {}: {}".format(r
.status_code
, r
.text
))
217 if isinstance(expected_codes
, int):
218 expected_codes
= (expected_codes
,)
219 if r
.status_code
not in expected_codes
:
221 "Got status {}. Expected {}. {}".format(r
.status_code
, expected_codes
, r
.text
))
224 for header_key
, header_val
in expected_headers
.items():
225 if header_key
.lower() not in r
.headers
:
226 raise TestException("Header {} not present".format(header_key
))
227 if header_val
and header_val
.lower() not in r
.headers
[header_key
]:
228 raise TestException("Header {} does not contain {} but {}".format(header_key
, header_val
,
229 r
.headers
[header_key
]))
231 if expected_payload
is not None:
232 if expected_payload
== 0 and len(r
.content
) > 0:
233 raise TestException("Expected empty payload")
234 elif expected_payload
== "json":
237 except Exception as e
:
238 raise TestException("Expected json response payload, but got Exception {}".format(e
))
239 elif expected_payload
== "yaml":
241 yaml
.safe_load(r
.text
)
242 except Exception as e
:
243 raise TestException("Expected yaml response payload, but got Exception {}".format(e
))
244 elif expected_payload
in ("zip", "octet-string"):
245 if len(r
.content
) == 0:
246 raise TestException("Expected some response payload, but got empty")
248 # tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
249 # for tarinfo in tar:
250 # tarname = tarinfo.name
252 # except Exception as e:
253 # raise TestException("Expected zip response payload, but got Exception {}".format(e))
254 elif expected_payload
== "text":
255 if len(r
.content
) == 0:
256 raise TestException("Expected some response payload, but got empty")
259 with
open(store_file
, 'wb') as fd
:
260 for chunk
in r
.iter_content(chunk_size
=128):
263 location
= r
.headers
.get("Location")
265 _id
= location
[location
.rfind("/") + 1:]
267 self
.last_id
= str(_id
)
269 self
.passed_tests
+= 1
271 except TestException
as e
:
272 self
.failed_tests
+= 1
276 r_status_code
= r
.status_code
278 logger
.error("{} \nRX code{}: {}".format(e
, r_status_code
, r_text
))
283 logger
.error("Cannot open file {}: {}".format(store_file
, e
))
285 logger
.error("Exception: {}".format(e
), exc_info
=True)
286 self
.failed_tests
+= 1
289 except requests
.exceptions
.RequestException
as e
:
290 logger
.error("Exception: {}".format(e
))
292 def get_autorization(self
): # user=None, password=None, project=None):
293 if self
.token
: # and self.user == user and self.password == password and self.project == project:
296 # self.password = password
297 # self.project = project
298 r
= self
.test("Obtain token", "POST", "/admin/v1/tokens", headers_json
,
299 {"username": self
.user
, "password": self
.password
, "project_id": self
.project
},
300 (200, 201), r_header_json
, "json")
304 self
.token
= response
["id"]
305 self
.set_header({"Authorization": "Bearer {}".format(self
.token
)})
307 def remove_authorization(self
):
309 self
.test("Delete token", "DELETE", "/admin/v1/tokens/{}".format(self
.token
), headers_json
,
310 None, (200, 201, 204), None, None)
312 self
.unset_header("Authorization")
314 def get_create_vim(self
, test_osm
):
317 self
.get_autorization()
319 vim_name
= os
.environ
.get("OSMNBITEST_VIM_NAME")
322 "Needed to define OSMNBITEST_VIM_XXX variables to create a real VIM for deployment")
326 r
= self
.test("Get VIM ID", "GET", "/admin/v1/vim_accounts?name={}".format(vim_name
), headers_json
,
327 None, 200, r_header_json
, "json")
332 return vims
[0]["_id"]
335 # check needed environ parameters:
336 if not os
.environ
.get("OSMNBITEST_VIM_URL") or not os
.environ
.get("OSMNBITEST_VIM_TENANT"):
337 raise TestException("Env OSMNBITEST_VIM_URL and OSMNBITEST_VIM_TENANT are needed for create a real VIM"
338 " to deploy on whit the --test-osm option")
339 vim_data
= "{{schema_version: '1.0', name: '{}', vim_type: {}, vim_url: '{}', vim_tenant_name: '{}', "\
340 "vim_user: {}, vim_password: {}".format(vim_name
,
341 os
.environ
.get("OSMNBITEST_VIM_TYPE", "openstack"),
342 os
.environ
.get("OSMNBITEST_VIM_URL"),
343 os
.environ
.get("OSMNBITEST_VIM_TENANT"),
344 os
.environ
.get("OSMNBITEST_VIM_USER"),
345 os
.environ
.get("OSMNBITEST_VIM_PASSWORD"))
346 if os
.environ
.get("OSMNBITEST_VIM_CONFIG"):
347 vim_data
+= " ,config: {}".format(os
.environ
.get("OSMNBITEST_VIM_CONFIG"))
350 vim_data
= "{schema_version: '1.0', name: fakeVim, vim_type: openstack, vim_url: 'http://10.11.12.13/fake'"\
351 ", vim_tenant_name: 'vimtenant', vim_user: vimuser, vim_password: vimpassword}"
352 self
.test("Create VIM", "POST", "/admin/v1/vim_accounts", headers_yaml
, vim_data
,
353 (201, 202), {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/yaml"}, "yaml")
356 def print_results(self
):
357 print("\n\n\n--------------------------------------------")
358 print("TEST RESULTS: Total: {}, Passed: {}, Failed: {}".format(self
.passed_tests
+ self
.failed_tests
,
359 self
.passed_tests
, self
.failed_tests
))
360 print("--------------------------------------------")
362 def wait_until_delete(self
, url_op
, timeout_delete
):
364 Make a pooling until topic is not present, because of deleted
366 :param timeout_delete:
369 description
= "Wait to topic being deleted"
370 test_description
= "Test {}{} {} {} {}".format(self
.test_name
, self
.step
, description
, "GET", url_op
)
371 logger
.warning(test_description
)
374 wait
= timeout_delete
376 r
= self
.test(description
, "GET", url_op
, headers_yaml
, None, (200, 404), None, r_header_yaml
, "yaml",
380 if r
.status_code
== 404:
381 self
.passed_tests
+= 1
383 elif r
.status_code
== 200:
387 raise TestException("Topic is not deleted after {} seconds".format(timeout_delete
))
388 self
.failed_tests
+= 1
390 def wait_operation_ready(self
, ns_nsi
, opp_id
, timeout
, expected_fail
=False):
392 Wait until nslcmop or nsilcmop finished
393 :param ns_nsi: "ns" o "nsi"
394 :param opp_id: Id o fthe operation
396 :param expected_fail:
397 :return: None. Updates passed/failed_tests
400 url_op
= "/nslcm/v1/ns_lcm_op_occs/{}".format(opp_id
)
402 url_op
= "/nsilcm/v1/nsi_lcm_op_occs/{}".format(opp_id
)
403 description
= "Wait to {} lcm operation complete".format(ns_nsi
)
404 test_description
= "Test {}{} {} {} {}".format(self
.test_name
, self
.step
, description
, "GET", url_op
)
405 logger
.warning(test_description
)
409 r
= self
.test(description
, "GET", url_op
, headers_json
, None,
410 200, r_header_json
, "json", pooling
=True)
414 if "COMPLETED" in nslcmop
["operationState"]:
416 logger
.error("NS terminate has success, expecting failing: {}".format(nslcmop
["detailed-status"]))
417 self
.failed_tests
+= 1
419 self
.passed_tests
+= 1
421 elif "FAILED" in nslcmop
["operationState"]:
422 if not expected_fail
:
423 logger
.error("NS terminate has failed: {}".format(nslcmop
["detailed-status"]))
424 self
.failed_tests
+= 1
426 self
.passed_tests
+= 1
429 print(".", end
="", file=stderr
)
433 self
.failed_tests
+= 1
434 logger
.error("NS instantiate is not terminate after {} seconds".format(timeout
))
436 print("", file=stderr
)
439 class TestNonAuthorized
:
440 description
= "Test invalid URLs. methods and no authorization"
443 def run(engine
, test_osm
, manual_check
, test_params
=None):
444 engine
.set_test_name("NonAuth")
445 engine
.remove_authorization()
446 test_not_authorized_list
= (
447 ("Invalid token", "GET", "/admin/v1/users", headers_json
, None, 401, r_header_json
, "json"),
448 ("Invalid URL", "POST", "/admin/v1/nonexist", headers_yaml
, None, 405, r_header_yaml
, "yaml"),
449 ("Invalid version", "DELETE", "/admin/v2/users", headers_yaml
, None, 405, r_header_yaml
, "yaml"),
451 for t
in test_not_authorized_list
:
455 class TestUsersProjects
:
456 description
= "test project and user creation"
459 def run(engine
, test_osm
, manual_check
, test_params
=None):
460 engine
.set_test_name("UserProject")
461 # backend = test_params.get("backend") if test_params else None # UNUSED
466 u1
= u2
= u3
= u4
= None
468 engine
.get_autorization()
470 res
= engine
.test("Create project non admin 1", "POST", "/admin/v1/projects", headers_json
, {"name": "P1"},
471 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
472 p1
= engine
.last_id
if res
else None
474 res
= engine
.test("Create project admin", "POST", "/admin/v1/projects", headers_json
,
475 {"name": "Padmin", "admin": True}, (201, 204),
476 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
477 padmin
= engine
.last_id
if res
else None
479 res
= engine
.test("Create project bad format", "POST", "/admin/v1/projects", headers_json
, {"name": 1},
480 (400, 422), r_header_json
, "json")
481 pbad
= engine
.last_id
if res
else None
483 res
= engine
.test("Get project admin role", "GET", "/admin/v1/roles?name=project_admin", headers_json
, {},
484 (200), {"Content-Type": "application/json"}, "json")
485 rpa
= res
.json()[0]["_id"] if res
else None
486 res
= engine
.test("Get project user role", "GET", "/admin/v1/roles?name=project_user", headers_json
, {},
487 (200), {"Content-Type": "application/json"}, "json")
488 rpu
= res
.json()[0]["_id"] if res
else None
489 res
= engine
.test("Get system admin role", "GET", "/admin/v1/roles?name=system_admin", headers_json
, {},
490 (200), {"Content-Type": "application/json"}, "json")
491 rsa
= res
.json()[0]["_id"] if res
else None
493 data
= {"username": "U1", "password": "pw1"}
495 data
["project_role_mappings"] = [
496 {"project": p1
, "role": rpa
},
497 {"project": p2
, "role": rpa
},
498 {"project": padmin
, "role": rpu
}
501 xhd
= {"Location": "/admin/v1/users/", "Content-Type": "application/json"}
502 res
= engine
.test("Create user with bad project and force", "POST", "/admin/v1/users?FORCE=True", headers_json
,
503 data
, rc
, xhd
, "json")
507 # User is created sometimes even though an exception is raised
508 res
= engine
.test("Get user U1", "GET", "/admin/v1/users?username=U1", headers_json
, {},
509 (200), {"Content-Type": "application/json"}, "json")
510 u1
= res
.json()[0]["_id"] if res
else None
512 data
= {"username": "U2", "password": "pw2"}
513 data
["project_role_mappings"] = [{"project": p1
, "role": rpa
}, {"project": padmin
, "role": rsa
}]
514 res
= engine
.test("Create user 2", "POST", "/admin/v1/users", headers_json
,
515 data
, 201, {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
516 u2
= engine
.last_id
if res
else None
519 ftt
= "project_role_mappings"
520 xpr
= [{"project": p1
, "role": rpa
}, {"project": padmin
, "role": rpu
}]
522 engine
.test("Edit user U1, delete P2 project", "PATCH", "/admin/v1/users/"+u1
, headers_json
,
523 data
, 204, None, None)
524 res
= engine
.test("Check user U1, contains the right projects", "GET", "/admin/v1/users/"+u1
,
525 headers_json
, None, 200, None, json
)
528 xpr
[0]["project_name"] = "P1"
529 xpr
[0]["role_name"] = "project_admin"
530 xpr
[1]["project_name"] = "Padmin"
531 xpr
[1]["role_name"] = "project_user"
537 if pr
not in rj
[ftt
]:
540 logger
.error("User {} '{}' are different than expected '{}'. Edition was not done properly"
541 .format(ftt
, rj
[ftt
], xpr
))
542 engine
.failed_tests
+= 1
544 p2
= None # To prevent deletion attempts
546 # Add a test of 'default project' for Keystone?
549 engine
.test("Edit user U2, change password", "PUT", "/admin/v1/users/"+u2
, headers_json
,
550 {"password": "pw2_new"}, 204, None, None)
553 engine
.test("Change to project P1 non existing", "POST", "/admin/v1/tokens/", headers_json
,
554 {"project_id": p1
}, 401, r_header_json
, "json")
557 res
= engine
.test("Change to user U2 project P1", "POST", "/admin/v1/tokens", headers_json
,
558 {"username": "U2", "password": "pw2_new", "project_id": "P1"}, (200, 201),
559 r_header_json
, "json")
562 engine
.set_header({"Authorization": "Bearer {}".format(rj
["id"])})
564 engine
.test("Edit user projects non admin", "PUT", "/admin/v1/users/U1", headers_json
,
565 {"remove_project_role_mappings": [{"project": "P1", "role": None}]},
566 401, r_header_json
, "json")
568 res
= engine
.test("Add new project non admin", "POST", "/admin/v1/projects", headers_json
,
569 {"name": "P2"}, 401, r_header_json
, "json")
570 if res
is None or res
.status_code
== 201:
571 # The project has been created even though it shouldn't
572 res
= engine
.test("Get project P2", "GET", "/admin/v1/projects/P2", headers_json
, None,
573 200, r_header_json
, "json")
574 p2
= res
.json()["_id"] if res
else None
577 data
= {"username": "U3", "password": "pw3"}
578 data
["project_role_mappings"] = [{"project": p1
, "role": rpu
}]
579 res
= engine
.test("Add new user non admin", "POST", "/admin/v1/users", headers_json
,
580 data
, 401, r_header_json
, "json")
581 if res
is None or res
.status_code
== 201:
582 # The user has been created even though it shouldn't
583 res
= engine
.test("Get user U3", "GET", "/admin/v1/users/U3", headers_json
, None,
584 200, r_header_json
, "json")
585 u3
= res
.json()["_id"] if res
else None
590 res
= engine
.test("Change to user U2 project Padmin", "POST", "/admin/v1/tokens", headers_json
,
591 {"project_id": "Padmin"}, # Caused a Keystone authentication error
592 # {"username": "U2", "password": "pw2_new", "project_id": "Padmin"},
593 (200, 201), r_header_json
, "json")
596 engine
.set_header({"Authorization": "Bearer {}".format(rj
["id"])})
598 res
= engine
.test("Add new project admin", "POST", "/admin/v1/projects", headers_json
,
599 {"name": "P3"}, (201, 204),
600 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
602 p3
= engine
.last_id
if res
else None
605 data
= {"username": "U4", "password": "pw4"}
606 data
["project_role_mappings"] = [{"project": p1
, "role": rpa
}]
607 res
= engine
.test("Add new user admin", "POST", "/admin/v1/users", headers_json
,
609 {"Location": "/admin/v1/users/", "Content-Type": "application/json"},
611 u4
= engine
.last_id
if res
else None
616 data
= {"project_role_mappings": [{"project": p3
, "role": rpa
}]}
617 engine
.test("Edit user projects admin", "PUT", "/admin/v1/users/U4", headers_json
,
618 data
, 204, None, None)
619 # Project is deleted even though it shouldn't - PROVISIONAL?
620 res
= engine
.test("Delete project P3 conflict", "DELETE", "/admin/v1/projects/"+p3
,
621 headers_json
, None, 409, None, None)
622 if res
and res
.status_code
in (200, 204):
625 res
= engine
.test("Delete project P3 forcing", "DELETE",
626 "/admin/v1/projects/"+p3
+"?FORCE=True", headers_json
, None, 204,
628 if res
and res
.status_code
in (200, 204):
632 res
= engine
.test("Delete user U2. Conflict deleting own user", "DELETE",
633 "/admin/v1/users/"+u2
, headers_json
, None, 409, r_header_json
, "json")
634 if res
is None or res
.status_code
in (200, 204):
637 res
= engine
.test("Delete user U4", "DELETE", "/admin/v1/users/"+u4
, headers_json
, None,
639 if res
and res
.status_code
in (200, 204):
642 res
= engine
.test("Delete project P3", "DELETE", "/admin/v1/projects/"+p3
, headers_json
,
643 None, 204, None, None)
644 if res
and res
.status_code
in (200, 204):
648 res
= engine
.test("Delete user U3", "DELETE", "/admin/v1/users/"+u3
, headers_json
, None,
654 engine
.remove_authorization() # To force get authorization
655 engine
.get_autorization()
657 engine
.test("Delete user U1", "DELETE", "/admin/v1/users/"+u1
, headers_json
, None, 204, None, None)
659 engine
.test("Delete user U2", "DELETE", "/admin/v1/users/"+u2
, headers_json
, None, 204, None, None)
661 engine
.test("Delete user U3", "DELETE", "/admin/v1/users/"+u3
, headers_json
, None, 204, None, None)
663 engine
.test("Delete user U4", "DELETE", "/admin/v1/users/"+u4
, headers_json
, None, 204, None, None)
665 engine
.test("Delete project P1", "DELETE", "/admin/v1/projects/"+p1
, headers_json
, None, 204, None, None)
667 engine
.test("Delete project P2", "DELETE", "/admin/v1/projects/"+p2
, headers_json
, None, 204, None, None)
669 engine
.test("Delete project P3", "DELETE", "/admin/v1/projects/"+p3
, headers_json
, None, 204, None, None)
671 engine
.test("Delete project Padmin", "DELETE", "/admin/v1/projects/"+padmin
, headers_json
, None, 204,
674 engine
.test("Delete bad project", "DELETE", "/admin/v1/projects/"+pbad
, headers_json
, None, 204,
677 # BEGIN New Tests - Addressing Projects/Users by Name/ID
680 res
= engine
.test("Create new project P1", "POST", "/admin/v1/projects", headers_json
, {"name": "P1"},
681 201, {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
683 pid1
= res
.json()["id"]
684 # print("# pid =", pid1)
685 res
= engine
.test("Create new project P2", "POST", "/admin/v1/projects", headers_json
, {"name": "P2"},
686 201, {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
688 pid2
= res
.json()["id"]
689 # print("# pid =", pid2)
690 data
= {"username": "U1", "password": "pw1"}
691 data
["project_role_mappings"] = [{"project": pid1
, "role": rpu
}]
692 res
= engine
.test("Create new user U1", "POST", "/admin/v1/users", headers_json
, data
, 201,
693 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
695 uid1
= res
.json()["id"]
696 # print("# uid =", uid1)
697 data
= {"username": "U2", "password": "pw2"}
698 data
["project_role_mappings"] = [{"project": pid2
, "role": rpu
}]
699 res
= engine
.test("Create new user U2", "POST", "/admin/v1/users", headers_json
, data
, 201,
700 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
702 uid2
= res
.json()["id"]
703 # print("# uid =", uid2)
705 engine
.test("Get Project P1 by Name", "GET", "/admin/v1/projects/P1", headers_json
, None,
707 engine
.test("Get Project P1 by ID", "GET", "/admin/v1/projects/"+pid1
, headers_json
, None,
710 engine
.test("Get User U1 by Name", "GET", "/admin/v1/users/U1", headers_json
, None, 200, None, "json")
711 engine
.test("Get User U1 by ID", "GET", "/admin/v1/users/"+uid1
, headers_json
, None, 200, None, "json")
713 res
= engine
.test("Rename Project P1 by Name", "PUT", "/admin/v1/projects/P1", headers_json
,
714 {"name": "P3"}, 204, None, None)
716 engine
.test("Get Project P1 by new Name", "GET", "/admin/v1/projects/P3", headers_json
, None,
719 res
= engine
.test("Rename Project P2 by ID", "PUT", "/admin/v1/projects/"+pid2
, headers_json
,
720 {"name": "P4"}, 204, None, None)
722 engine
.test("Get Project P2 by new Name", "GET", "/admin/v1/projects/P4", headers_json
, None,
726 res
= engine
.test("Rename User U1 by Name", "PUT", "/admin/v1/users/U1", headers_json
,
727 {"username": "U3"}, 204, None, None)
729 engine
.test("Get User U1 by new Name", "GET", "/admin/v1/users/U3", headers_json
, None,
733 res
= engine
.test("Rename User U2 by ID", "PUT", "/admin/v1/users/"+uid2
, headers_json
,
734 {"username": "U4"}, 204, None, None)
736 engine
.test("Get User U2 by new Name", "GET", "/admin/v1/users/U4", headers_json
, None,
739 res
= engine
.test("Delete User U1 by Name", "DELETE", "/admin/v1/users/U3", headers_json
, None,
745 res
= engine
.test("Delete User U2 by ID", "DELETE", "/admin/v1/users/"+uid2
, headers_json
, None,
751 res
= engine
.test("Delete Project P1 by Name", "DELETE", "/admin/v1/projects/P3", headers_json
, None,
757 res
= engine
.test("Delete Project P2 by ID", "DELETE", "/admin/v1/projects/"+pid2
, headers_json
, None,
762 # END New Tests - Addressing Projects/Users by Name
766 engine
.test("Delete Project P1", "DELETE", "/admin/v1/projects/"+pid1
, headers_json
, None, 204, None, None)
768 engine
.test("Delete Project P2", "DELETE", "/admin/v1/projects/"+pid2
, headers_json
, None, 204, None, None)
770 engine
.test("Delete User U1", "DELETE", "/admin/v1/users/"+uid1
, headers_json
, None, 204, None, None)
772 engine
.test("Delete User U2", "DELETE", "/admin/v1/users/"+uid2
, headers_json
, None, 204, None, None)
774 engine
.remove_authorization() # To finish
777 class TestProjectsDescriptors
:
778 description
= "test descriptors visibility among projects"
781 def run(engine
, test_osm
, manual_check
, test_params
=None):
783 engine
.set_test_name("ProjectDescriptors")
784 engine
.get_autorization()
786 project_admin_id
= None
787 res
= engine
.test("Get my project Padmin", "GET", "/admin/v1/projects/{}".format(engine
.project
), headers_json
,
788 None, 200, r_header_json
, "json")
790 response
= res
.json()
791 project_admin_id
= response
["_id"]
792 engine
.test("Create project Padmin", "POST", "/admin/v1/projects", headers_json
,
793 {"name": "Padmin", "admin": True}, (201, 204),
794 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
795 engine
.test("Create project P2", "POST", "/admin/v1/projects", headers_json
, {"name": "P2"},
796 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
797 engine
.test("Create project P3", "POST", "/admin/v1/projects", headers_json
, {"name": "P3"},
798 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
800 engine
.test("Create user U1", "POST", "/admin/v1/users", headers_json
,
801 {"username": "U1", "password": "pw1",
802 "project_role_mappings": [{"project": "Padmin", "role": "system_admin"},
803 {"project": "P2", "role": "project_admin"},
804 {"project": "P3", "role": "project_admin"}],
805 }, 201, {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
807 engine
.test("Onboard VNFD id1", "POST", "/vnfpkgm/v1/vnf_packages_content?id=id1", headers_yaml
,
808 TestDescriptors
.vnfd_empty
, 201, r_headers_yaml_location_vnfd
, "yaml")
809 vnfd_ids
.append(engine
.last_id
)
810 engine
.test("Onboard VNFD id2 PUBLIC", "POST", "/vnfpkgm/v1/vnf_packages_content?id=id2&PUBLIC=TRUE",
811 headers_yaml
, TestDescriptors
.vnfd_empty
, 201, r_headers_yaml_location_vnfd
, "yaml")
812 vnfd_ids
.append(engine
.last_id
)
813 engine
.test("Onboard VNFD id3", "POST", "/vnfpkgm/v1/vnf_packages_content?id=id3&PUBLIC=FALSE", headers_yaml
,
814 TestDescriptors
.vnfd_empty
, 201, r_headers_yaml_location_vnfd
, "yaml")
815 vnfd_ids
.append(engine
.last_id
)
817 res
= engine
.test("Get VNFD descriptors", "GET", "/vnfpkgm/v1/vnf_packages?id=id1,id2,id3",
818 headers_json
, None, 200, r_header_json
, "json")
819 response
= res
.json()
820 if len(response
) != 3:
821 logger
.error("Only 3 vnfds should be present for project admin. {} listed".format(len(response
)))
822 engine
.failed_tests
+= 1
824 # Change to other project Padmin
825 res
= engine
.test("Change to user U1 project Padmin", "POST", "/admin/v1/tokens", headers_json
,
826 {"username": "U1", "password": "pw1", "project_id": "Padmin"}, (200, 201),
827 r_header_json
, "json")
829 response
= res
.json()
830 engine
.set_header({"Authorization": "Bearer {}".format(response
["id"])})
833 res
= engine
.test("List VNFD descriptors for Padmin", "GET", "/vnfpkgm/v1/vnf_packages",
834 headers_json
, None, 200, r_header_json
, "json")
835 response
= res
.json()
836 if len(response
) != 0:
837 logger
.error("Only 0 vnfds should be present for project Padmin. {} listed".format(len(response
)))
838 engine
.failed_tests
+= 1
841 res
= engine
.test("List VNFD public descriptors", "GET", "/vnfpkgm/v1/vnf_packages?PUBLIC=True",
842 headers_json
, None, 200, r_header_json
, "json")
843 response
= res
.json()
844 if len(response
) != 1:
845 logger
.error("Only 1 vnfds should be present for project Padmin. {} listed".format(len(response
)))
846 engine
.failed_tests
+= 1
848 # list vnfds belonging to project "admin"
849 res
= engine
.test("List VNFD of admin project", "GET",
850 "/vnfpkgm/v1/vnf_packages?ADMIN={}".format(project_admin_id
),
851 headers_json
, None, 200, r_header_json
, "json")
853 response
= res
.json()
854 if len(response
) != 3:
855 logger
.error("Only 3 vnfds should be present for project Padmin. {} listed".format(len(response
)))
856 engine
.failed_tests
+= 1
859 engine
.test("Get VNFD public descriptors", "GET", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids
[1]),
860 headers_json
, None, 200, r_header_json
, "json")
861 # Edit not owned vnfd
862 engine
.test("Edit VNFD ", "PATCH", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids
[0]),
863 headers_yaml
, '{name: pepe}', 404, r_header_yaml
, "yaml")
866 engine
.test("Add VNFD id2 to my catalog", "PATCH", "/vnfpkgm/v1/vnf_packages/{}?SET_PROJECT".
867 format(vnfd_ids
[1]), headers_json
, None, 204, None, 0)
870 engine
.test("Onboard VNFD id4", "POST", "/vnfpkgm/v1/vnf_packages_content?id=id4", headers_yaml
,
871 TestDescriptors
.vnfd_empty
, 201, r_headers_yaml_location_vnfd
, "yaml")
872 vnfd_ids
.append(engine
.last_id
)
875 res
= engine
.test("List VNFD public descriptors", "GET", "/vnfpkgm/v1/vnf_packages",
876 headers_json
, None, 200, r_header_json
, "json")
877 response
= res
.json()
878 if len(response
) != 2:
879 logger
.error("Only 2 vnfds should be present for project Padmin. {} listed".format(len(response
)))
880 engine
.failed_tests
+= 1
883 input('VNFDs have been omboarded. Perform manual check and press enter to resume')
885 test_rest
.test("Delete VNFD id2", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids
[1]),
886 headers_yaml
, None, 204, None, 0)
888 # change to admin project
889 engine
.remove_authorization() # To force get authorization
890 engine
.get_autorization()
891 test_rest
.test("Delete VNFD id1", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids
[0]),
892 headers_yaml
, None, 204, None, 0)
893 test_rest
.test("Delete VNFD id2", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids
[1]),
894 headers_yaml
, None, 204, None, 0)
895 test_rest
.test("Delete VNFD id3", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids
[2]),
896 headers_yaml
, None, 204, None, 0)
897 test_rest
.test("Delete VNFD id4", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids
[3]),
898 headers_yaml
, None, 404, r_header_yaml
, "yaml")
899 test_rest
.test("Delete VNFD id4", "DELETE", "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids
[3]),
900 headers_yaml
, None, 204, None, 0)
902 engine
.test("Get VNFD deleted id1", "GET", "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids
[0]),
903 headers_json
, None, 404, r_header_json
, "json")
904 engine
.test("Get VNFD deleted id2", "GET", "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids
[1]),
905 headers_json
, None, 404, r_header_json
, "json")
906 engine
.test("Get VNFD deleted id3", "GET", "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids
[2]),
907 headers_json
, None, 404, r_header_json
, "json")
908 engine
.test("Get VNFD deleted id4", "GET", "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids
[3]),
909 headers_json
, None, 404, r_header_json
, "json")
911 engine
.test("Delete user U1", "DELETE", "/admin/v1/users/U1", headers_json
, None, 204, None, None)
912 engine
.test("Delete project Padmin", "DELETE", "/admin/v1/projects/Padmin", headers_json
, None, 204, None, None)
913 engine
.test("Delete project P2", "DELETE", "/admin/v1/projects/P2", headers_json
, None, 204, None, None)
914 engine
.test("Delete project P3", "DELETE", "/admin/v1/projects/P3", headers_json
, None, 204, None, None)
918 description
= "Creates/edit/delete fake VIMs and SDN controllers"
922 "schema_version": "1.0",
923 "schema_type": "No idea",
925 "description": "Descriptor name",
926 "vim_type": "openstack",
927 "vim_url": "http://localhost:/vim",
928 "vim_tenant_name": "vimTenant",
930 "vim_password": "password",
931 "config": {"config_param": 1}
935 "description": "sdn-description",
936 "dpid": "50:50:52:54:00:94:21:21",
937 "ip": "192.168.15.17",
939 "type": "opendaylight",
944 self
.port_mapping
= [
945 {"compute_node": "compute node 1",
946 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/1", "switch_mac": "52:54:00:94:21:21"},
947 {"pci": "0000:81:00.1", "switch_port": "port-2/2", "switch_mac": "52:54:00:94:21:22"}
949 {"compute_node": "compute node 2",
950 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/3", "switch_mac": "52:54:00:94:21:23"},
951 {"pci": "0000:81:00.1", "switch_port": "port-2/4", "switch_mac": "52:54:00:94:21:24"}
955 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
957 vim_bad
= self
.vim
.copy()
960 engine
.set_test_name("FakeVim")
961 engine
.get_autorization()
962 engine
.test("Create VIM", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
, (201, 202),
963 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json")
964 vim_id
= engine
.last_id
965 engine
.test("Create VIM without name, bad schema", "POST", "/admin/v1/vim_accounts", headers_json
,
966 vim_bad
, 422, None, headers_json
)
967 engine
.test("Create VIM name repeated", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
,
968 409, None, headers_json
)
969 engine
.test("Show VIMs", "GET", "/admin/v1/vim_accounts", headers_yaml
, None, 200, r_header_yaml
,
971 engine
.test("Show VIM", "GET", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_yaml
, None, 200,
972 r_header_yaml
, "yaml")
975 engine
.test("Delete VIM", "DELETE", "/admin/v1/vim_accounts/{}?FORCE=True".format(vim_id
), headers_yaml
,
977 engine
.test("Check VIM is deleted", "GET", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_yaml
, None,
978 404, r_header_yaml
, "yaml")
980 # delete and wait until is really deleted
981 engine
.test("Delete VIM", "DELETE", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_yaml
, None, 202,
983 engine
.wait_until_delete("/admin/v1/vim_accounts/{}".format(vim_id
), timeout
)
986 class TestVIMSDN(TestFakeVim
):
987 description
= "Creates VIM with SDN editing SDN controllers and port_mapping"
990 TestFakeVim
.__init
__(self
)
992 "schema_version": "1.0",
993 "schema_type": "No idea",
995 "description": "Descriptor name",
997 "wim_url": "http://localhost:/wim",
999 "password": "password",
1000 "config": {"config_param": 1}
1003 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1004 engine
.set_test_name("VimSdn")
1005 engine
.get_autorization()
1007 engine
.test("Create SDN", "POST", "/admin/v1/sdns", headers_json
, self
.sdn
, (201, 202),
1008 {"Location": "/admin/v1/sdns/", "Content-Type": "application/json"}, "json")
1009 sdnc_id
= engine
.last_id
1012 engine
.test("Edit SDN", "PATCH", "/admin/v1/sdns/{}".format(sdnc_id
), headers_json
, {"name": "new_sdn_name"},
1013 (202, 204), None, None)
1016 self
.vim
["config"]["sdn-controller"] = sdnc_id
1017 self
.vim
["config"]["sdn-port-mapping"] = self
.port_mapping
1018 engine
.test("Create VIM", "POST", "/admin/v1/vim_accounts", headers_json
, self
.vim
, (200, 202, 201),
1019 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json"),
1021 vim_id
= engine
.last_id
1022 self
.port_mapping
[0]["compute_node"] = "compute node XX"
1023 engine
.test("Edit VIM change port-mapping", "PUT", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_json
,
1024 {"config": {"sdn-port-mapping": self
.port_mapping
}}, (202, 204), None, None)
1025 engine
.test("Edit VIM remove port-mapping", "PUT", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_json
,
1026 {"config": {"sdn-port-mapping": None}}, (202, 204), None, None)
1028 engine
.test("Create WIM", "POST", "/admin/v1/wim_accounts", headers_json
, self
.wim
, (200, 202, 201),
1029 {"Location": "/admin/v1/wim_accounts/", "Content-Type": "application/json"}, "json"),
1030 wim_id
= engine
.last_id
1034 engine
.test("Delete VIM remove port-mapping", "DELETE",
1035 "/admin/v1/vim_accounts/{}?FORCE=True".format(vim_id
), headers_json
, None, 202, None, 0)
1036 engine
.test("Delete SDNC", "DELETE", "/admin/v1/sdns/{}?FORCE=True".format(sdnc_id
), headers_json
, None,
1039 engine
.test("Delete WIM", "DELETE",
1040 "/admin/v1/wim_accounts/{}?FORCE=True".format(wim_id
), headers_json
, None, 202, None, 0)
1041 engine
.test("Check VIM is deleted", "GET", "/admin/v1/vim_accounts/{}".format(vim_id
), headers_yaml
,
1042 None, 404, r_header_yaml
, "yaml")
1043 engine
.test("Check SDN is deleted", "GET", "/admin/v1/sdns/{}".format(sdnc_id
), headers_yaml
, None,
1044 404, r_header_yaml
, "yaml")
1045 engine
.test("Check WIM is deleted", "GET", "/admin/v1/wim_accounts/{}".format(wim_id
), headers_yaml
,
1046 None, 404, r_header_yaml
, "yaml")
1049 input('VIM, SDN, WIM has been deployed. Perform manual check and press enter to resume')
1050 # delete and wait until is really deleted
1051 engine
.test("Delete VIM remove port-mapping", "DELETE", "/admin/v1/vim_accounts/{}".format(vim_id
),
1052 headers_json
, None, (202, 201, 204), None, 0)
1053 engine
.test("Delete SDN", "DELETE", "/admin/v1/sdns/{}".format(sdnc_id
), headers_json
, None,
1054 (202, 201, 204), None, 0)
1055 engine
.test("Delete VIM", "DELETE", "/admin/v1/wim_accounts/{}".format(wim_id
),
1056 headers_json
, None, (202, 201, 204), None, 0)
1057 engine
.wait_until_delete("/admin/v1/vim_accounts/{}".format(vim_id
), timeout
)
1058 engine
.wait_until_delete("/admin/v1/sdns/{}".format(sdnc_id
), timeout
)
1059 engine
.wait_until_delete("/admin/v1/wim_accounts/{}".format(wim_id
), timeout
)
1063 description
= "Base class for downloading descriptors from ETSI, onboard and deploy in real VIM"
1066 self
.test_name
= "DEPLOY"
1071 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
1072 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
1073 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
1074 self
.descriptor_edit
= None
1075 self
.uses_configuration
= False
1082 self
.ns_params
= None
1083 self
.vnfr_ip_list
= {}
1085 def create_descriptors(self
, engine
):
1086 temp_dir
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/temp/"
1087 if not os
.path
.exists(temp_dir
):
1088 os
.makedirs(temp_dir
)
1089 for vnfd_index
, vnfd_filename
in enumerate(self
.vnfd_filenames
):
1090 if "/" in vnfd_filename
:
1091 vnfd_filename_path
= vnfd_filename
1092 if not os
.path
.exists(vnfd_filename_path
):
1093 raise TestException("File '{}' does not exist".format(vnfd_filename_path
))
1095 vnfd_filename_path
= temp_dir
+ vnfd_filename
1096 if not os
.path
.exists(vnfd_filename_path
):
1097 with
open(vnfd_filename_path
, "wb") as file:
1098 response
= requests
.get(self
.descriptor_url
+ vnfd_filename
)
1099 if response
.status_code
>= 300:
1100 raise TestException("Error downloading descriptor from '{}': {}".format(
1101 self
.descriptor_url
+ vnfd_filename
, response
.status_code
))
1102 file.write(response
.content
)
1103 if vnfd_filename_path
.endswith(".yaml"):
1104 headers
= headers_yaml
1106 headers
= headers_zip_yaml
1107 if randint(0, 1) == 0:
1108 # vnfd CREATE AND UPLOAD in one step:
1109 engine
.test("Onboard VNFD in one step", "POST",
1110 "/vnfpkgm/v1/vnf_packages_content" + self
.qforce
, headers
, "@b" + vnfd_filename_path
, 201,
1111 r_headers_yaml_location_vnfd
,
1113 self
.vnfds_id
.append(engine
.last_id
)
1115 # vnfd CREATE AND UPLOAD ZIP
1116 engine
.test("Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages",
1117 headers_json
, None, 201,
1118 {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
1119 self
.vnfds_id
.append(engine
.last_id
)
1120 engine
.test("Onboard VNFD step 2 as ZIP", "PUT",
1121 "/vnfpkgm/v1/vnf_packages/<>/package_content" + self
.qforce
,
1122 headers
, "@b" + vnfd_filename_path
, 204, None, 0)
1124 if self
.descriptor_edit
:
1125 if "vnfd{}".format(vnfd_index
) in self
.descriptor_edit
:
1127 engine
.test("Edit VNFD ", "PATCH",
1128 "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfds_id
[-1]),
1129 headers_yaml
, self
.descriptor_edit
["vnfd{}".format(vnfd_index
)], 204, None, None)
1131 if "/" in self
.nsd_filename
:
1132 nsd_filename_path
= self
.nsd_filename
1133 if not os
.path
.exists(nsd_filename_path
):
1134 raise TestException("File '{}' does not exist".format(nsd_filename_path
))
1136 nsd_filename_path
= temp_dir
+ self
.nsd_filename
1137 if not os
.path
.exists(nsd_filename_path
):
1138 with
open(nsd_filename_path
, "wb") as file:
1139 response
= requests
.get(self
.descriptor_url
+ self
.nsd_filename
)
1140 if response
.status_code
>= 300:
1141 raise TestException("Error downloading descriptor from '{}': {}".format(
1142 self
.descriptor_url
+ self
.nsd_filename
, response
.status_code
))
1143 file.write(response
.content
)
1144 if nsd_filename_path
.endswith(".yaml"):
1145 headers
= headers_yaml
1147 headers
= headers_zip_yaml
1149 if randint(0, 1) == 0:
1150 # nsd CREATE AND UPLOAD in one step:
1151 engine
.test("Onboard NSD in one step", "POST",
1152 "/nsd/v1/ns_descriptors_content" + self
.qforce
, headers
, "@b" + nsd_filename_path
, 201,
1153 r_headers_yaml_location_nsd
, yaml
)
1154 self
.nsd_id
= engine
.last_id
1156 # nsd CREATE AND UPLOAD ZIP
1157 engine
.test("Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors",
1158 headers_json
, None, 201,
1159 {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
1160 self
.nsd_id
= engine
.last_id
1161 engine
.test("Onboard NSD step 2 as ZIP", "PUT",
1162 "/nsd/v1/ns_descriptors/<>/nsd_content" + self
.qforce
,
1163 headers
, "@b" + nsd_filename_path
, 204, None, 0)
1165 if self
.descriptor_edit
and "nsd" in self
.descriptor_edit
:
1167 engine
.test("Edit NSD ", "PATCH",
1168 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
),
1169 headers_yaml
, self
.descriptor_edit
["nsd"], 204, None, None)
1171 def delete_descriptors(self
, engine
):
1172 # delete descriptors
1173 engine
.test("Delete NSSD SOL005", "DELETE",
1174 "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
),
1175 headers_yaml
, None, 204, None, 0)
1176 for vnfd_id
in self
.vnfds_id
:
1177 engine
.test("Delete VNFD SOL005", "DELETE",
1178 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id
), headers_yaml
, None, 204, None, 0)
1180 def instantiate(self
, engine
, ns_data
):
1181 ns_data_text
= yaml
.safe_dump(ns_data
, default_flow_style
=True, width
=256)
1182 # create NS Two steps
1183 r
= engine
.test("Create NS step 1", "POST", "/nslcm/v1/ns_instances",
1184 headers_yaml
, ns_data_text
, (201, 202),
1185 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"}, "yaml")
1188 self
.ns_id
= engine
.last_id
1189 engine
.test("Instantiate NS step 2", "POST",
1190 "/nslcm/v1/ns_instances/{}/instantiate".format(self
.ns_id
), headers_yaml
, ns_data_text
,
1191 (201, 202), r_headers_yaml_location_nslcmop
, "yaml")
1192 nslcmop_id
= engine
.last_id
1195 # Wait until status is Ok
1196 timeout
= timeout_configure
if self
.uses_configuration
else timeout_deploy
1197 engine
.wait_operation_ready("ns", nslcmop_id
, timeout
)
1199 def terminate(self
, engine
):
1202 engine
.test("Terminate NS", "POST", "/nslcm/v1/ns_instances/{}/terminate".format(self
.ns_id
), headers_yaml
,
1203 None, (201, 202), r_headers_yaml_location_nslcmop
, "yaml")
1204 nslcmop2_id
= engine
.last_id
1205 # Wait until status is Ok
1206 engine
.wait_operation_ready("ns", nslcmop2_id
, timeout_deploy
)
1208 engine
.test("Delete NS", "DELETE", "/nslcm/v1/ns_instances/{}".format(self
.ns_id
), headers_yaml
, None,
1211 engine
.test("Delete NS with FORCE", "DELETE", "/nslcm/v1/ns_instances/{}?FORCE=True".format(self
.ns_id
),
1212 headers_yaml
, None, 204, None, 0)
1214 # check all it is deleted
1215 engine
.test("Check NS is deleted", "GET", "/nslcm/v1/ns_instances/{}".format(self
.ns_id
), headers_yaml
, None,
1217 r
= engine
.test("Check NSLCMOPs are deleted", "GET",
1218 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId={}".format(self
.ns_id
), headers_json
, None,
1223 if not isinstance(nslcmops
, list) or nslcmops
:
1224 raise TestException("NS {} deleted but with ns_lcm_op_occ active: {}".format(self
.ns_id
, nslcmops
))
1226 def test_ns(self
, engine
, test_osm
, commands
=None, users
=None, passwds
=None, keys
=None, timeout
=0):
1228 r
= engine
.test("GET VNFR IDs", "GET",
1229 "/nslcm/v1/ns_instances/{}".format(self
.ns_id
), headers_json
, None,
1230 200, r_header_json
, "json")
1235 vnfr_list
= ns_data
['constituent-vnfr-ref']
1237 _commands
= commands
if commands
is not None else self
.commands
1238 _users
= users
if users
is not None else self
.users
1239 _passwds
= passwds
if passwds
is not None else self
.passwords
1240 _keys
= keys
if keys
is not None else self
.keys
1241 _timeout
= timeout
if timeout
!= 0 else self
.timeout
1243 # vnfr_list=[d8272263-6bd3-4680-84ca-6a4be23b3f2d, 88b22e2f-994a-4b61-94fd-4a3c90de3dc4]
1244 for vnfr_id
in vnfr_list
:
1245 r
= engine
.test("Get VNFR to get IP_ADDRESS", "GET",
1246 "/nslcm/v1/vnfrs/{}".format(vnfr_id
), headers_json
, None,
1247 200, r_header_json
, "json")
1250 vnfr_data
= r
.json()
1252 vnf_index
= str(vnfr_data
["member-vnf-index-ref"])
1254 ip_address
= self
.get_vnfr_ip(engine
, vnf_index
)
1255 description
= "Exec command='{}' at VNFR={} IP={}".format(_commands
.get(vnf_index
)[0], vnf_index
,
1258 test_description
= "{}{} {}".format(engine
.test_name
, engine
.step
, description
)
1259 logger
.warning(test_description
)
1260 while _timeout
>= time
:
1261 result
, message
= self
.do_checks([ip_address
],
1262 vnf_index
=vnfr_data
["member-vnf-index-ref"],
1263 commands
=_commands
.get(vnf_index
), user
=_users
.get(vnf_index
),
1264 passwd
=_passwds
.get(vnf_index
), key
=_keys
.get(vnf_index
))
1266 engine
.passed_tests
+= 1
1267 logger
.debug(message
)
1273 engine
.failed_tests
+= 1
1274 logger
.error(message
)
1278 engine
.failed_tests
+= 1
1279 logger
.error(message
)
1281 engine
.failed_tests
+= 1
1282 logger
.error("VNFR {} has not mgmt address. Check failed".format(vnf_index
))
1284 def do_checks(self
, ip
, vnf_index
, commands
=[], user
=None, passwd
=None, key
=None):
1287 from pssh
.clients
import ParallelSSHClient
1288 from pssh
.utils
import load_private_key
1289 from ssh2
import exceptions
as ssh2Exception
1290 except ImportError as e
:
1291 logger
.critical("Package <pssh> or/and <urllib3> is not installed. Please add them with 'pip3 install "
1292 "parallel-ssh urllib3': {}".format(e
))
1293 return -1, "install needed packages 'pip3 install parallel-ssh urllib3'"
1294 urllib3
.disable_warnings(urllib3
.exceptions
.InsecureRequestWarning
)
1296 p_host
= os
.environ
.get("PROXY_HOST")
1297 p_user
= os
.environ
.get("PROXY_USER")
1298 p_password
= os
.environ
.get("PROXY_PASSWD")
1301 pkey
= load_private_key(key
)
1305 client
= ParallelSSHClient(ip
, user
=user
, password
=passwd
, pkey
=pkey
, proxy_host
=p_host
,
1306 proxy_user
=p_user
, proxy_password
=p_password
, timeout
=10, num_retries
=0)
1307 for cmd
in commands
:
1308 output
= client
.run_command(cmd
)
1310 if output
[ip
[0]].exit_code
:
1311 return -1, "VNFR {} command '{}' returns error: '{}'".format(ip
[0], cmd
,
1312 "\n".join(output
[ip
[0]].stderr
))
1314 return 1, "VNFR {} command '{}' successful".format(ip
[0], cmd
)
1315 except (ssh2Exception
.ChannelFailure
, ssh2Exception
.SocketDisconnectError
, ssh2Exception
.SocketTimeout
,
1316 ssh2Exception
.SocketRecvError
) as e
:
1317 return 0, "Timeout accessing the VNFR {}: {}".format(ip
[0], str(e
))
1318 except Exception as e
:
1319 return -1, "ERROR checking the VNFR {}: {}".format(ip
[0], str(e
))
1321 def additional_operations(self
, engine
, test_osm
, manual_check
):
1324 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1325 engine
.set_test_name(self
.test_name
)
1326 engine
.get_autorization()
1327 nsname
= os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
1329 if "vnfd-files" in test_params
:
1330 self
.vnfd_filenames
= test_params
["vnfd-files"].split(",")
1331 if "nsd-file" in test_params
:
1332 self
.nsd_filename
= test_params
["nsd-file"]
1333 if test_params
.get("ns-name"):
1334 nsname
= test_params
["ns-name"]
1335 self
.create_descriptors(engine
)
1337 # create real VIM if not exist
1338 self
.vim_id
= engine
.get_create_vim(test_osm
)
1339 ns_data
= {"nsDescription": "default description", "nsName": nsname
, "nsdId": self
.nsd_id
,
1340 "vimAccountId": self
.vim_id
}
1342 ns_data
.update(self
.ns_params
)
1343 if test_params
and test_params
.get("ns-config"):
1344 if isinstance(test_params
["ns-config"], str):
1345 ns_data
.update(yaml
.load(test_params
["ns-config"]))
1347 ns_data
.update(test_params
["ns-config"])
1348 self
.instantiate(engine
, ns_data
)
1351 input('NS has been deployed. Perform manual check and press enter to resume')
1352 if test_osm
and self
.commands
:
1353 self
.test_ns(engine
, test_osm
)
1354 self
.additional_operations(engine
, test_osm
, manual_check
)
1355 self
.terminate(engine
)
1356 self
.delete_descriptors(engine
)
1358 def get_first_ip(self
, ip_string
):
1359 # When using a floating IP, the vnfr_data['ip-address'] contains a semicolon-separated list of IP:s.
1360 first_ip
= ip_string
.split(";")[0] if ip_string
else ""
1363 def get_vnfr_ip(self
, engine
, vnfr_index_wanted
):
1364 # If the IP address list has been obtained before, it has been stored in 'vnfr_ip_list'
1365 ip
= self
.vnfr_ip_list
.get(vnfr_index_wanted
, "")
1367 return self
.get_first_ip(ip
)
1368 r
= engine
.test("Get VNFR to get IP_ADDRESS", "GET",
1369 "/nslcm/v1/vnfrs?member-vnf-index-ref={}&nsr-id-ref={}".format(
1370 vnfr_index_wanted
, self
.ns_id
), headers_json
, None,
1371 200, r_header_json
, "json")
1374 vnfr_data
= r
.json()
1375 if not (vnfr_data
and vnfr_data
[0]):
1377 # Store the IP (or list of IPs) in 'vnfr_ip_list'
1378 ip_list
= vnfr_data
[0].get("ip-address", "")
1380 self
.vnfr_ip_list
[vnfr_index_wanted
] = ip_list
1381 ip
= self
.get_first_ip(ip_list
)
1385 class TestDeployHackfestCirros(TestDeploy
):
1386 description
= "Load and deploy Hackfest cirros_2vnf_ns example"
1390 self
.test_name
= "CIRROS"
1391 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
1392 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
1393 self
.commands
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1394 self
.users
= {'1': "cirros", '2': "cirros"}
1395 self
.passwords
= {'1': "cubswin:)", '2': "cubswin:)"}
1397 def terminate(self
, engine
):
1398 # Make a delete in one step, overriding the normal two step of TestDeploy that launched terminate and delete
1400 engine
.test("Terminate and delete NS in one step", "DELETE", "/nslcm/v1/ns_instances_content/{}".
1401 format(self
.ns_id
), headers_yaml
, None, 202, None, "yaml")
1403 engine
.wait_until_delete("/nslcm/v1/ns_instances/{}".format(self
.ns_id
), timeout_deploy
)
1405 engine
.test("Delete NS with FORCE", "DELETE", "/nslcm/v1/ns_instances/{}?FORCE=True".format(self
.ns_id
),
1406 headers_yaml
, None, 204, None, 0)
1408 # check all it is deleted
1409 engine
.test("Check NS is deleted", "GET", "/nslcm/v1/ns_instances/{}".format(self
.ns_id
), headers_yaml
, None,
1411 r
= engine
.test("Check NSLCMOPs are deleted", "GET",
1412 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId={}".format(self
.ns_id
), headers_json
, None,
1417 if not isinstance(nslcmops
, list) or nslcmops
:
1418 raise TestException("NS {} deleted but with ns_lcm_op_occ active: {}".format(self
.ns_id
, nslcmops
))
1421 class TestDeployHackfest1(TestDeploy
):
1422 description
= "Load and deploy Hackfest_1_vnfd example"
1426 self
.test_name
= "HACKFEST1-"
1427 self
.vnfd_filenames
= ("hackfest_1_vnfd.tar.gz",)
1428 self
.nsd_filename
= "hackfest_1_nsd.tar.gz"
1429 # self.commands = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1430 # self.users = {'1': "cirros", '2': "cirros"}
1431 # self.passwords = {'1': "cubswin:)", '2': "cubswin:)"}
1434 class TestDeployHackfestCirrosScaling(TestDeploy
):
1435 description
= "Load and deploy Hackfest cirros_2vnf_ns example with scaling modifications"
1439 self
.test_name
= "CIRROS-SCALE"
1440 self
.vnfd_filenames
= ("cirros_vnf.tar.gz",)
1441 self
.nsd_filename
= "cirros_2vnf_ns.tar.gz"
1442 # Modify VNFD to add scaling and count=2
1443 self
.descriptor_edit
= {
1446 "$id: 'cirros_vnfd-VM'": {"count": 2}
1448 "scaling-group-descriptor": [{
1449 "name": "scale_cirros",
1450 "max-instance-count": 2,
1452 "vdu-id-ref": "cirros_vnfd-VM",
1459 def additional_operations(self
, engine
, test_osm
, manual_check
):
1462 # 2 perform scale out twice
1463 payload
= '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1464 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
1465 for i
in range(0, 2):
1466 engine
.test("Execute scale action over NS", "POST",
1467 "/nslcm/v1/ns_instances/{}/scale".format(self
.ns_id
), headers_yaml
, payload
,
1468 (201, 202), r_headers_yaml_location_nslcmop
, "yaml")
1469 nslcmop2_scale_out
= engine
.last_id
1470 engine
.wait_operation_ready("ns", nslcmop2_scale_out
, timeout_deploy
)
1472 input('NS scale out done. Check that two more vdus are there')
1473 # TODO check automatic
1475 # 2 perform scale in
1476 payload
= '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1477 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
1478 for i
in range(0, 2):
1479 engine
.test("Execute scale IN action over NS", "POST",
1480 "/nslcm/v1/ns_instances/{}/scale".format(self
.ns_id
), headers_yaml
, payload
,
1481 (201, 202), r_headers_yaml_location_nslcmop
, "yaml")
1482 nslcmop2_scale_in
= engine
.last_id
1483 engine
.wait_operation_ready("ns", nslcmop2_scale_in
, timeout_deploy
)
1485 input('NS scale in done. Check that two less vdus are there')
1486 # TODO check automatic
1488 # perform scale in that must fail as reached limit
1489 engine
.test("Execute scale IN out of limit action over NS", "POST",
1490 "/nslcm/v1/ns_instances/{}/scale".format(self
.ns_id
), headers_yaml
, payload
,
1491 (201, 202), r_headers_yaml_location_nslcmop
, "yaml")
1492 nslcmop2_scale_in
= engine
.last_id
1493 engine
.wait_operation_ready("ns", nslcmop2_scale_in
, timeout_deploy
, expected_fail
=True)
1496 class TestDeployIpMac(TestDeploy
):
1497 description
= "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
1501 self
.test_name
= "SetIpMac"
1502 self
.vnfd_filenames
= ("vnfd_2vdu_set_ip_mac2.yaml", "vnfd_2vdu_set_ip_mac.yaml")
1503 self
.nsd_filename
= "scenario_2vdu_set_ip_mac.yaml"
1504 self
.descriptor_url
= \
1505 "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
1506 self
.commands
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1507 self
.users
= {'1': "osm", '2': "osm"}
1508 self
.passwords
= {'1': "osm4u", '2': "osm4u"}
1511 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
1512 # super().run(engine, test_osm, manual_check, test_params)
1513 # run again setting IPs with instantiate parameters
1514 instantiation_params
= {
1517 "member-vnf-index": "1",
1520 "name": "internal_vld1", # net_internal
1522 "ip-version": "ipv4",
1523 "subnet-address": "10.9.8.0/24",
1524 "dhcp-params": {"count": 100, "start-address": "10.9.8.100"}
1526 "internal-connection-point": [
1529 "ip-address": "10.9.8.2",
1533 "ip-address": "10.9.8.3",
1544 # "name": "iface11",
1545 # "floating-ip-required": True,
1549 "mac-address": "52:33:44:55:66:13"
1558 "ip-address": "10.31.31.22",
1559 "mac-address": "52:33:44:55:66:21"
1568 super().run(engine
, test_osm
, manual_check
, test_params
={"ns-config": instantiation_params
})
1571 class TestDeployHackfest4(TestDeploy
):
1572 description
= "Load and deploy Hackfest 4 example."
1576 self
.test_name
= "HACKFEST4-"
1577 self
.vnfd_filenames
= ("hackfest_4_vnfd.tar.gz",)
1578 self
.nsd_filename
= "hackfest_4_nsd.tar.gz"
1579 self
.uses_configuration
= True
1580 self
.commands
= {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1581 self
.users
= {'1': "ubuntu", '2': "ubuntu"}
1582 self
.passwords
= {'1': "osm4u", '2': "osm4u"}
1583 # Modify VNFD to add scaling
1584 # self.descriptor_edit = {
1586 # 'vnf-configuration': {
1587 # 'config-primitive': [{
1590 # 'name': 'filename',
1591 # 'data-type': 'STRING',
1592 # 'default-value': '/home/ubuntu/touched'
1596 # 'scaling-group-descriptor': [{
1597 # 'name': 'scale_dataVM',
1598 # 'scaling-policy': [{
1599 # 'threshold-time': 0,
1600 # 'name': 'auto_cpu_util_above_threshold',
1601 # 'scaling-type': 'automatic',
1602 # 'scaling-criteria': [{
1603 # 'name': 'cpu_util_above_threshold',
1604 # 'vnf-monitoring-param-ref': 'all_aaa_cpu_util',
1605 # 'scale-out-relational-operation': 'GE',
1606 # 'scale-in-threshold': 15,
1607 # 'scale-out-threshold': 60,
1608 # 'scale-in-relational-operation': 'LE'
1610 # 'cooldown-time': 60
1612 # 'max-instance-count': 10,
1613 # 'scaling-config-action': [
1614 # {'vnf-config-primitive-name-ref': 'touch',
1615 # 'trigger': 'post-scale-out'},
1616 # {'vnf-config-primitive-name-ref': 'touch',
1617 # 'trigger': 'pre-scale-in'}
1620 # 'vdu-id-ref': 'dataVM',
1628 class TestDeployHackfest3Charmed(TestDeploy
):
1629 description
= "Load and deploy Hackfest 3charmed_ns example"
1633 self
.test_name
= "HACKFEST3-"
1634 self
.vnfd_filenames
= ("hackfest_3charmed_vnfd.tar.gz",)
1635 self
.nsd_filename
= "hackfest_3charmed_nsd.tar.gz"
1636 self
.uses_configuration
= True
1637 self
.commands
= {'1': ['ls -lrt /home/ubuntu/first-touch'], '2': ['ls -lrt /home/ubuntu/first-touch']}
1638 self
.users
= {'1': "ubuntu", '2': "ubuntu"}
1639 self
.passwords
= {'1': "osm4u", '2': "osm4u"}
1640 self
.descriptor_edit
= {
1641 "vnfd0": yaml
.safe_load(
1644 terminate-config-primitive:
1649 value: '/home/ubuntu/last-touch1'
1654 value: '/home/ubuntu/last-touch3'
1659 value: '/home/ubuntu/last-touch2'
1663 def additional_operations(self
, engine
, test_osm
, manual_check
):
1667 vnfr_index_selected
= "2"
1668 payload
= '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
1669 engine
.test("Exec service primitive over NS", "POST",
1670 "/nslcm/v1/ns_instances/{}/action".format(self
.ns_id
), headers_yaml
, payload
,
1671 (201, 202), r_headers_yaml_location_nslcmop
, "yaml")
1672 nslcmop2_action
= engine
.last_id
1673 # Wait until status is Ok
1674 engine
.wait_operation_ready("ns", nslcmop2_action
, timeout_deploy
)
1675 vnfr_ip
= self
.get_vnfr_ip(engine
, vnfr_index_selected
)
1678 "NS service primitive has been executed."
1679 "Check that file /home/ubuntu/OSMTESTNBI is present at {}".
1682 commands
= {'1': [''], '2': ['ls -lrt /home/ubuntu/OSMTESTNBI', ]}
1683 self
.test_ns(engine
, test_osm
, commands
=commands
)
1685 # # 2 perform scale out
1686 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1687 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1688 # engine.test("Execute scale action over NS", "POST",
1689 # "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1690 # (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1691 # nslcmop2_scale_out = engine.last_id
1692 # engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
1694 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
1695 # # TODO check automatic
1697 # # 2 perform scale in
1698 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1699 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1700 # engine.test("Execute scale action over NS", "POST",
1701 # "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1702 # (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1703 # nslcmop2_scale_in = engine.last_id
1704 # engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
1706 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
1707 # # TODO check automatic
1710 class TestDeployHackfest3Charmed2(TestDeployHackfest3Charmed
):
1711 description
= "Load and deploy Hackfest 3charmed_ns example modified version of descriptors to have dots in " \
1712 "ids and member-vnf-index."
1716 self
.test_name
= "HACKFEST3v2-"
1717 self
.qforce
= "?FORCE=True"
1718 self
.descriptor_edit
= {
1722 "interface": {"$[0]": {"external-connection-point-ref": "pdu-mgmt"}}
1726 "vnf-configuration": None,
1727 "connection-point": {
1731 "short-name": "pdu-mgmt"
1735 "mgmt-interface": {"cp": "pdu-mgmt"},
1736 "description": "A vnf single vdu to be used as PDU",
1740 "id": "pdu_internal",
1741 "name": "pdu_internal",
1742 "internal-connection-point": {"$[1]": None},
1743 "short-name": "pdu_internal",
1749 # Modify NSD accordingly
1751 "constituent-vnfd": {
1752 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
1755 "description": "A nsd to deploy the vnf to act as as PDU",
1757 "name": "nsd-as-pdu",
1758 "short-name": "nsd-as-pdu",
1763 "short-name": "mgmt_pdu",
1764 "vnfd-connection-point-ref": {
1766 "vnfd-connection-point-ref": "pdu-mgmt",
1767 "vnfd-id-ref": "vdu-as-pdu",
1779 class TestDeployHackfest3Charmed3(TestDeployHackfest3Charmed
):
1780 description
= "Load and deploy Hackfest 3charmed_ns example modified version to test scaling and NS parameters"
1784 self
.test_name
= "HACKFEST3v3-"
1785 self
.commands
= {'1': ['ls -lrt /home/ubuntu/first-touch-1'], '2': ['ls -lrt /home/ubuntu/first-touch-2']}
1786 self
.descriptor_edit
= {
1789 scaling-group-descriptor:
1790 - name: "scale_dataVM"
1791 max-instance-count: 10
1793 - name: "auto_cpu_util_above_threshold"
1794 scaling-type: "automatic"
1798 - name: "cpu_util_above_threshold"
1799 scale-in-threshold: 15
1800 scale-in-relational-operation: "LE"
1801 scale-out-threshold: 60
1802 scale-out-relational-operation: "GE"
1803 vnf-monitoring-param-ref: "monitor1"
1805 - vdu-id-ref: dataVM
1807 scaling-config-action:
1808 - trigger: post-scale-out
1809 vnf-config-primitive-name-ref: touch
1810 - trigger: pre-scale-in
1811 vnf-config-primitive-name-ref: touch
1815 - id: "dataVM_cpu_util"
1816 nfvi-metric: "cpu_utilization"
1821 aggregation-type: AVERAGE
1822 vdu-monitoring-param:
1824 vdu-monitoring-param-ref: "dataVM_cpu_util"
1826 initial-config-primitive:
1830 value: "<touch_filename>" # default-value: /home/ubuntu/first-touch
1835 default-value: "<touch_filename2>"
1839 "additionalParamsForVnf": [
1840 {"member-vnf-index": "1", "additionalParams": {"touch_filename": "/home/ubuntu/first-touch-1",
1841 "touch_filename2": "/home/ubuntu/second-touch-1"}},
1842 {"member-vnf-index": "2", "additionalParams": {"touch_filename": "/home/ubuntu/first-touch-2",
1843 "touch_filename2": "/home/ubuntu/second-touch-2"}},
1847 def additional_operations(self
, engine
, test_osm
, manual_check
):
1848 super().additional_operations(engine
, test_osm
, manual_check
)
1852 # 2 perform scale out
1853 payload
= '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1854 '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1855 engine
.test("Execute scale action over NS", "POST",
1856 "/nslcm/v1/ns_instances/{}/scale".format(self
.ns_id
), headers_yaml
, payload
,
1857 (201, 202), r_headers_yaml_location_nslcmop
, "yaml")
1858 nslcmop2_scale_out
= engine
.last_id
1859 engine
.wait_operation_ready("ns", nslcmop2_scale_out
, timeout_deploy
)
1861 input('NS scale out done. Check that file /home/ubuntu/second-touch-1 is present and new VM is created')
1863 commands
= {'1': ['ls -lrt /home/ubuntu/second-touch-1', ]}
1864 self
.test_ns(engine
, test_osm
, commands
=commands
)
1865 # TODO check automatic connection to scaled VM
1867 # 2 perform scale in
1868 payload
= '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1869 '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1870 engine
.test("Execute scale action over NS", "POST",
1871 "/nslcm/v1/ns_instances/{}/scale".format(self
.ns_id
), headers_yaml
, payload
,
1872 (201, 202), r_headers_yaml_location_nslcmop
, "yaml")
1873 nslcmop2_scale_in
= engine
.last_id
1874 engine
.wait_operation_ready("ns", nslcmop2_scale_in
, timeout_deploy
)
1876 input('NS scale in done. Check that file /home/ubuntu/second-touch-1 is updated and new VM is deleted')
1877 # TODO check automatic
1880 class TestDeploySimpleCharm(TestDeploy
):
1881 description
= "Deploy hackfest-4 hackfest_simplecharm example"
1885 self
.test_name
= "HACKFEST-SIMPLE"
1886 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-4.0-four/4th-hackfest/packages/"
1887 self
.vnfd_filenames
= ("hackfest_simplecharm_vnf.tar.gz",)
1888 self
.nsd_filename
= "hackfest_simplecharm_ns.tar.gz"
1889 self
.uses_configuration
= True
1890 self
.commands
= {'1': [''], '2': ['ls -lrt /home/ubuntu/first-touch', ]}
1891 self
.users
= {'1': "ubuntu", '2': "ubuntu"}
1892 self
.passwords
= {'1': "osm4u", '2': "osm4u"}
1895 class TestDeploySimpleCharm2(TestDeploySimpleCharm
):
1896 description
= "Deploy hackfest-4 hackfest_simplecharm example changing naming to contain dots on ids and " \
1901 self
.test_name
= "HACKFEST-SIMPLE2-"
1902 self
.qforce
= "?FORCE=True"
1903 self
.descriptor_edit
= {
1905 "id": "hackfest.simplecharm.vnf"
1909 "id": "hackfest.simplecharm.ns",
1910 "constituent-vnfd": {
1911 "$[0]": {"vnfd-id-ref": "hackfest.simplecharm.vnf", "member-vnf-index": "$1"},
1912 "$[1]": {"vnfd-id-ref": "hackfest.simplecharm.vnf", "member-vnf-index": "$2"},
1916 "vnfd-connection-point-ref": {"$[0]": {"member-vnf-index-ref": "$1",
1917 "vnfd-id-ref": "hackfest.simplecharm.vnf"},
1918 "$[1]": {"member-vnf-index-ref": "$2",
1919 "vnfd-id-ref": "hackfest.simplecharm.vnf"}},
1922 "vnfd-connection-point-ref": {"$[0]": {"member-vnf-index-ref": "$1",
1923 "vnfd-id-ref": "hackfest.simplecharm.vnf"},
1924 "$[1]": {"member-vnf-index-ref": "$2",
1925 "vnfd-id-ref": "hackfest.simplecharm.vnf"}},
1932 class TestDeploySingleVdu(TestDeployHackfest3Charmed
):
1933 description
= "Generate a single VDU base on editing Hackfest3Charmed descriptors and deploy"
1937 self
.test_name
= "SingleVDU"
1938 self
.qforce
= "?FORCE=True"
1939 self
.descriptor_edit
= {
1940 # Modify VNFD to remove one VDU
1944 "interface": {"$[0]": {"external-connection-point-ref": "pdu-mgmt"}}
1948 "vnf-configuration": None,
1949 "connection-point": {
1953 "short-name": "pdu-mgmt"
1957 "mgmt-interface": {"cp": "pdu-mgmt"},
1958 "description": "A vnf single vdu to be used as PDU",
1962 "id": "pdu_internal",
1963 "name": "pdu_internal",
1964 "internal-connection-point": {"$[1]": None},
1965 "short-name": "pdu_internal",
1971 # Modify NSD accordingly
1973 "constituent-vnfd": {
1974 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
1977 "description": "A nsd to deploy the vnf to act as as PDU",
1979 "name": "nsd-as-pdu",
1980 "short-name": "nsd-as-pdu",
1985 "short-name": "mgmt_pdu",
1986 "vnfd-connection-point-ref": {
1988 "vnfd-connection-point-ref": "pdu-mgmt",
1989 "vnfd-id-ref": "vdu-as-pdu",
2001 class TestDeployHnfd(TestDeployHackfest3Charmed
):
2002 description
= "Generate a HNFD base on editing Hackfest3Charmed descriptors and deploy"
2006 self
.test_name
= "HNFD"
2007 self
.pduDeploy
= TestDeploySingleVdu()
2008 self
.pdu_interface_0
= {}
2009 self
.pdu_interface_1
= {}
2012 # self.vnf_to_pdu = """
2015 # pdu-type: PDU-TYPE-1
2020 # name: pdu-iface-internal
2022 # description: HFND, one PDU + One VDU
2028 self
.pdu_descriptor
= {
2030 "type": "PDU-TYPE-1",
2031 "vim_accounts": "to-override",
2034 "name": "mgmt-iface",
2037 "ip-address": "to override",
2038 "mac-address": "mac_address",
2039 "vim-network-name": "mgmt",
2042 "name": "pdu-iface-internal",
2045 "ip-address": "to override",
2046 "mac-address": "mac_address",
2047 "vim-network-name": "pdu_internal", # OSMNBITEST-PDU-pdu_internal
2051 self
.vnfd_filenames
= ("hackfest_3charmed_vnfd.tar.gz", "hackfest_3charmed_vnfd.tar.gz")
2053 self
.descriptor_edit
= {
2057 "short-name": "hfn1",
2060 "pdu-type": "PDU-TYPE-1",
2062 "$[0]": {"name": "mgmt-iface"},
2063 "$[1]": {"name": "pdu-iface-internal"},
2069 "constituent-vnfd": {
2070 "$[1]": {"vnfd-id-ref": "hfnd1"}
2073 "$[0]": {"vnfd-connection-point-ref": {"$[1]": {"vnfd-id-ref": "hfnd1"}}},
2074 "$[1]": {"vnfd-connection-point-ref": {"$[1]": {"vnfd-id-ref": "hfnd1"}}}
2079 def create_descriptors(self
, engine
):
2080 super().create_descriptors(engine
)
2083 self
.pdu_descriptor
["interfaces"][0].update(self
.pdu_interface_0
)
2084 self
.pdu_descriptor
["interfaces"][1].update(self
.pdu_interface_1
)
2085 self
.pdu_descriptor
["vim_accounts"] = [self
.vim_id
]
2086 # TODO get vim-network-name from vnfr.vld.name
2087 self
.pdu_descriptor
["interfaces"][1]["vim-network-name"] = "{}-{}-{}".format(
2088 os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST"),
2089 "PDU", self
.pdu_descriptor
["interfaces"][1]["vim-network-name"])
2090 engine
.test("Onboard PDU descriptor", "POST", "/pdu/v1/pdu_descriptors",
2091 {"Location": "/pdu/v1/pdu_descriptors/", "Content-Type": "application/yaml"}, self
.pdu_descriptor
,
2092 201, r_header_yaml
, "yaml")
2093 self
.pdu_id
= engine
.last_id
2095 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
2096 engine
.get_autorization()
2097 engine
.set_test_name(self
.test_name
)
2098 nsname
= os
.environ
.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
2100 # create real VIM if not exist
2101 self
.vim_id
= engine
.get_create_vim(test_osm
)
2103 self
.pduDeploy
.create_descriptors(engine
)
2104 self
.pduDeploy
.instantiate(engine
, {"nsDescription": "to be used as PDU", "nsName": nsname
+ "-PDU",
2105 "nsdId": self
.pduDeploy
.nsd_id
, "vimAccountId": self
.vim_id
})
2107 input('VNF to be used as PDU has been deployed. Perform manual check and press enter to resume')
2109 self
.pduDeploy
.test_ns(engine
, test_osm
)
2112 r
= engine
.test("Get VNFR to obtain IP_ADDRESS", "GET",
2113 "/nslcm/v1/vnfrs?nsr-id-ref={}".format(self
.pduDeploy
.ns_id
), headers_json
, None,
2114 200, r_header_json
, "json")
2117 vnfr_data
= r
.json()
2120 self
.pdu_interface_0
["ip-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][0].get("ip-address")
2121 self
.pdu_interface_1
["ip-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][1].get("ip-address")
2122 self
.pdu_interface_0
["mac-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][0].get("mac-address")
2123 self
.pdu_interface_1
["mac-address"] = vnfr_data
[0]["vdur"][0]["interfaces"][1].get("mac-address")
2124 if not self
.pdu_interface_0
["ip-address"]:
2125 raise TestException("Vnfr has not managment ip address")
2127 self
.pdu_interface_0
["ip-address"] = "192.168.10.10"
2128 self
.pdu_interface_1
["ip-address"] = "192.168.11.10"
2129 self
.pdu_interface_0
["mac-address"] = "52:33:44:55:66:13"
2130 self
.pdu_interface_1
["mac-address"] = "52:33:44:55:66:14"
2132 self
.create_descriptors(engine
)
2134 ns_data
= {"nsDescription": "default description", "nsName": nsname
, "nsdId": self
.nsd_id
,
2135 "vimAccountId": self
.vim_id
}
2136 if test_params
and test_params
.get("ns-config"):
2137 if isinstance(test_params
["ns-config"], str):
2138 ns_data
.update(yaml
.load(test_params
["ns-config"]))
2140 ns_data
.update(test_params
["ns-config"])
2142 self
.instantiate(engine
, ns_data
)
2144 input('NS has been deployed. Perform manual check and press enter to resume')
2146 self
.test_ns(engine
, test_osm
)
2147 self
.additional_operations(engine
, test_osm
, manual_check
)
2148 self
.terminate(engine
)
2149 self
.pduDeploy
.terminate(engine
)
2150 self
.delete_descriptors(engine
)
2151 self
.pduDeploy
.delete_descriptors(engine
)
2153 def delete_descriptors(self
, engine
):
2154 super().delete_descriptors(engine
)
2156 engine
.test("Delete PDU SOL005", "DELETE",
2157 "/pdu/v1/pdu_descriptors/{}".format(self
.pdu_id
),
2158 headers_yaml
, None, 204, None, 0)
2161 class TestDescriptors
:
2162 description
= "Test VNFD, NSD, PDU descriptors CRUD and dependencies"
2163 vnfd_empty
= """vnfd:vnfd-catalog:
2169 vnfd_prova
= """vnfd:vnfd-catalog:
2181 - external-connection-point-ref: cp_0h8m
2190 self
.vnfd_filename
= "hackfest_3charmed_vnfd.tar.gz"
2191 self
.nsd_filename
= "hackfest_3charmed_nsd.tar.gz"
2192 self
.descriptor_url
= "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
2196 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
2197 engine
.set_test_name("Descriptors")
2198 engine
.get_autorization()
2199 temp_dir
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/temp/"
2200 if not os
.path
.exists(temp_dir
):
2201 os
.makedirs(temp_dir
)
2204 for filename
in (self
.vnfd_filename
, self
.nsd_filename
):
2205 filename_path
= temp_dir
+ filename
2206 if not os
.path
.exists(filename_path
):
2207 with
open(filename_path
, "wb") as file:
2208 response
= requests
.get(self
.descriptor_url
+ filename
)
2209 if response
.status_code
>= 300:
2210 raise TestException("Error downloading descriptor from '{}': {}".format(
2211 self
.descriptor_url
+ filename
, response
.status_code
))
2212 file.write(response
.content
)
2214 vnfd_filename_path
= temp_dir
+ self
.vnfd_filename
2215 nsd_filename_path
= temp_dir
+ self
.nsd_filename
2217 engine
.test("Onboard empty VNFD in one step", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml
,
2218 self
.vnfd_empty
, 201, r_headers_yaml_location_vnfd
, "yaml")
2219 self
.vnfd_id
= engine
.last_id
2222 engine
.test("Upload invalid VNFD ", "PUT", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self
.vnfd_id
),
2223 headers_yaml
, self
.vnfd_prova
, 422, r_header_yaml
, "yaml")
2225 engine
.test("Upload VNFD {}".format(self
.vnfd_filename
), "PUT",
2226 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self
.vnfd_id
), headers_zip_yaml
,
2227 "@b" + vnfd_filename_path
, 204, None, 0)
2229 queries
= ["mgmt-interface.cp=mgmt", "vdu.0.interface.0.external-connection-point-ref=mgmt",
2230 "vdu.0.interface.1.internal-connection-point-ref=internal",
2231 "internal-vld.0.internal-connection-point.0.id-ref=internal",
2232 # Detection of duplicated VLD names in VNF Descriptors
2233 # URL: internal-vld=[
2234 # {id: internal1, name: internal, type:ELAN,
2235 # internal-connection-point: [{id-ref: mgmtVM-internal}, {id-ref: dataVM-internal}]},
2236 # {id: internal2, name: internal, type:ELAN,
2237 # internal-connection-point: [{id-ref: mgmtVM-internal}, {id-ref: dataVM-internal}]}
2239 "internal-vld=%5B%7Bid%3A%20internal1%2C%20name%3A%20internal%2C%20type%3A%20ELAN%2C%20"
2240 "internal-connection-point%3A%20%5B%7Bid-ref%3A%20mgmtVM-internal%7D%2C%20%7Bid-ref%3A%20"
2241 "dataVM-internal%7D%5D%7D%2C%20%7Bid%3A%20internal2%2C%20name%3A%20internal%2C%20type%3A%20"
2242 "ELAN%2C%20internal-connection-point%3A%20%5B%7Bid-ref%3A%20mgmtVM-internal%7D%2C%20%7B"
2243 "id-ref%3A%20dataVM-internal%7D%5D%7D%5D"
2245 for query
in queries
:
2246 engine
.test("Upload invalid VNFD ", "PUT",
2247 "/vnfpkgm/v1/vnf_packages/{}/package_content?{}".format(self
.vnfd_id
, query
),
2248 headers_zip_yaml
, "@b" + vnfd_filename_path
, 422, r_header_yaml
, "yaml")
2251 engine
.test("Upload invalid VNFD ", "PUT", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self
.vnfd_id
),
2252 headers_yaml
, self
.vnfd_prova
, 422, r_header_yaml
, "yaml")
2254 # get vnfd descriptor
2255 engine
.test("Get VNFD descriptor", "GET", "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_id
),
2256 headers_yaml
, None, 200, r_header_yaml
, "yaml")
2258 # get vnfd file descriptor
2259 engine
.test("Get VNFD file descriptor", "GET", "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(self
.vnfd_id
),
2260 headers_text
, None, 200, r_header_text
, "text", temp_dir
+"vnfd-yaml")
2261 # TODO compare files: diff vnfd-yaml hackfest_3charmed_vnfd/hackfest_3charmed_vnfd.yaml
2263 # get vnfd zip file package
2264 engine
.test("Get VNFD zip package", "GET",
2265 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self
.vnfd_id
), headers_zip
, None, 200,
2266 r_header_zip
, "zip", temp_dir
+"vnfd-zip")
2267 # TODO compare files: diff vnfd-zip hackfest_3charmed_vnfd.tar.gz
2270 engine
.test("Get VNFD artifact package", "GET",
2271 "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/osm.png".format(self
.vnfd_id
), headers_zip
, None, 200,
2272 r_header_octect
, "octet-string", temp_dir
+"vnfd-icon")
2273 # TODO compare files: diff vnfd-icon hackfest_3charmed_vnfd/icons/osm.png
2275 # nsd CREATE AND UPLOAD in one step:
2276 engine
.test("Onboard NSD in one step", "POST", "/nsd/v1/ns_descriptors_content", headers_zip_yaml
,
2277 "@b" + nsd_filename_path
, 201, r_headers_yaml_location_nsd
, "yaml")
2278 self
.nsd_id
= engine
.last_id
2280 queries
= ["vld.0.vnfd-connection-point-ref.0.vnfd-id-ref=hf"]
2281 for query
in queries
:
2282 engine
.test("Upload invalid NSD ", "PUT",
2283 "/nsd/v1/ns_descriptors/{}/nsd_content?{}".format(self
.nsd_id
, query
),
2284 headers_zip_yaml
, "@b" + nsd_filename_path
, 422, r_header_yaml
, "yaml")
2286 # get nsd descriptor
2287 engine
.test("Get NSD descriptor", "GET", "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
), headers_yaml
,
2288 None, 200, r_header_yaml
, "yaml")
2290 # get nsd file descriptor
2291 engine
.test("Get NSD file descriptor", "GET", "/nsd/v1/ns_descriptors/{}/nsd".format(self
.nsd_id
), headers_text
,
2292 None, 200, r_header_text
, "text", temp_dir
+"nsd-yaml")
2293 # TODO compare files: diff nsd-yaml hackfest_3charmed_nsd/hackfest_3charmed_nsd.yaml
2295 # get nsd zip file package
2296 engine
.test("Get NSD zip package", "GET", "/nsd/v1/ns_descriptors/{}/nsd_content".format(self
.nsd_id
),
2297 headers_zip
, None, 200, r_header_zip
, "zip", temp_dir
+"nsd-zip")
2298 # TODO compare files: diff nsd-zip hackfest_3charmed_nsd.tar.gz
2301 engine
.test("Get NSD artifact package", "GET",
2302 "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm.png".format(self
.nsd_id
), headers_zip
, None, 200,
2303 r_header_octect
, "octet-string", temp_dir
+"nsd-icon")
2304 # TODO compare files: diff nsd-icon hackfest_3charmed_nsd/icons/osm.png
2307 test_rest
.test("Delete VNFD conflict", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_id
),
2308 headers_yaml
, None, 409, None, None)
2310 test_rest
.test("Delete VNFD force", "DELETE", "/vnfpkgm/v1/vnf_packages/{}?FORCE=TRUE".format(self
.vnfd_id
),
2311 headers_yaml
, None, 204, None, 0)
2314 test_rest
.test("Delete NSD", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self
.nsd_id
), headers_yaml
, None, 204,
2318 class TestNetSliceTemplates
:
2319 description
= "Upload a NST to OSM"
2322 self
.vnfd_filename
= ("@./slice_shared/vnfd/slice_shared_vnfd.yaml")
2323 self
.vnfd_filename_middle
= ("@./slice_shared/vnfd/slice_shared_middle_vnfd.yaml")
2324 self
.nsd_filename
= ("@./slice_shared/nsd/slice_shared_nsd.yaml")
2325 self
.nsd_filename_middle
= ("@./slice_shared/nsd/slice_shared_middle_nsd.yaml")
2326 self
.nst_filenames
= ("@./slice_shared/slice_shared_nstd.yaml")
2328 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
2330 engine
.set_test_name("NST step ")
2331 engine
.get_autorization()
2332 temp_dir
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/temp/"
2333 if not os
.path
.exists(temp_dir
):
2334 os
.makedirs(temp_dir
)
2337 engine
.test("Onboard edge VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml
,
2338 self
.vnfd_filename
, 201, r_headers_yaml_location_vnfd
, "yaml")
2339 self
.vnfd_edge_id
= engine
.last_id
2341 engine
.test("Onboard middle VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml
,
2342 self
.vnfd_filename_middle
, 201, r_headers_yaml_location_vnfd
, "yaml")
2343 self
.vnfd_middle_id
= engine
.last_id
2346 engine
.test("Onboard NSD edge", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml
,
2347 self
.nsd_filename
, 201, r_headers_yaml_location_nsd
, "yaml")
2348 self
.nsd_edge_id
= engine
.last_id
2350 engine
.test("Onboard NSD middle", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml
,
2351 self
.nsd_filename_middle
, 201, r_headers_yaml_location_nsd
, "yaml")
2352 self
.nsd_middle_id
= engine
.last_id
2355 engine
.test("Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml
, self
.nst_filenames
,
2356 201, r_headers_yaml_location_nst
, "yaml")
2357 nst_id
= engine
.last_id
2359 # nstd SHOW OSM format
2360 engine
.test("Show NSTD OSM format", "GET", "/nst/v1/netslice_templates/{}".format(nst_id
), headers_json
, None,
2361 200, r_header_json
, "json")
2364 engine
.test("Delete NSTD", "DELETE", "/nst/v1/netslice_templates/{}".format(nst_id
), headers_json
, None,
2368 test_rest
.test("Delete NSD middle", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self
.nsd_middle_id
),
2369 headers_json
, None, 204, None, 0)
2371 test_rest
.test("Delete NSD edge", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self
.nsd_edge_id
), headers_json
,
2375 test_rest
.test("Delete VNFD edge", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_edge_id
),
2376 headers_yaml
, None, 204, None, 0)
2378 test_rest
.test("Delete VNFD middle", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_middle_id
),
2379 headers_yaml
, None, 204, None, 0)
2382 class TestNetSliceInstances
:
2385 1. Populate databases with VNFD, NSD, NST with the following scenario
2386 +-----------------management-----------------+
2388 +--+---+ +----+----+ +---+--+
2390 | edge +---data1----+ middle +---data2-----+ edge |
2392 +------+ +---------+ +------+
2395 3. Instantiate NSI-1
2397 5. Instantiate NSI-2
2398 Manual check - Are 2 slices instantiated correctly?
2399 NSI-1 3 nss (2 nss-edges + 1 nss-middle)
2400 NSI-2 2 nss (2 nss-edge sharing nss-middle)
2403 Manual check - Is slice NSI-1 deleted correctly?
2404 NSI-2 with 2 nss-edge + 1 nss-middle (The one from NSI-1)
2406 9. Instantiate NSI-3
2407 Manual check - Is slice NSI-3 instantiated correctly?
2408 NSI-3 reuse nss-middle. NSI-3 only create 2 nss-edge
2413 Manual check - All cleaned correctly?
2414 NSI-2 and NSI-3 were terminated and deleted
2415 14. Cleanup database
2418 description
= "Upload a NST to OSM"
2422 self
.vnfd_filename
= ("@./slice_shared/vnfd/slice_shared_vnfd.yaml")
2423 self
.vnfd_filename_middle
= ("@./slice_shared/vnfd/slice_shared_middle_vnfd.yaml")
2424 self
.nsd_filename
= ("@./slice_shared/nsd/slice_shared_nsd.yaml")
2425 self
.nsd_filename_middle
= ("@./slice_shared/nsd/slice_shared_middle_nsd.yaml")
2426 self
.nst_filenames
= ("@./slice_shared/slice_shared_nstd.yaml")
2428 def create_slice(self
, engine
, nsi_data
, name
):
2429 ns_data_text
= yaml
.safe_dump(nsi_data
, default_flow_style
=True, width
=256)
2430 r
= engine
.test(name
, "POST", "/nsilcm/v1/netslice_instances",
2431 headers_yaml
, ns_data_text
, (201, 202),
2432 {"Location": "nsilcm/v1/netslice_instances/", "Content-Type": "application/yaml"}, "yaml")
2435 def instantiate_slice(self
, engine
, nsi_data
, nsi_id
, name
):
2436 ns_data_text
= yaml
.safe_dump(nsi_data
, default_flow_style
=True, width
=256)
2437 engine
.test(name
, "POST",
2438 "/nsilcm/v1/netslice_instances/{}/instantiate".format(nsi_id
), headers_yaml
, ns_data_text
,
2439 (201, 202), r_headers_yaml_location_nsilcmop
, "yaml")
2441 def terminate_slice(self
, engine
, nsi_id
, name
):
2442 engine
.test(name
, "POST", "/nsilcm/v1/netslice_instances/{}/terminate".format(nsi_id
),
2443 headers_yaml
, None, (201, 202), r_headers_yaml_location_nsilcmop
, "yaml")
2445 def delete_slice(self
, engine
, nsi_id
, name
):
2446 engine
.test(name
, "DELETE", "/nsilcm/v1/netslice_instances/{}".format(nsi_id
), headers_yaml
, None,
2449 def run(self
, engine
, test_osm
, manual_check
, test_params
=None):
2451 engine
.set_test_name("NSI")
2452 engine
.get_autorization()
2455 engine
.test("Onboard edge VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml
,
2456 self
.vnfd_filename
, 201, r_headers_yaml_location_vnfd
, "yaml")
2457 self
.vnfd_edge_id
= engine
.last_id
2459 engine
.test("Onboard middle VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml
,
2460 self
.vnfd_filename_middle
, 201, r_headers_yaml_location_vnfd
, "yaml")
2461 self
.vnfd_middle_id
= engine
.last_id
2464 engine
.test("Onboard NSD edge", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml
,
2465 self
.nsd_filename
, 201, r_headers_yaml_location_nsd
, "yaml")
2466 self
.nsd_edge_id
= engine
.last_id
2468 engine
.test("Onboard NSD middle", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml
,
2469 self
.nsd_filename_middle
, 201, r_headers_yaml_location_nsd
, "yaml")
2470 self
.nsd_middle_id
= engine
.last_id
2473 engine
.test("Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml
, self
.nst_filenames
,
2474 201, r_headers_yaml_location_nst
, "yaml")
2475 nst_id
= engine
.last_id
2477 self
.vim_id
= engine
.get_create_vim(test_osm
)
2480 ns_data
= {'nsiName': 'Deploy-NSI-1', 'vimAccountId': self
.vim_id
, 'nstId': nst_id
, 'nsiDescription': 'default'}
2481 r
= self
.create_slice(engine
, ns_data
, "Create NSI-1 step 1")
2484 self
.nsi_id1
= engine
.last_id
2487 self
.instantiate_slice(engine
, ns_data
, self
.nsi_id1
, "Instantiate NSI-1 step 2")
2488 nsilcmop_id1
= engine
.last_id
2492 engine
.wait_operation_ready("nsi", nsilcmop_id1
, timeout_deploy
)
2495 ns_data
= {'nsiName': 'Deploy-NSI-2', 'vimAccountId': self
.vim_id
, 'nstId': nst_id
, 'nsiDescription': 'default'}
2496 r
= self
.create_slice(engine
, ns_data
, "Create NSI-2 step 1")
2499 self
.nsi_id2
= engine
.last_id
2502 self
.instantiate_slice(engine
, ns_data
, self
.nsi_id2
, "Instantiate NSI-2 step 2")
2503 nsilcmop_id2
= engine
.last_id
2507 engine
.wait_operation_ready("nsi", nsilcmop_id2
, timeout_deploy
)
2510 input('NSI-1 AND NSI-2 has been deployed. Perform manual check and press enter to resume')
2514 self
.terminate_slice(engine
, self
.nsi_id1
, "Terminate NSI-1")
2515 nsilcmop1_id
= engine
.last_id
2517 # Wait terminate NSI-1
2518 engine
.wait_operation_ready("nsi", nsilcmop1_id
, timeout_deploy
)
2521 self
.delete_slice(engine
, self
.nsi_id1
, "Delete NS")
2524 input('NSI-1 has been deleted. Perform manual check and press enter to resume')
2527 ns_data
= {'nsiName': 'Deploy-NSI-3', 'vimAccountId': self
.vim_id
, 'nstId': nst_id
, 'nsiDescription': 'default'}
2528 r
= self
.create_slice(engine
, ns_data
, "Create NSI-3 step 1")
2532 self
.nsi_id3
= engine
.last_id
2535 self
.instantiate_slice(engine
, ns_data
, self
.nsi_id3
, "Instantiate NSI-3 step 2")
2536 nsilcmop_id3
= engine
.last_id
2538 # Wait Instantiate NSI-3
2540 engine
.wait_operation_ready("nsi", nsilcmop_id3
, timeout_deploy
)
2543 input('NSI-3 has been deployed. Perform manual check and press enter to resume')
2547 self
.terminate_slice(engine
, self
.nsi_id2
, "Terminate NSI-2")
2548 nsilcmop2_id
= engine
.last_id
2550 # Wait terminate NSI-2
2551 engine
.wait_operation_ready("nsi", nsilcmop2_id
, timeout_deploy
)
2554 self
.delete_slice(engine
, self
.nsi_id2
, "DELETE NSI-2")
2558 self
. terminate_slice(engine
, self
.nsi_id3
, "Terminate NSI-3")
2559 nsilcmop3_id
= engine
.last_id
2561 # Wait terminate NSI-3
2562 engine
.wait_operation_ready("nsi", nsilcmop3_id
, timeout_deploy
)
2565 self
.delete_slice(engine
, self
.nsi_id3
, "DELETE NSI-3")
2568 input('NSI-2 and NSI-3 has been deleted. Perform manual check and press enter to resume')
2571 engine
.test("Delete NSTD", "DELETE", "/nst/v1/netslice_templates/{}".format(nst_id
), headers_json
, None,
2575 test_rest
.test("Delete NSD middle", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self
.nsd_middle_id
),
2576 headers_json
, None, 204, None, 0)
2578 test_rest
.test("Delete NSD edge", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self
.nsd_edge_id
), headers_json
,
2582 test_rest
.test("Delete VNFD edge", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_edge_id
),
2583 headers_yaml
, None, 204, None, 0)
2585 test_rest
.test("Delete VNFD middle", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self
.vnfd_middle_id
),
2586 headers_yaml
, None, 204, None, 0)
2589 class TestAuthentication
:
2590 description
= "Test Authentication"
2593 def run(engine
, test_osm
, manual_check
, test_params
=None):
2594 engine
.set_test_name("Authentication")
2595 # backend = test_params.get("backend") if test_params else None # UNUSED
2597 admin_project_id
= test_project_id
= None
2598 project_admin_role_id
= project_user_role_id
= None
2599 test_user_id
= empty_user_id
= None
2600 default_role_id
= empty_role_id
= token_role_id
= None
2602 engine
.get_autorization()
2605 engine
.test("Get tokens", "GET", "/admin/v1/tokens", headers_json
, {},
2606 (200), {"Content-Type": "application/json"}, "json")
2607 engine
.test("Get projects", "GET", "/admin/v1/projects", headers_json
, {},
2608 (200), {"Content-Type": "application/json"}, "json")
2609 engine
.test("Get users", "GET", "/admin/v1/users", headers_json
, {},
2610 (200), {"Content-Type": "application/json"}, "json")
2611 engine
.test("Get roles", "GET", "/admin/v1/roles", headers_json
, {},
2612 (200), {"Content-Type": "application/json"}, "json")
2613 res
= engine
.test("Get admin project", "GET", "/admin/v1/projects?name=admin", headers_json
, {},
2614 (200), {"Content-Type": "application/json"}, "json")
2615 admin_project_id
= res
.json()[0]["_id"] if res
else None
2616 res
= engine
.test("Get project admin role", "GET", "/admin/v1/roles?name=project_admin", headers_json
, {},
2617 (200), {"Content-Type": "application/json"}, "json")
2618 project_admin_role_id
= res
.json()[0]["_id"] if res
else None
2619 res
= engine
.test("Get project user role", "GET", "/admin/v1/roles?name=project_user", headers_json
, {},
2620 (200), {"Content-Type": "application/json"}, "json")
2621 project_user_role_id
= res
.json()[0]["_id"] if res
else None
2624 res
= engine
.test("Create test project", "POST", "/admin/v1/projects", headers_json
, {"name": "test"},
2625 (201), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
2626 test_project_id
= engine
.last_id
if res
else None
2627 res
= engine
.test("Create role without permissions", "POST", "/admin/v1/roles", headers_json
, {"name": "empty"},
2628 (201), {"Content-Type": "application/json"}, "json")
2629 empty_role_id
= engine
.last_id
if res
else None
2630 res
= engine
.test("Create role with default permissions", "POST", "/admin/v1/roles", headers_json
,
2631 {"name": "default", "permissions": {"default": True}},
2632 (201), {"Location": "/admin/v1/roles/", "Content-Type": "application/json"}, "json")
2633 default_role_id
= engine
.last_id
if res
else None
2634 res
= engine
.test("Create role with token permissions", "POST", "/admin/v1/roles", headers_json
,
2635 {"name": "tokens", "permissions": {"tokens": True}}, # is default required ?
2636 (201), {"Location": "/admin/v1/roles/", "Content-Type": "application/json"}, "json")
2637 token_role_id
= engine
.last_id
if res
else None
2638 pr
= "project-role mappings"
2639 res
= engine
.test("Create user without "+pr
, "POST", "/admin/v1/users", headers_json
,
2640 {"username": "empty", "password": "empty"},
2641 201, {"Content-Type": "application/json"}, "json")
2642 empty_user_id
= engine
.last_id
if res
else None
2643 if admin_project_id
and test_project_id
and project_admin_role_id
and project_user_role_id
:
2644 data
= {"username": "test", "password": "test"}
2645 data
["project_role_mappings"] = [
2646 {"project": test_project_id
, "role": project_admin_role_id
},
2647 {"project": admin_project_id
, "role": project_user_role_id
}
2649 res
= engine
.test("Create user with "+pr
, "POST", "/admin/v1/users", headers_json
, data
,
2650 (201), {"Content-Type": "application/json"}, "json")
2651 test_user_id
= engine
.last_id
if res
else None
2655 engine
.test("Modify test user's password", "PUT", "/admin/v1/users/"+test_user_id
, headers_json
,
2656 {"password": "password"},
2658 if empty_user_id
and admin_project_id
and test_project_id
and project_admin_role_id
and project_user_role_id
:
2659 data
= {"project_role_mappings": [
2660 {"project": test_project_id
, "role": project_admin_role_id
},
2661 {"project": admin_project_id
, "role": project_user_role_id
}
2663 engine
.test("Modify empty user's "+pr
, "PUT", "/admin/v1/users/"+empty_user_id
,
2670 engine
.test("Delete empty user", "DELETE", "/admin/v1/users/"+empty_user_id
, headers_json
, {},
2673 engine
.test("Delete test user", "DELETE", "/admin/v1/users/"+test_user_id
, headers_json
, {},
2676 engine
.test("Delete empty role", "DELETE", "/admin/v1/roles/"+empty_role_id
, headers_json
, {},
2679 engine
.test("Delete default role", "DELETE", "/admin/v1/roles/"+default_role_id
, headers_json
, {},
2682 engine
.test("Delete token role", "DELETE", "/admin/v1/roles/"+token_role_id
, headers_json
, {},
2685 engine
.test("Delete test project", "DELETE", "/admin/v1/projects/"+test_project_id
, headers_json
, {},
2690 engine
.remove_authorization() # To finish
2693 class TestNbiQuotas():
2694 description
= "Test NBI Quotas"
2697 def run(engine
, test_osm
, manual_check
, test_params
=None):
2698 engine
.set_test_name("NBI-Quotas_")
2699 # backend = test_params.get("backend") if test_params else None # UNUSED
2701 test_username
= "test-nbi-quotas"
2702 test_password
= "test-nbi-quotas"
2703 test_project
= "test-nbi-quotas"
2705 test_vim
= "test-nbi-quotas"
2706 test_wim
= "test-nbi-quotas"
2707 test_sdn
= "test-nbi-quotas"
2710 test_project_id
= None
2722 # Save admin access data
2723 admin_username
= engine
.user
2724 admin_password
= engine
.password
2725 admin_project
= engine
.project
2728 engine
.get_autorization()
2729 admin_token
= engine
.last_id
2731 # Check that test project,user do not exist
2732 res1
= engine
.test("Check that test project doesn't exist", "GET", "/admin/v1/projects/"+test_project
,
2733 headers_json
, {}, (404), {}, True)
2734 res2
= engine
.test("Check that test user doesn't exist", "GET", "/admin/v1/users/"+test_username
,
2735 headers_json
, {}, (404), {}, True)
2736 if None in [res1
, res2
]:
2737 engine
.remove_authorization()
2738 logger
.error("Test project and/or user already exist")
2741 # Create test project&user
2742 res
= engine
.test("Create test project", "POST", "/admin/v1/projects", headers_json
,
2743 {"name": test_username
,
2756 (201), r_header_json
, "json")
2757 test_project_id
= engine
.last_id
if res
else None
2758 res
= engine
.test("Create test user", "POST", "/admin/v1/users", headers_json
,
2759 {"username": test_username
, "password": test_password
,
2760 "project_role_mappings": [{"project": test_project
, "role": "project_admin"}]},
2761 (201), r_header_json
, "json")
2762 test_user_id
= engine
.last_id
if res
else None
2764 if test_project_id
and test_user_id
:
2768 engine
.user
= test_username
2769 engine
.password
= test_password
2770 engine
.project
= test_project
2771 engine
.get_autorization()
2772 user_token
= engine
.last_id
2775 res
= engine
.test("Create test VIM", "POST", "/admin/v1/vim_accounts", headers_json
,
2777 "vim_type": "openvim",
2778 "vim_user": test_username
,
2779 "vim_password": test_password
,
2780 "vim_tenant_name": test_project
,
2781 "vim_url": "https://0.0.0.0:0/v0.0",
2783 (202), r_header_json
, "json")
2784 test_vim_ids
+= [engine
.last_id
if res
else None]
2786 res
= engine
.test("Try to create second test VIM", "POST", "/admin/v1/vim_accounts", headers_json
,
2787 {"name": test_vim
+ "_2",
2788 "vim_type": "openvim",
2789 "vim_user": test_username
,
2790 "vim_password": test_password
,
2791 "vim_tenant_name": test_project
,
2792 "vim_url": "https://0.0.0.0:0/v0.0",
2794 (422), r_header_json
, "json")
2795 test_vim_ids
+= [engine
.last_id
if res
is None else None]
2797 res
= engine
.test("Try to create second test VIM with FORCE",
2798 "POST", "/admin/v1/vim_accounts?FORCE", headers_json
,
2799 {"name": test_vim
+ "_3",
2800 "vim_type": "openvim",
2801 "vim_user": test_username
,
2802 "vim_password": test_password
,
2803 "vim_tenant_name": test_project
,
2804 "vim_url": "https://0.0.0.0:0/v0.0",
2806 (202), r_header_json
, "json")
2807 test_vim_ids
+= [engine
.last_id
if res
else None]
2811 # Download descriptor files (if required)
2812 test_dir
= "/tmp/"+test_username
+"/"
2813 test_url
= "https://osm-download.etsi.org/ftp/osm-6.0-six/7th-hackfest/packages/"
2814 vnfd_filenames
= ["slice_hackfest_vnfd.tar.gz", "slice_hackfest_middle_vnfd.tar.gz"]
2815 nsd_filenames
= ["slice_hackfest_nsd.tar.gz", "slice_hackfest_middle_nsd.tar.gz"]
2816 nst_filenames
= ["slice_hackfest_nstd.yaml"]
2817 pdu_filenames
= ["PDU_router.yaml"]
2818 desc_filenames
= vnfd_filenames
+ nsd_filenames
+ nst_filenames
+ pdu_filenames
2819 if not os
.path
.exists(test_dir
):
2820 os
.makedirs(test_dir
)
2821 for filename
in desc_filenames
:
2822 if not os
.path
.exists(test_dir
+filename
):
2823 res
= requests
.get(test_url
+filename
)
2824 if res
.status_code
< 300:
2825 with
open(test_dir
+filename
, "wb") as file:
2826 file.write(res
.content
)
2828 if all([os
.path
.exists(test_dir
+p
) for p
in desc_filenames
]):
2831 res
= engine
.test("Create test VNFD #1", "POST", "/vnfpkgm/v1/vnf_packages_content",
2832 headers_zip_json
, "@b"+test_dir
+vnfd_filenames
[0],
2833 (201), r_header_json
, "json")
2834 test_vnfd_ids
+= [engine
.last_id
if res
else None]
2835 res
= engine
.test("Create test VNFD #2", "POST", "/vnfpkgm/v1/vnf_packages_content",
2836 headers_zip_json
, "@b"+test_dir
+vnfd_filenames
[1],
2837 (201), r_header_json
, "json")
2838 test_vnfd_ids
+= [engine
.last_id
if res
else None]
2839 res
= engine
.test("Try to create extra test VNFD", "POST",
2840 "/vnfpkgm/v1/vnf_packages_content",
2841 headers_zip_json
, "@b"+test_dir
+vnfd_filenames
[0],
2842 (422), r_header_json
, "json")
2843 test_vnfd_ids
+= [engine
.last_id
if res
is None else None]
2844 res
= engine
.test("Try to create extra test VNFD with FORCE",
2845 "POST", "/vnfpkgm/v1/vnf_packages_content?FORCE",
2846 headers_zip_json
, "@b"+test_dir
+vnfd_filenames
[0],
2847 (201), r_header_json
, "json")
2848 test_vnfd_ids
+= [engine
.last_id
if res
else None]
2850 # Remove extra VNFDs to prevent further errors
2852 if test_vnfd_ids
[i
]:
2853 res
= engine
.test("Delete test VNFD #" + str(i
), "DELETE",
2854 "/vnfpkgm/v1/vnf_packages_content/"+test_vnfd_ids
[i
]+"?FORCE",
2855 headers_json
, {}, (204), {}, 0)
2857 test_vnfd_ids
[i
] = None
2859 if test_vnfd_ids
[0] and test_vnfd_ids
[1]:
2862 res
= engine
.test("Create test NSD #1", "POST", "/nsd/v1/ns_descriptors_content",
2863 headers_zip_json
, "@b"+test_dir
+nsd_filenames
[0],
2864 (201), r_header_json
, "json")
2865 test_nsd_ids
+= [engine
.last_id
if res
else None]
2866 res
= engine
.test("Create test NSD #2", "POST", "/nsd/v1/ns_descriptors_content",
2867 headers_zip_json
, "@b"+test_dir
+nsd_filenames
[1],
2868 (201), r_header_json
, "json")
2869 test_nsd_ids
+= [engine
.last_id
if res
else None]
2870 res
= engine
.test("Try to create extra test NSD", "POST", "/nsd/v1/ns_descriptors_content",
2871 headers_zip_json
, "@b"+test_dir
+nsd_filenames
[0],
2872 (422), r_header_json
, "json")
2873 test_nsd_ids
+= [engine
.last_id
if res
is None else None]
2874 res
= engine
.test("Try to create extra test NSD with FORCE",
2875 "POST", "/nsd/v1/ns_descriptors_content?FORCE",
2876 headers_zip_json
, "@b"+test_dir
+nsd_filenames
[0],
2877 (201), r_header_json
, "json")
2878 test_nsd_ids
+= [engine
.last_id
if res
else None]
2880 # Remove extra NSDs to prevent further errors
2883 res
= engine
.test("Delete test NSD #" + str(i
), "DELETE",
2884 "/nsd/v1/ns_descriptors_content/"+test_nsd_ids
[i
]+"?FORCE",
2885 headers_json
, {}, (204), {}, 0)
2887 test_nsd_ids
[i
] = None
2889 if test_nsd_ids
[0] and test_nsd_ids
[1]:
2892 res
= engine
.test("Create test NSR #1", "POST", "/nslcm/v1/ns_instances_content",
2894 {"nsName": test_username
+"_1",
2895 "nsdId": test_nsd_ids
[0],
2896 "vimAccountId": test_vim_ids
[0],
2898 (201), r_header_json
, "json")
2899 test_nsr_ids
+= [engine
.last_id
if res
else None]
2900 res
= engine
.test("Create test NSR #2", "POST", "/nslcm/v1/ns_instances_content",
2902 {"nsName": test_username
+"_2",
2903 "nsdId": test_nsd_ids
[1],
2904 "vimAccountId": test_vim_ids
[0],
2906 (201), r_header_json
, "json")
2907 test_nsr_ids
+= [engine
.last_id
if res
else None]
2908 res
= engine
.test("Try to create extra test NSR", "POST", "/nslcm/v1/ns_instances_content",
2910 {"nsName": test_username
+"_3",
2911 "nsdId": test_nsd_ids
[0],
2912 "vimAccountId": test_vim_ids
[0],
2914 (422), r_header_json
, "json")
2915 test_nsr_ids
+= [engine
.last_id
if res
is None else None]
2916 res
= engine
.test("Try to create test NSR with FORCE", "POST",
2917 "/nslcm/v1/ns_instances_content?FORCE", headers_json
,
2918 {"nsName": test_username
+"_4",
2919 "nsdId": test_nsd_ids
[0],
2920 "vimAccountId": test_vim_ids
[0],
2922 (201), r_header_json
, "json")
2923 test_nsr_ids
+= [engine
.last_id
if res
else None]
2926 res
= engine
.test("Create test NST", "POST", "/nst/v1/netslice_templates_content",
2927 headers_txt_json
, "@b"+test_dir
+nst_filenames
[0],
2928 (201), r_header_json
, "json")
2929 test_nst_ids
+= [engine
.last_id
if res
else None]
2930 res
= engine
.test("Try to create extra test NST", "POST",
2931 "/nst/v1/netslice_templates_content",
2932 headers_txt_json
, "@b"+test_dir
+nst_filenames
[0],
2933 (422), r_header_json
, "json")
2934 test_nst_ids
+= [engine
.last_id
if res
is None else None]
2935 res
= engine
.test("Try to create extra test NST with FORCE", "POST",
2936 "/nst/v1/netslice_templates_content?FORCE",
2937 headers_txt_json
, "@b"+test_dir
+nst_filenames
[0],
2938 (201), r_header_json
, "json")
2939 test_nst_ids
+= [engine
.last_id
if res
else None]
2943 engine
.set_header({"Authorization": "Bearer {}".format(admin_token
)})
2944 res
= engine
.test("Remove NSR Quota", "PUT", "/admin/v1/projects/"+test_project_id
,
2946 {"quotas": {"nsrs": None}},
2948 engine
.set_header({"Authorization": "Bearer {}".format(user_token
)})
2951 res
= engine
.test("Create test NSI", "POST",
2952 "/nsilcm/v1/netslice_instances_content", headers_json
,
2953 {"nsiName": test_username
,
2954 "nstId": test_nst_ids
[0],
2955 "vimAccountId": test_vim_ids
[0],
2957 (201), r_header_json
, "json")
2958 test_nsi_ids
+= [engine
.last_id
if res
else None]
2959 res
= engine
.test("Try to create extra test NSI", "POST",
2960 "/nsilcm/v1/netslice_instances_content", headers_json
,
2961 {"nsiName": test_username
,
2962 "nstId": test_nst_ids
[0],
2963 "vimAccountId": test_vim_ids
[0],
2965 (400), r_header_json
, "json")
2966 test_nsi_ids
+= [engine
.last_id
if res
is None else None]
2967 res
= engine
.test("Try to create extra test NSI with FORCE", "POST",
2968 "/nsilcm/v1/netslice_instances_content?FORCE", headers_json
,
2969 {"nsiName": test_username
,
2970 "nstId": test_nst_ids
[0],
2971 "vimAccountId": test_vim_ids
[0],
2973 (201), r_header_json
, "json")
2974 test_nsi_ids
+= [engine
.last_id
if res
else None]
2977 with
open(test_dir
+pdu_filenames
[0], "rb") as file:
2978 pdu_text
= re
.sub(r
"ip-address: *\[[^\]]*\]", "ip-address: '0.0.0.0'",
2979 file.read().decode("utf-8"))
2980 with
open(test_dir
+pdu_filenames
[0], "wb") as file:
2981 file.write(pdu_text
.encode("utf-8"))
2982 res
= engine
.test("Create test PDU", "POST", "/pdu/v1/pdu_descriptors",
2983 headers_yaml
, "@b"+test_dir
+pdu_filenames
[0],
2984 (201), r_header_yaml
, "yaml")
2985 test_pdu_ids
+= [engine
.last_id
if res
else None]
2986 res
= engine
.test("Try to create extra test PDU", "POST", "/pdu/v1/pdu_descriptors",
2987 headers_yaml
, "@b"+test_dir
+pdu_filenames
[0],
2988 (422), r_header_yaml
, "yaml")
2989 test_pdu_ids
+= [engine
.last_id
if res
is None else None]
2990 res
= engine
.test("Try to create extra test PDU with FORCE", "POST",
2991 "/pdu/v1/pdu_descriptors?FORCE",
2992 headers_yaml
, "@b"+test_dir
+pdu_filenames
[0],
2993 (201), r_header_yaml
, "yaml")
2994 test_pdu_ids
+= [engine
.last_id
if res
else None]
2997 for i
, id in enumerate(test_nsi_ids
):
2999 engine
.test("Delete test NSI #"+str(i
), "DELETE",
3000 "/nsilcm/v1/netslice_instances_content/"+id+"?FORCE",
3001 headers_json
, {}, (204), {}, 0)
3002 for i
, id in enumerate(test_nsr_ids
):
3004 engine
.test("Delete test NSR #"+str(i
), "DELETE",
3005 "/nslcm/v1/ns_instances_content/"+id+"?FORCE",
3006 headers_json
, {}, (204), {}, 0)
3007 for i
, id in enumerate(test_nst_ids
):
3009 engine
.test("Delete test NST #"+str(i
), "DELETE",
3010 "/nst/v1/netslice_templates_content/"+id+"?FORCE",
3011 headers_json
, {}, (204), {}, 0)
3012 for i
, id in enumerate(test_nsd_ids
):
3014 engine
.test("Delete test NSD #"+str(i
), "DELETE",
3015 "/nsd/v1/ns_descriptors_content/"+id+"?FORCE",
3016 headers_json
, {}, (204), {}, 0)
3017 for i
, id in enumerate(test_vnfd_ids
):
3019 engine
.test("Delete test VNFD #"+str(i
), "DELETE",
3020 "/vnfpkgm/v1/vnf_packages_content/"+id+"?FORCE",
3021 headers_json
, {}, (204), {}, 0)
3022 for i
, id in enumerate(test_pdu_ids
):
3024 engine
.test("Delete test PDU #"+str(i
), "DELETE",
3025 "/pdu/v1/pdu_descriptors/"+id+"?FORCE",
3026 headers_json
, {}, (204), {}, 0)
3028 # END Test NBI Quotas
3031 res
= engine
.test("Create test WIM", "POST", "/admin/v1/wim_accounts", headers_json
,
3034 "wim_url": "https://0.0.0.0:0/v0.0",
3036 (202), r_header_json
, "json")
3037 test_wim_ids
+= [engine
.last_id
if res
else None]
3038 res
= engine
.test("Try to create second test WIM", "POST", "/admin/v1/wim_accounts", headers_json
,
3039 {"name": test_wim
+ "_2",
3041 "wim_url": "https://0.0.0.0:0/v0.0",
3043 (422), r_header_json
, "json")
3044 test_wim_ids
+= [engine
.last_id
if res
is None else None]
3045 res
= engine
.test("Try to create second test WIM with FORCE", "POST", "/admin/v1/wim_accounts?FORCE",
3047 {"name": test_wim
+ "_3",
3049 "wim_url": "https://0.0.0.0:0/v0.0",
3051 (202), r_header_json
, "json")
3052 test_wim_ids
+= [engine
.last_id
if res
else None]
3055 res
= engine
.test("Create test SDN", "POST", "/admin/v1/sdns", headers_json
,
3060 "dpid": "00:00:00:00:00:00:00:00",
3062 (202), r_header_json
, "json")
3063 test_sdn_ids
+= [engine
.last_id
if res
else None]
3064 res
= engine
.test("Try to create second test SDN", "POST", "/admin/v1/sdns", headers_json
,
3065 {"name": test_sdn
+ "_2",
3069 "dpid": "00:00:00:00:00:00:00:00",
3071 (422), r_header_json
, "json")
3072 test_sdn_ids
+= [engine
.last_id
if res
is None else None]
3073 res
= engine
.test("Try to create second test SDN with FORCE", "POST", "/admin/v1/sdns?FORCE", headers_json
,
3074 {"name": test_sdn
+ "_3",
3078 "dpid": "00:00:00:00:00:00:00:00",
3080 (202), r_header_json
, "json")
3081 test_sdn_ids
+= [engine
.last_id
if res
else None]
3084 for i
, id in enumerate(test_vim_ids
):
3086 engine
.test("Delete test VIM #"+str(i
), "DELETE", "/admin/v1/vim_accounts/"+id+"?FORCE",
3087 headers_json
, {}, (202), {}, 0)
3088 for i
, id in enumerate(test_wim_ids
):
3090 engine
.test("Delete test WIM #"+str(i
), "DELETE", "/admin/v1/wim_accounts/"+id+"?FORCE",
3091 headers_json
, {}, (202), {}, 0)
3092 for i
, id in enumerate(test_sdn_ids
):
3094 engine
.test("Delete test SDN #"+str(i
), "DELETE", "/admin/v1/sdns/"+id+"?FORCE",
3095 headers_json
, {}, (202), {}, 0)
3097 # Release user access
3098 engine
.remove_authorization()
3101 engine
.user
= admin_username
3102 engine
.password
= admin_password
3103 engine
.project
= admin_project
3104 engine
.get_autorization()
3106 engine
.test("Delete test user", "DELETE", "/admin/v1/users/"+test_user_id
+"?FORCE",
3107 headers_json
, {}, (204), {}, 0)
3109 engine
.test("Delete test project", "DELETE", "/admin/v1/projects/"+test_project_id
+"?FORCE",
3110 headers_json
, {}, (204), {}, 0)
3111 engine
.remove_authorization()
3113 # END class TestNbiQuotas
3116 if __name__
== "__main__":
3120 # Disable warnings from self-signed certificates.
3121 requests
.packages
.urllib3
.disable_warnings()
3123 logging
.basicConfig(format
="%(levelname)s %(message)s", level
=logging
.ERROR
)
3124 logger
= logging
.getLogger('NBI')
3125 # load parameters and configuration
3126 opts
, args
= getopt
.getopt(sys
.argv
[1:], "hvu:p:",
3127 ["url=", "user=", "password=", "help", "version", "verbose", "no-verbose",
3128 "project=", "insecure", "timeout", "timeout-deploy", "timeout-configure",
3129 "test=", "list", "test-osm", "manual-check", "params=", 'fail-fast'])
3130 url
= "https://localhost:9999/osm"
3131 user
= password
= project
= "admin"
3133 manual_check
= False
3138 "NonAuthorized": TestNonAuthorized
,
3139 "FakeVIM": TestFakeVim
,
3140 "Users-Projects": TestUsersProjects
,
3141 "Projects-Descriptors": TestProjectsDescriptors
,
3142 "VIM-SDN": TestVIMSDN
,
3143 "Deploy-Custom": TestDeploy
,
3144 "Deploy-Hackfest-Cirros": TestDeployHackfestCirros
,
3145 "Deploy-Hackfest-Cirros-Scaling": TestDeployHackfestCirrosScaling
,
3146 "Deploy-Hackfest-3Charmed": TestDeployHackfest3Charmed
,
3147 "Deploy-Hackfest-3Charmed2": TestDeployHackfest3Charmed2
,
3148 "Deploy-Hackfest-3Charmed3": TestDeployHackfest3Charmed3
,
3149 "Deploy-Hackfest-4": TestDeployHackfest4
,
3150 "Deploy-CirrosMacIp": TestDeployIpMac
,
3151 "Descriptors": TestDescriptors
,
3152 "Deploy-Hackfest1": TestDeployHackfest1
,
3153 # "Deploy-MultiVIM": TestDeployMultiVIM,
3154 "Deploy-SingleVdu": TestDeploySingleVdu
,
3155 "Deploy-Hnfd": TestDeployHnfd
,
3156 "Upload-Slice-Template": TestNetSliceTemplates
,
3157 "Deploy-Slice-Instance": TestNetSliceInstances
,
3158 "Deploy-SimpleCharm": TestDeploySimpleCharm
,
3159 "Deploy-SimpleCharm2": TestDeploySimpleCharm2
,
3160 "Authentication": TestAuthentication
,
3161 "NBI-Quotas": TestNbiQuotas
,
3167 # print("parameter:", o, a)
3168 if o
== "--version":
3169 print("test version " + __version__
+ ' ' + version_date
)
3172 for test
, test_class
in sorted(test_classes
.items()):
3173 print("{:32} {}".format(test
+ ":", test_class
.description
))
3175 elif o
in ("-v", "--verbose"):
3177 elif o
== "no-verbose":
3179 elif o
in ("-h", "--help"):
3182 elif o
== "--test-osm":
3184 elif o
== "--manual-check":
3188 elif o
in ("-u", "--user"):
3190 elif o
in ("-p", "--password"):
3192 elif o
== "--project":
3194 elif o
== "--fail-fast":
3197 # print("asdfadf", o, a, a.split(","))
3198 for _test
in a
.split(","):
3199 if _test
not in test_classes
:
3200 print("Invalid test name '{}'. Use option '--list' to show available tests".format(_test
),
3203 test_to_do
.append(_test
)
3204 elif o
== "--params":
3205 param_key
, _
, param_value
= a
.partition("=")
3206 text_index
= len(test_to_do
)
3207 if text_index
not in test_params
:
3208 test_params
[text_index
] = {}
3209 test_params
[text_index
][param_key
] = param_value
3210 elif o
== "--insecure":
3212 elif o
== "--timeout":
3214 elif o
== "--timeout-deploy":
3215 timeout_deploy
= int(a
)
3216 elif o
== "--timeout-configure":
3217 timeout_configure
= int(a
)
3219 assert False, "Unhandled option"
3221 logger
.setLevel(logging
.WARNING
)
3223 logger
.setLevel(logging
.DEBUG
)
3225 logger
.setLevel(logging
.ERROR
)
3227 test_rest
= TestRest(url
, user
=user
, password
=password
, project
=project
)
3228 # print("tests to do:", test_to_do)
3231 for test
in test_to_do
:
3232 if fail_fast
and test_rest
.failed_tests
:
3235 test_class
= test_classes
[test
]
3236 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(text_index
))
3238 for test
, test_class
in sorted(test_classes
.items()):
3239 if fail_fast
and test_rest
.failed_tests
:
3241 test_class().run(test_rest
, test_osm
, manual_check
, test_params
.get(0))
3242 test_rest
.print_results()
3243 exit(1 if test_rest
.failed_tests
else 0)
3245 except TestException
as e
:
3246 logger
.error(test
+ "Test {} Exception: {}".format(test
, str(e
)))
3248 except getopt
.GetoptError
as e
:
3250 print(e
, file=sys
.stderr
)
3252 except Exception as e
:
3253 logger
.critical(test
+ " Exception: " + str(e
), exc_info
=True)