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