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