c9a37d8a226d232021ae4089cc5ec933972d58ea
[osm/NBI.git] / osm_nbi / tests / test.py
1 #! /usr/bin/python3
2 # -*- coding: utf-8 -*-
3
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 # http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
13 # implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16
17 import getopt
18 import sys
19 import requests
20 import json
21 import logging
22 import yaml
23 # import json
24 # import tarfile
25 from time import sleep
26 from random import randint
27 import os
28 from sys import stderr
29 from uuid import uuid4
30 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"]))
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 }
1838 self.ns_params = {
1839 "additionalParamsForVnf": [
1840 {"member-vnf-index": "1", "additionalParams": {"touch_filename": "/home/ubuntu/first-touch-1",
1841 "touch_filename2": "/home/ubuntu/second-touch-1"}},
1842 {"member-vnf-index": "2", "additionalParams": {"touch_filename": "/home/ubuntu/first-touch-2",
1843 "touch_filename2": "/home/ubuntu/second-touch-2"}},
1844 ]
1845 }
1846
1847 def additional_operations(self, engine, test_osm, manual_check):
1848 super().additional_operations(engine, test_osm, manual_check)
1849 if not test_osm:
1850 return
1851
1852 # 2 perform scale out
1853 payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_OUT, scaleByStepData: ' \
1854 '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1855 engine.test("Execute scale action over NS", "POST",
1856 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1857 (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1858 nslcmop2_scale_out = engine.last_id
1859 engine.wait_operation_ready("ns", nslcmop2_scale_out, timeout_deploy)
1860 if manual_check:
1861 input('NS scale out done. Check that file /home/ubuntu/second-touch-1 is present and new VM is created')
1862 if test_osm:
1863 commands = {'1': ['ls -lrt /home/ubuntu/second-touch-1', ]}
1864 self.test_ns(engine, test_osm, commands=commands)
1865 # TODO check automatic connection to scaled VM
1866
1867 # 2 perform scale in
1868 payload = '{scaleType: SCALE_VNF, scaleVnfData: {scaleVnfType: SCALE_IN, scaleByStepData: ' \
1869 '{scaling-group-descriptor: scale_dataVM, member-vnf-index: "1"}}}'
1870 engine.test("Execute scale action over NS", "POST",
1871 "/nslcm/v1/ns_instances/{}/scale".format(self.ns_id), headers_yaml, payload,
1872 (201, 202), r_headers_yaml_location_nslcmop, "yaml")
1873 nslcmop2_scale_in = engine.last_id
1874 engine.wait_operation_ready("ns", nslcmop2_scale_in, timeout_deploy)
1875 if manual_check:
1876 input('NS scale in done. Check that file /home/ubuntu/second-touch-1 is updated and new VM is deleted')
1877 # TODO check automatic
1878
1879
1880 class TestDeploySimpleCharm(TestDeploy):
1881 description = "Deploy hackfest-4 hackfest_simplecharm example"
1882
1883 def __init__(self):
1884 super().__init__()
1885 self.test_name = "HACKFEST-SIMPLE"
1886 self.descriptor_url = "https://osm-download.etsi.org/ftp/osm-4.0-four/4th-hackfest/packages/"
1887 self.vnfd_filenames = ("hackfest_simplecharm_vnf.tar.gz",)
1888 self.nsd_filename = "hackfest_simplecharm_ns.tar.gz"
1889 self.uses_configuration = True
1890 self.commands = {'1': [''], '2': ['ls -lrt /home/ubuntu/first-touch', ]}
1891 self.users = {'1': "ubuntu", '2': "ubuntu"}
1892 self.passwords = {'1': "osm4u", '2': "osm4u"}
1893
1894
1895 class TestDeploySimpleCharm2(TestDeploySimpleCharm):
1896 description = "Deploy hackfest-4 hackfest_simplecharm example changing naming to contain dots on ids and " \
1897 "vnf-member-index"
1898
1899 def __init__(self):
1900 super().__init__()
1901 self.test_name = "HACKFEST-SIMPLE2-"
1902 self.qforce = "?FORCE=True"
1903 self.descriptor_edit = {
1904 "vnfd0": {
1905 "id": "hackfest.simplecharm.vnf"
1906 },
1907
1908 "nsd": {
1909 "id": "hackfest.simplecharm.ns",
1910 "constituent-vnfd": {
1911 "$[0]": {"vnfd-id-ref": "hackfest.simplecharm.vnf", "member-vnf-index": "$1"},
1912 "$[1]": {"vnfd-id-ref": "hackfest.simplecharm.vnf", "member-vnf-index": "$2"},
1913 },
1914 "vld": {
1915 "$[0]": {
1916 "vnfd-connection-point-ref": {"$[0]": {"member-vnf-index-ref": "$1",
1917 "vnfd-id-ref": "hackfest.simplecharm.vnf"},
1918 "$[1]": {"member-vnf-index-ref": "$2",
1919 "vnfd-id-ref": "hackfest.simplecharm.vnf"}},
1920 },
1921 "$[1]": {
1922 "vnfd-connection-point-ref": {"$[0]": {"member-vnf-index-ref": "$1",
1923 "vnfd-id-ref": "hackfest.simplecharm.vnf"},
1924 "$[1]": {"member-vnf-index-ref": "$2",
1925 "vnfd-id-ref": "hackfest.simplecharm.vnf"}},
1926 },
1927 }
1928 }
1929 }
1930
1931
1932 class TestDeploySingleVdu(TestDeployHackfest3Charmed):
1933 description = "Generate a single VDU base on editing Hackfest3Charmed descriptors and deploy"
1934
1935 def __init__(self):
1936 super().__init__()
1937 self.test_name = "SingleVDU"
1938 self.qforce = "?FORCE=True"
1939 self.descriptor_edit = {
1940 # Modify VNFD to remove one VDU
1941 "vnfd0": {
1942 "vdu": {
1943 "$[0]": {
1944 "interface": {"$[0]": {"external-connection-point-ref": "pdu-mgmt"}}
1945 },
1946 "$[1]": None
1947 },
1948 "vnf-configuration": None,
1949 "connection-point": {
1950 "$[0]": {
1951 "id": "pdu-mgmt",
1952 "name": "pdu-mgmt",
1953 "short-name": "pdu-mgmt"
1954 },
1955 "$[1]": None
1956 },
1957 "mgmt-interface": {"cp": "pdu-mgmt"},
1958 "description": "A vnf single vdu to be used as PDU",
1959 "id": "vdu-as-pdu",
1960 "internal-vld": {
1961 "$[0]": {
1962 "id": "pdu_internal",
1963 "name": "pdu_internal",
1964 "internal-connection-point": {"$[1]": None},
1965 "short-name": "pdu_internal",
1966 "type": "ELAN"
1967 }
1968 }
1969 },
1970
1971 # Modify NSD accordingly
1972 "nsd": {
1973 "constituent-vnfd": {
1974 "$[0]": {"vnfd-id-ref": "vdu-as-pdu"},
1975 "$[1]": None,
1976 },
1977 "description": "A nsd to deploy the vnf to act as as PDU",
1978 "id": "nsd-as-pdu",
1979 "name": "nsd-as-pdu",
1980 "short-name": "nsd-as-pdu",
1981 "vld": {
1982 "$[0]": {
1983 "id": "mgmt_pdu",
1984 "name": "mgmt_pdu",
1985 "short-name": "mgmt_pdu",
1986 "vnfd-connection-point-ref": {
1987 "$[0]": {
1988 "vnfd-connection-point-ref": "pdu-mgmt",
1989 "vnfd-id-ref": "vdu-as-pdu",
1990 },
1991 "$[1]": None
1992 },
1993 "type": "ELAN"
1994 },
1995 "$[1]": None,
1996 }
1997 }
1998 }
1999
2000
2001 class TestDeployHnfd(TestDeployHackfest3Charmed):
2002 description = "Generate a HNFD base on editing Hackfest3Charmed descriptors and deploy"
2003
2004 def __init__(self):
2005 super().__init__()
2006 self.test_name = "HNFD"
2007 self.pduDeploy = TestDeploySingleVdu()
2008 self.pdu_interface_0 = {}
2009 self.pdu_interface_1 = {}
2010
2011 self.pdu_id = None
2012 # self.vnf_to_pdu = """
2013 # vdu:
2014 # "$[0]":
2015 # pdu-type: PDU-TYPE-1
2016 # interface:
2017 # "$[0]":
2018 # name: mgmt-iface
2019 # "$[1]":
2020 # name: pdu-iface-internal
2021 # id: hfn1
2022 # description: HFND, one PDU + One VDU
2023 # name: hfn1
2024 # short-name: hfn1
2025 #
2026 # """
2027
2028 self.pdu_descriptor = {
2029 "name": "my-PDU",
2030 "type": "PDU-TYPE-1",
2031 "vim_accounts": "to-override",
2032 "interfaces": [
2033 {
2034 "name": "mgmt-iface",
2035 "mgmt": True,
2036 "type": "overlay",
2037 "ip-address": "to override",
2038 "mac-address": "mac_address",
2039 "vim-network-name": "mgmt",
2040 },
2041 {
2042 "name": "pdu-iface-internal",
2043 "mgmt": False,
2044 "type": "overlay",
2045 "ip-address": "to override",
2046 "mac-address": "mac_address",
2047 "vim-network-name": "pdu_internal", # OSMNBITEST-PDU-pdu_internal
2048 },
2049 ]
2050 }
2051 self.vnfd_filenames = ("hackfest_3charmed_vnfd.tar.gz", "hackfest_3charmed_vnfd.tar.gz")
2052
2053 self.descriptor_edit = {
2054 "vnfd0": {
2055 "id": "hfnd1",
2056 "name": "hfn1",
2057 "short-name": "hfn1",
2058 "vdu": {
2059 "$[0]": {
2060 "pdu-type": "PDU-TYPE-1",
2061 "interface": {
2062 "$[0]": {"name": "mgmt-iface"},
2063 "$[1]": {"name": "pdu-iface-internal"},
2064 }
2065 }
2066 }
2067 },
2068 "nsd": {
2069 "constituent-vnfd": {
2070 "$[1]": {"vnfd-id-ref": "hfnd1"}
2071 },
2072 "vld": {
2073 "$[0]": {"vnfd-connection-point-ref": {"$[1]": {"vnfd-id-ref": "hfnd1"}}},
2074 "$[1]": {"vnfd-connection-point-ref": {"$[1]": {"vnfd-id-ref": "hfnd1"}}}
2075 }
2076 }
2077 }
2078
2079 def create_descriptors(self, engine):
2080 super().create_descriptors(engine)
2081
2082 # Create PDU
2083 self.pdu_descriptor["interfaces"][0].update(self.pdu_interface_0)
2084 self.pdu_descriptor["interfaces"][1].update(self.pdu_interface_1)
2085 self.pdu_descriptor["vim_accounts"] = [self.vim_id]
2086 # TODO get vim-network-name from vnfr.vld.name
2087 self.pdu_descriptor["interfaces"][1]["vim-network-name"] = "{}-{}-{}".format(
2088 os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST"),
2089 "PDU", self.pdu_descriptor["interfaces"][1]["vim-network-name"])
2090 engine.test("Onboard PDU descriptor", "POST", "/pdu/v1/pdu_descriptors",
2091 {"Location": "/pdu/v1/pdu_descriptors/", "Content-Type": "application/yaml"}, self.pdu_descriptor,
2092 201, r_header_yaml, "yaml")
2093 self.pdu_id = engine.last_id
2094
2095 def run(self, engine, test_osm, manual_check, test_params=None):
2096 engine.get_autorization()
2097 engine.set_test_name(self.test_name)
2098 nsname = os.environ.get("OSMNBITEST_NS_NAME", "OSMNBITEST")
2099
2100 # create real VIM if not exist
2101 self.vim_id = engine.get_create_vim(test_osm)
2102 # instantiate PDU
2103 self.pduDeploy.create_descriptors(engine)
2104 self.pduDeploy.instantiate(engine, {"nsDescription": "to be used as PDU", "nsName": nsname + "-PDU",
2105 "nsdId": self.pduDeploy.nsd_id, "vimAccountId": self.vim_id})
2106 if manual_check:
2107 input('VNF to be used as PDU has been deployed. Perform manual check and press enter to resume')
2108 if test_osm:
2109 self.pduDeploy.test_ns(engine, test_osm)
2110
2111 if test_osm:
2112 r = engine.test("Get VNFR to obtain IP_ADDRESS", "GET",
2113 "/nslcm/v1/vnfrs?nsr-id-ref={}".format(self.pduDeploy.ns_id), headers_json, None,
2114 200, r_header_json, "json")
2115 if not r:
2116 return
2117 vnfr_data = r.json()
2118 # print(vnfr_data)
2119
2120 self.pdu_interface_0["ip-address"] = vnfr_data[0]["vdur"][0]["interfaces"][0].get("ip-address")
2121 self.pdu_interface_1["ip-address"] = vnfr_data[0]["vdur"][0]["interfaces"][1].get("ip-address")
2122 self.pdu_interface_0["mac-address"] = vnfr_data[0]["vdur"][0]["interfaces"][0].get("mac-address")
2123 self.pdu_interface_1["mac-address"] = vnfr_data[0]["vdur"][0]["interfaces"][1].get("mac-address")
2124 if not self.pdu_interface_0["ip-address"]:
2125 raise TestException("Vnfr has not managment ip address")
2126 else:
2127 self.pdu_interface_0["ip-address"] = "192.168.10.10"
2128 self.pdu_interface_1["ip-address"] = "192.168.11.10"
2129 self.pdu_interface_0["mac-address"] = "52:33:44:55:66:13"
2130 self.pdu_interface_1["mac-address"] = "52:33:44:55:66:14"
2131
2132 self.create_descriptors(engine)
2133
2134 ns_data = {"nsDescription": "default description", "nsName": nsname, "nsdId": self.nsd_id,
2135 "vimAccountId": self.vim_id}
2136 if test_params and test_params.get("ns-config"):
2137 if isinstance(test_params["ns-config"], str):
2138 ns_data.update(yaml.load(test_params["ns-config"]))
2139 else:
2140 ns_data.update(test_params["ns-config"])
2141
2142 self.instantiate(engine, ns_data)
2143 if manual_check:
2144 input('NS has been deployed. Perform manual check and press enter to resume')
2145 if test_osm:
2146 self.test_ns(engine, test_osm)
2147 self.additional_operations(engine, test_osm, manual_check)
2148 self.terminate(engine)
2149 self.pduDeploy.terminate(engine)
2150 self.delete_descriptors(engine)
2151 self.pduDeploy.delete_descriptors(engine)
2152
2153 def delete_descriptors(self, engine):
2154 super().delete_descriptors(engine)
2155 # delete pdu
2156 engine.test("Delete PDU SOL005", "DELETE",
2157 "/pdu/v1/pdu_descriptors/{}".format(self.pdu_id),
2158 headers_yaml, None, 204, None, 0)
2159
2160
2161 class TestDescriptors:
2162 description = "Test VNFD, NSD, PDU descriptors CRUD and dependencies"
2163 vnfd_empty = """vnfd:vnfd-catalog:
2164 vnfd:
2165 - name: prova
2166 short-name: prova
2167 id: prova
2168 """
2169 vnfd_prova = """vnfd:vnfd-catalog:
2170 vnfd:
2171 - connection-point:
2172 - name: cp_0h8m
2173 type: VPORT
2174 id: prova
2175 name: prova
2176 short-name: prova
2177 vdu:
2178 - id: vdu_z4bm
2179 image: ubuntu
2180 interface:
2181 - external-connection-point-ref: cp_0h8m
2182 name: eth0
2183 virtual-interface:
2184 type: VIRTIO
2185 name: vdu_z4bm
2186 version: '1.0'
2187 """
2188
2189 def __init__(self):
2190 self.vnfd_filename = "hackfest_3charmed_vnfd.tar.gz"
2191 self.nsd_filename = "hackfest_3charmed_nsd.tar.gz"
2192 self.descriptor_url = "https://osm-download.etsi.org/ftp/osm-3.0-three/2nd-hackfest/packages/"
2193 self.vnfd_id = None
2194 self.nsd_id = None
2195
2196 def run(self, engine, test_osm, manual_check, test_params=None):
2197 engine.set_test_name("Descriptors")
2198 engine.get_autorization()
2199 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
2200 if not os.path.exists(temp_dir):
2201 os.makedirs(temp_dir)
2202
2203 # download files
2204 for filename in (self.vnfd_filename, self.nsd_filename):
2205 filename_path = temp_dir + filename
2206 if not os.path.exists(filename_path):
2207 with open(filename_path, "wb") as file:
2208 response = requests.get(self.descriptor_url + filename)
2209 if response.status_code >= 300:
2210 raise TestException("Error downloading descriptor from '{}': {}".format(
2211 self.descriptor_url + filename, response.status_code))
2212 file.write(response.content)
2213
2214 vnfd_filename_path = temp_dir + self.vnfd_filename
2215 nsd_filename_path = temp_dir + self.nsd_filename
2216
2217 engine.test("Onboard empty VNFD in one step", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml,
2218 self.vnfd_empty, 201, r_headers_yaml_location_vnfd, "yaml")
2219 self.vnfd_id = engine.last_id
2220
2221 # test bug 605
2222 engine.test("Upload invalid VNFD ", "PUT", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id),
2223 headers_yaml, self.vnfd_prova, 422, r_header_yaml, "yaml")
2224
2225 engine.test("Upload VNFD {}".format(self.vnfd_filename), "PUT",
2226 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id), headers_zip_yaml,
2227 "@b" + vnfd_filename_path, 204, None, 0)
2228
2229 queries = ["mgmt-interface.cp=mgmt", "vdu.0.interface.0.external-connection-point-ref=mgmt",
2230 "vdu.0.interface.1.internal-connection-point-ref=internal",
2231 "internal-vld.0.internal-connection-point.0.id-ref=internal",
2232 # Detection of duplicated VLD names in VNF Descriptors
2233 # URL: internal-vld=[
2234 # {id: internal1, name: internal, type:ELAN,
2235 # internal-connection-point: [{id-ref: mgmtVM-internal}, {id-ref: dataVM-internal}]},
2236 # {id: internal2, name: internal, type:ELAN,
2237 # internal-connection-point: [{id-ref: mgmtVM-internal}, {id-ref: dataVM-internal}]}
2238 # ]
2239 "internal-vld=%5B%7Bid%3A%20internal1%2C%20name%3A%20internal%2C%20type%3A%20ELAN%2C%20"
2240 "internal-connection-point%3A%20%5B%7Bid-ref%3A%20mgmtVM-internal%7D%2C%20%7Bid-ref%3A%20"
2241 "dataVM-internal%7D%5D%7D%2C%20%7Bid%3A%20internal2%2C%20name%3A%20internal%2C%20type%3A%20"
2242 "ELAN%2C%20internal-connection-point%3A%20%5B%7Bid-ref%3A%20mgmtVM-internal%7D%2C%20%7B"
2243 "id-ref%3A%20dataVM-internal%7D%5D%7D%5D"
2244 ]
2245 for query in queries:
2246 engine.test("Upload invalid VNFD ", "PUT",
2247 "/vnfpkgm/v1/vnf_packages/{}/package_content?{}".format(self.vnfd_id, query),
2248 headers_zip_yaml, "@b" + vnfd_filename_path, 422, r_header_yaml, "yaml")
2249
2250 # test bug 605
2251 engine.test("Upload invalid VNFD ", "PUT", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id),
2252 headers_yaml, self.vnfd_prova, 422, r_header_yaml, "yaml")
2253
2254 # get vnfd descriptor
2255 engine.test("Get VNFD descriptor", "GET", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_id),
2256 headers_yaml, None, 200, r_header_yaml, "yaml")
2257
2258 # get vnfd file descriptor
2259 engine.test("Get VNFD file descriptor", "GET", "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(self.vnfd_id),
2260 headers_text, None, 200, r_header_text, "text", temp_dir+"vnfd-yaml")
2261 # TODO compare files: diff vnfd-yaml hackfest_3charmed_vnfd/hackfest_3charmed_vnfd.yaml
2262
2263 # get vnfd zip file package
2264 engine.test("Get VNFD zip package", "GET",
2265 "/vnfpkgm/v1/vnf_packages/{}/package_content".format(self.vnfd_id), headers_zip, None, 200,
2266 r_header_zip, "zip", temp_dir+"vnfd-zip")
2267 # TODO compare files: diff vnfd-zip hackfest_3charmed_vnfd.tar.gz
2268
2269 # get vnfd artifact
2270 engine.test("Get VNFD artifact package", "GET",
2271 "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/osm.png".format(self.vnfd_id), headers_zip, None, 200,
2272 r_header_octect, "octet-string", temp_dir+"vnfd-icon")
2273 # TODO compare files: diff vnfd-icon hackfest_3charmed_vnfd/icons/osm.png
2274
2275 # nsd CREATE AND UPLOAD in one step:
2276 engine.test("Onboard NSD in one step", "POST", "/nsd/v1/ns_descriptors_content", headers_zip_yaml,
2277 "@b" + nsd_filename_path, 201, r_headers_yaml_location_nsd, "yaml")
2278 self.nsd_id = engine.last_id
2279
2280 queries = ["vld.0.vnfd-connection-point-ref.0.vnfd-id-ref=hf"]
2281 for query in queries:
2282 engine.test("Upload invalid NSD ", "PUT",
2283 "/nsd/v1/ns_descriptors/{}/nsd_content?{}".format(self.nsd_id, query),
2284 headers_zip_yaml, "@b" + nsd_filename_path, 422, r_header_yaml, "yaml")
2285
2286 # get nsd descriptor
2287 engine.test("Get NSD descriptor", "GET", "/nsd/v1/ns_descriptors/{}".format(self.nsd_id), headers_yaml,
2288 None, 200, r_header_yaml, "yaml")
2289
2290 # get nsd file descriptor
2291 engine.test("Get NSD file descriptor", "GET", "/nsd/v1/ns_descriptors/{}/nsd".format(self.nsd_id), headers_text,
2292 None, 200, r_header_text, "text", temp_dir+"nsd-yaml")
2293 # TODO compare files: diff nsd-yaml hackfest_3charmed_nsd/hackfest_3charmed_nsd.yaml
2294
2295 # get nsd zip file package
2296 engine.test("Get NSD zip package", "GET", "/nsd/v1/ns_descriptors/{}/nsd_content".format(self.nsd_id),
2297 headers_zip, None, 200, r_header_zip, "zip", temp_dir+"nsd-zip")
2298 # TODO compare files: diff nsd-zip hackfest_3charmed_nsd.tar.gz
2299
2300 # get nsd artifact
2301 engine.test("Get NSD artifact package", "GET",
2302 "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm.png".format(self.nsd_id), headers_zip, None, 200,
2303 r_header_octect, "octet-string", temp_dir+"nsd-icon")
2304 # TODO compare files: diff nsd-icon hackfest_3charmed_nsd/icons/osm.png
2305
2306 # vnfd DELETE
2307 test_rest.test("Delete VNFD conflict", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_id),
2308 headers_yaml, None, 409, None, None)
2309
2310 test_rest.test("Delete VNFD force", "DELETE", "/vnfpkgm/v1/vnf_packages/{}?FORCE=TRUE".format(self.vnfd_id),
2311 headers_yaml, None, 204, None, 0)
2312
2313 # nsd DELETE
2314 test_rest.test("Delete NSD", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self.nsd_id), headers_yaml, None, 204,
2315 None, 0)
2316
2317
2318 class TestNetSliceTemplates:
2319 description = "Upload a NST to OSM"
2320
2321 def __init__(self):
2322 self.vnfd_filename = ("@./slice_shared/vnfd/slice_shared_vnfd.yaml")
2323 self.vnfd_filename_middle = ("@./slice_shared/vnfd/slice_shared_middle_vnfd.yaml")
2324 self.nsd_filename = ("@./slice_shared/nsd/slice_shared_nsd.yaml")
2325 self.nsd_filename_middle = ("@./slice_shared/nsd/slice_shared_middle_nsd.yaml")
2326 self.nst_filenames = ("@./slice_shared/slice_shared_nstd.yaml")
2327
2328 def run(self, engine, test_osm, manual_check, test_params=None):
2329 # nst CREATE
2330 engine.set_test_name("NST step ")
2331 engine.get_autorization()
2332 temp_dir = os.path.dirname(os.path.abspath(__file__)) + "/temp/"
2333 if not os.path.exists(temp_dir):
2334 os.makedirs(temp_dir)
2335
2336 # Onboard VNFDs
2337 engine.test("Onboard edge VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml,
2338 self.vnfd_filename, 201, r_headers_yaml_location_vnfd, "yaml")
2339 self.vnfd_edge_id = engine.last_id
2340
2341 engine.test("Onboard middle VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml,
2342 self.vnfd_filename_middle, 201, r_headers_yaml_location_vnfd, "yaml")
2343 self.vnfd_middle_id = engine.last_id
2344
2345 # Onboard NSDs
2346 engine.test("Onboard NSD edge", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml,
2347 self.nsd_filename, 201, r_headers_yaml_location_nsd, "yaml")
2348 self.nsd_edge_id = engine.last_id
2349
2350 engine.test("Onboard NSD middle", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml,
2351 self.nsd_filename_middle, 201, r_headers_yaml_location_nsd, "yaml")
2352 self.nsd_middle_id = engine.last_id
2353
2354 # Onboard NST
2355 engine.test("Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml, self.nst_filenames,
2356 201, r_headers_yaml_location_nst, "yaml")
2357 nst_id = engine.last_id
2358
2359 # nstd SHOW OSM format
2360 engine.test("Show NSTD OSM format", "GET", "/nst/v1/netslice_templates/{}".format(nst_id), headers_json, None,
2361 200, r_header_json, "json")
2362
2363 # nstd DELETE
2364 engine.test("Delete NSTD", "DELETE", "/nst/v1/netslice_templates/{}".format(nst_id), headers_json, None,
2365 204, None, 0)
2366
2367 # NSDs DELETE
2368 test_rest.test("Delete NSD middle", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self.nsd_middle_id),
2369 headers_json, None, 204, None, 0)
2370
2371 test_rest.test("Delete NSD edge", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self.nsd_edge_id), headers_json,
2372 None, 204, None, 0)
2373
2374 # VNFDs DELETE
2375 test_rest.test("Delete VNFD edge", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_edge_id),
2376 headers_yaml, None, 204, None, 0)
2377
2378 test_rest.test("Delete VNFD middle", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_middle_id),
2379 headers_yaml, None, 204, None, 0)
2380
2381
2382 class TestNetSliceInstances:
2383 '''
2384 Test procedure:
2385 1. Populate databases with VNFD, NSD, NST with the following scenario
2386 +-----------------management-----------------+
2387 | | |
2388 +--+---+ +----+----+ +---+--+
2389 | | | | | |
2390 | edge +---data1----+ middle +---data2-----+ edge |
2391 | | | | | |
2392 +------+ +---------+ +------+
2393 shared-nss
2394 2. Create NSI-1
2395 3. Instantiate NSI-1
2396 4. Create NSI-2
2397 5. Instantiate NSI-2
2398 Manual check - Are 2 slices instantiated correctly?
2399 NSI-1 3 nss (2 nss-edges + 1 nss-middle)
2400 NSI-2 2 nss (2 nss-edge sharing nss-middle)
2401 6. Terminate NSI-1
2402 7. Delete NSI-1
2403 Manual check - Is slice NSI-1 deleted correctly?
2404 NSI-2 with 2 nss-edge + 1 nss-middle (The one from NSI-1)
2405 8. Create NSI-3
2406 9. Instantiate NSI-3
2407 Manual check - Is slice NSI-3 instantiated correctly?
2408 NSI-3 reuse nss-middle. NSI-3 only create 2 nss-edge
2409 10. Delete NSI-2
2410 11. Terminate NSI-2
2411 12. Delete NSI-3
2412 13. Terminate NSI-3
2413 Manual check - All cleaned correctly?
2414 NSI-2 and NSI-3 were terminated and deleted
2415 14. Cleanup database
2416 '''
2417
2418 description = "Upload a NST to OSM"
2419
2420 def __init__(self):
2421 self.vim_id = None
2422 self.vnfd_filename = ("@./slice_shared/vnfd/slice_shared_vnfd.yaml")
2423 self.vnfd_filename_middle = ("@./slice_shared/vnfd/slice_shared_middle_vnfd.yaml")
2424 self.nsd_filename = ("@./slice_shared/nsd/slice_shared_nsd.yaml")
2425 self.nsd_filename_middle = ("@./slice_shared/nsd/slice_shared_middle_nsd.yaml")
2426 self.nst_filenames = ("@./slice_shared/slice_shared_nstd.yaml")
2427
2428 def create_slice(self, engine, nsi_data, name):
2429 ns_data_text = yaml.safe_dump(nsi_data, default_flow_style=True, width=256)
2430 r = engine.test(name, "POST", "/nsilcm/v1/netslice_instances",
2431 headers_yaml, ns_data_text, (201, 202),
2432 {"Location": "nsilcm/v1/netslice_instances/", "Content-Type": "application/yaml"}, "yaml")
2433 return r
2434
2435 def instantiate_slice(self, engine, nsi_data, nsi_id, name):
2436 ns_data_text = yaml.safe_dump(nsi_data, default_flow_style=True, width=256)
2437 engine.test(name, "POST",
2438 "/nsilcm/v1/netslice_instances/{}/instantiate".format(nsi_id), headers_yaml, ns_data_text,
2439 (201, 202), r_headers_yaml_location_nsilcmop, "yaml")
2440
2441 def terminate_slice(self, engine, nsi_id, name):
2442 engine.test(name, "POST", "/nsilcm/v1/netslice_instances/{}/terminate".format(nsi_id),
2443 headers_yaml, None, (201, 202), r_headers_yaml_location_nsilcmop, "yaml")
2444
2445 def delete_slice(self, engine, nsi_id, name):
2446 engine.test(name, "DELETE", "/nsilcm/v1/netslice_instances/{}".format(nsi_id), headers_yaml, None,
2447 204, None, 0)
2448
2449 def run(self, engine, test_osm, manual_check, test_params=None):
2450 # nst CREATE
2451 engine.set_test_name("NSI")
2452 engine.get_autorization()
2453
2454 # Onboard VNFDs
2455 engine.test("Onboard edge VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml,
2456 self.vnfd_filename, 201, r_headers_yaml_location_vnfd, "yaml")
2457 self.vnfd_edge_id = engine.last_id
2458
2459 engine.test("Onboard middle VNFD", "POST", "/vnfpkgm/v1/vnf_packages_content", headers_yaml,
2460 self.vnfd_filename_middle, 201, r_headers_yaml_location_vnfd, "yaml")
2461 self.vnfd_middle_id = engine.last_id
2462
2463 # Onboard NSDs
2464 engine.test("Onboard NSD edge", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml,
2465 self.nsd_filename, 201, r_headers_yaml_location_nsd, "yaml")
2466 self.nsd_edge_id = engine.last_id
2467
2468 engine.test("Onboard NSD middle", "POST", "/nsd/v1/ns_descriptors_content", headers_yaml,
2469 self.nsd_filename_middle, 201, r_headers_yaml_location_nsd, "yaml")
2470 self.nsd_middle_id = engine.last_id
2471
2472 # Onboard NST
2473 engine.test("Onboard NST", "POST", "/nst/v1/netslice_templates_content", headers_yaml, self.nst_filenames,
2474 201, r_headers_yaml_location_nst, "yaml")
2475 nst_id = engine.last_id
2476
2477 self.vim_id = engine.get_create_vim(test_osm)
2478
2479 # CREATE NSI-1
2480 ns_data = {'nsiName': 'Deploy-NSI-1', 'vimAccountId': self.vim_id, 'nstId': nst_id, 'nsiDescription': 'default'}
2481 r = self.create_slice(engine, ns_data, "Create NSI-1 step 1")
2482 if not r:
2483 return
2484 self.nsi_id1 = engine.last_id
2485
2486 # INSTANTIATE NSI-1
2487 self.instantiate_slice(engine, ns_data, self.nsi_id1, "Instantiate NSI-1 step 2")
2488 nsilcmop_id1 = engine.last_id
2489
2490 # Waiting for NSI-1
2491 if test_osm:
2492 engine.wait_operation_ready("nsi", nsilcmop_id1, timeout_deploy)
2493
2494 # CREATE NSI-2
2495 ns_data = {'nsiName': 'Deploy-NSI-2', 'vimAccountId': self.vim_id, 'nstId': nst_id, 'nsiDescription': 'default'}
2496 r = self.create_slice(engine, ns_data, "Create NSI-2 step 1")
2497 if not r:
2498 return
2499 self.nsi_id2 = engine.last_id
2500
2501 # INSTANTIATE NSI-2
2502 self.instantiate_slice(engine, ns_data, self.nsi_id2, "Instantiate NSI-2 step 2")
2503 nsilcmop_id2 = engine.last_id
2504
2505 # Waiting for NSI-2
2506 if test_osm:
2507 engine.wait_operation_ready("nsi", nsilcmop_id2, timeout_deploy)
2508
2509 if manual_check:
2510 input('NSI-1 AND NSI-2 has been deployed. Perform manual check and press enter to resume')
2511
2512 # TERMINATE NSI-1
2513 if test_osm:
2514 self.terminate_slice(engine, self.nsi_id1, "Terminate NSI-1")
2515 nsilcmop1_id = engine.last_id
2516
2517 # Wait terminate NSI-1
2518 engine.wait_operation_ready("nsi", nsilcmop1_id, timeout_deploy)
2519
2520 # DELETE NSI-1
2521 self.delete_slice(engine, self.nsi_id1, "Delete NS")
2522
2523 if manual_check:
2524 input('NSI-1 has been deleted. Perform manual check and press enter to resume')
2525
2526 # CREATE NSI-3
2527 ns_data = {'nsiName': 'Deploy-NSI-3', 'vimAccountId': self.vim_id, 'nstId': nst_id, 'nsiDescription': 'default'}
2528 r = self.create_slice(engine, ns_data, "Create NSI-3 step 1")
2529
2530 if not r:
2531 return
2532 self.nsi_id3 = engine.last_id
2533
2534 # INSTANTIATE NSI-3
2535 self.instantiate_slice(engine, ns_data, self.nsi_id3, "Instantiate NSI-3 step 2")
2536 nsilcmop_id3 = engine.last_id
2537
2538 # Wait Instantiate NSI-3
2539 if test_osm:
2540 engine.wait_operation_ready("nsi", nsilcmop_id3, timeout_deploy)
2541
2542 if manual_check:
2543 input('NSI-3 has been deployed. Perform manual check and press enter to resume')
2544
2545 # TERMINATE NSI-2
2546 if test_osm:
2547 self.terminate_slice(engine, self.nsi_id2, "Terminate NSI-2")
2548 nsilcmop2_id = engine.last_id
2549
2550 # Wait terminate NSI-2
2551 engine.wait_operation_ready("nsi", nsilcmop2_id, timeout_deploy)
2552
2553 # DELETE NSI-2
2554 self.delete_slice(engine, self.nsi_id2, "DELETE NSI-2")
2555
2556 # TERMINATE NSI-3
2557 if test_osm:
2558 self. terminate_slice(engine, self.nsi_id3, "Terminate NSI-3")
2559 nsilcmop3_id = engine.last_id
2560
2561 # Wait terminate NSI-3
2562 engine.wait_operation_ready("nsi", nsilcmop3_id, timeout_deploy)
2563
2564 # DELETE NSI-3
2565 self.delete_slice(engine, self.nsi_id3, "DELETE NSI-3")
2566
2567 if manual_check:
2568 input('NSI-2 and NSI-3 has been deleted. Perform manual check and press enter to resume')
2569
2570 # nstd DELETE
2571 engine.test("Delete NSTD", "DELETE", "/nst/v1/netslice_templates/{}".format(nst_id), headers_json, None,
2572 204, None, 0)
2573
2574 # NSDs DELETE
2575 test_rest.test("Delete NSD middle", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self.nsd_middle_id),
2576 headers_json, None, 204, None, 0)
2577
2578 test_rest.test("Delete NSD edge", "DELETE", "/nsd/v1/ns_descriptors/{}".format(self.nsd_edge_id), headers_json,
2579 None, 204, None, 0)
2580
2581 # VNFDs DELETE
2582 test_rest.test("Delete VNFD edge", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_edge_id),
2583 headers_yaml, None, 204, None, 0)
2584
2585 test_rest.test("Delete VNFD middle", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(self.vnfd_middle_id),
2586 headers_yaml, None, 204, None, 0)
2587
2588
2589 class TestAuthentication:
2590 description = "Test Authentication"
2591
2592 @staticmethod
2593 def run(engine, test_osm, manual_check, test_params=None):
2594 engine.set_test_name("Authentication")
2595 # backend = test_params.get("backend") if test_params else None # UNUSED
2596
2597 admin_project_id = test_project_id = None
2598 project_admin_role_id = project_user_role_id = None
2599 test_user_id = empty_user_id = None
2600 default_role_id = empty_role_id = token_role_id = None
2601
2602 engine.get_autorization()
2603
2604 # GET
2605 engine.test("Get tokens", "GET", "/admin/v1/tokens", headers_json, {},
2606 (200), {"Content-Type": "application/json"}, "json")
2607 engine.test("Get projects", "GET", "/admin/v1/projects", headers_json, {},
2608 (200), {"Content-Type": "application/json"}, "json")
2609 engine.test("Get users", "GET", "/admin/v1/users", headers_json, {},
2610 (200), {"Content-Type": "application/json"}, "json")
2611 engine.test("Get roles", "GET", "/admin/v1/roles", headers_json, {},
2612 (200), {"Content-Type": "application/json"}, "json")
2613 res = engine.test("Get admin project", "GET", "/admin/v1/projects?name=admin", headers_json, {},
2614 (200), {"Content-Type": "application/json"}, "json")
2615 admin_project_id = res.json()[0]["_id"] if res else None
2616 res = engine.test("Get project admin role", "GET", "/admin/v1/roles?name=project_admin", headers_json, {},
2617 (200), {"Content-Type": "application/json"}, "json")
2618 project_admin_role_id = res.json()[0]["_id"] if res else None
2619 res = engine.test("Get project user role", "GET", "/admin/v1/roles?name=project_user", headers_json, {},
2620 (200), {"Content-Type": "application/json"}, "json")
2621 project_user_role_id = res.json()[0]["_id"] if res else None
2622
2623 # POST
2624 res = engine.test("Create test project", "POST", "/admin/v1/projects", headers_json, {"name": "test"},
2625 (201), {"Location": "/admin/v1/projects/", "Content-Type": "application/json"}, "json")
2626 test_project_id = engine.last_id if res else None
2627 res = engine.test("Create role without permissions", "POST", "/admin/v1/roles", headers_json, {"name": "empty"},
2628 (201), {"Content-Type": "application/json"}, "json")
2629 empty_role_id = engine.last_id if res else None
2630 res = engine.test("Create role with default permissions", "POST", "/admin/v1/roles", headers_json,
2631 {"name": "default", "permissions": {"default": True}},
2632 (201), {"Location": "/admin/v1/roles/", "Content-Type": "application/json"}, "json")
2633 default_role_id = engine.last_id if res else None
2634 res = engine.test("Create role with token permissions", "POST", "/admin/v1/roles", headers_json,
2635 {"name": "tokens", "permissions": {"tokens": True}}, # is default required ?
2636 (201), {"Location": "/admin/v1/roles/", "Content-Type": "application/json"}, "json")
2637 token_role_id = engine.last_id if res else None
2638 pr = "project-role mappings"
2639 res = engine.test("Create user without "+pr, "POST", "/admin/v1/users", headers_json,
2640 {"username": "empty", "password": "empty"},
2641 201, {"Content-Type": "application/json"}, "json")
2642 empty_user_id = engine.last_id if res else None
2643 if admin_project_id and test_project_id and project_admin_role_id and project_user_role_id:
2644 data = {"username": "test", "password": "test"}
2645 data["project_role_mappings"] = [
2646 {"project": test_project_id, "role": project_admin_role_id},
2647 {"project": admin_project_id, "role": project_user_role_id}
2648 ]
2649 res = engine.test("Create user with "+pr, "POST", "/admin/v1/users", headers_json, data,
2650 (201), {"Content-Type": "application/json"}, "json")
2651 test_user_id = engine.last_id if res else None
2652
2653 # PUT
2654 if test_user_id:
2655 engine.test("Modify test user's password", "PUT", "/admin/v1/users/"+test_user_id, headers_json,
2656 {"password": "password"},
2657 (204), {}, 0)
2658 if empty_user_id and admin_project_id and test_project_id and project_admin_role_id and project_user_role_id:
2659 data = {"project_role_mappings": [
2660 {"project": test_project_id, "role": project_admin_role_id},
2661 {"project": admin_project_id, "role": project_user_role_id}
2662 ]}
2663 engine.test("Modify empty user's "+pr, "PUT", "/admin/v1/users/"+empty_user_id,
2664 headers_json,
2665 data,
2666 (204), {}, 0)
2667
2668 # DELETE
2669 if empty_user_id:
2670 engine.test("Delete empty user", "DELETE", "/admin/v1/users/"+empty_user_id, headers_json, {},
2671 (204), {}, 0)
2672 if test_user_id:
2673 engine.test("Delete test user", "DELETE", "/admin/v1/users/"+test_user_id, headers_json, {},
2674 (204), {}, 0)
2675 if empty_role_id:
2676 engine.test("Delete empty role", "DELETE", "/admin/v1/roles/"+empty_role_id, headers_json, {},
2677 (204), {}, 0)
2678 if default_role_id:
2679 engine.test("Delete default role", "DELETE", "/admin/v1/roles/"+default_role_id, headers_json, {},
2680 (204), {}, 0)
2681 if token_role_id:
2682 engine.test("Delete token role", "DELETE", "/admin/v1/roles/"+token_role_id, headers_json, {},
2683 (204), {}, 0)
2684 if test_project_id:
2685 engine.test("Delete test project", "DELETE", "/admin/v1/projects/"+test_project_id, headers_json, {},
2686 (204), {}, 0)
2687
2688 # END Tests
2689
2690 engine.remove_authorization() # To finish
2691
2692
2693 class TestNbiQuotas():
2694 description = "Test NBI Quotas"
2695
2696 @staticmethod
2697 def run(engine, test_osm, manual_check, test_params=None):
2698 engine.set_test_name("NBI-Quotas_")
2699 # backend = test_params.get("backend") if test_params else None # UNUSED
2700
2701 test_username = "test-nbi-quotas"
2702 test_password = "test-nbi-quotas"
2703 test_project = "test-nbi-quotas"
2704
2705 test_vim = "test-nbi-quotas"
2706 test_wim = "test-nbi-quotas"
2707 test_sdn = "test-nbi-quotas"
2708
2709 test_user_id = None
2710 test_project_id = None
2711
2712 test_vim_ids = []
2713 test_wim_ids = []
2714 test_sdn_ids = []
2715 test_vnfd_ids = []
2716 test_nsd_ids = []
2717 test_nst_ids = []
2718 test_pdu_ids = []
2719 test_nsr_ids = []
2720 test_nsi_ids = []
2721
2722 # Save admin access data
2723 admin_username = engine.user
2724 admin_password = engine.password
2725 admin_project = engine.project
2726
2727 # Get admin access
2728 engine.get_autorization()
2729 admin_token = engine.last_id
2730
2731 # Check that test project,user do not exist
2732 res1 = engine.test("Check that test project doesn't exist", "GET", "/admin/v1/projects/"+test_project,
2733 headers_json, {}, (404), {}, True)
2734 res2 = engine.test("Check that test user doesn't exist", "GET", "/admin/v1/users/"+test_username,
2735 headers_json, {}, (404), {}, True)
2736 if None in [res1, res2]:
2737 engine.remove_authorization()
2738 logger.error("Test project and/or user already exist")
2739 return
2740
2741 # Create test project&user
2742 res = engine.test("Create test project", "POST", "/admin/v1/projects", headers_json,
2743 {"name": test_username,
2744 "quotas": {
2745 "vnfds": 2,
2746 "nsds": 2,
2747 "nsts": 1,
2748 "pdus": 1,
2749 "nsrs": 2,
2750 "nsis": 1,
2751 "vim_accounts": 1,
2752 "wim_accounts": 1,
2753 "sdns": 1,
2754 }
2755 },
2756 (201), r_header_json, "json")
2757 test_project_id = engine.last_id if res else None
2758 res = engine.test("Create test user", "POST", "/admin/v1/users", headers_json,
2759 {"username": test_username, "password": test_password,
2760 "project_role_mappings": [{"project": test_project, "role": "project_admin"}]},
2761 (201), r_header_json, "json")
2762 test_user_id = engine.last_id if res else None
2763
2764 if test_project_id and test_user_id:
2765
2766 # Get user access
2767 engine.token = None
2768 engine.user = test_username
2769 engine.password = test_password
2770 engine.project = test_project
2771 engine.get_autorization()
2772 user_token = engine.last_id
2773
2774 # Create test VIM
2775 res = engine.test("Create test VIM", "POST", "/admin/v1/vim_accounts", headers_json,
2776 {"name": test_vim,
2777 "vim_type": "openvim",
2778 "vim_user": test_username,
2779 "vim_password": test_password,
2780 "vim_tenant_name": test_project,
2781 "vim_url": "https://0.0.0.0:0/v0.0",
2782 },
2783 (202), r_header_json, "json")
2784 test_vim_ids += [engine.last_id if res else None]
2785
2786 res = engine.test("Try to create second test VIM", "POST", "/admin/v1/vim_accounts", headers_json,
2787 {"name": test_vim + "_2",
2788 "vim_type": "openvim",
2789 "vim_user": test_username,
2790 "vim_password": test_password,
2791 "vim_tenant_name": test_project,
2792 "vim_url": "https://0.0.0.0:0/v0.0",
2793 },
2794 (422), r_header_json, "json")
2795 test_vim_ids += [engine.last_id if res is None else None]
2796
2797 res = engine.test("Try to create second test VIM with FORCE",
2798 "POST", "/admin/v1/vim_accounts?FORCE", headers_json,
2799 {"name": test_vim + "_3",
2800 "vim_type": "openvim",
2801 "vim_user": test_username,
2802 "vim_password": test_password,
2803 "vim_tenant_name": test_project,
2804 "vim_url": "https://0.0.0.0:0/v0.0",
2805 },
2806 (202), r_header_json, "json")
2807 test_vim_ids += [engine.last_id if res else None]
2808
2809 if test_vim_ids[0]:
2810
2811 # Download descriptor files (if required)
2812 test_dir = "/tmp/"+test_username+"/"
2813 test_url = "https://osm-download.etsi.org/ftp/osm-6.0-six/7th-hackfest/packages/"
2814 vnfd_filenames = ["slice_hackfest_vnfd.tar.gz", "slice_hackfest_middle_vnfd.tar.gz"]
2815 nsd_filenames = ["slice_hackfest_nsd.tar.gz", "slice_hackfest_middle_nsd.tar.gz"]
2816 nst_filenames = ["slice_hackfest_nstd.yaml"]
2817 pdu_filenames = ["PDU_router.yaml"]
2818 desc_filenames = vnfd_filenames + nsd_filenames + nst_filenames + pdu_filenames
2819 if not os.path.exists(test_dir):
2820 os.makedirs(test_dir)
2821 for filename in desc_filenames:
2822 if not os.path.exists(test_dir+filename):
2823 res = requests.get(test_url+filename)
2824 if res.status_code < 300:
2825 with open(test_dir+filename, "wb") as file:
2826 file.write(res.content)
2827
2828 if all([os.path.exists(test_dir+p) for p in desc_filenames]):
2829
2830 # Test VNFD Quotas
2831 res = engine.test("Create test VNFD #1", "POST", "/vnfpkgm/v1/vnf_packages_content",
2832 headers_zip_json, "@b"+test_dir+vnfd_filenames[0],
2833 (201), r_header_json, "json")
2834 test_vnfd_ids += [engine.last_id if res else None]
2835 res = engine.test("Create test VNFD #2", "POST", "/vnfpkgm/v1/vnf_packages_content",
2836 headers_zip_json, "@b"+test_dir+vnfd_filenames[1],
2837 (201), r_header_json, "json")
2838 test_vnfd_ids += [engine.last_id if res else None]
2839 res = engine.test("Try to create extra test VNFD", "POST",
2840 "/vnfpkgm/v1/vnf_packages_content",
2841 headers_zip_json, "@b"+test_dir+vnfd_filenames[0],
2842 (422), r_header_json, "json")
2843 test_vnfd_ids += [engine.last_id if res is None else None]
2844 res = engine.test("Try to create extra test VNFD with FORCE",
2845 "POST", "/vnfpkgm/v1/vnf_packages_content?FORCE",
2846 headers_zip_json, "@b"+test_dir+vnfd_filenames[0],
2847 (201), r_header_json, "json")
2848 test_vnfd_ids += [engine.last_id if res else None]
2849
2850 # Remove extra VNFDs to prevent further errors
2851 for i in [2, 3]:
2852 if test_vnfd_ids[i]:
2853 res = engine.test("Delete test VNFD #" + str(i), "DELETE",
2854 "/vnfpkgm/v1/vnf_packages_content/"+test_vnfd_ids[i]+"?FORCE",
2855 headers_json, {}, (204), {}, 0)
2856 if res:
2857 test_vnfd_ids[i] = None
2858
2859 if test_vnfd_ids[0] and test_vnfd_ids[1]:
2860
2861 # Test NSD Quotas
2862 res = engine.test("Create test NSD #1", "POST", "/nsd/v1/ns_descriptors_content",
2863 headers_zip_json, "@b"+test_dir+nsd_filenames[0],
2864 (201), r_header_json, "json")
2865 test_nsd_ids += [engine.last_id if res else None]
2866 res = engine.test("Create test NSD #2", "POST", "/nsd/v1/ns_descriptors_content",
2867 headers_zip_json, "@b"+test_dir+nsd_filenames[1],
2868 (201), r_header_json, "json")
2869 test_nsd_ids += [engine.last_id if res else None]
2870 res = engine.test("Try to create extra test NSD", "POST", "/nsd/v1/ns_descriptors_content",
2871 headers_zip_json, "@b"+test_dir+nsd_filenames[0],
2872 (422), r_header_json, "json")
2873 test_nsd_ids += [engine.last_id if res is None else None]
2874 res = engine.test("Try to create extra test NSD with FORCE",
2875 "POST", "/nsd/v1/ns_descriptors_content?FORCE",
2876 headers_zip_json, "@b"+test_dir+nsd_filenames[0],
2877 (201), r_header_json, "json")
2878 test_nsd_ids += [engine.last_id if res else None]
2879
2880 # Remove extra NSDs to prevent further errors
2881 for i in [2, 3]:
2882 if test_nsd_ids[i]:
2883 res = engine.test("Delete test NSD #" + str(i), "DELETE",
2884 "/nsd/v1/ns_descriptors_content/"+test_nsd_ids[i]+"?FORCE",
2885 headers_json, {}, (204), {}, 0)
2886 if res:
2887 test_nsd_ids[i] = None
2888
2889 if test_nsd_ids[0] and test_nsd_ids[1]:
2890
2891 # Test NSR Quotas
2892 res = engine.test("Create test NSR #1", "POST", "/nslcm/v1/ns_instances_content",
2893 headers_json,
2894 {"nsName": test_username+"_1",
2895 "nsdId": test_nsd_ids[0],
2896 "vimAccountId": test_vim_ids[0],
2897 },
2898 (201), r_header_json, "json")
2899 test_nsr_ids += [engine.last_id if res else None]
2900 res = engine.test("Create test NSR #2", "POST", "/nslcm/v1/ns_instances_content",
2901 headers_json,
2902 {"nsName": test_username+"_2",
2903 "nsdId": test_nsd_ids[1],
2904 "vimAccountId": test_vim_ids[0],
2905 },
2906 (201), r_header_json, "json")
2907 test_nsr_ids += [engine.last_id if res else None]
2908 res = engine.test("Try to create extra test NSR", "POST", "/nslcm/v1/ns_instances_content",
2909 headers_json,
2910 {"nsName": test_username+"_3",
2911 "nsdId": test_nsd_ids[0],
2912 "vimAccountId": test_vim_ids[0],
2913 },
2914 (422), r_header_json, "json")
2915 test_nsr_ids += [engine.last_id if res is None else None]
2916 res = engine.test("Try to create test NSR with FORCE", "POST",
2917 "/nslcm/v1/ns_instances_content?FORCE", headers_json,
2918 {"nsName": test_username+"_4",
2919 "nsdId": test_nsd_ids[0],
2920 "vimAccountId": test_vim_ids[0],
2921 },
2922 (201), r_header_json, "json")
2923 test_nsr_ids += [engine.last_id if res else None]
2924
2925 # Test NST Quotas
2926 res = engine.test("Create test NST", "POST", "/nst/v1/netslice_templates_content",
2927 headers_txt_json, "@b"+test_dir+nst_filenames[0],
2928 (201), r_header_json, "json")
2929 test_nst_ids += [engine.last_id if res else None]
2930 res = engine.test("Try to create extra test NST", "POST",
2931 "/nst/v1/netslice_templates_content",
2932 headers_txt_json, "@b"+test_dir+nst_filenames[0],
2933 (422), r_header_json, "json")
2934 test_nst_ids += [engine.last_id if res is None else None]
2935 res = engine.test("Try to create extra test NST with FORCE", "POST",
2936 "/nst/v1/netslice_templates_content?FORCE",
2937 headers_txt_json, "@b"+test_dir+nst_filenames[0],
2938 (201), r_header_json, "json")
2939 test_nst_ids += [engine.last_id if res else None]
2940
2941 if test_nst_ids[0]:
2942 # Remove NSR Quota
2943 engine.set_header({"Authorization": "Bearer {}".format(admin_token)})
2944 res = engine.test("Remove NSR Quota", "PUT", "/admin/v1/projects/"+test_project_id,
2945 headers_json,
2946 {"quotas": {"nsrs": None}},
2947 (204), {}, 0)
2948 engine.set_header({"Authorization": "Bearer {}".format(user_token)})
2949 if res:
2950 # Test NSI Quotas
2951 res = engine.test("Create test NSI", "POST",
2952 "/nsilcm/v1/netslice_instances_content", headers_json,
2953 {"nsiName": test_username,
2954 "nstId": test_nst_ids[0],
2955 "vimAccountId": test_vim_ids[0],
2956 },
2957 (201), r_header_json, "json")
2958 test_nsi_ids += [engine.last_id if res else None]
2959 res = engine.test("Try to create extra test NSI", "POST",
2960 "/nsilcm/v1/netslice_instances_content", headers_json,
2961 {"nsiName": test_username,
2962 "nstId": test_nst_ids[0],
2963 "vimAccountId": test_vim_ids[0],
2964 },
2965 (400), r_header_json, "json")
2966 test_nsi_ids += [engine.last_id if res is None else None]
2967 res = engine.test("Try to create extra test NSI with FORCE", "POST",
2968 "/nsilcm/v1/netslice_instances_content?FORCE", headers_json,
2969 {"nsiName": test_username,
2970 "nstId": test_nst_ids[0],
2971 "vimAccountId": test_vim_ids[0],
2972 },
2973 (201), r_header_json, "json")
2974 test_nsi_ids += [engine.last_id if res else None]
2975
2976 # Test PDU Quotas
2977 with open(test_dir+pdu_filenames[0], "rb") as file:
2978 pdu_text = re.sub(r"ip-address: *\[[^\]]*\]", "ip-address: '0.0.0.0'",
2979 file.read().decode("utf-8"))
2980 with open(test_dir+pdu_filenames[0], "wb") as file:
2981 file.write(pdu_text.encode("utf-8"))
2982 res = engine.test("Create test PDU", "POST", "/pdu/v1/pdu_descriptors",
2983 headers_yaml, "@b"+test_dir+pdu_filenames[0],
2984 (201), r_header_yaml, "yaml")
2985 test_pdu_ids += [engine.last_id if res else None]
2986 res = engine.test("Try to create extra test PDU", "POST", "/pdu/v1/pdu_descriptors",
2987 headers_yaml, "@b"+test_dir+pdu_filenames[0],
2988 (422), r_header_yaml, "yaml")
2989 test_pdu_ids += [engine.last_id if res is None else None]
2990 res = engine.test("Try to create extra test PDU with FORCE", "POST",
2991 "/pdu/v1/pdu_descriptors?FORCE",
2992 headers_yaml, "@b"+test_dir+pdu_filenames[0],
2993 (201), r_header_yaml, "yaml")
2994 test_pdu_ids += [engine.last_id if res else None]
2995
2996 # Cleanup
2997 for i, id in enumerate(test_nsi_ids):
2998 if id:
2999 engine.test("Delete test NSI #"+str(i), "DELETE",