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