osm_nbi/test/test.py

   1 #! /usr/bin/python3
   2 # -*- coding: utf-8 -*-
   3
   4 import getopt
   5 import sys
   6 import requests
   7 #import base64
   8 #from os.path import getsize, basename
   9 #from hashlib import md5
  10 import json
  11 import logging
  12 import yaml
  13 #import json
  14 import tarfile
  15 from os import makedirs
  16 from copy import deepcopy
  17
  18 __author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com"
  19 __date__ = "$2018-03-01$"
  20 __version__ = "0.1"
  21 version_date = "Mar 2018"
  22
  23
  24 def usage():
  25     print("Usage: ", sys.argv[0], "[options]")
  26     print("      --version: prints current version")
  27     print("      -f|--file FILE: file to be sent")
  28     print("      -h|--help: shows this help")
  29     print("      -u|--url URL: complete server URL")
  30     print("      -s|--chunk-size SIZE: size of chunks, by default 1000")
  31     print("      -t|--token TOKEN: Authorizaton token, previously obtained from server")
  32     print("      -v|--verbose print debug information, can be used several times")
  33     return
  34
  35
  36 r_header_json = {"Content-type": "application/json"}
  37 headers_json = {
  38     "Content-type": "application/json",
  39     "Accept": "application/json",
  40 }
  41 r_header_yaml = {"Content-type": "application/yaml"}
  42 headers_yaml = {
  43     "Content-type": "application/yaml",
  44     "Accept": "application/yaml",
  45 }
  46 r_header_text = {"Content-type": "text/plain"}
  47 r_header_octect = {"Content-type": "application/octet-stream"}
  48 headers_text = {
  49     "Accept": "text/plain",
  50 }
  51 r_header_zip = {"Content-type": "application/zip"}
  52 headers_zip = {
  53     "Accept": "application/zip",
  54 }
  55 # test without authorization
  56 test_not_authorized_list = (
  57     ("NA1", "Invalid token", "GET", "/admin/v1/users", headers_json, None, 401, r_header_json, "json"),
  58     ("NA2", "Invalid URL", "POST", "/admin/v1/nonexist", headers_yaml, None, 405, r_header_yaml, "yaml"),
  59     ("NA3", "Invalid version", "DELETE", "/admin/v2/users", headers_yaml, None, 405, r_header_yaml, "yaml"),
  60 )
  61
  62 # test ones authorized
  63 test_authorized_list = (
  64     ("AU1", "Invalid vnfd id", "GET", "/vnfpkgm/v1/vnf_packages/non-existing-id", headers_json, None, 404, r_header_json, "json"),
  65     ("AU2","Invalid nsd id", "GET", "/nsd/v1/ns_descriptors/non-existing-id", headers_yaml, None, 404, r_header_yaml, "yaml"),
  66     ("AU3","Invalid nsd id", "DELETE", "/nsd/v1/ns_descriptors_content/non-existing-id", headers_yaml, None, 404, r_header_yaml, "yaml"),
  67 )
  68
  69 vim = {
  70     "schema_version": "1.0",
  71     "schema_type": "No idea",
  72     "name": "myVim",
  73     "description": "Descriptor name",
  74     "vim_type": "openstack",
  75     "vim_url": "http://localhost:/vim",
  76     "vim_tenant_name": "vimTenant",
  77     "vim_user": "user",
  78     "vim_password": "password",
  79     "config": {"config_param": 1}
  80 }
  81
  82 vim_bad = vim.copy()
  83 vim_bad.pop("name")
  84
  85 test_admin_list1 = (
  86     ("VIM1", "Create VIM", "POST", "/admin/v1/vim_accounts", headers_json, vim, (201, 204), {"Location": "/admin/v1/vim_accounts/", "Content-Type": "application/json"}, "json"),
  87     ("VIM2", "Create VIM bad schema", "POST", "/admin/v1/vim_accounts", headers_json, vim_bad, 422, None, headers_json),
  88     ("VIM2", "Create VIM name repeated", "POST", "/admin/v1/vim_accounts", headers_json, vim, 409, None, headers_json),
  89     ("VIM4", "Show VIMs", "GET", "/admin/v1/vim_accounts", headers_yaml, None, 200, r_header_yaml, "yaml"),
  90     ("VIM5", "Show VIM", "GET", "/admin/v1/vim_accounts/{VIM1}", headers_yaml, None, 200, r_header_yaml, "yaml"),
  91     ("VIM6", "Delete VIM", "DELETE", "/admin/v1/vim_accounts/{VIM1}", headers_yaml, None, 202, None, 0),
  92 )
  93
  94 class TestException(Exception):
  95     pass
  96
  97
  98 class TestRest:
  99     def __init__(self, url_base, header_base={}, verify=False):
 100         self.url_base = url_base
 101         self.header_base = header_base
 102         self.s = requests.session()
 103         self.s.headers = header_base
 104         self.verify = verify
 105         # contains ID of tests obtained from Location response header. "" key contains last obtained id
 106         self.test_ids = {}
 107
 108     def set_header(self, header):
 109         self.s.headers.update(header)
 110
 111     def test(self, name, description, method, url, headers, payload, expected_codes, expected_headers, expected_payload):
 112         """
 113         Performs an http request and check http code response. Exit if different than allowed. It get the returned id
 114         that can be used by following test in the URL with {name} where name is the name of the test
 115         :param name:  short name of the test
 116         :param description:  description of the test
 117         :param method: HTTP method: GET,PUT,POST,DELETE,...
 118         :param url: complete URL or relative URL
 119         :param headers: request headers to add to the base headers
 120         :param payload: Can be a dict, transformed to json, a text or a file if starts with '@'
 121         :param expected_codes: expected response codes, can be int, int tuple or int range
 122         :param expected_headers: expected response headers, dict with key values
 123         :param expected_payload: expected payload, 0 if empty, 'yaml', 'json', 'text', 'zip'
 124         :return: requests response
 125         """
 126         try:
 127             if not self.s:
 128                 self.s = requests.session()
 129             # URL
 130             if not url:
 131                 url = self.url_base
 132             elif not url.startswith("http"):
 133                 url = self.url_base + url
 134
 135             var_start = url.find("{") + 1
 136             while var_start:
 137                 var_end = url.find("}", var_start)
 138                 if var_end == -1:
 139                     break
 140                 var_name = url[var_start:var_end]
 141                 if var_name in self.test_ids:
 142                     url = url[:var_start-1] + self.test_ids[var_name] + url[var_end+1:]
 143                     var_start += len(self.test_ids[var_name])
 144                 var_start = url.find("{", var_start) + 1
 145             if payload:
 146                 if isinstance(payload, str):
 147                     if payload.startswith("@"):
 148                         mode = "r"
 149                         file_name = payload[1:]
 150                         if payload.startswith("@b"):
 151                             mode = "rb"
 152                             file_name = payload[2:]
 153                         with open(file_name, mode) as f:
 154                             payload = f.read()
 155                 elif isinstance(payload, dict):
 156                     payload = json.dumps(payload)
 157
 158             test = "Test {} {} {} {}".format(name, description, method, url)
 159             logger.warning(test)
 160             stream = False
 161             # if expected_payload == "zip":
 162             #     stream = True
 163             r = getattr(self.s, method.lower())(url, data=payload, headers=headers, verify=self.verify, stream=stream)
 164             logger.debug("RX {}: {}".format(r.status_code, r.text))
 165
 166             # check response
 167             if expected_codes:
 168                 if isinstance(expected_codes, int):
 169                     expected_codes = (expected_codes,)
 170                 if r.status_code not in expected_codes:
 171                     raise TestException(
 172                         "Got status {}. Expected {}. {}".format(r.status_code, expected_codes, r.text))
 173
 174             if expected_headers:
 175                 for header_key, header_val in expected_headers.items():
 176                     if header_key.lower() not in r.headers:
 177                         raise TestException("Header {} not present".format(header_key))
 178                     if header_val and header_val.lower() not in r.headers[header_key]:
 179                         raise TestException("Header {} does not contain {} but {}".format(header_key, header_val,
 180                                             r.headers[header_key]))
 181
 182             if expected_payload is not None:
 183                 if expected_payload == 0 and len(r.content) > 0:
 184                     raise TestException("Expected empty payload")
 185                 elif expected_payload == "json":
 186                     try:
 187                         r.json()
 188                     except Exception as e:
 189                         raise TestException("Expected json response payload, but got Exception {}".format(e))
 190                 elif expected_payload == "yaml":
 191                     try:
 192                         yaml.safe_load(r.text)
 193                     except Exception as e:
 194                         raise TestException("Expected yaml response payload, but got Exception {}".format(e))
 195                 elif expected_payload == "zip":
 196                     if len(r.content) == 0:
 197                         raise TestException("Expected some response payload, but got empty")
 198                     # try:
 199                     #     tar = tarfile.open(None, 'r:gz', fileobj=r.raw)
 200                     #     for tarinfo in tar:
 201                     #         tarname = tarinfo.name
 202                     #         print(tarname)
 203                     # except Exception as e:
 204                     #     raise TestException("Expected zip response payload, but got Exception {}".format(e))
 205                 elif expected_payload == "text":
 206                     if len(r.content) == 0:
 207                         raise TestException("Expected some response payload, but got empty")
 208                     #r.text
 209             location = r.headers.get("Location")
 210             if location:
 211                 _id = location[location.rfind("/") + 1:]
 212                 if _id:
 213                     self.test_ids[name] = str(_id)
 214                     self.test_ids[""] = str(_id)  # last id
 215             return r
 216         except TestException as e:
 217             logger.error("{} \nRX code{}: {}".format(e, r.status_code, r.text))
 218             exit(1)
 219         except IOError as e:
 220             logger.error("Cannot open file {}".format(e))
 221             exit(1)
 222
 223
 224 if __name__ == "__main__":
 225     global logger
 226     test = ""
 227
 228     # Disable warnings from self-signed certificates.
 229     requests.packages.urllib3.disable_warnings()
 230     try:
 231         logging.basicConfig(format="%(levelname)s %(message)s", level=logging.ERROR)
 232         logger = logging.getLogger('NBI')
 233         # load parameters and configuration
 234         opts, args = getopt.getopt(sys.argv[1:], "hvu:p:",
 235                                    ["url=", "user=", "password=", "help", "version", "verbose", "project=", "insecure"])
 236         url = "https://localhost:9999/osm"
 237         user = password = project = "admin"
 238         verbose = 0
 239         verify = True
 240
 241         for o, a in opts:
 242             if o == "--version":
 243                 print ("test version " + __version__ + ' ' + version_date)
 244                 sys.exit()
 245             elif o in ("-v", "--verbose"):
 246                 verbose += 1
 247             elif o in ("no-verbose"):
 248                 verbose = -1
 249             elif o in ("-h", "--help"):
 250                 usage()
 251                 sys.exit()
 252             elif o in ("--url"):
 253                 url = a
 254             elif o in ("-u", "--user"):
 255                 user = a
 256             elif o in ("-p", "--password"):
 257                 password = a
 258             elif o in ("--project"):
 259                 project = a
 260             elif o in ("--insecure"):
 261                 verify = False
 262             else:
 263                 assert False, "Unhandled option"
 264         if verbose == 0:
 265             logger.setLevel(logging.WARNING)
 266         elif verbose > 1:
 267             logger.setLevel(logging.DEBUG)
 268         else:
 269             logger.setLevel(logging.ERROR)
 270
 271         test_rest = TestRest(url)
 272
 273         # tests without authorization
 274         for t in test_not_authorized_list:
 275             test_rest.test(*t)
 276
 277         # get token
 278         r = test_rest.test("token1", "Obtain token", "POST", "/admin/v1/tokens", headers_json,
 279                            {"username": user, "password": password, "project_id": project},
 280                            (200, 201), {"Content-Type": "application/json"}, "json")
 281         response = r.json()
 282         token = response["id"]
 283         test_rest.set_header({"Authorization": "Bearer {}".format(token)})
 284
 285         # tests once authorized
 286         for t in test_authorized_list:
 287             test_rest.test(*t)
 288
 289         # tests admin
 290         for t in test_admin_list1:
 291             test_rest.test(*t)
 292
 293
 294         # nsd CREATE
 295         r = test_rest.test("NSD1", "Onboard NSD step 1", "POST", "/nsd/v1/ns_descriptors", headers_json, None,
 296                            201, {"Location": "/nsd/v1/ns_descriptors/", "Content-Type": "application/json"}, "json")
 297         location = r.headers["Location"]
 298         nsd_id = location[location.rfind("/")+1:]
 299         # print(location, nsd_id)
 300
 301         # nsd UPLOAD test
 302         r = test_rest.test("NSD2", "Onboard NSD step 2 as TEXT", "PUT", "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
 303                            r_header_text, "@./cirros_ns/cirros_nsd.yaml", 204, None, 0)
 304
 305         # nsd SHOW OSM format
 306         r = test_rest.test("NSD3", "Show NSD OSM format", "GET", "/nsd/v1/ns_descriptors_content/{}".format(nsd_id),
 307                            headers_json, None, 200, r_header_json, "json")
 308
 309         # nsd SHOW text
 310         r = test_rest.test("NSD4", "Show NSD SOL005 text", "GET", "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
 311                            headers_text, None, 200, r_header_text, "text")
 312
 313         # nsd UPLOAD ZIP
 314         makedirs("temp", exist_ok=True)
 315         tar = tarfile.open("temp/cirros_ns.tar.gz", "w:gz")
 316         tar.add("cirros_ns")
 317         tar.close()
 318         r = test_rest.test("NSD5", "Onboard NSD step 3 replace with ZIP", "PUT", "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
 319                            r_header_zip, "@b./temp/cirros_ns.tar.gz", 204, None, 0)
 320
 321         # nsd SHOW OSM format
 322         r = test_rest.test("NSD6", "Show NSD OSM format", "GET", "/nsd/v1/ns_descriptors_content/{}".format(nsd_id),
 323                            headers_json, None, 200, r_header_json, "json")
 324
 325         # nsd SHOW zip
 326         r = test_rest.test("NSD7", "Show NSD SOL005 zip", "GET", "/nsd/v1/ns_descriptors/{}/nsd_content".format(nsd_id),
 327                            headers_zip, None, 200, r_header_zip, "zip")
 328
 329         # nsd SHOW descriptor
 330         r = test_rest.test("NSD8", "Show NSD descriptor", "GET", "/nsd/v1/ns_descriptors/{}/nsd".format(nsd_id),
 331                            headers_text, None, 200, r_header_text, "text")
 332         # nsd SHOW actifact
 333         r = test_rest.test("NSD9", "Show NSD artifact", "GET", "/nsd/v1/ns_descriptors/{}/artifacts/icons/osm_2x.png".format(nsd_id),
 334                            headers_text, None, 200, r_header_octect, "text")
 335
 336         # nsd DELETE
 337         r = test_rest.test("NSD10", "Delete NSD SOL005 text", "DELETE", "/nsd/v1/ns_descriptors/{}".format(nsd_id),
 338                            headers_yaml, None, 204, None, 0)
 339
 340         # vnfd CREATE
 341         r = test_rest.test("VNFD1", "Onboard VNFD step 1", "POST", "/vnfpkgm/v1/vnf_packages", headers_json, None,
 342                            201, {"Location": "/vnfpkgm/v1/vnf_packages/", "Content-Type": "application/json"}, "json")
 343         location = r.headers["Location"]
 344         vnfd_id = location[location.rfind("/")+1:]
 345         # print(location, vnfd_id)
 346
 347         # vnfd UPLOAD test
 348         r = test_rest.test("VNFD2", "Onboard VNFD step 2 as TEXT", "PUT", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
 349                            r_header_text, "@./cirros_vnf/cirros_vnfd.yaml", 204, None, 0)
 350
 351         # vnfd SHOW OSM format
 352         r = test_rest.test("VNFD3", "Show VNFD OSM format", "GET", "/vnfpkgm/v1/vnf_packages_content/{}".format(vnfd_id),
 353                            headers_json, None, 200, r_header_json, "json")
 354
 355         # vnfd SHOW text
 356         r = test_rest.test("VNFD4", "Show VNFD SOL005 text", "GET", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
 357                            headers_text, None, 200, r_header_text, "text")
 358
 359         # vnfd UPLOAD ZIP
 360         makedirs("temp", exist_ok=True)
 361         tar = tarfile.open("temp/cirros_vnf.tar.gz", "w:gz")
 362         tar.add("cirros_vnf")
 363         tar.close()
 364         r = test_rest.test("VNFD5", "Onboard VNFD step 3 replace with ZIP", "PUT", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
 365                            r_header_zip, "@b./temp/cirros_vnf.tar.gz", 204, None, 0)
 366
 367         # vnfd SHOW OSM format
 368         r = test_rest.test("VNFD6", "Show VNFD OSM format", "GET", "/vnfpkgm/v1/vnf_packages_content/{}".format(vnfd_id),
 369                            headers_json, None, 200, r_header_json, "json")
 370
 371         # vnfd SHOW zip
 372         r = test_rest.test("VNFD7", "Show VNFD SOL005 zip", "GET", "/vnfpkgm/v1/vnf_packages/{}/package_content".format(vnfd_id),
 373                            headers_zip, None, 200, r_header_zip, "zip")
 374         # vnfd SHOW descriptor
 375         r = test_rest.test("VNFD8", "Show VNFD descriptor", "GET", "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(vnfd_id),
 376                            headers_text, None, 200, r_header_text, "text")
 377         # vnfd SHOW actifact
 378         r = test_rest.test("VNFD9", "Show VNFD artifact", "GET", "/vnfpkgm/v1/vnf_packages/{}/artifacts/icons/cirros-64.png".format(vnfd_id),
 379                            headers_text, None, 200, r_header_octect, "text")
 380
 381         # vnfd DELETE
 382         r = test_rest.test("VNFD10", "Delete VNFD SOL005 text", "DELETE", "/vnfpkgm/v1/vnf_packages/{}".format(vnfd_id),
 383                            headers_yaml, None, 204, None, 0)
 384
 385         print("PASS")
 386
 387     except Exception as e:
 388         if test:
 389             logger.error(test + " Exception: " + str(e))
 390             exit(1)
 391         else:
 392             logger.critical(test + " Exception: " + str(e), exc_info=True)