dd3c796838387ce2e73bc9b2815ba72d5b36f501
[osm/NBI.git] / osm_nbi / tests / test.py
1 #! /usr/bin/python3
2 # -*- coding: utf-8 -*-
3
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
7 #
8 # http://www.apache.org/licenses/LICENSE-2.0
9 #
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
13 # implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16
17 import getopt
18 import sys
19 import requests
20 import json
21 import logging
22 import yaml
23 # import json
24 # import tarfile
25 from time import sleep
26 from random import randint
27 import os
28 from sys import stderr
29 from uuid import uuid4
30
31 __author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
32 __date__ = "$2018-03-01$"
33 __version__ = "0.3"
34 version_date = "Oct 2018"
35
36
37 def usage():
38 print("Usage: ", sys.argv[0], "[options]")
39 print(" Performs system tests over running NBI. It can be used for real OSM test using option '--test-osm'")
40 print(" If this is the case env variables 'OSMNBITEST_VIM_NAME' must be supplied to create a VIM if not exist "
41 "where deployment is done")
42 print("OPTIONS")
43 print(" -h|--help: shows this help")
44 print(" --insecure: Allows non trusted https NBI server")
45 print(" --list: list available tests")
46 print(" --manual-check: Deployment tests stop after deployed to allow manual inspection. Only make sense with "
47 "'--test-osm'")
48 print(" -p|--password PASSWORD: NBI access password. 'admin' by default")
49 print(" ---project PROJECT: NBI access project. 'admin' by default")
50 print(" --test TEST[,...]: Execute only a test or a comma separated list of tests")
51 print(" --params key=val: params to the previous test. key can be vnfd-files, nsd-file, ns-name, ns-config")
52 print(" --test-osm: If missing this test is intended for NBI only, no other OSM components are expected. Use "
53 "this flag to test the system. LCM and RO components are expected to be up and running")
54 print(" --timeout TIMEOUT: General NBI timeout, by default {}s".format(timeout))
55 print(" --timeout-deploy TIMEOUT: Timeout used for getting NS deployed, by default {}s".format(timeout_deploy))
56 print(" --timeout-configure TIMEOUT: Timeout used for getting NS deployed and configured,"
57 " by default {}s".format(timeout_configure))
58 print(" -u|--user USERNAME: NBI access username. 'admin' by default")
59 print(" --url URL: complete NBI server URL. 'https//localhost:9999/osm' by default")
60 print(" -v|--verbose print debug information, can be used several times")
61 print(" --no-verbose remove verbosity")
62 print(" --version: prints current version")
63 print("ENV variables used for real deployment tests with option osm-test.")
64 print(" export OSMNBITEST_VIM_NAME=vim-name")
65 print(" export OSMNBITEST_VIM_URL=vim-url")
66 print(" export OSMNBITEST_VIM_TYPE=vim-type")
67 print(" export OSMNBITEST_VIM_TENANT=vim-tenant")
68 print(" export OSMNBITEST_VIM_USER=vim-user")
69 print(" export OSMNBITEST_VIM_PASSWORD=vim-password")
70 print(" export OSMNBITEST_VIM_CONFIG=\"vim-config\"")
71 print(" export OSMNBITEST_NS_NAME=\"vim-config\"")
72 return
73
74
75 r_header_json = {"Content-type": "application/json"}
76 headers_json = {"Content-type": "application/json", "Accept": "application/json"}
77 r_header_yaml = {"Content-type": "application/yaml"}
78 headers_yaml = {"Content-type": "application/yaml", "Accept": "application/yaml"}
79 r_header_text = {"Content-type": "text/plain"}
80 r_header_octect = {"Content-type": "application/octet-stream"}
81 headers_text = {"Accept": "text/plain,application/yaml"}
82 r_header_zip = {"Content-type": "application/zip"}
83 headers_zip = {"Accept": "application/zip,application/yaml"}
84 headers_zip_yaml = {"Accept": "application/yaml", "Content-type": "application/zip"}
85 r_headers_yaml_location_vnfd = {"Location": "/vnfpkgm/v1/vnf_packages_content/", "Content-Type": "application/yaml"}
86 r_headers_yaml_location_nsd = {"Location": "/nsd/v1/ns_descriptors_content/", "Content-Type": "application/yaml"}
87 r_headers_yaml_location_nst = {"Location": "/nst/v1/netslice_templates_content", "Content-Type": "application/yaml"}
88 r_headers_yaml_location_nslcmop = {"Location": "nslcm/v1/ns_lcm_op_occs/", "Content-Type": "application/yaml"}
89 r_headers_yaml_location_nsilcmop = {"Location": "/osm/nsilcm/v1/nsi_lcm_op_occs/", "Content-Type": "application/yaml"}
90
91 # test ones authorized
92 test_authorized_list = (
93 ("AU1", "Invalid vnfd id", "GET", "/vnfpkgm/v1/vnf_packages/non-existing-id",
94 headers_json, None, 404, r_header_json, "json"),
95 ("AU2", "Invalid nsd id", "GET", "/nsd/v1/ns_descriptors/non-existing-id",
96 headers_yaml, None, 404, r_header_yaml, "yaml"),
97 ("AU3", "Invalid nsd id", "DELETE", "/nsd/v1/ns_descriptors_content/non-existing-id",
98 headers_yaml, None, 404, r_header_yaml, "yaml"),
99 )
100 timeout = 120 # general timeout
101 timeout_deploy = 60*10 # timeout for NS deploying without charms
102 timeout_configure = 60*20 # timeout for NS deploying and configuring
103
104
105 class TestException(Exception):
106 pass
107
108
109 class TestRest:
110 def __init__(self, url_base, header_base=None, verify=False, user="admin", password="admin", project="admin"):
111 self.url_base = url_base
112 if header_base is None:
113 self.header_base = {}
114 else:
115 self.header_base = header_base.copy()
116 self.s = requests.session()
117 self.s.headers = self.header_base
118 self.verify = verify
119 self.token = False
120 self.user = user
121 self.password = password
122 self.project = project
123 self.vim_id = None
124 # contains ID of tests obtained from Location response header. "" key contains last obtained id
125 self.last_id = ""
126 self.test_name = None
127 self.step = 0 # number of subtest under test
128 self.passed_tests = 0
129 self.failed_tests = 0
130
131 def set_test_name(self, test_name):
132 self.test_name = test_name
133 self.step = 0
134 self.last_id = ""
135
136 def set_header(self, header):
137 self.s.headers.update(header)
138
139 def set_tet_name(self, test_name):
140 self.test_name = test_name
141
142 def unset_header(self, key):
143 if key in self.s.headers:
144 del self.s.headers[key]
145
146 def test(self, description, method, url, headers, payload, expected_codes, expected_headers,
147 expected_payload, store_file=None, pooling=False):
148 """
149 Performs an http request and check http code response. Exit if different than allowed. It get the returned id
150 that can be used by following test in the URL with {name} where name is the name of the test
151 :param description: description of the test
152 :param method: HTTP method: GET,PUT,POST,DELETE,...
153 :param url: complete URL or relative URL
154 :param headers: request headers to add to the base headers
155 :param payload: Can be a dict, transformed to json, a text or a file if starts with '@'
156 :param expected_codes: expected response codes, can be int, int tuple or int range
157 :param expected_headers: expected response headers, dict with key values
158 :param expected_payload: expected payload, 0 if empty, 'yaml', 'json', 'text', 'zip', 'octet-stream'
159 :param store_file: filename to store content
160 :param pooling: if True do not count neither log this test. Because a pooling is done with many equal requests
161 :return: requests response
162 """
163 r = None
164 try:
165 if not self.s:
166 self.s = requests.session()
167 # URL
168 if not url:
169 url = self.url_base
170 elif not url.startswith("http"):
171 url = self.url_base + url
172
173 # replace url <> with the last ID
174 url = url.replace("<>", self.last_id)
175 if payload:
176 if isinstance(payload, str):
177 if payload.startswith("@"):
178 mode = "r"
179 file_name = payload[1:]
180 if payload.startswith("@b"):
181 mode = "rb"
182 file_name = payload[2:]
183 with open(file_name, mode) as f:
184 payload = f.read()
185 elif isinstance(payload, dict):
186 payload = json.dumps(payload)
187
188 if not pooling:
189 test_description = "Test {}{} {} {} {}".format(self.test_name, self.step, description, method, url)
190 logger.warning(test_description)
191 self.step += 1
192 stream = False
193 if expected_payload in ("zip", "octet-string") or store_file:
194 stream = True
195 __retry = 0
196 while True:
197 try:
198 r = getattr(self.s, method.lower())(url, data=payload, headers=headers, verify=self.verify,
199 stream=stream)
200 break
201 except requests.exceptions.ConnectionError as e:
202 if __retry == 2:
203 raise
204 logger.error("Exception {}. Retrying".format(e))
205 __retry += 1
206
207 if expected_payload in ("zip", "octet-string") or store_file:
208 logger.debug("RX {}".format(r.status_code))
209 else:
210 logger.debug("RX {}: {}".format(r.status_code, r.text))
211
212 # check response
213 if expected_codes:
214 if isinstance(expected_codes, int):
215 expected_codes = (expected_codes,)
216 if r.status_code not in expected_codes:
217 raise TestException(
218 "Got status {}. Expected {}. {}".format(r.status_code, expected_codes, r.text))
219
220 if expected_headers:
221 for header_key, header_val in expected_headers.items():
222 if header_key.lower() not in r.headers:
223 raise TestException("Header {} not present".format(header_key))
224 if header_val and header_val.lower() not in r.headers[header_key]:
225 raise TestException("Header {} does not contain {} but {}".format(header_key, header_val,
226 r.headers[header_key]))
227
228 if expected_payload is not None:
229 if expected_payload == 0 and len(r.content) > 0:
230 raise TestException("Expected empty payload")
231 elif expected_payload == "json":
232 try:
233 r.json()
234 except Exception as e:
235 raise TestException("Expected json response payload, but got Exception {}".format(e))
236 elif expected_payload == "yaml":
237 try:
238 yaml.safe_load(r.text)
239 except Exception as e:
240 raise TestException("Expected yaml response payload, but got Exception {}".format(e))
241 elif expected_payload in ("zip", "octet-string"):
242 if len(r.content) == 0:
243 raise TestException("Expected some response payload, but got empty")
244 # try:
245 # tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
246 # for tarinfo in tar:
247 # tarname = tarinfo.name
248 # print(tarname)
249 # except Exception as e:
250 # raise TestException("Expected zip response payload, but got Exception {}".format(e))
251 elif expected_payload == "text":
252 if len(r.content) == 0:
253 raise TestException("Expected some response payload, but got empty")
254 # r.text
255 if store_file:
256 with open(store_file, 'wb') as fd:
257 for chunk in r.iter_content(chunk_size=128):
258 fd.write(chunk)
259
260 location = r.headers.get("Location")
261 if location:
262 _id = location[location.rfind("/") + 1:]
263 if _id:
264 self.last_id = str(_id)
265 if not pooling:
266 self.passed_tests += 1
267 return r
268 except TestException as e:
269 self.failed_tests += 1
270 r_status_code = None
271 r_text = None
272 if r:
273 r_status_code = r.status_code
274 r_text = r.text
275 logger.error("{} \nRX code{}: {}".format(e, r_status_code, r_text))
276 return None
277 # exit(1)
278 except IOError as e:
279 if store_file:
280 logger.error("Cannot open file {}: {}".format(store_file, e))
281 else:
282 logger.error("Exception: {}".format(e), exc_info=True)
283 self.failed_tests += 1
284 return None
285 # exit(1)
286 except requests.exceptions.RequestException as e:
287 logger.error("Exception: {}".format(e))
288
289 def get_autorization(self): # user=None, password=None, project=None):
290 if self.token: # and self.user == user and self.password == password and self.project == project:
291 return
292 # self.user = user
293 # self.password = password
294 # self.project = project
295 r = self.test("Obtain token", "POST", "/admin/v1/tokens", headers_json,
296 {"username": self.user, "password": self.password, "project_id": self.project},
297 (200, 201), r_header_json, "json")
298 if not r:
299 return
300 response = r.json()
301 self.token = response["id"]
302 self.set_header({"Authorization": "Bearer {}".format(self.token)})
303
304 def remove_authorization(self):
305 if self.token:
306 self.test("Delete token", "DELETE", "/admin/v1/tokens/{}".format(self.token), headers_json,
307 None, (200, 201, 204), None, None)
308 self.token = None
309 self.unset_header("Authorization")
310
311 def get_create_vim(self, test_osm):
312 if self.vim_id:
313 return self.vim_id
314 self.get_autorization()
315 if test_osm:
316 vim_name = os.environ.get("OSMNBITEST_VIM_NAME")
317 if not vim_name:
318 raise TestException(
319 "Needed to define OSMNBITEST_VIM_XXX variables to create a real VIM for deployment")
320 else:
321 vim_name = "fakeVim"
322 # Get VIM
323 r = self.test("Get VIM ID", "GET", "/admin/v1/vim_accounts?name={}".format(vim_name), headers_json,
324 None, 200, r_header_json, "json")
325 if not r:
326 return
327 vims = r.json()
328 if vims:
329 return vims[0]["_id"]
330 # Add VIM
331 if test_osm:
332 # check needed environ parameters:
333 if not os.environ.get("OSMNBITEST_VIM_URL") or not os.environ.get("OSMNBITEST_VIM_TENANT"):
334 raise TestException("Env OSMNBITEST_VIM_URL and OSMNBITEST_VIM_TENANT are needed for create a real VIM"
335 " to deploy on whit the --test-osm option")
336 vim_data = "{{schema_version: '1.0', name: '{}', vim_type: {}, vim_url: '{}', vim_tenant_name: '{}', "\
337 "vim_user: {}, vim_password: {}".format(vim_name,
338 os.environ.get("OSMNBITEST_VIM_TYPE", "openstack"),
339 os.environ.get("OSMNBITEST_VIM_URL"),
340 os.environ.get("OSMNBITEST_VIM_TENANT"),
341 os.environ.get("OSMNBITEST_VIM_USER"),
342 os.environ.get("OSMNBITEST_VIM_PASSWORD"))
343 if os.environ.get("OSMNBITEST_VIM_CONFIG"):
344 vim_data += " ,config: {}".format(os.environ.get("OSMNBITEST_VIM_CONFIG"))
345 vim_data += "}"
346 else:
347 vim_data = "{schema_version: '1.0', name: fakeVim, vim_type: openstack, vim_url: 'http://10.11.12.13/fake'"\
348 ", vim_tenant_name: 'vimtenant', vim_user: vimuser, vim_password: vimpassword}"
349 self.test("Create VIM", "POST", "/admin/v1/vim_accounts", headers_yaml, vim_data,
350 (201, 202), {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/yaml"}, "yaml")
351 return self.last_id
352
353 def print_results(self):
354 print("\n\n\n--------------------------------------------")
355 print("TEST RESULTS: Total: {}, Passed: {}, Failed: {}".format(self.passed_tests + self.failed_tests,
356 self.passed_tests, self.failed_tests))
357 print("--------------------------------------------")
358
359 def wait_until_delete(self, url_op, timeout_delete):
360 """
361 Make a pooling until topic is not present, because of deleted
362 :param url_op:
363 :param timeout_delete:
364 :return:
365 """
366 description = "Wait to topic being deleted"
367 test_description = "Test {}{} {} {} {}".format(self.test_name, self.step, description, "GET", url_op)
368 logger.warning(test_description)
369 self.step += 1
370
371 wait = timeout_delete
372 while wait >= 0:
373 r = self.test(description, "GET", url_op, headers_yaml, None, (200, 404), None, r_header_yaml, "yaml",
374 pooling=True)
375 if not r:
376 return
377 if r.status_code == 404:
378 self.passed_tests += 1
379 break
380 elif r.status_code == 200:
381 wait -= 5
382 sleep(5)
383 else:
384 raise TestException("Topic is not deleted after {} seconds".format(timeout_delete))
385 self.failed_tests += 1
386
387 def wait_operation_ready(self, ns_nsi, opp_id, timeout, expected_fail=False):
388 """
389 Wait until nslcmop or nsilcmop finished
390 :param ns_nsi: "ns" o "nsi"
391 :param opp_id: Id o fthe operation
392 :param timeout:
393 :param expected_fail:
394 :return: None. Updates passed/failed_tests
395 """
396 if ns_nsi == "ns":
397 url_op = "/nslcm/v1/ns_lcm_op_occs/{}".format(opp_id)
398 else:
399 url_op = "/nsilcm/v1/nsi_lcm_op_occs/{}".format(opp_id)
400 description = "Wait to {} lcm operation complete".format(ns_nsi)
401 test_description = "Test {}{} {} {} {}".format(self.test_name, self.step, description, "GET", url_op)
402 logger.warning(test_description)
403 self.step += 1
404 wait = timeout
405 while wait >= 0:
406 r = self.test(description, "GET", url_op, headers_json, None,
407 200, r_header_json, "json", pooling=True)
408 if not r:
409 return
410 nslcmop = r.json()
411 if "COMPLETED" in nslcmop["operationState"]:
412 if expected_fail:
413 logger.error("NS terminate has success, expecting failing: {}".format(nslcmop["detailed-status"]))
414 self.failed_tests += 1
415 else:
416 self.passed_tests += 1
417 break
418 elif "FAILED" in nslcmop["operationState"]:
419 if not expected_fail:
420 logger.error("NS terminate has failed: {}".format(nslcmop["detailed-status"]))
421 self.failed_tests += 1
422 else:
423 self.passed_tests += 1
424 break
425
426 print(".", end="", file=stderr)
427 wait -= 10
428 sleep(10)
429 else:
430 self.failed_tests += 1
431 logger.error("NS instantiate is not terminate after {} seconds".format(timeout))
432 return
433 print("", file=stderr)
434
435
436 class TestNonAuthorized:
437 description = "Test invalid URLs. methods and no authorization"
438
439 @staticmethod
440 def run(engine, test_osm, manual_check, test_params=None):
441 engine.set_test_name("NonAuth")
442 engine.remove_authorization()
443 test_not_authorized_list = (
444 ("Invalid token", "GET", "/admin/v1/users", headers_json, None, 401, r_header_json, "json"),
445 ("Invalid URL", "POST", "/admin/v1/nonexist", headers_yaml, None, 405, r_header_yaml, "yaml"),
446 ("Invalid version", "DELETE", "/admin/v2/users", headers_yaml, None, 405, r_header_yaml, "yaml"),
447 )
448 for t in test_not_authorized_list:
449 engine.test(*t)
450
451
452 class TestUsersProjects:
453 description = "test project and user creation"
454
455 @staticmethod
456 def run(engine, test_osm, manual_check, test_params=None):
457 engine.set_test_name("UserProject")
458 # backend = test_params.get("backend") if test_params else None # UNUSED
459
460 # Initialisation
461 p1 = p2 = p3 = None
462 padmin = pbad = None
463 u1 = u2 = u3 = u4 = None
464
465 engine.get_autorization()
466
467 res = engine.test("Create project non admin 1", "POST", "/admin/v1/projects", headers_json, {"name": "P1"},
468 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
469 p1 = engine.last_id if res else None
470
471 res = engine.test("Create project admin", "POST", "/admin/v1/projects", headers_json,
472 {"name": "Padmin", "admin": True}, (201, 204),
473 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
474 padmin = engine.last_id if res else None
475
476 res = engine.test("Create project bad format", "POST", "/admin/v1/projects", headers_json, {"name": 1},
477 (400, 422), r_header_json, "json")
478 pbad = engine.last_id if res else None
479
480 res = engine.test("Get project admin role", "GET", "/admin/v1/roles?name=project_admin", headers_json, {},
481 (200), {"Content-Type": "application/json"}, "json")
482 rpa = res.json()[0]["_id"] if res else None
483 res = engine.test("Get project user role", "GET", "/admin/v1/roles?name=project_user", headers_json, {},
484 (200), {"Content-Type": "application/json"}, "json")
485 rpu = res.json()[0]["_id"] if res else None
486 res = engine.test("Get system admin role", "GET", "/admin/v1/roles?name=system_admin", headers_json, {},
487 (200), {"Content-Type": "application/json"}, "json")
488 rsa = res.json()[0]["_id"] if res else None
489
490 data = {"username": "U1", "password": "pw1"}
491 p2 = uuid4().hex
492 data["project_role_mappings"] = [
493 {"project": p1, "role": rpa},
494 {"project": p2, "role": rpa},
495 {"project": padmin, "role": rpu}
496 ]
497 rc = 201
498 xhd = {"Location": "/admin/v1/users/", "Content-Type": "application/json"}
499 res = engine.test("Create user with bad project and force", "POST", "/admin/v1/users?FORCE=True", headers_json,
500 data, rc, xhd, "json")
501 if res:
502 u1 = engine.last_id
503 else:
504 # User is created sometimes even though an exception is raised
505 res = engine.test("Get user U1", "GET", "/admin/v1/users?username=U1", headers_json, {},
506 (200), {"Content-Type": "application/json"}, "json")
507 u1 = res.json()[0]["_id"] if res else None
508
509 data = {"username": "U2", "password": "pw2"}
510 data["project_role_mappings"] = [{"project": p1, "role": rpa}, {"project": padmin, "role": rsa}]
511 res = engine.test("Create user 2", "POST", "/admin/v1/users", headers_json,
512 data, 201, {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
513 u2 = engine.last_id if res else None
514
515 if u1:
516 ftt = "project_role_mappings"
517 xpr = [{"project": p1, "role": rpa}, {"project": padmin, "role": rpu}]
518 data = {ftt: xpr}
519 engine.test("Edit user U1, delete P2 project", "PATCH", "/admin/v1/users/"+u1, headers_json,
520 data, 204, None, None)
521 res = engine.test("Check user U1, contains the right projects", "GET", "/admin/v1/users/"+u1,
522 headers_json, None, 200, None, json)
523 if res:
524 rj = res.json()
525 xpr[0]["project_name"] = "P1"
526 xpr[0]["role_name"] = "project_admin"
527 xpr[1]["project_name"] = "Padmin"
528 xpr[1]["role_name"] = "project_user"
529 ok = True
530 for pr in rj[ftt]:
531 if pr not in xpr:
532 ok = False
533 for pr in xpr:
534 if pr not in rj[ftt]:
535 ok = False
536 if not ok:
537 logger.error("User {} '{}' are different than expected '{}'. Edition was not done properly"
538 .format(ftt, rj[ftt], xpr))
539 engine.failed_tests += 1
540
541 p2 = None # To prevent deletion attempts
542
543 # Add a test of 'default project' for Keystone?
544
545 if u2:
546 engine.test("Edit user U2, change password", "PUT", "/admin/v1/users/"+u2, headers_json,
547 {"password": "pw2_new"}, 204, None, None)
548
549 if p1:
550 engine.test("Change to project P1 non existing", "POST", "/admin/v1/tokens/", headers_json,
551 {"project_id": p1}, 401, r_header_json, "json")
552
553 if u2 and p1:
554 res = engine.test("Change to user U2 project P1", "POST", "/admin/v1/tokens", headers_json,
555 {"username": "U2", "password": "pw2_new", "project_id": "P1"}, (200, 201),
556 r_header_json, "json")
557 if res:
558 rj = res.json()
559 engine.set_header({"Authorization": "Bearer {}".format(rj["id"])})
560
561 engine.test("Edit user projects non admin", "PUT", "/admin/v1/users/U1", headers_json,
562 {"remove_project_role_mappings": [{"project": "P1", "role": None}]},
563 401, r_header_json, "json")
564
565 res = engine.test("Add new project non admin", "POST", "/admin/v1/projects", headers_json,
566 {"name": "P2"}, 401, r_header_json, "json")
567 if res is None or res.status_code == 201:
568 # The project has been created even though it shouldn't
569 res = engine.test("Get project P2", "GET", "/admin/v1/projects/P2", headers_json, None,
570 200, r_header_json, "json")
571 p2 = res.json()["_id"] if res else None
572
573 if p1:
574 data = {"username": "U3", "password": "pw3"}
575 data["project_role_mappings"] = [{"project": p1, "role": rpu}]
576 res = engine.test("Add new user non admin", "POST", "/admin/v1/users", headers_json,
577 data, 401, r_header_json, "json")
578 if res is None or res.status_code == 201:
579 # The user has been created even though it shouldn't
580 res = engine.test("Get user U3", "GET", "/admin/v1/users/U3", headers_json, None,
581 200, r_header_json, "json")
582 u3 = res.json()["_id"] if res else None
583 else:
584 u3 = None
585
586 if padmin:
587 res = engine.test("Change to user U2 project Padmin", "POST", "/admin/v1/tokens", headers_json,
588 {"project_id": "Padmin"}, # Caused a Keystone authentication error
589 # {"username": "U2", "password": "pw2_new", "project_id": "Padmin"},
590 (200, 201), r_header_json, "json")
591 if res:
592 rj = res.json()
593 engine.set_header({"Authorization": "Bearer {}".format(rj["id"])})
594
595 res = engine.test("Add new project admin", "POST", "/admin/v1/projects", headers_json,
596 {"name": "P3"}, (201, 204),
597 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"},
598 "json")
599 p3 = engine.last_id if res else None
600
601 if p1:
602 data = {"username": "U4", "password": "pw4"}
603 data["project_role_mappings"] = [{"project": p1, "role": rpa}]
604 res = engine.test("Add new user admin", "POST", "/admin/v1/users", headers_json,
605 data, (201, 204),
606 {"Location": "/admin/v1/users/", "Content-Type": "application/json"},
607 "json")
608 u4 = engine.last_id if res else None
609 else:
610 u4 = None
611
612 if u4 and p3:
613 data = {"project_role_mappings": [{"project": p3, "role": rpa}]}
614 engine.test("Edit user projects admin", "PUT", "/admin/v1/users/U4", headers_json,
615 data, 204, None, None)
616 # Project is deleted even though it shouldn't - PROVISIONAL?
617 res = engine.test("Delete project P3 conflict", "DELETE", "/admin/v1/projects/"+p3,
618 headers_json, None, 409, None, None)
619 if res and res.status_code in (200, 204):
620 p3 = None
621 if p3:
622 res = engine.test("Delete project P3 forcing", "DELETE",
623 "/admin/v1/projects/"+p3+"?FORCE=True", headers_json, None, 204,
624 None, None)
625 if res and res.status_code in (200, 204):
626 p3 = None
627
628 if u2:
629 res = engine.test("Delete user U2. Conflict deleting own user", "DELETE",
630 "/admin/v1/users/"+u2, headers_json, None, 409, r_header_json, "json")
631 if res is None or res.status_code in (200, 204):
632 u2 = None
633 if u4:
634 res = engine.test("Delete user U4", "DELETE", "/admin/v1/users/"+u4, headers_json, None,
635 204, None, None)
636 if res and res.status_code in (200, 204):
637 u4 = None
638 if p3:
639 res = engine.test("Delete project P3", "DELETE", "/admin/v1/projects/"+p3, headers_json,
640 None, 204, None, None)
641 if res and res.status_code in (200, 204):
642 p3 = None
643
644 if u3:
645 res = engine.test("Delete user U3", "DELETE", "/admin/v1/users/"+u3, headers_json, None,
646 204, None, None)
647 if res:
648 u3 = None
649
650 # change to admin
651 engine.remove_authorization() # To force get authorization
652 engine.get_autorization()
653 if u1:
654 engine.test("Delete user U1", "DELETE", "/admin/v1/users/"+u1, headers_json, None, 204, None, None)
655 if u2:
656 engine.test("Delete user U2", "DELETE", "/admin/v1/users/"+u2, headers_json, None, 204, None, None)
657 if u3:
658 engine.test("Delete user U3", "DELETE", "/admin/v1/users/"+u3, headers_json, None, 204, None, None)
659 if u4:
660 engine.test("Delete user U4", "DELETE", "/admin/v1/users/"+u4, headers_json, None, 204, None, None)
661 if p1:
662 engine.test("Delete project P1", "DELETE", "/admin/v1/projects/"+p1, headers_json, None, 204, None, None)
663 if p2:
664 engine.test("Delete project P2", "DELETE", "/admin/v1/projects/"+p2, headers_json, None, 204, None, None)
665 if p3:
666 engine.test("Delete project P3", "DELETE", "/admin/v1/projects/"+p3, headers_json, None, 204, None, None)
667 if padmin:
668 engine.test("Delete project Padmin", "DELETE", "/admin/v1/projects/"+padmin, headers_json, None, 204,
669 None, None)
670 if pbad:
671 engine.test("Delete bad project", "DELETE", "/admin/v1/projects/"+pbad, headers_json, None, 204,
672 None, None)
673
674 # BEGIN New Tests - Addressing Projects/Users by Name/ID
675 pid1 = pid2 = None
676 uid1 = uid2 = None
677 res = engine.test("Create new project P1", "POST", "/admin/v1/projects", headers_json, {"name": "P1"},
678 201, {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
679 if res:
680 pid1 = res.json()["id"]
681 # print("# pid =", pid1)
682 res = engine.test("Create new project P2", "POST", "/admin/v1/projects", headers_json, {"name": "P2"},
683 201, {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
684 if res:
685 pid2 = res.json()["id"]
686 # print("# pid =", pid2)
687 data = {"username": "U1", "password": "pw1"}
688 data["project_role_mappings"] = [{"project": pid1, "role": rpu}]
689 res = engine.test("Create new user U1", "POST", "/admin/v1/users", headers_json, data, 201,
690 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
691 if res:
692 uid1 = res.json()["id"]
693 # print("# uid =", uid1)
694 data = {"username": "U2", "password": "pw2"}
695 data["project_role_mappings"] = [{"project": pid2, "role": rpu}]
696 res = engine.test("Create new user U2", "POST", "/admin/v1/users", headers_json, data, 201,
697 {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
698 if res:
699 uid2 = res.json()["id"]
700 # print("# uid =", uid2)
701 if pid1:
702 engine.test("Get Project P1 by Name", "GET", "/admin/v1/projects/P1", headers_json, None,
703 200, None, "json")
704 engine.test("Get Project P1 by ID", "GET", "/admin/v1/projects/"+pid1, headers_json, None,
705 200, None, "json")
706 if uid1:
707 engine.test("Get User U1 by Name", "GET", "/admin/v1/users/U1", headers_json, None, 200, None, "json")
708 engine.test("Get User U1 by ID", "GET", "/admin/v1/users/"+uid1, headers_json, None, 200, None, "json")
709 if pid1:
710 res = engine.test("Rename Project P1 by Name", "PUT", "/admin/v1/projects/P1", headers_json,
711 {"name": "P3"}, 204, None, None)
712 if res:
713 engine.test("Get Project P1 by new Name", "GET", "/admin/v1/projects/P3", headers_json, None,
714 200, None, "json")
715 if pid2:
716 res = engine.test("Rename Project P2 by ID", "PUT", "/admin/v1/projects/"+pid2, headers_json,
717 {"name": "P4"}, 204, None, None)
718 if res:
719 engine.test("Get Project P2 by new Name", "GET", "/admin/v1/projects/P4", headers_json, None,
720 200, None, "json")
721
722 if uid1:
723 res = engine.test("Rename User U1 by Name", "PUT", "/admin/v1/users/U1", headers_json,
724 {"username": "U3"}, 204, None, None)
725 if res:
726 engine.test("Get User U1 by new Name", "GET", "/admin/v1/users/U3", headers_json, None,
727 200, None, "json")
728
729 if uid2:
730 res = engine.test("Rename User U2 by ID", "PUT", "/admin/v1/users/"+uid2, headers_json,
731 {"username": "U4"}, 204, None, None)
732 if res:
733 engine.test("Get User U2 by new Name", "GET", "/admin/v1/users/U4", headers_json, None,
734 200, None, "json")
735 if uid1:
736 res = engine.test("Delete User U1 by Name", "DELETE", "/admin/v1/users/U3", headers_json, None,
737 204, None, None)
738 if res:
739 uid1 = None
740
741 if uid2:
742 res = engine.test("Delete User U2 by ID", "DELETE", "/admin/v1/users/"+uid2, headers_json, None,
743 204, None, None)
744 if res:
745 uid2 = None
746
747 if pid1:
748 res = engine.test("Delete Project P1 by Name", "DELETE", "/admin/v1/projects/P3", headers_json, None,
749 204, None, None)
750 if res:
751 pid1 = None
752
753 if pid2:
754 res = engine.test("Delete Project P2 by ID", "DELETE", "/admin/v1/projects/"+pid2, headers_json, None,
755 204, None, None)
756 if res:
757 pid2 = None
758
759 # END New Tests - Addressing Projects/Users by Name
760
761 # CLEANUP
762 if pid1:
763 engine.test("Delete Project P1", "DELETE", "/admin/v1/projects/"+pid1, headers_json, None, 204, None, None)
764 if pid2:
765 engine.test("Delete Project P2", "DELETE", "/admin/v1/projects/"+pid2, headers_json, None, 204, None, None)
766 if uid1:
767 engine.test("Delete User U1", "DELETE", "/admin/v1/users/"+uid1, headers_json, None, 204, None, None)
768 if uid2:
769 engine.test("Delete User U2", "DELETE", "/admin/v1/users/"+uid2, headers_json, None, 204, None, None)
770
771 engine.remove_authorization() # To finish
772
773
774 class TestProjectsDescriptors:
775 description = "test descriptors visibility among projects"
776
777 @staticmethod
778 def run(engine, test_osm, manual_check, test_params=None):
779 vnfd_ids = []
780 engine.set_test_name("ProjectDescriptors")
781 engine.get_autorization()
782
783 project_admin_id = None
784 res = engine.test("Get my project Padmin", "GET", "/admin/v1/projects/{}".format(engine.project), headers_json,
785 None, 200, r_header_json, "json")
786 if res:
787 response = res.json()
788 project_admin_id = response["_id"]
789 engine.test("Create project Padmin", "POST", "/admin/v1/projects", headers_json,
790 {"name": "Padmin", "admin": True}, (201, 204),
791 {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
792 engine.test("Create project P2", "POST", "/admin/v1/projects", headers_json, {"name": "P2"},
793 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
794 engine.test("Create project P3", "POST", "/admin/v1/projects", headers_json, {"name": "P3"},
795 (201, 204), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
796
797 engine.test("Create user U1", "POST", "/admin/v1/users", headers_json,
798 {"username": "U1", "password": "pw1",
799 "project_role_mappings": [{"project": "Padmin", "role": "system_admin"},
800 {"project": "P2", "role": "project_admin"},
801 {"project": "P3", "role": "project_admin"}],
802 }, 201, {"Location": "/admin/v1/users/", "Content-Type": "application/json"}, "json")
803
804 engine.test("Onboard VNFD id1", "POST", "/vnfpkgm/v1/vnf_packages_content?id=id1", headers_yaml,
805 TestDescriptors.vnfd_empty, 201, r_headers_yaml_location_vnfd, "yaml")
806 vnfd_ids.append(engine.last_id)
807 engine.test("Onboard VNFD id2 PUBLIC", "POST", "/vnfpkgm/v1/vnf_packages_content?id=id2&PUBLIC=TRUE",
808 headers_yaml, TestDescriptors.vnfd_empty, 201, r_headers_yaml_location_vnfd, "yaml")
809 vnfd_ids.append(engine.last_id)
810 engine.test("Onboard VNFD id3", "POST", "/vnfpkgm/v1/vnf_packages_content?id=id3&PUBLIC=FALSE", headers_yaml,
811 TestDescriptors.vnfd_empty, 201, r_headers_yaml_location_vnfd, "yaml")
812 vnfd_ids.append(engine.last_id)
813
814 res = engine.test("Get VNFD descriptors", "GET", "/vnfpkgm/v1/vnf_packages?id=id1,id2,id3",
815 headers_json, None, 200, r_header_json, "json")
816 response = res.json()
817 if len(response) != 3:
818 logger.error("Only 3 vnfds should be present for project admin. {} listed".format(len(response)))
819 engine.failed_tests += 1
820
821 # Change to other project Padmin
822 res = engine.test("Change to user U1 project Padmin", "POST", "/admin/v1/tokens", headers_json,
823 {"username": "U1", "password": "pw1", "project_id": "Padmin"}, (200, 201),
824 r_header_json, "json")
825 if res:
826 response = res.json()
827 engine.set_header({"Authorization": "Bearer {}".format(response["id"])})
828
829 # list vnfds
830 res = engine.test("List VNFD descriptors for Padmin", "GET", "/vnfpkgm/v1/vnf_packages",
831 headers_json, None, 200, r_header_json, "json")
832 response = res.json()
833 if len(response) != 0:
834 logger.error("Only 0 vnfds should be present for project Padmin. {} listed".format(len(response)))
835 engine.failed_tests += 1
836
837 # list Public vnfds
838 res = engine.test("List VNFD public descriptors", "GET", "/vnfpkgm/v1/vnf_packages?PUBLIC=True",
839 headers_json, None, 200, r_header_json, "json")
840 response = res.json()
841 if len(response) != 1:
842 logger.error("Only 1 vnfds should be present for project Padmin. {} listed".format(len(response)))
843 engine.failed_tests += 1
844
845 # list vnfds belonging to project "admin"
846 res = engine.test("List VNFD of admin project", "GET",
847 "/vnfpkgm/v1/vnf_packages?ADMIN={}".format(project_admin_id),
848 headers_json, None, 200, r_header_json, "json")
849 if res:
850 response = res.json()
851 if len(response) != 3:
852 logger.error("Only 3 vnfds should be present for project Padmin. {} listed".format(len(response)))
853 engine.failed_tests += 1
854
855 # Get Public vnfds
856 engine.test("Get VNFD public descriptors", "GET", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[1]),
857 headers_json, None, 200, r_header_json, "json")
858 # Edit not owned vnfd
859 engine.test("Edit VNFD ", "PATCH", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[0]),
860 headers_yaml, '{name: pepe}', 404, r_header_yaml, "yaml")
861
862 # Add to my catalog
863 engine.test("Add VNFD id2 to my catalog", "PATCH", "/vnfpkgm/v1/vnf_packages/{}?SET_PROJECT".
864 format(vnfd_ids[1]), headers_json, None, 204, None, 0)
865
866 # Add a new vnfd
867 engine.test("Onboard VNFD id4", "POST", "/vnfpkgm/v1/vnf_packages_content?id=id4", headers_yaml,
868 TestDescriptors.vnfd_empty, 201, r_headers_yaml_location_vnfd, "yaml")
869 vnfd_ids.append(engine.last_id)
870
871 # list vnfds
872 res = engine.test("List VNFD public descriptors", "GET", "/vnfpkgm/v1/vnf_packages",
873 headers_json, None, 200, r_header_json, "json")
874 response = res.json()
875 if len(response) != 2:
876 logger.error("Only 2 vnfds should be present for project Padmin. {} listed".format(len(response)))
877 engine.failed_tests += 1
878
879 if manual_check:
880 input('VNFDs have been omboarded. Perform manual check and press enter to resume')
881
882 test_rest.test("Delete VNFD id2", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[1]),
883 headers_yaml, None, 204, None, 0)
884
885 # change to admin project
886 engine.remove_authorization() # To force get authorization
887 engine.get_autorization()
888 test_rest.test("Delete VNFD id1", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[0]),
889 headers_yaml, None, 204, None, 0)
890 test_rest.test("Delete VNFD id2", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[1]),
891 headers_yaml, None, 204, None, 0)
892 test_rest.test("Delete VNFD id3", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[2]),
893 headers_yaml, None, 204, None, 0)
894 test_rest.test("Delete VNFD id4", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_ids[3]),
895 headers_yaml, None, 404, r_header_yaml, "yaml")
896 test_rest.test("Delete VNFD id4", "DELETE", "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[3]),
897 headers_yaml, None, 204, None, 0)
898 # Get Public vnfds
899 engine.test("Get VNFD deleted id1", "GET", "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[0]),
900 headers_json, None, 404, r_header_json, "json")
901 engine.test("Get VNFD deleted id2", "GET", "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[1]),
902 headers_json, None, 404, r_header_json, "json")
903 engine.test("Get VNFD deleted id3", "GET", "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[2]),
904 headers_json, None, 404, r_header_json, "json")
905 engine.test("Get VNFD deleted id4", "GET", "/vnfpkgm/v1/vnf_packages/{}?ADMIN".format(vnfd_ids[3]),
906 headers_json, None, 404, r_header_json, "json")
907
908 engine.test("Delete user U1", "DELETE", "/admin/v1/users/U1", headers_json, None, 204, None, None)
909 engine.test("Delete project Padmin", "DELETE", "/admin/v1/projects/Padmin", headers_json, None, 204, None, None)
910 engine.test("Delete project P2", "DELETE", "/admin/v1/projects/P2", headers_json, None, 204, None, None)
911 engine.test("Delete project P3", "DELETE", "/admin/v1/projects/P3", headers_json, None, 204, None, None)
912
913
914 class TestFakeVim:
915 description = "Creates/edit/delete fake VIMs and SDN controllers"
916
917 def __init__(self):
918 self.vim = {
919 "schema_version": "1.0",
920 "schema_type": "No idea",
921 "name": "myVim",
922 "description": "Descriptor name",
923 "vim_type": "openstack",
924 "vim_url": "http://localhost:/vim",
925 "vim_tenant_name": "vimTenant",
926 "vim_user": "user",
927 "vim_password": "password",
928 "config": {"config_param": 1}
929 }
930 self.sdn = {
931 "name": "sdn-name",
932 "description": "sdn-description",
933 "dpid": "50:50:52:54:00:94:21:21",
934 "ip": "192.168.15.17",
935 "port": 8080,
936 "type": "opendaylight",
937 "version": "3.5.6",
938 "user": "user",
939 "password": "passwd"
940 }
941 self.port_mapping = [
942 {"compute_node": "compute node 1",
943 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/1", "switch_mac": "52:54:00:94:21:21"},
944 {"pci": "0000:81:00.1", "switch_port": "port-2/2", "switch_mac": "52:54:00:94:21:22"}
945 ]},
946 {"compute_node": "compute node 2",
947 "ports": [{"pci": "0000:81:00.0", "switch_port": "port-2/3", "switch_mac": "52:54:00:94:21:23"},
948 {"pci": "0000:81:00.1", "switch_port": "port-2/4", "switch_mac": "52:54:00:94:21:24"}
949 ]}
950 ]
951
952 def run(self, engine, test_osm, manual_check, test_params=None):
953
954 vim_bad = self.vim.copy()
955 vim_bad.pop("name")
956
957 engine.set_test_name("FakeVim")
958 engine.get_autorization()
959 engine.test("Create VIM", "POST", "/admin/v1/vim_accounts", headers_json, self.vim, (201, 202),
960 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json")
961 vim_id = engine.last_id
962 engine.test("Create VIM without name, bad schema", "POST", "/admin/v1/vim_accounts", headers_json,
963 vim_bad, 422, None, headers_json)
964 engine.test("Create VIM name repeated", "POST", "/admin/v1/vim_accounts", headers_json, self.vim,
965 409, None, headers_json)
966 engine.test("Show VIMs", "GET", "/admin/v1/vim_accounts", headers_yaml, None, 200, r_header_yaml,
967 "yaml")
968 engine.test("Show VIM", "GET", "/admin/v1/vim_accounts/{}".format(vim_id), headers_yaml, None, 200,
969 r_header_yaml, "yaml")
970 if not test_osm:
971 # delete with FORCE
972 engine.test("Delete VIM", "DELETE", "/admin/v1/vim_accounts/{}?FORCE=True".format(vim_id), headers_yaml,
973 None, 202, None, 0)
974 engine.test("Check VIM is deleted", "GET", "/admin/v1/vim_accounts/{}".format(vim_id), headers_yaml, None,
975 404, r_header_yaml, "yaml")
976 else:
977 # delete and wait until is really deleted
978 engine.test("Delete VIM", "DELETE", "/admin/v1/vim_accounts/{}".format(vim_id), headers_yaml, None, 202,
979 None, 0)
980 engine.wait_until_delete("/admin/v1/vim_accounts/{}".format(vim_id), timeout)
981
982
983 class TestVIMSDN(TestFakeVim):
984 description = "Creates VIM with SDN editing SDN controllers and port_mapping"
985
986 def __init__(self):
987 TestFakeVim.__init__(self)
988 self.wim = {
989 "schema_version": "1.0",
990 "schema_type": "No idea",
991 "name": "myWim",
992 "description": "Descriptor name",
993 "wim_type": "odl",
994 "wim_url": "http://localhost:/wim",
995 "user": "user",
996 "password": "password",
997 "config": {"config_param": 1}
998 }
999
1000 def run(self, engine, test_osm, manual_check, test_params=None):
1001 engine.set_test_name("VimSdn")
1002 engine.get_autorization()
1003 # Added SDN
1004 engine.test("Create SDN", "POST", "/admin/v1/sdns", headers_json, self.sdn, (201, 202),
1005 {"Location": "/admin/v1/sdns/", "Content-Type": "application/json"}, "json")
1006 sdnc_id = engine.last_id
1007 # sleep(5)
1008 # Edit SDN
1009 engine.test("Edit SDN", "PATCH", "/admin/v1/sdns/{}".format(sdnc_id), headers_json, {"name": "new_sdn_name"},
1010 (202, 204), None, None)
1011 # sleep(5)
1012 # VIM with SDN
1013 self.vim["config"]["sdn-controller"] = sdnc_id
1014 self.vim["config"]["sdn-port-mapping"] = self.port_mapping
1015 engine.test("Create VIM", "POST", "/admin/v1/vim_accounts", headers_json, self.vim, (200, 202, 201),
1016 {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json"),
1017
1018 vim_id = engine.last_id
1019 self.port_mapping[0]["compute_node"] = "compute node XX"
1020 engine.test("Edit VIM change port-mapping", "PUT", "/admin/v1/vim_accounts/{}".format(vim_id), headers_json,
1021 {"config": {"sdn-port-mapping": self.port_mapping}}, (202, 204), None, None)
1022 engine.test("Edit VIM remove port-mapping", "PUT", "/admin/v1/vim_accounts/{}".format(vim_id), headers_json,
1023 {"config": {"sdn-port-mapping": None}}, (202, 204), None, None)
1024
1025 engine.test("Create WIM", "POST", "/admin/v1/wim_accounts", headers_json, self.wim, (200, 202, 201),
1026 {"Location": "/admin/v1/wim_accounts/", "Content-Type": "application/json"}, "json"),
1027 wim_id = engine.last_id
1028
1029 if not test_osm:
1030 # delete with FORCE
1031 engine.test("Delete VIM remove port-mapping", "DELETE",
1032 "/admin/v1/vim_accounts/{}?FORCE=True".format(vim_id), headers_json, None, 202, None, 0)
1033 engine.test("Delete SDNC", "DELETE", "/admin/v1/sdns/{}?FORCE=True".format(sdnc_id), headers_json, None,
1034 202, None, 0)
1035
1036 engine.test("Delete WIM", "DELETE",
1037 "/admin/v1/wim_accounts/{}?FORCE=True".format(wim_id), headers_json, None, 202, None, 0)
1038 engine.test("Check VIM is deleted", "GET", "/admin/v1/vim_accounts/{}".format(vim_id), headers_yaml,
1039 None, 404, r_header_yaml, "yaml")
1040 engine.test("Check SDN is deleted", "GET", "/admin/v1/sdns/{}".format(sdnc_id), headers_yaml, None,
1041 404, r_header_yaml, "yaml")
1042 engine.test("Check WIM is deleted", "GET", "/admin/v1/wim_accounts/{}".format(wim_id), headers_yaml,
1043 None, 404, r_header_yaml, "yaml")
1044 else:
1045 if manual_check:
1046 input('VIM, SDN, WIM has been deployed. Perform manual check and press enter to resume')
1047 # delete and wait until is really deleted
1048 engine.test("Delete VIM remove port-mapping", "DELETE", "/admin/v1/vim_accounts/{}".format(vim_id),
1049 headers_json, None, (202, 201, 204), None, 0)
1050 engine.test("Delete SDN", "DELETE", "/admin/v1/sdns/{}".format(sdnc_id), headers_json, None,
1051 (202, 201, 204), None, 0)
1052 engine.test("Delete VIM", "DELETE", "/admin/v1/wim_accounts/{}".format(wim_id),
1053 headers_json, None, (202, 201, 204), None, 0)
1054 engine.wait_until_delete("/admin/v1/vim_accounts/{}".format(vim_id), timeout)
1055 engine.wait_until_delete("/admin/v1/sdns/{}".format(sdnc_id), timeout)
1056 engine.wait_until_delete("/admin/v1/wim_accounts/{}".format(wim_id), timeout)
1057
1058
1059 class TestDeploy:
1060 description = "Base class for downloading descriptors from ETSI, onboard and deploy in real VIM"
1061
1062 def __init__(self):
1063 self.test_name = "DEPLOY"
1064 self.nsd_id = None
1065 self.vim_id = None
1066 self.ns_id = None
1067 self.vnfds_id = []
1068 self.descriptor_url = "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
1069 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
1070 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
1071 self.descriptor_edit = None
1072 self.uses_configuration = False
1073 self.users = {}
1074 self.passwords = {}
1075 self.commands = {}
1076 self.keys = {}
1077 self.timeout = 120
1078 self.qforce = ""
1079 self.ns_params = None
1080 self.vnfr_ip_list = {}
1081
1082 def create_descriptors(self, engine):
1083 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
1084 if not os.path.exists(temp_dir):
1085 os.makedirs(temp_dir)
1086 for vnfd_index, vnfd_filename in enumerate(self.vnfd_filenames):
1087 if "/" in vnfd_filename:
1088 vnfd_filename_path = vnfd_filename
1089 if not os.path.exists(vnfd_filename_path):
1090 raise TestException("File '{}' does not exist".format(vnfd_filename_path))
1091 else:
1092 vnfd_filename_path = temp_dir + vnfd_filename
1093 if not os.path.exists(vnfd_filename_path):
1094 with open(vnfd_filename_path, "wb") as file:
1095 response = requests.get(self.descriptor_url + vnfd_filename)
1096 if response.status_code >= 300:
1097 raise TestException("Error downloading descriptor from '{}': {}".format(
1098 self.descriptor_url + vnfd_filename, response.status_code))
1099 file.write(response.content)
1100 if vnfd_filename_path.endswith(".yaml"):
1101 headers = headers_yaml
1102 else:
1103 headers = headers_zip_yaml
1104 if randint(0, 1) == 0:
1105 # vnfd CREATE AND UPLOAD in one step:
1106 engine.test("Onboard VNFD in one step", "POST",
1107 "/vnfpkgm/v1/vnf_packages_content" + self.qforce, headers, "@b" + vnfd_filename_path, 201,
1108 r_headers_yaml_location_vnfd,
1109 "yaml")
1110 self.vnfds_id.append(engine.last_id)
1111 else:
1112 # vnfd CREATE AND UPLOAD ZIP
1113 engine.test("Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages",
1114 headers_json, None, 201,
1115 {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
1116 self.vnfds_id.append(engine.last_id)
1117 engine.test("Onboard VNFD step 2 as ZIP", "PUT",
1118 "/vnfpkgm/v1/vnf_packages/<>/package_content" + self.qforce,
1119 headers, "@b" + vnfd_filename_path, 204, None, 0)
1120
1121 if self.descriptor_edit:
1122 if "vnfd{}".format(vnfd_index) in self.descriptor_edit:
1123 # Modify VNFD
1124 engine.test("Edit VNFD ", "PATCH",
1125 "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfds_id[-1]),
1126 headers_yaml, self.descriptor_edit["vnfd{}".format(vnfd_index)], 204, None, None)
1127
1128 if "/" in self.nsd_filename:
1129 nsd_filename_path = self.nsd_filename
1130 if not os.path.exists(nsd_filename_path):
1131 raise TestException("File '{}' does not exist".format(nsd_filename_path))
1132 else:
1133 nsd_filename_path = temp_dir + self.nsd_filename
1134 if not os.path.exists(nsd_filename_path):
1135 with open(nsd_filename_path, "wb") as file:
1136 response = requests.get(self.descriptor_url + self.nsd_filename)
1137 if response.status_code >= 300:
1138 raise TestException("Error downloading descriptor from '{}': {}".format(
1139 self.descriptor_url + self.nsd_filename, response.status_code))
1140 file.write(response.content)
1141 if nsd_filename_path.endswith(".yaml"):
1142 headers = headers_yaml
1143 else:
1144 headers = headers_zip_yaml
1145
1146 if randint(0, 1) == 0:
1147 # nsd CREATE AND UPLOAD in one step:
1148 engine.test("Onboard NSD in one step", "POST",
1149 "/nsd/v1/ns_descriptors_content" + self.qforce, headers, "@b" + nsd_filename_path, 201,
1150 r_headers_yaml_location_nsd, yaml)
1151 self.nsd_id = engine.last_id
1152 else:
1153 # nsd CREATE AND UPLOAD ZIP
1154 engine.test("Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors",
1155 headers_json, None, 201,
1156 {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
1157 self.nsd_id = engine.last_id
1158 engine.test("Onboard NSD step 2 as ZIP", "PUT",
1159 "/nsd/v1/ns_descriptors/<>/nsd_content" + self.qforce,
1160 headers, "@b" + nsd_filename_path, 204, None, 0)
1161
1162 if self.descriptor_edit and "nsd" in self.descriptor_edit:
1163 # Modify NSD
1164 engine.test("Edit NSD ", "PATCH",
1165 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id),
1166 headers_yaml, self.descriptor_edit["nsd"], 204, None, None)
1167
1168 def delete_descriptors(self, engine):
1169 # delete descriptors
1170 engine.test("Delete NSSD SOL005", "DELETE",
1171 "/nsd/v1/ns_descriptors/{}".format(self.nsd_id),
1172 headers_yaml, None, 204, None, 0)
1173 for vnfd_id in self.vnfds_id:
1174 engine.test("Delete VNFD SOL005", "DELETE",
1175 "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id), headers_yaml, None, 204, None, 0)
1176
1177 def instantiate(self, engine, ns_data):
1178 ns_data_text = yaml.safe_dump(ns_data, default_flow_style=True, width=256)
1179 # create NS Two steps
1180 r = engine.test("Create NS step 1", "POST", "/nslcm/v1/ns_instances",
1181 headers_yaml, ns_data_text, (201, 202),
1182 {"Location": "nslcm/v1/ns_instances/", "Content-Type": "application/yaml"}, "yaml")
1183 if not r:
1184 return
1185 self.ns_id = engine.last_id
1186 engine.test("Instantiate NS step 2", "POST",
1187 "/nslcm/v1/ns_instances/{}/instantiate".format(self.ns_id), headers_yaml, ns_data_text,
1188 (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1189 nslcmop_id = engine.last_id
1190
1191 if test_osm:
1192 # Wait until status is Ok
1193 timeout = timeout_configure if self.uses_configuration else timeout_deploy
1194 engine.wait_operation_ready("ns", nslcmop_id, timeout)
1195
1196 def terminate(self, engine):
1197 # remove deployment
1198 if test_osm:
1199 engine.test("Terminate NS", "POST", "/nslcm/v1/ns_instances/{}/terminate".format(self.ns_id), headers_yaml,
1200 None, (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1201 nslcmop2_id = engine.last_id
1202 # Wait until status is Ok
1203 engine.wait_operation_ready("ns", nslcmop2_id, timeout_deploy)
1204
1205 engine.test("Delete NS", "DELETE", "/nslcm/v1/ns_instances/{}".format(self.ns_id), headers_yaml, None,
1206 204, None, 0)
1207 else:
1208 engine.test("Delete NS with FORCE", "DELETE", "/nslcm/v1/ns_instances/{}?FORCE=True".format(self.ns_id),
1209 headers_yaml, None, 204, None, 0)
1210
1211 # check all it is deleted
1212 engine.test("Check NS is deleted", "GET", "/nslcm/v1/ns_instances/{}".format(self.ns_id), headers_yaml, None,
1213 404, None, "yaml")
1214 r = engine.test("Check NSLCMOPs are deleted", "GET",
1215 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId={}".format(self.ns_id), headers_json, None,
1216 200, None, "json")
1217 if not r:
1218 return
1219 nslcmops = r.json()
1220 if not isinstance(nslcmops, list) or nslcmops:
1221 raise TestException("NS {} deleted but with ns_lcm_op_occ active: {}".format(self.ns_id, nslcmops))
1222
1223 def test_ns(self, engine, test_osm, commands=None, users=None, passwds=None, keys=None, timeout=0):
1224
1225 r = engine.test("GET VNFR IDs", "GET",
1226 "/nslcm/v1/ns_instances/{}".format(self.ns_id), headers_json, None,
1227 200, r_header_json, "json")
1228 if not r:
1229 return
1230 ns_data = r.json()
1231
1232 vnfr_list = ns_data['constituent-vnfr-ref']
1233 time = 0
1234 _commands = commands if commands is not None else self.commands
1235 _users = users if users is not None else self.users
1236 _passwds = passwds if passwds is not None else self.passwords
1237 _keys = keys if keys is not None else self.keys
1238 _timeout = timeout if timeout != 0 else self.timeout
1239
1240 # vnfr_list=[d8272263-6bd3-4680-84ca-6a4be23b3f2d, 88b22e2f-994a-4b61-94fd-4a3c90de3dc4]
1241 for vnfr_id in vnfr_list:
1242 r = engine.test("Get VNFR to get IP_ADDRESS", "GET",
1243 "/nslcm/v1/vnfrs/{}".format(vnfr_id), headers_json, None,
1244 200, r_header_json, "json")
1245 if not r:
1246 continue
1247 vnfr_data = r.json()
1248
1249 vnf_index = str(vnfr_data["member-vnf-index-ref"])
1250
1251 ip_address = self.get_vnfr_ip(engine, vnf_index)
1252 description = "Exec command='{}' at VNFR={} IP={}".format(_commands.get(vnf_index)[0], vnf_index,
1253 ip_address)
1254 engine.step += 1
1255 test_description = "{}{} {}".format(engine.test_name, engine.step, description)
1256 logger.warning(test_description)
1257 while _timeout >= time:
1258 result, message = self.do_checks([ip_address],
1259 vnf_index=vnfr_data["member-vnf-index-ref"],
1260 commands=_commands.get(vnf_index), user=_users.get(vnf_index),
1261 passwd=_passwds.get(vnf_index), key=_keys.get(vnf_index))
1262 if result == 1:
1263 engine.passed_tests += 1
1264 logger.debug(message)
1265 break
1266 elif result == 0:
1267 time += 20
1268 sleep(20)
1269 elif result == -1:
1270 engine.failed_tests += 1
1271 logger.error(message)
1272 break
1273 else:
1274 time -= 20
1275 engine.failed_tests += 1
1276 logger.error(message)
1277 else:
1278 engine.failed_tests += 1
1279 logger.error("VNFR {} has not mgmt address. Check failed".format(vnf_index))
1280
1281 def do_checks(self, ip, vnf_index, commands=[], user=None, passwd=None, key=None):
1282 try:
1283 import urllib3
1284 from pssh.clients import ParallelSSHClient
1285 from pssh.utils import load_private_key
1286 from ssh2 import exceptions as ssh2Exception
1287 except ImportError as e:
1288 logger.critical("Package <pssh> or/and <urllib3> is not installed. Please add them with 'pip3 install "
1289 "parallel-ssh urllib3': {}".format(e))
1290 return -1, "install needed packages 'pip3 install parallel-ssh urllib3'"
1291 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
1292 try:
1293 p_host = os.environ.get("PROXY_HOST")
1294 p_user = os.environ.get("PROXY_USER")
1295 p_password = os.environ.get("PROXY_PASSWD")
1296
1297 if key:
1298 pkey = load_private_key(key)
1299 else:
1300 pkey = None
1301
1302 client = ParallelSSHClient(ip, user=user, password=passwd, pkey=pkey, proxy_host=p_host,
1303 proxy_user=p_user, proxy_password=p_password, timeout=10, num_retries=0)
1304 for cmd in commands:
1305 output = client.run_command(cmd)
1306 client.join(output)
1307 if output[ip[0]].exit_code:
1308 return -1, "VNFR {} command '{}' returns error: '{}'".format(ip[0], cmd,
1309 "\n".join(output[ip[0]].stderr))
1310 else:
1311 return 1, "VNFR {} command '{}' successful".format(ip[0], cmd)
1312 except (ssh2Exception.ChannelFailure, ssh2Exception.SocketDisconnectError, ssh2Exception.SocketTimeout,
1313 ssh2Exception.SocketRecvError) as e:
1314 return 0, "Timeout accessing the VNFR {}: {}".format(ip[0], str(e))
1315 except Exception as e:
1316 return -1, "ERROR checking the VNFR {}: {}".format(ip[0], str(e))
1317
1318 def additional_operations(self, engine, test_osm, manual_check):
1319 pass
1320
1321 def run(self, engine, test_osm, manual_check, test_params=None):
1322 engine.set_test_name(self.test_name)
1323 engine.get_autorization()
1324 nsname = os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
1325 if test_params:
1326 if "vnfd-files" in test_params:
1327 self.vnfd_filenames = test_params["vnfd-files"].split(",")
1328 if "nsd-file" in test_params:
1329 self.nsd_filename = test_params["nsd-file"]
1330 if test_params.get("ns-name"):
1331 nsname = test_params["ns-name"]
1332 self.create_descriptors(engine)
1333
1334 # create real VIM if not exist
1335 self.vim_id = engine.get_create_vim(test_osm)
1336 ns_data = {"nsDescription": "default description", "nsName": nsname, "nsdId": self.nsd_id,
1337 "vimAccountId": self.vim_id}
1338 if self.ns_params:
1339 ns_data.update(self.ns_params)
1340 if test_params and test_params.get("ns-config"):
1341 if isinstance(test_params["ns-config"], str):
1342 ns_data.update(yaml.load(test_params["ns-config"]))
1343 else:
1344 ns_data.update(test_params["ns-config"])
1345 self.instantiate(engine, ns_data)
1346
1347 if manual_check:
1348 input('NS has been deployed. Perform manual check and press enter to resume')
1349 if test_osm and self.commands:
1350 self.test_ns(engine, test_osm)
1351 self.additional_operations(engine, test_osm, manual_check)
1352 self.terminate(engine)
1353 self.delete_descriptors(engine)
1354
1355 def get_first_ip(self, ip_string):
1356 # When using a floating IP, the vnfr_data['ip-address'] contains a semicolon-separated list of IP:s.
1357 first_ip = ip_string.split(";")[0] if ip_string else ""
1358 return first_ip
1359
1360 def get_vnfr_ip(self, engine, vnfr_index_wanted):
1361 # If the IP address list has been obtained before, it has been stored in 'vnfr_ip_list'
1362 ip = self.vnfr_ip_list.get(vnfr_index_wanted, "")
1363 if (ip):
1364 return self.get_first_ip(ip)
1365 r = engine.test("Get VNFR to get IP_ADDRESS", "GET",
1366 "/nslcm/v1/vnfrs?member-vnf-index-ref={}&nsr-id-ref={}".format(
1367 vnfr_index_wanted, self.ns_id), headers_json, None,
1368 200, r_header_json, "json")
1369 if not r:
1370 return ""
1371 vnfr_data = r.json()
1372 if not (vnfr_data and vnfr_data[0]):
1373 return ""
1374 # Store the IP (or list of IPs) in 'vnfr_ip_list'
1375 ip_list = vnfr_data[0].get("ip-address", "")
1376 if ip_list:
1377 self.vnfr_ip_list[vnfr_index_wanted] = ip_list
1378 ip = self.get_first_ip(ip_list)
1379 return ip
1380
1381
1382 class TestDeployHackfestCirros(TestDeploy):
1383 description = "Load and deploy Hackfest cirros_2vnf_ns example"
1384
1385 def __init__(self):
1386 super().__init__()
1387 self.test_name = "CIRROS"
1388 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
1389 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
1390 self.commands = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1391 self.users = {'1': "cirros", '2': "cirros"}
1392 self.passwords = {'1': "cubswin:)", '2': "cubswin:)"}
1393
1394 def terminate(self, engine):
1395 # Make a delete in one step, overriding the normal two step of TestDeploy that launched terminate and delete
1396 if test_osm:
1397 engine.test("Terminate and delete NS in one step", "DELETE", "/nslcm/v1/ns_instances_content/{}".
1398 format(self.ns_id), headers_yaml, None, 202, None, "yaml")
1399
1400 engine .wait_until_delete("/nslcm/v1/ns_instances/{}".format(self.ns_id), timeout_deploy)
1401 else:
1402 engine.test("Delete NS with FORCE", "DELETE", "/nslcm/v1/ns_instances/{}?FORCE=True".format(self.ns_id),
1403 headers_yaml, None, 204, None, 0)
1404
1405 # check all it is deleted
1406 engine.test("Check NS is deleted", "GET", "/nslcm/v1/ns_instances/{}".format(self.ns_id), headers_yaml, None,
1407 404, None, "yaml")
1408 r = engine.test("Check NSLCMOPs are deleted", "GET",
1409 "/nslcm/v1/ns_lcm_op_occs?nsInstanceId={}".format(self.ns_id), headers_json, None,
1410 200, None, "json")
1411 if not r:
1412 return
1413 nslcmops = r.json()
1414 if not isinstance(nslcmops, list) or nslcmops:
1415 raise TestException("NS {} deleted but with ns_lcm_op_occ active: {}".format(self.ns_id, nslcmops))
1416
1417
1418 class TestDeployHackfest1(TestDeploy):
1419 description = "Load and deploy Hackfest_1_vnfd example"
1420
1421 def __init__(self):
1422 super().__init__()
1423 self.test_name = "HACKFEST1-"
1424 self.vnfd_filenames = ("hackfest_1_vnfd.tar.gz",)
1425 self.nsd_filename = "hackfest_1_nsd.tar.gz"
1426 # self.commands = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1427 # self.users = {'1': "cirros", '2': "cirros"}
1428 # self.passwords = {'1': "cubswin:)", '2': "cubswin:)"}
1429
1430
1431 class TestDeployHackfestCirrosScaling(TestDeploy):
1432 description = "Load and deploy Hackfest cirros_2vnf_ns example with scaling modifications"
1433
1434 def __init__(self):
1435 super().__init__()
1436 self.test_name = "CIRROS-SCALE"
1437 self.vnfd_filenames = ("cirros_vnf.tar.gz",)
1438 self.nsd_filename = "cirros_2vnf_ns.tar.gz"
1439 # Modify VNFD to add scaling and count=2
1440 self.descriptor_edit = {
1441 "vnfd0": {
1442 "vdu": {
1443 "$id: 'cirros_vnfd-VM'": {"count": 2}
1444 },
1445 "scaling-group-descriptor": [{
1446 "name": "scale_cirros",
1447 "max-instance-count": 2,
1448 "vdu": [{
1449 "vdu-id-ref": "cirros_vnfd-VM",
1450 "count": 2
1451 }]
1452 }]
1453 }
1454 }
1455
1456 def additional_operations(self, engine, test_osm, manual_check):
1457 if not test_osm:
1458 return
1459 # 2 perform scale out twice
1460 payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1461 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
1462 for i in range(0, 2):
1463 engine.test("Execute scale action over NS", "POST",
1464 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1465 (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1466 nslcmop2_scale_out = engine.last_id
1467 engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
1468 if manual_check:
1469 input('NS scale out done. Check that two more vdus are there')
1470 # TODO check automatic
1471
1472 # 2 perform scale in
1473 payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1474 '{scaling-group-descriptor: scale_cirros, member-vnf-index: "1"}}}'
1475 for i in range(0, 2):
1476 engine.test("Execute scale IN action over NS", "POST",
1477 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1478 (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1479 nslcmop2_scale_in = engine.last_id
1480 engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
1481 if manual_check:
1482 input('NS scale in done. Check that two less vdus are there')
1483 # TODO check automatic
1484
1485 # perform scale in that must fail as reached limit
1486 engine.test("Execute scale IN out of limit action over NS", "POST",
1487 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1488 (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1489 nslcmop2_scale_in = engine.last_id
1490 engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy, expected_fail=True)
1491
1492
1493 class TestDeployIpMac(TestDeploy):
1494 description = "Load and deploy descriptor examples setting mac, ip address at descriptor and instantiate params"
1495
1496 def __init__(self):
1497 super().__init__()
1498 self.test_name = "SetIpMac"
1499 self.vnfd_filenames = ("vnfd_2vdu_set_ip_mac2.yaml", "vnfd_2vdu_set_ip_mac.yaml")
1500 self.nsd_filename = "scenario_2vdu_set_ip_mac.yaml"
1501 self.descriptor_url = \
1502 "https://osm.etsi.org/gitweb/?p=osm/RO.git;a=blob_plain;f=test/RO_tests/v3_2vdu_set_ip_mac/"
1503 self.commands = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1504 self.users = {'1': "osm", '2': "osm"}
1505 self.passwords = {'1': "osm4u", '2': "osm4u"}
1506 self.timeout = 360
1507
1508 def run(self, engine, test_osm, manual_check, test_params=None):
1509 # super().run(engine, test_osm, manual_check, test_params)
1510 # run again setting IPs with instantiate parameters
1511 instantiation_params = {
1512 "vnf": [
1513 {
1514 "member-vnf-index": "1",
1515 "internal-vld": [
1516 {
1517 "name": "internal_vld1", # net_internal
1518 "ip-profile": {
1519 "ip-version": "ipv4",
1520 "subnet-address": "10.9.8.0/24",
1521 "dhcp-params": {"count": 100, "start-address": "10.9.8.100"}
1522 },
1523 "internal-connection-point": [
1524 {
1525 "id-ref": "eth2",
1526 "ip-address": "10.9.8.2",
1527 },
1528 {
1529 "id-ref": "eth3",
1530 "ip-address": "10.9.8.3",
1531 }
1532 ]
1533 },
1534 ],
1535
1536 "vdu": [
1537 {
1538 "id": "VM1",
1539 "interface": [
1540 # {
1541 # "name": "iface11",
1542 # "floating-ip-required": True,
1543 # },
1544 {
1545 "name": "iface13",
1546 "mac-address": "52:33:44:55:66:13"
1547 },
1548 ],
1549 },
1550 {
1551 "id": "VM2",
1552 "interface": [
1553 {
1554 "name": "iface21",
1555 "ip-address": "10.31.31.22",
1556 "mac-address": "52:33:44:55:66:21"
1557 },
1558 ],
1559 },
1560 ]
1561 },
1562 ]
1563 }
1564
1565 super().run(engine, test_osm, manual_check, test_params={"ns-config": instantiation_params})
1566
1567
1568 class TestDeployHackfest4(TestDeploy):
1569 description = "Load and deploy Hackfest 4 example."
1570
1571 def __init__(self):
1572 super().__init__()
1573 self.test_name = "HACKFEST4-"
1574 self.vnfd_filenames = ("hackfest_4_vnfd.tar.gz",)
1575 self.nsd_filename = "hackfest_4_nsd.tar.gz"
1576 self.uses_configuration = True
1577 self.commands = {'1': ['ls -lrt', ], '2': ['ls -lrt', ]}
1578 self.users = {'1': "ubuntu", '2': "ubuntu"}
1579 self.passwords = {'1': "osm4u", '2': "osm4u"}
1580 # Modify VNFD to add scaling
1581 # self.descriptor_edit = {
1582 # "vnfd0": {
1583 # 'vnf-configuration': {
1584 # 'config-primitive': [{
1585 # 'name': 'touch',
1586 # 'parameter': [{
1587 # 'name': 'filename',
1588 # 'data-type': 'STRING',
1589 # 'default-value': '/home/ubuntu/touched'
1590 # }]
1591 # }]
1592 # },
1593 # 'scaling-group-descriptor': [{
1594 # 'name': 'scale_dataVM',
1595 # 'scaling-policy': [{
1596 # 'threshold-time': 0,
1597 # 'name': 'auto_cpu_util_above_threshold',
1598 # 'scaling-type': 'automatic',
1599 # 'scaling-criteria': [{
1600 # 'name': 'cpu_util_above_threshold',
1601 # 'vnf-monitoring-param-ref': 'all_aaa_cpu_util',
1602 # 'scale-out-relational-operation': 'GE',
1603 # 'scale-in-threshold': 15,
1604 # 'scale-out-threshold': 60,
1605 # 'scale-in-relational-operation': 'LE'
1606 # }],
1607 # 'cooldown-time': 60
1608 # }],
1609 # 'max-instance-count': 10,
1610 # 'scaling-config-action': [
1611 # {'vnf-config-primitive-name-ref': 'touch',
1612 # 'trigger': 'post-scale-out'},
1613 # {'vnf-config-primitive-name-ref': 'touch',
1614 # 'trigger': 'pre-scale-in'}
1615 # ],
1616 # 'vdu': [{
1617 # 'vdu-id-ref': 'dataVM',
1618 # 'count': 1
1619 # }]
1620 # }]
1621 # }
1622 # }
1623
1624
1625 class TestDeployHackfest3Charmed(TestDeploy):
1626 description = "Load and deploy Hackfest 3charmed_ns example"
1627
1628 def __init__(self):
1629 super().__init__()
1630 self.test_name = "HACKFEST3-"
1631 self.vnfd_filenames = ("hackfest_3charmed_vnfd.tar.gz",)
1632 self.nsd_filename = "hackfest_3charmed_nsd.tar.gz"
1633 self.uses_configuration = True
1634 self.commands = {'1': ['ls -lrt /home/ubuntu/first-touch'], '2': ['ls -lrt /home/ubuntu/first-touch']}
1635 self.users = {'1': "ubuntu", '2': "ubuntu"}
1636 self.passwords = {'1': "osm4u", '2': "osm4u"}
1637 self.descriptor_edit = {
1638 "vnfd0": yaml.safe_load(
1639 """
1640 vnf-configuration:
1641 terminate-config-primitive:
1642 - seq: '1'
1643 name: touch
1644 parameter:
1645 - name: filename
1646 value: '/home/ubuntu/last-touch1'
1647 - seq: '3'
1648 name: touch
1649 parameter:
1650 - name: filename
1651 value: '/home/ubuntu/last-touch3'
1652 - seq: '2'
1653 name: touch
1654 parameter:
1655 - name: filename
1656 value: '/home/ubuntu/last-touch2'
1657 """)
1658 }
1659
1660 def additional_operations(self, engine, test_osm, manual_check):
1661 if not test_osm:
1662 return
1663 # 1 perform action
1664 vnfr_index_selected = "2"
1665 payload = '{member_vnf_index: "2", primitive: touch, primitive_params: { filename: /home/ubuntu/OSMTESTNBI }}'
1666 engine.test("Exec service primitive over NS", "POST",
1667 "/nslcm/v1/ns_instances/{}/action".format(self.ns_id), headers_yaml, payload,
1668 (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1669 nslcmop2_action = engine.last_id
1670 # Wait until status is Ok
1671 engine.wait_operation_ready("ns", nslcmop2_action, timeout_deploy)
1672 vnfr_ip = self.get_vnfr_ip(engine, vnfr_index_selected)
1673 if manual_check:
1674 input(
1675 "NS service primitive has been executed."
1676 "Check that file /home/ubuntu/OSMTESTNBI is present at {}".
1677 format(vnfr_ip))
1678 if test_osm:
1679 commands = {'1': [''], '2': ['ls -lrt /home/ubuntu/OSMTESTNBI', ]}
1680 self.test_ns(engine, test_osm, commands=commands)
1681
1682 # # 2 perform scale out
1683 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1684 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1685 # engine.test("Execute scale action over NS", "POST",
1686 # "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1687 # (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1688 # nslcmop2_scale_out = engine.last_id
1689 # engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
1690 # if manual_check:
1691 # input('NS scale out done. Check that file /home/ubuntu/touched is present and new VM is created')
1692 # # TODO check automatic
1693 #
1694 # # 2 perform scale in
1695 # payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1696 # '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1697 # engine.test("Execute scale action over NS", "POST",
1698 # "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1699 # (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1700 # nslcmop2_scale_in = engine.last_id
1701 # engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
1702 # if manual_check:
1703 # input('NS scale in done. Check that file /home/ubuntu/touched is updated and new VM is deleted')
1704 # # TODO check automatic
1705
1706
1707 class TestDeployHackfest3Charmed2(TestDeployHackfest3Charmed):
1708 description = "Load and deploy Hackfest 3charmed_ns example modified version of descriptors to have dots in " \
1709 "ids and member-vnf-index."
1710
1711 def __init__(self):
1712 super().__init__()
1713 self.test_name = "HACKFEST3v2-"
1714 self.qforce = "?FORCE=True"
1715 self.descriptor_edit = {
1716 "vnfd0": {
1717 "vdu": {
1718 "$[0]": {
1719 "interface": {"$[0]": {"external-connection-point-ref": "pdu-mgmt"}}
1720 },
1721 "$[1]": None
1722 },
1723 "vnf-configuration": None,
1724 "connection-point": {
1725 "$[0]": {
1726 "id": "pdu-mgmt",
1727 "name": "pdu-mgmt",
1728 "short-name": "pdu-mgmt"
1729 },
1730 "$[1]": None
1731 },
1732 "mgmt-interface": {"cp": "pdu-mgmt"},
1733 "description": "A vnf single vdu to be used as PDU",
1734 "id": "vdu-as-pdu",
1735 "internal-vld": {
1736 "$[0]": {
1737 "id": "pdu_internal",
1738 "name": "pdu_internal",
1739 "internal-connection-point": {"$[1]": None},
1740 "short-name": "pdu_internal",
1741 "type": "ELAN"
1742 }
1743 }
1744 },
1745
1746 # Modify NSD accordingly
1747 "nsd": {
1748 "constituent-vnfd": {
1749 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
1750 "$[1]": None,
1751 },
1752 "description": "A nsd to deploy the vnf to act as as PDU",
1753 "id": "nsd-as-pdu",
1754 "name": "nsd-as-pdu",
1755 "short-name": "nsd-as-pdu",
1756 "vld": {
1757 "$[0]": {
1758 "id": "mgmt_pdu",
1759 "name": "mgmt_pdu",
1760 "short-name": "mgmt_pdu",
1761 "vnfd-connection-point-ref": {
1762 "$[0]": {
1763 "vnfd-connection-point-ref": "pdu-mgmt",
1764 "vnfd-id-ref": "vdu-as-pdu",
1765 },
1766 "$[1]": None
1767 },
1768 "type": "ELAN"
1769 },
1770 "$[1]": None,
1771 }
1772 }
1773 }
1774
1775
1776 class TestDeployHackfest3Charmed3(TestDeployHackfest3Charmed):
1777 description = "Load and deploy Hackfest 3charmed_ns example modified version to test scaling and NS parameters"
1778
1779 def __init__(self):
1780 super().__init__()
1781 self.test_name = "HACKFEST3v3-"
1782 self.commands = {'1': ['ls -lrt /home/ubuntu/first-touch-1'], '2': ['ls -lrt /home/ubuntu/first-touch-2']}
1783 self.descriptor_edit = {
1784 "vnfd0": yaml.load(
1785 """
1786 scaling-group-descriptor:
1787 - name: "scale_dataVM"
1788 max-instance-count: 10
1789 scaling-policy:
1790 - name: "auto_cpu_util_above_threshold"
1791 scaling-type: "automatic"
1792 threshold-time: 0
1793 cooldown-time: 60
1794 scaling-criteria:
1795 - name: "cpu_util_above_threshold"
1796 scale-in-threshold: 15
1797 scale-in-relational-operation: "LE"
1798 scale-out-threshold: 60
1799 scale-out-relational-operation: "GE"
1800 vnf-monitoring-param-ref: "monitor1"
1801 vdu:
1802 - vdu-id-ref: dataVM
1803 count: 1
1804 scaling-config-action:
1805 - trigger: post-scale-out
1806 vnf-config-primitive-name-ref: touch
1807 - trigger: pre-scale-in
1808 vnf-config-primitive-name-ref: touch
1809 vdu:
1810 "$id: dataVM":
1811 monitoring-param:
1812 - id: "dataVM_cpu_util"
1813 nfvi-metric: "cpu_utilization"
1814
1815 monitoring-param:
1816 - id: "monitor1"
1817 name: "monitor1"
1818 aggregation-type: AVERAGE
1819 vdu-monitoring-param:
1820 vdu-ref: "dataVM"
1821 vdu-monitoring-param-ref: "dataVM_cpu_util"
1822 vnf-configuration:
1823 initial-config-primitive:
1824 "$[1]":
1825 parameter:
1826 "$[0]":
1827 value: "<touch_filename>" # default-value: /home/ubuntu/first-touch
1828 config-primitive:
1829 "$[0]":
1830 parameter:
1831 "$[0]":
1832 default-value: "<touch_filename2>"
1833 """)
1834 }
1835 self.ns_params = {
1836 "additionalParamsForVnf": [
1837 {"member-vnf-index": "1", "additionalParams": {"touch_filename": "/home/ubuntu/first-touch-1",
1838 "touch_filename2": "/home/ubuntu/second-touch-1"}},
1839 {"member-vnf-index": "2", "additionalParams": {"touch_filename": "/home/ubuntu/first-touch-2",
1840 "touch_filename2": "/home/ubuntu/second-touch-2"}},
1841 ]
1842 }
1843
1844 def additional_operations(self, engine, test_osm, manual_check):
1845 super().additional_operations(engine, test_osm, manual_check)
1846 if not test_osm:
1847 return
1848
1849 # 2 perform scale out
1850 payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1851 '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1852 engine.test("Execute scale action over NS", "POST",
1853 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1854 (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1855 nslcmop2_scale_out = engine.last_id
1856 engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
1857 if manual_check:
1858 input('NS scale out done. Check that file /home/ubuntu/second-touch-1 is present and new VM is created')
1859 if test_osm:
1860 commands = {'1': ['ls -lrt /home/ubuntu/second-touch-1', ]}
1861 self.test_ns(engine, test_osm, commands=commands)
1862 # TODO check automatic connection to scaled VM
1863
1864 # 2 perform scale in
1865 payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1866 '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1867 engine.test("Execute scale action over NS", "POST",
1868 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1869 (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1870 nslcmop2_scale_in = engine.last_id
1871 engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
1872 if manual_check:
1873 input('NS scale in done. Check that file /home/ubuntu/second-touch-1 is updated and new VM is deleted')
1874 # TODO check automatic
1875
1876
1877 class TestDeploySimpleCharm(TestDeploy):
1878 description = "Deploy hackfest-4 hackfest_simplecharm example"
1879
1880 def __init__(self):
1881 super().__init__()
1882 self.test_name = "HACKFEST-SIMPLE"
1883 self.descriptor_url = "https://osm-download.etsi.org/ftp/osm-4.0-four/4th-hackfest/packages/"
1884 self.vnfd_filenames = ("hackfest_simplecharm_vnf.tar.gz",)
1885 self.nsd_filename = "hackfest_simplecharm_ns.tar.gz"
1886 self.uses_configuration = True
1887 self.commands = {'1': [''], '2': ['ls -lrt /home/ubuntu/first-touch', ]}
1888 self.users = {'1': "ubuntu", '2': "ubuntu"}
1889 self.passwords = {'1': "osm4u", '2': "osm4u"}
1890
1891
1892 class TestDeploySimpleCharm2(TestDeploySimpleCharm):
1893 description = "Deploy hackfest-4 hackfest_simplecharm example changing naming to contain dots on ids and " \
1894 "vnf-member-index"
1895
1896 def __init__(self):
1897 super().__init__()
1898 self.test_name = "HACKFEST-SIMPLE2-"
1899 self.qforce = "?FORCE=True"
1900 self.descriptor_edit = {
1901 "vnfd0": {
1902 "id": "hackfest.simplecharm.vnf"
1903 },
1904
1905 "nsd": {
1906 "id": "hackfest.simplecharm.ns",
1907 "constituent-vnfd": {
1908 "$[0]": {"vnfd-id-ref": "hackfest.simplecharm.vnf", "member-vnf-index": "$1"},
1909 "$[1]": {"vnfd-id-ref": "hackfest.simplecharm.vnf", "member-vnf-index": "$2"},
1910 },
1911 "vld": {
1912 "$[0]": {
1913 "vnfd-connection-point-ref": {"$[0]": {"member-vnf-index-ref": "$1",
1914 "vnfd-id-ref": "hackfest.simplecharm.vnf"},
1915 "$[1]": {"member-vnf-index-ref": "$2",
1916 "vnfd-id-ref": "hackfest.simplecharm.vnf"}},
1917 },
1918 "$[1]": {
1919 "vnfd-connection-point-ref": {"$[0]": {"member-vnf-index-ref": "$1",
1920 "vnfd-id-ref": "hackfest.simplecharm.vnf"},
1921 "$[1]": {"member-vnf-index-ref": "$2",
1922 "vnfd-id-ref": "hackfest.simplecharm.vnf"}},
1923 },
1924 }
1925 }
1926 }
1927
1928
1929 class TestDeploySingleVdu(TestDeployHackfest3Charmed):
1930 description = "Generate a single VDU base on editing Hackfest3Charmed descriptors and deploy"
1931
1932 def __init__(self):
1933 super().__init__()
1934 self.test_name = "SingleVDU"
1935 self.qforce = "?FORCE=True"
1936 self.descriptor_edit = {
1937 # Modify VNFD to remove one VDU
1938 "vnfd0": {
1939 "vdu": {
1940 "$[0]": {
1941 "interface": {"$[0]": {"external-connection-point-ref": "pdu-mgmt"}}
1942 },
1943 "$[1]": None
1944 },
1945 "vnf-configuration": None,
1946 "connection-point": {
1947 "$[0]": {
1948 "id": "pdu-mgmt",
1949 "name": "pdu-mgmt",
1950 "short-name": "pdu-mgmt"
1951 },
1952 "$[1]": None
1953 },
1954 "mgmt-interface": {"cp": "pdu-mgmt"},
1955 "description": "A vnf single vdu to be used as PDU",
1956 "id": "vdu-as-pdu",
1957 "internal-vld": {
1958 "$[0]": {
1959 "id": "pdu_internal",
1960 "name": "pdu_internal",
1961 "internal-connection-point": {"$[1]": None},
1962 "short-name": "pdu_internal",
1963 "type": "ELAN"
1964 }
1965 }
1966 },
1967
1968 # Modify NSD accordingly
1969 "nsd": {
1970 "constituent-vnfd": {
1971 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
1972 "$[1]": None,
1973 },
1974 "description": "A nsd to deploy the vnf to act as as PDU",
1975 "id": "nsd-as-pdu",
1976 "name": "nsd-as-pdu",
1977 "short-name": "nsd-as-pdu",
1978 "vld": {
1979 "$[0]": {
1980 "id": "mgmt_pdu",
1981 "name": "mgmt_pdu",
1982 "short-name": "mgmt_pdu",
1983 "vnfd-connection-point-ref": {
1984 "$[0]": {
1985 "vnfd-connection-point-ref": "pdu-mgmt",
1986 "vnfd-id-ref": "vdu-as-pdu",
1987 },
1988 "$[1]": None
1989 },
1990 "type": "ELAN"
1991 },
1992 "$[1]": None,
1993 }
1994 }
1995 }
1996
1997
1998 class TestDeployHnfd(TestDeployHackfest3Charmed):
1999 description = "Generate a HNFD base on editing Hackfest3Charmed descriptors and deploy"
2000
2001 def __init__(self):
2002 super().__init__()
2003 self.test_name = "HNFD"
2004 self.pduDeploy = TestDeploySingleVdu()
2005 self.pdu_interface_0 = {}
2006 self.pdu_interface_1 = {}
2007
2008 self.pdu_id = None
2009 # self.vnf_to_pdu = """
2010 # vdu:
2011 # "$[0]":
2012 # pdu-type: PDU-TYPE-1
2013 # interface:
2014 # "$[0]":
2015 # name: mgmt-iface
2016 # "$[1]":
2017 # name: pdu-iface-internal
2018 # id: hfn1
2019 # description: HFND, one PDU + One VDU
2020 # name: hfn1
2021 # short-name: hfn1
2022 #
2023 # """
2024
2025 self.pdu_descriptor = {
2026 "name": "my-PDU",
2027 "type": "PDU-TYPE-1",
2028 "vim_accounts": "to-override",
2029 "interfaces": [
2030 {
2031 "name": "mgmt-iface",
2032 "mgmt": True,
2033 "type": "overlay",
2034 "ip-address": "to override",
2035 "mac-address": "mac_address",
2036 "vim-network-name": "mgmt",
2037 },
2038 {
2039 "name": "pdu-iface-internal",
2040 "mgmt": False,
2041 "type": "overlay",
2042 "ip-address": "to override",
2043 "mac-address": "mac_address",
2044 "vim-network-name": "pdu_internal", # OSMNBITEST-PDU-pdu_internal
2045 },
2046 ]
2047 }
2048 self.vnfd_filenames = ("hackfest_3charmed_vnfd.tar.gz", "hackfest_3charmed_vnfd.tar.gz")
2049
2050 self.descriptor_edit = {
2051 "vnfd0": {
2052 "id": "hfnd1",
2053 "name": "hfn1",
2054 "short-name": "hfn1",
2055 "vdu": {
2056 "$[0]": {
2057 "pdu-type": "PDU-TYPE-1",
2058 "interface": {
2059 "$[0]": {"name": "mgmt-iface"},
2060 "$[1]": {"name": "pdu-iface-internal"},
2061 }
2062 }
2063 }
2064 },
2065 "nsd": {
2066 "constituent-vnfd": {
2067 "$[1]": {"vnfd-id-ref": "hfnd1"}
2068 },
2069 "vld": {
2070 "$[0]": {"vnfd-connection-point-ref": {"$[1]": {"vnfd-id-ref": "hfnd1"}}},
2071 "$[1]": {"vnfd-connection-point-ref": {"$[1]": {"vnfd-id-ref": "hfnd1"}}}
2072 }
2073 }
2074 }
2075
2076 def create_descriptors(self, engine):
2077 super().create_descriptors(engine)
2078
2079 # Create PDU
2080 self.pdu_descriptor["interfaces"][0].update(self.pdu_interface_0)
2081 self.pdu_descriptor["interfaces"][1].update(self.pdu_interface_1)
2082 self.pdu_descriptor["vim_accounts"] = [self.vim_id]
2083 # TODO get vim-network-name from vnfr.vld.name
2084 self.pdu_descriptor["interfaces"][1]["vim-network-name"] = "{}-{}-{}".format(
2085 os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST"),
2086 "PDU", self.pdu_descriptor["interfaces"][1]["vim-network-name"])
2087 engine.test("Onboard PDU descriptor", "POST", "/pdu/v1/pdu_descriptors",
2088 {"Location": "/pdu/v1/pdu_descriptors/", "Content-Type": "application/yaml"}, self.pdu_descriptor,
2089 201, r_header_yaml, "yaml")
2090 self.pdu_id = engine.last_id
2091
2092 def run(self, engine, test_osm, manual_check, test_params=None):
2093 engine.get_autorization()
2094 engine.set_test_name(self.test_name)
2095 nsname = os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
2096
2097 # create real VIM if not exist
2098 self.vim_id = engine.get_create_vim(test_osm)
2099 # instantiate PDU
2100 self.pduDeploy.create_descriptors(engine)
2101 self.pduDeploy.instantiate(engine, {"nsDescription": "to be used as PDU", "nsName": nsname + "-PDU",
2102 "nsdId": self.pduDeploy.nsd_id, "vimAccountId": self.vim_id})
2103 if manual_check:
2104 input('VNF to be used as PDU has been deployed. Perform manual check and press enter to resume')
2105 if test_osm:
2106 self.pduDeploy.test_ns(engine, test_osm)
2107
2108 if test_osm:
2109 r = engine.test("Get VNFR to obtain IP_ADDRESS", "GET",
2110 "/nslcm/v1/vnfrs?nsr-id-ref={}".format(self.pduDeploy.ns_id), headers_json, None,
2111 200, r_header_json, "json")
2112 if not r:
2113 return
2114 vnfr_data = r.json()
2115 # print(vnfr_data)
2116
2117 self.pdu_interface_0["ip-address"] = vnfr_data[0]["vdur"][0]["interfaces"][0].get("ip-address")
2118 self.pdu_interface_1["ip-address"] = vnfr_data[0]["vdur"][0]["interfaces"][1].get("ip-address")
2119 self.pdu_interface_0["mac-address"] = vnfr_data[0]["vdur"][0]["interfaces"][0].get("mac-address")
2120 self.pdu_interface_1["mac-address"] = vnfr_data[0]["vdur"][0]["interfaces"][1].get("mac-address")
2121 if not self.pdu_interface_0["ip-address"]:
2122 raise TestException("Vnfr has not managment ip address")
2123 else:
2124 self.pdu_interface_0["ip-address"] = "192.168.10.10"
2125 self.pdu_interface_1["ip-address"] = "192.168.11.10"
2126 self.pdu_interface_0["mac-address"] = "52:33:44:55:66:13"
2127 self.pdu_interface_1["mac-address"] = "52:33:44:55:66:14"
2128
2129 self.create_descriptors(engine)
2130
2131 ns_data = {"nsDescription": "default description", "nsName": nsname, "nsdId": self.nsd_id,
2132 "vimAccountId": self.vim_id}
2133 if test_params and test_params.get("ns-config"):
2134 if isinstance(test_params["ns-config"], str):
2135 ns_data.update(yaml.load(test_params["ns-config"]))
2136 else:
2137 ns_data.update(test_params["ns-config"])
2138
2139 self.instantiate(engine, ns_data)
2140 if manual_check:
2141 input('NS has been deployed. Perform manual check and press enter to resume')
2142 if test_osm:
2143 self.test_ns(engine, test_osm)
2144 self.additional_operations(engine, test_osm, manual_check)
2145 self.terminate(engine)
2146 self.pduDeploy.terminate(engine)
2147 self.delete_descriptors(engine)
2148 self.pduDeploy.delete_descriptors(engine)
2149
2150 def delete_descriptors(self, engine):
2151 super().delete_descriptors(engine)
2152 # delete pdu
2153 engine.test("Delete PDU SOL005", "DELETE",
2154 "/pdu/v1/pdu_descriptors/{}".format(self.pdu_id),
2155 headers_yaml, None, 204, None, 0)
2156
2157
2158 class TestDescriptors:
2159 description = "Test VNFD, NSD, PDU descriptors CRUD and dependencies"
2160 vnfd_empty = """vnfd:vnfd-catalog:
2161 vnfd:
2162 - name: prova
2163 short-name: prova
2164 id: prova
2165 """
2166 vnfd_prova = """vnfd:vnfd-catalog:
2167 vnfd:
2168 - connection-point:
2169 - name: cp_0h8m
2170 type: VPORT
2171 id: prova
2172 name: prova
2173 short-name: prova
2174 vdu:
2175 - id: vdu_z4bm
2176 image: ubuntu
2177 interface:
2178 - external-connection-point-ref: cp_0h8m
2179 name: eth0
2180 virtual-interface:
2181 type: VIRTIO
2182 name: vdu_z4bm
2183 version: '1.0'
2184 """
2185
2186 def __init__(self):
2187 self.vnfd_filename = "hackfest_3charmed_vnfd.tar.gz"
2188 self.nsd_filename = "hackfest_3charmed_nsd.tar.gz"
2189 self.descriptor_url = "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
2190 self.vnfd_id = None
2191 self.nsd_id = None
2192
2193 def run(self, engine, test_osm, manual_check, test_params=None):
2194 engine.set_test_name("Descriptors")
2195 engine.get_autorization()
2196 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
2197 if not os.path.exists(temp_dir):
2198 os.makedirs(temp_dir)
2199
2200 # download files
2201 for filename in (self.vnfd_filename, self.nsd_filename):
2202 filename_path = temp_dir + filename
2203 if not os.path.exists(filename_path):
2204 with open(filename_path, "wb") as file:
2205 response = requests.get(self.descriptor_url + filename)
2206 if response.status_code >= 300:
2207 raise TestException("Error downloading descriptor from '{}': {}".format(
2208 self.descriptor_url + filename, response.status_code))
2209 file.write(response.content)
2210
2211 vnfd_filename_path = temp_dir + self.vnfd_filename
2212 nsd_filename_path = temp_dir + self.nsd_filename
2213
2214 engine.test("Onboard empty VNFD in one step", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml,
2215 self.vnfd_empty, 201, r_headers_yaml_location_vnfd, "yaml")
2216 self.vnfd_id = engine.last_id
2217
2218 # test bug 605
2219 engine.test("Upload invalid VNFD ", "PUT", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id),
2220 headers_yaml, self.vnfd_prova, 422, r_header_yaml, "yaml")
2221
2222 engine.test("Upload VNFD {}".format(self.vnfd_filename), "PUT",
2223 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id), headers_zip_yaml,
2224 "@b" + vnfd_filename_path, 204, None, 0)
2225
2226 queries = ["mgmt-interface.cp=mgmt", "vdu.0.interface.0.external-connection-point-ref=mgmt",
2227 "vdu.0.interface.1.internal-connection-point-ref=internal",
2228 "internal-vld.0.internal-connection-point.0.id-ref=internal",
2229 # Detection of duplicated VLD names in VNF Descriptors
2230 # URL: internal-vld=[
2231 # {id: internal1, name: internal, type:ELAN,
2232 # internal-connection-point: [{id-ref: mgmtVM-internal}, {id-ref: dataVM-internal}]},
2233 # {id: internal2, name: internal, type:ELAN,
2234 # internal-connection-point: [{id-ref: mgmtVM-internal}, {id-ref: dataVM-internal}]}
2235 # ]
2236 "internal-vld=%5B%7Bid%3A%20internal1%2C%20name%3A%20internal%2C%20type%3A%20ELAN%2C%20"
2237 "internal-connection-point%3A%20%5B%7Bid-ref%3A%20mgmtVM-internal%7D%2C%20%7Bid-ref%3A%20"
2238 "dataVM-internal%7D%5D%7D%2C%20%7Bid%3A%20internal2%2C%20name%3A%20internal%2C%20type%3A%20"
2239 "ELAN%2C%20internal-connection-point%3A%20%5B%7Bid-ref%3A%20mgmtVM-internal%7D%2C%20%7B"
2240 "id-ref%3A%20dataVM-internal%7D%5D%7D%5D"
2241 ]
2242 for query in queries:
2243 engine.test("Upload invalid VNFD ", "PUT",
2244 "/vnfpkgm/v1/vnf_packages/{}/package_content?{}".format(self.vnfd_id, query),
2245 headers_zip_yaml, "@b" + vnfd_filename_path, 422, r_header_yaml, "yaml")
2246
2247 # test bug 605
2248 engine.test("Upload invalid VNFD ", "PUT", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id),
2249 headers_yaml, self.vnfd_prova, 422, r_header_yaml, "yaml")
2250
2251 # get vnfd descriptor
2252 engine.test("Get VNFD descriptor", "GET", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_id),
2253 headers_yaml, None, 200, r_header_yaml, "yaml")
2254
2255 # get vnfd file descriptor
2256 engine.test("Get VNFD file descriptor", "GET", "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(self.vnfd_id),
2257 headers_text, None, 200, r_header_text, "text", temp_dir+"vnfd-yaml")
2258 # TODO compare files: diff vnfd-yaml hackfest_3charmed_vnfd/hackfest_3charmed_vnfd.yaml
2259
2260 # get vnfd zip file package
2261 engine.test("Get VNFD zip package", "GET",
2262 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id), headers_zip, None, 200,
2263 r_header_zip, "zip", temp_dir+"vnfd-zip")
2264 # TODO compare files: diff vnfd-zip hackfest_3charmed_vnfd.tar.gz
2265
2266 # get vnfd artifact
2267 engine.test("Get VNFD artifact package", "GET",
2268 "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/osm.png".format(self.vnfd_id), headers_zip, None, 200,
2269 r_header_octect, "octet-string", temp_dir+"vnfd-icon")
2270 # TODO compare files: diff vnfd-icon hackfest_3charmed_vnfd/icons/osm.png
2271
2272 # nsd CREATE AND UPLOAD in one step:
2273 engine.test("Onboard NSD in one step", "POST", "/nsd/v1/ns_descriptors_content", headers_zip_yaml,
2274 "@b" + nsd_filename_path, 201, r_headers_yaml_location_nsd, "yaml")
2275 self.nsd_id = engine.last_id
2276
2277 queries = ["vld.0.vnfd-connection-point-ref.0.vnfd-id-ref=hf"]
2278 for query in queries:
2279 engine.test("Upload invalid NSD ", "PUT",
2280 "/nsd/v1/ns_descriptors/{}/nsd_content?{}".format(self.nsd_id, query),
2281 headers_zip_yaml, "@b" + nsd_filename_path, 422, r_header_yaml, "yaml")
2282
2283 # get nsd descriptor
2284 engine.test("Get NSD descriptor", "GET", "/nsd/v1/ns_descriptors/{}".format(self.nsd_id), headers_yaml,
2285 None, 200, r_header_yaml, "yaml")
2286
2287 # get nsd file descriptor
2288 engine.test("Get NSD file descriptor", "GET", "/nsd/v1/ns_descriptors/{}/nsd".format(self.nsd_id), headers_text,
2289 None, 200, r_header_text, "text", temp_dir+"nsd-yaml")
2290 # TODO compare files: diff nsd-yaml hackfest_3charmed_nsd/hackfest_3charmed_nsd.yaml
2291
2292 # get nsd zip file package
2293 engine.test("Get NSD zip package", "GET", "/nsd/v1/ns_descriptors/{}/nsd_content".format(self.nsd_id),
2294 headers_zip, None, 200, r_header_zip, "zip", temp_dir+"nsd-zip")
2295 # TODO compare files: diff nsd-zip hackfest_3charmed_nsd.tar.gz
2296
2297 # get nsd artifact
2298 engine.test("Get NSD artifact package", "GET",
2299 "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm.png".format(self.nsd_id), headers_zip, None, 200,
2300 r_header_octect, "octet-string", temp_dir+"nsd-icon")
2301 # TODO compare files: diff nsd-icon hackfest_3charmed_nsd/icons/osm.png
2302
2303 # vnfd DELETE
2304 test_rest.test("Delete VNFD conflict", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_id),
2305 headers_yaml, None, 409, None, None)
2306
2307 test_rest.test("Delete VNFD force", "DELETE", "/vnfpkgm/v1/vnf_packages/{}?FORCE=TRUE".format(self.vnfd_id),
2308 headers_yaml, None, 204, None, 0)
2309
2310 # nsd DELETE
2311 test_rest.test("Delete NSD", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self.nsd_id), headers_yaml, None, 204,
2312 None, 0)
2313
2314
2315 class TestNetSliceTemplates:
2316 description = "Upload a NST to OSM"
2317
2318 def __init__(self):
2319 self.vnfd_filename = ("@./slice_shared/vnfd/slice_shared_vnfd.yaml")
2320 self.vnfd_filename_middle = ("@./slice_shared/vnfd/slice_shared_middle_vnfd.yaml")
2321 self.nsd_filename = ("@./slice_shared/nsd/slice_shared_nsd.yaml")
2322 self.nsd_filename_middle = ("@./slice_shared/nsd/slice_shared_middle_nsd.yaml")
2323 self.nst_filenames = ("@./slice_shared/slice_shared_nstd.yaml")
2324
2325 def run(self, engine, test_osm, manual_check, test_params=None):
2326 # nst CREATE
2327 engine.set_test_name("NST step ")
2328 engine.get_autorization()
2329 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
2330 if not os.path.exists(temp_dir):
2331 os.makedirs(temp_dir)
2332
2333 # Onboard VNFDs
2334 engine.test("Onboard edge VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml,
2335 self.vnfd_filename, 201, r_headers_yaml_location_vnfd, "yaml")
2336 self.vnfd_edge_id = engine.last_id
2337
2338 engine.test("Onboard middle VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml,
2339 self.vnfd_filename_middle, 201, r_headers_yaml_location_vnfd, "yaml")
2340 self.vnfd_middle_id = engine.last_id
2341
2342 # Onboard NSDs
2343 engine.test("Onboard NSD edge", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml,
2344 self.nsd_filename, 201, r_headers_yaml_location_nsd, "yaml")
2345 self.nsd_edge_id = engine.last_id
2346
2347 engine.test("Onboard NSD middle", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml,
2348 self.nsd_filename_middle, 201, r_headers_yaml_location_nsd, "yaml")
2349 self.nsd_middle_id = engine.last_id
2350
2351 # Onboard NST
2352 engine.test("Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml, self.nst_filenames,
2353 201, r_headers_yaml_location_nst, "yaml")
2354 nst_id = engine.last_id
2355
2356 # nstd SHOW OSM format
2357 engine.test("Show NSTD OSM format", "GET", "/nst/v1/netslice_templates/{}".format(nst_id), headers_json, None,
2358 200, r_header_json, "json")
2359
2360 # nstd DELETE
2361 engine.test("Delete NSTD", "DELETE", "/nst/v1/netslice_templates/{}".format(nst_id), headers_json, None,
2362 204, None, 0)
2363
2364 # NSDs DELETE
2365 test_rest.test("Delete NSD middle", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self.nsd_middle_id),
2366 headers_json, None, 204, None, 0)
2367
2368 test_rest.test("Delete NSD edge", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self.nsd_edge_id), headers_json,
2369 None, 204, None, 0)
2370
2371 # VNFDs DELETE
2372 test_rest.test("Delete VNFD edge", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_edge_id),
2373 headers_yaml, None, 204, None, 0)
2374
2375 test_rest.test("Delete VNFD middle", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_middle_id),
2376 headers_yaml, None, 204, None, 0)
2377
2378
2379 class TestNetSliceInstances:
2380 '''
2381 Test procedure:
2382 1. Populate databases with VNFD, NSD, NST with the following scenario
2383 +-----------------management-----------------+
2384 | | |
2385 +--+---+ +----+----+ +---+--+
2386 | | | | | |
2387 | edge +---data1----+ middle +---data2-----+ edge |
2388 | | | | | |
2389 +------+ +---------+ +------+
2390 shared-nss
2391 2. Create NSI-1
2392 3. Instantiate NSI-1
2393 4. Create NSI-2
2394 5. Instantiate NSI-2
2395 Manual check - Are 2 slices instantiated correctly?
2396 NSI-1 3 nss (2 nss-edges + 1 nss-middle)
2397 NSI-2 2 nss (2 nss-edge sharing nss-middle)
2398 6. Terminate NSI-1
2399 7. Delete NSI-1
2400 Manual check - Is slice NSI-1 deleted correctly?
2401 NSI-2 with 2 nss-edge + 1 nss-middle (The one from NSI-1)
2402 8. Create NSI-3
2403 9. Instantiate NSI-3
2404 Manual check - Is slice NSI-3 instantiated correctly?
2405 NSI-3 reuse nss-middle. NSI-3 only create 2 nss-edge
2406 10. Delete NSI-2
2407 11. Terminate NSI-2
2408 12. Delete NSI-3
2409 13. Terminate NSI-3
2410 Manual check - All cleaned correctly?
2411 NSI-2 and NSI-3 were terminated and deleted
2412 14. Cleanup database
2413 '''
2414
2415 description = "Upload a NST to OSM"
2416
2417 def __init__(self):
2418 self.vim_id = None
2419 self.vnfd_filename = ("@./slice_shared/vnfd/slice_shared_vnfd.yaml")
2420 self.vnfd_filename_middle = ("@./slice_shared/vnfd/slice_shared_middle_vnfd.yaml")
2421 self.nsd_filename = ("@./slice_shared/nsd/slice_shared_nsd.yaml")
2422 self.nsd_filename_middle = ("@./slice_shared/nsd/slice_shared_middle_nsd.yaml")
2423 self.nst_filenames = ("@./slice_shared/slice_shared_nstd.yaml")
2424
2425 def create_slice(self, engine, nsi_data, name):
2426 ns_data_text = yaml.safe_dump(nsi_data, default_flow_style=True, width=256)
2427 r = engine.test(name, "POST", "/nsilcm/v1/netslice_instances",
2428 headers_yaml, ns_data_text, (201, 202),
2429 {"Location": "nsilcm/v1/netslice_instances/", "Content-Type": "application/yaml"}, "yaml")
2430 return r
2431
2432 def instantiate_slice(self, engine, nsi_data, nsi_id, name):
2433 ns_data_text = yaml.safe_dump(nsi_data, default_flow_style=True, width=256)
2434 engine.test(name, "POST",
2435 "/nsilcm/v1/netslice_instances/{}/instantiate".format(nsi_id), headers_yaml, ns_data_text,
2436 (201, 202), r_headers_yaml_location_nsilcmop, "yaml")
2437
2438 def terminate_slice(self, engine, nsi_id, name):
2439 engine.test(name, "POST", "/nsilcm/v1/netslice_instances/{}/terminate".format(nsi_id),
2440 headers_yaml, None, (201, 202), r_headers_yaml_location_nsilcmop, "yaml")
2441
2442 def delete_slice(self, engine, nsi_id, name):
2443 engine.test(name, "DELETE", "/nsilcm/v1/netslice_instances/{}".format(nsi_id), headers_yaml, None,
2444 204, None, 0)
2445
2446 def run(self, engine, test_osm, manual_check, test_params=None):
2447 # nst CREATE
2448 engine.set_test_name("NSI")
2449 engine.get_autorization()
2450
2451 # Onboard VNFDs
2452 engine.test("Onboard edge VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml,
2453 self.vnfd_filename, 201, r_headers_yaml_location_vnfd, "yaml")
2454 self.vnfd_edge_id = engine.last_id
2455
2456 engine.test("Onboard middle VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml,
2457 self.vnfd_filename_middle, 201, r_headers_yaml_location_vnfd, "yaml")
2458 self.vnfd_middle_id = engine.last_id
2459
2460 # Onboard NSDs
2461 engine.test("Onboard NSD edge", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml,
2462 self.nsd_filename, 201, r_headers_yaml_location_nsd, "yaml")
2463 self.nsd_edge_id = engine.last_id
2464
2465 engine.test("Onboard NSD middle", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml,
2466 self.nsd_filename_middle, 201, r_headers_yaml_location_nsd, "yaml")
2467 self.nsd_middle_id = engine.last_id
2468
2469 # Onboard NST
2470 engine.test("Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml, self.nst_filenames,
2471 201, r_headers_yaml_location_nst, "yaml")
2472 nst_id = engine.last_id
2473
2474 self.vim_id = engine.get_create_vim(test_osm)
2475
2476 # CREATE NSI-1
2477 ns_data = {'nsiName': 'Deploy-NSI-1', 'vimAccountId': self.vim_id, 'nstId': nst_id, 'nsiDescription': 'default'}
2478 r = self.create_slice(engine, ns_data, "Create NSI-1 step 1")
2479 if not r:
2480 return
2481 self.nsi_id1 = engine.last_id
2482
2483 # INSTANTIATE NSI-1
2484 self.instantiate_slice(engine, ns_data, self.nsi_id1, "Instantiate NSI-1 step 2")
2485 nsilcmop_id1 = engine.last_id
2486
2487 # Waiting for NSI-1
2488 if test_osm:
2489 engine.wait_operation_ready("nsi", nsilcmop_id1, timeout_deploy)
2490
2491 # CREATE NSI-2
2492 ns_data = {'nsiName': 'Deploy-NSI-2', 'vimAccountId': self.vim_id, 'nstId': nst_id, 'nsiDescription': 'default'}
2493 r = self.create_slice(engine, ns_data, "Create NSI-2 step 1")
2494 if not r:
2495 return
2496 self.nsi_id2 = engine.last_id
2497
2498 # INSTANTIATE NSI-2
2499 self.instantiate_slice(engine, ns_data, self.nsi_id2, "Instantiate NSI-2 step 2")
2500 nsilcmop_id2 = engine.last_id
2501
2502 # Waiting for NSI-2
2503 if test_osm:
2504 engine.wait_operation_ready("nsi", nsilcmop_id2, timeout_deploy)
2505
2506 if manual_check:
2507 input('NSI-1 AND NSI-2 has been deployed. Perform manual check and press enter to resume')
2508
2509 # TERMINATE NSI-1
2510 if test_osm:
2511 self.terminate_slice(engine, self.nsi_id1, "Terminate NSI-1")
2512 nsilcmop1_id = engine.last_id
2513
2514 # Wait terminate NSI-1
2515 engine.wait_operation_ready("nsi", nsilcmop1_id, timeout_deploy)
2516
2517 # DELETE NSI-1
2518 self.delete_slice(engine, self.nsi_id1, "Delete NS")
2519
2520 if manual_check:
2521 input('NSI-1 has been deleted. Perform manual check and press enter to resume')
2522
2523 # CREATE NSI-3
2524 ns_data = {'nsiName': 'Deploy-NSI-3', 'vimAccountId': self.vim_id, 'nstId': nst_id, 'nsiDescription': 'default'}
2525 r = self.create_slice(engine, ns_data, "Create NSI-3 step 1")
2526
2527 if not r:
2528 return
2529 self.nsi_id3 = engine.last_id
2530
2531 # INSTANTIATE NSI-3
2532 self.instantiate_slice(engine, ns_data, self.nsi_id3, "Instantiate NSI-3 step 2")
2533 nsilcmop_id3 = engine.last_id
2534
2535 # Wait Instantiate NSI-3
2536 if test_osm:
2537 engine.wait_operation_ready("nsi", nsilcmop_id3, timeout_deploy)
2538
2539 if manual_check:
2540 input('NSI-3 has been deployed. Perform manual check and press enter to resume')
2541
2542 # TERMINATE NSI-2
2543 if test_osm:
2544 self.terminate_slice(engine, self.nsi_id2, "Terminate NSI-2")
2545 nsilcmop2_id = engine.last_id
2546
2547 # Wait terminate NSI-2
2548 engine.wait_operation_ready("nsi", nsilcmop2_id, timeout_deploy)
2549
2550 # DELETE NSI-2
2551 self.delete_slice(engine, self.nsi_id2, "DELETE NSI-2")
2552
2553 # TERMINATE NSI-3
2554 if test_osm:
2555 self. terminate_slice(engine, self.nsi_id3, "Terminate NSI-3")
2556 nsilcmop3_id = engine.last_id
2557
2558 # Wait terminate NSI-3
2559 engine.wait_operation_ready("nsi", nsilcmop3_id, timeout_deploy)
2560
2561 # DELETE NSI-3
2562 self.delete_slice(engine, self.nsi_id3, "DELETE NSI-3")
2563
2564 if manual_check:
2565 input('NSI-2 and NSI-3 has been deleted. Perform manual check and press enter to resume')
2566
2567 # nstd DELETE
2568 engine.test("Delete NSTD", "DELETE", "/nst/v1/netslice_templates/{}".format(nst_id), headers_json, None,
2569 204, None, 0)
2570
2571 # NSDs DELETE
2572 test_rest.test("Delete NSD middle", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self.nsd_middle_id),
2573 headers_json, None, 204, None, 0)
2574
2575 test_rest.test("Delete NSD edge", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self.nsd_edge_id), headers_json,
2576 None, 204, None, 0)
2577
2578 # VNFDs DELETE
2579 test_rest.test("Delete VNFD edge", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_edge_id),
2580 headers_yaml, None, 204, None, 0)
2581
2582 test_rest.test("Delete VNFD middle", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_middle_id),
2583 headers_yaml, None, 204, None, 0)
2584
2585
2586 class TestAuthentication:
2587 description = "Test Authentication"
2588
2589 @staticmethod
2590 def run(engine, test_osm, manual_check, test_params=None):
2591 engine.set_test_name("Authentication")
2592 # backend = test_params.get("backend") if test_params else None # UNUSED
2593
2594 admin_project_id = test_project_id = None
2595 project_admin_role_id = project_user_role_id = None
2596 test_user_id = empty_user_id = None
2597 default_role_id = empty_role_id = token_role_id = None
2598
2599 engine.get_autorization()
2600
2601 # GET
2602 engine.test("Get tokens", "GET", "/admin/v1/tokens", headers_json, {},
2603 (200), {"Content-Type": "application/json"}, "json")
2604 engine.test("Get projects", "GET", "/admin/v1/projects", headers_json, {},
2605 (200), {"Content-Type": "application/json"}, "json")
2606 engine.test("Get users", "GET", "/admin/v1/users", headers_json, {},
2607 (200), {"Content-Type": "application/json"}, "json")
2608 engine.test("Get roles", "GET", "/admin/v1/roles", headers_json, {},
2609 (200), {"Content-Type": "application/json"}, "json")
2610 res = engine.test("Get admin project", "GET", "/admin/v1/projects?name=admin", headers_json, {},
2611 (200), {"Content-Type": "application/json"}, "json")
2612 admin_project_id = res.json()[0]["_id"] if res else None
2613 res = engine.test("Get project admin role", "GET", "/admin/v1/roles?name=project_admin", headers_json, {},
2614 (200), {"Content-Type": "application/json"}, "json")
2615 project_admin_role_id = res.json()[0]["_id"] if res else None
2616 res = engine.test("Get project user role", "GET", "/admin/v1/roles?name=project_user", headers_json, {},
2617 (200), {"Content-Type": "application/json"}, "json")
2618 project_user_role_id = res.json()[0]["_id"] if res else None
2619
2620 # POST
2621 res = engine.test("Create test project", "POST", "/admin/v1/projects", headers_json, {"name": "test"},
2622 (201), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
2623 test_project_id = engine.last_id if res else None
2624 res = engine.test("Create role without permissions", "POST", "/admin/v1/roles", headers_json, {"name": "empty"},
2625 (201), {"Content-Type": "application/json"}, "json")
2626 empty_role_id = engine.last_id if res else None
2627 res = engine.test("Create role with default permissions", "POST", "/admin/v1/roles", headers_json,
2628 {"name": "default", "permissions": {"default": True}},
2629 (201), {"Location": "/admin/v1/roles/", "Content-Type": "application/json"}, "json")
2630 default_role_id = engine.last_id if res else None
2631 res = engine.test("Create role with token permissions", "POST", "/admin/v1/roles", headers_json,
2632 {"name": "tokens", "permissions": {"tokens": True}}, # is default required ?
2633 (201), {"Location": "/admin/v1/roles/", "Content-Type": "application/json"}, "json")
2634 token_role_id = engine.last_id if res else None
2635 pr = "project-role mappings"
2636 res = engine.test("Create user without "+pr, "POST", "/admin/v1/users", headers_json,
2637 {"username": "empty", "password": "empty"},
2638 201, {"Content-Type": "application/json"}, "json")
2639 empty_user_id = engine.last_id if res else None
2640 if admin_project_id and test_project_id and project_admin_role_id and project_user_role_id:
2641 data = {"username": "test", "password": "test"}
2642 data["project_role_mappings"] = [
2643 {"project": test_project_id, "role": project_admin_role_id},
2644 {"project": admin_project_id, "role": project_user_role_id}
2645 ]
2646 res = engine.test("Create user with "+pr, "POST", "/admin/v1/users", headers_json, data,
2647 (201), {"Content-Type": "application/json"}, "json")
2648 test_user_id = engine.last_id if res else None
2649
2650 # PUT
2651 if test_user_id:
2652 engine.test("Modify test user's password", "PUT", "/admin/v1/users/"+test_user_id, headers_json,
2653 {"password": "password"},
2654 (204), {}, 0)
2655 if empty_user_id and admin_project_id and test_project_id and project_admin_role_id and project_user_role_id:
2656 data = {"project_role_mappings": [
2657 {"project": test_project_id, "role": project_admin_role_id},
2658 {"project": admin_project_id, "role": project_user_role_id}
2659 ]}
2660 engine.test("Modify empty user's "+pr, "PUT", "/admin/v1/users/"+empty_user_id,
2661 headers_json,
2662 data,
2663 (204), {}, 0)
2664
2665 # DELETE
2666 if empty_user_id:
2667 engine.test("Delete empty user", "DELETE", "/admin/v1/users/"