osm_nbi/tests/test.py

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