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