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