osm_nbi/tests/test_instance_topics.py

   1 #
   2 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   3 # not use this file except in compliance with the License. You may obtain
   4 # a copy of the License at
   5 #
   6 #         http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  10 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  11 # License for the specific language governing permissions and limitations
  12 # under the License.
  13 #
  14 # For those usages not covered by the Apache License, Version 2.0 please
  15 # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
  16 ##
  17
  18 import unittest
  19 from unittest.mock import Mock, mock_open   # patch, MagicMock
  20 from osm_common.dbbase import DbException
  21 from osm_nbi.engine import EngineException
  22 from osm_common.dbmemory import DbMemory
  23 from osm_common.fsbase import FsBase
  24 from osm_common.msgbase import MsgBase
  25 from http import HTTPStatus
  26 from osm_nbi.instance_topics import NsLcmOpTopic, NsrTopic
  27 from osm_nbi.tests.test_db_descriptors import db_vim_accounts_text, db_nsds_text, db_vnfds_text, db_nsrs_text,\
  28     db_vnfrs_text
  29 from copy import deepcopy
  30 import yaml
  31
  32
  33 class TestNsLcmOpTopic(unittest.TestCase):
  34
  35     def setUp(self):
  36         self.db = DbMemory()
  37         self.fs = Mock(FsBase())
  38         self.fs.get_params.return_value = {"./fake/folder"}
  39         self.fs.file_open = mock_open()
  40         self.msg = Mock(MsgBase())
  41         # create class
  42         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
  43         self.nslcmop_topic.check_quota = Mock(return_value=None)  # skip quota
  44
  45         self.db.create_list("vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader))
  46         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
  47         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
  48         self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
  49         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
  50         self.db.create = Mock(return_value="created_id")
  51         self.db.set_one = Mock(return_value={"updated": 1})
  52         self.nsd = self.db.get_list("nsds")[0]
  53         self.nsd_id = self.nsd["_id"]
  54         self.nsr = self.db.get_list("nsrs")[0]
  55         self.nsr_id = self.nsr["_id"]
  56         self.nsr_project = self.nsr["_admin"]["projects_read"][0]
  57
  58         self.vim = self.db.get_list("vim_accounts")[0]
  59         self.vim_id = self.vim["_id"]
  60
  61     def test_create_instantiate(self):
  62         session = {"force": False, "admin": False, "public": False, "project_id": [self.nsr_project], "method": "write"}
  63         indata = {
  64             "nsdId": self.nsd_id,
  65             "nsInstanceId": self.nsr_id,
  66             "nsName": "name",
  67             "vimAccountId": self.vim_id,
  68             "additionalParamsForVnf": [{"member-vnf-index": "1", "additionalParams": {"touch_filename": "file"}},
  69                                        {"member-vnf-index": "2", "additionalParams": {"touch_filename": "file"}}],
  70             "vnf": [{"member-vnf-index": "1",
  71                      "vdu": [{"id": "dataVM", "interface": [{"name": "dataVM-eth0",
  72                                                              "ip-address": "10.11.12.13",
  73                                                              "floating-ip-required": True}]
  74                               }],
  75                      "internal-vld": [{"name": "internal", "vim-network-id": "vim-net-id"}]
  76                      }],
  77             "lcmOperationType": "instantiate",
  78
  79         }
  80         rollback = []
  81         headers = {}
  82
  83         nslcmop_id, _ = self.nslcmop_topic.new(rollback, session, indata=indata, kwargs=None, headers=headers)
  84
  85         # check nslcmop is created at database
  86         self.assertEqual(self.db.create.call_count, 1, "database create not called, or called more than once")
  87         _call = self.db.create.call_args_list[0]
  88         self.assertEqual(_call[0][0], "nslcmops", "must be create a nslcmops entry at database")
  89
  90         created_nslcmop = _call[0][1]
  91         self.assertEqual(nslcmop_id, created_nslcmop["_id"], "mismatch between return id and database '_id'")
  92         self.assertEqual(self.nsr_id, created_nslcmop["nsInstanceId"], "bad reference id from nslcmop to nsr")
  93         self.assertTrue(created_nslcmop["_admin"].get("projects_read"),
  94                         "Database record must contain '_amdin.projects_read'")
  95         self.assertIn("created", created_nslcmop["_admin"], "Database record must contain '_admin.created'")
  96         self.assertTrue(created_nslcmop["lcmOperationType"] == "instantiate",
  97                         "Database record must contain 'lcmOperationType=instantiate'")
  98
  99         self.assertEqual(len(rollback), len(self.db.set_one.call_args_list) + 1,
 100                          "rollback mismatch with created/set items at database")
 101
 102         # test parameters with error
 103         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
 104         test_set = (
 105             ("nsr not found", {"nsInstanceId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),
 106             # TODO add "nsd"
 107             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
 108             ("bad member-vnf-index", {"vnf.0.member-vnf-index": "k"}, EngineException, HTTPStatus.BAD_REQUEST,
 109              ("k",)),
 110         )
 111         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
 112             with self.assertRaises(expect_exc, msg=message) as e:
 113                 self.nslcmop_topic.new(rollback, session, indata=deepcopy(indata), kwargs=kwargs_, headers=headers)
 114             if expect_code:
 115                 self.assertTrue(e.exception.http_code == expect_code)
 116             if expect_text_list:
 117                 for expect_text in expect_text_list:
 118                     self.assertIn(expect_text, str(e.exception).lower(),
 119                                   "Expected '{}' at exception text".format(expect_text))
 120
 121     def test_check_ns_operation_action(self):
 122         nsrs = self.db.get_list("nsrs")[0]
 123         session = {}
 124
 125         indata = {
 126             "member_vnf_index": "1",
 127             "vdu_id": None,
 128             "primitive": "touch",
 129             "primitive_params": {"filename": "file"}
 130         }
 131
 132         self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata)
 133         for k in indata:
 134             indata_copy = indata.copy()
 135             if k == "primitive_params":
 136                 continue
 137             indata_copy[k] = "non_existing"
 138             with self.assertRaises(EngineException) as exc_manager:
 139                 self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata_copy)
 140             exc = exc_manager.exception
 141             self.assertEqual(exc.http_code, HTTPStatus.BAD_REQUEST, "Engine exception bad http_code with {}".
 142                              format(indata_copy))
 143
 144
 145 class TestNsrTopic(unittest.TestCase):
 146
 147     def setUp(self):
 148         self.db = DbMemory()
 149         self.fs = Mock(FsBase())
 150         self.fs.get_params.return_value = {"./fake/folder"}
 151         self.fs.file_open = mock_open()
 152         self.msg = Mock(MsgBase())
 153         # create class
 154         self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None)
 155         self.nsr_topic.check_quota = Mock(return_value=None)  # skip quota
 156
 157         self.db.create_list("vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader))
 158         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
 159         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
 160         self.db.create = Mock(return_value="created_id")
 161         self.nsd = self.db.get_list("nsds")[0]
 162         self.nsd_id = self.nsd["_id"]
 163         self.nsd_project = self.nsd["_admin"]["projects_read"][0]
 164
 165         self.vim = self.db.get_list("vim_accounts")[0]
 166         self.vim_id = self.vim["_id"]
 167
 168     def test_create(self):
 169         session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
 170         indata = {
 171             "nsdId": self.nsd_id,
 172             "nsName": "name",
 173             "vimAccountId": self.vim_id,
 174             "additionalParamsForVnf": [{"member-vnf-index": "1", "additionalParams": {"touch_filename": "file"}},
 175                                        {"member-vnf-index": "2", "additionalParams": {"touch_filename": "file"}}]
 176         }
 177         rollback = []
 178         headers = {}
 179
 180         self.nsr_topic.new(rollback, session, indata=indata, kwargs=None, headers=headers)
 181
 182         # check vnfrs and nsrs created in whatever order
 183         created_vnfrs = []
 184         created_nsrs = []
 185         nsr_id = None
 186         for _call in self.db.create.call_args_list:
 187             assert len(_call[0]) >= 2, "called db.create with few parameters"
 188             created_item = _call[0][1]
 189             if _call[0][0] == "vnfrs":
 190                 created_vnfrs.append(created_item)
 191                 self.assertIn("member-vnf-index-ref", created_item,
 192                               "Created item must contain member-vnf-index-ref section")
 193                 if nsr_id:
 194                     self.assertEqual(nsr_id, created_item["nsr-id-ref"], "bad reference id from vnfr to nsr")
 195                 else:
 196                     nsr_id = created_item["nsr-id-ref"]
 197
 198             elif _call[0][0] == "nsrs":
 199                 created_nsrs.append(created_item)
 200                 if nsr_id:
 201                     self.assertEqual(nsr_id, created_item["_id"], "bad reference id from vnfr to nsr")
 202                 else:
 203                     nsr_id = created_item["_id"]
 204             else:
 205                 assert True, "created an unknown record {} at database".format(_call[0][0])
 206
 207             self.assertTrue(created_item["_admin"].get("projects_read"),
 208                             "Database record must contain '_amdin.projects_read'")
 209             self.assertIn("created", created_item["_admin"], "Database record must contain '_admin.created'")
 210             self.assertTrue(created_item["_admin"]["nsState"] == "NOT_INSTANTIATED",
 211                             "Database record must contain '_admin.nstate=NOT INSTANTIATE'")
 212
 213         self.assertEqual(len(created_vnfrs), len(self.nsd["constituent-vnfd"]),
 214                          "created a mismatch number of vnfr at database")
 215         self.assertEqual(len(created_nsrs), 1, "Only one nsrs must be created at database")
 216         self.assertEqual(len(rollback), len(created_vnfrs) + 1, "rollback mismatch with created items at database")
 217
 218         # test parameters with error
 219         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
 220         test_set = (
 221             # TODO add "nsd"
 222             ("nsd not found", {"nsdId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),
 223             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
 224             ("additional params not supply", {"additionalParamsForVnf.0.member-vnf-index": "k"}, EngineException,
 225              HTTPStatus.BAD_REQUEST, None),
 226         )
 227         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
 228             with self.assertRaises(expect_exc, msg=message) as e:
 229                 self.nsr_topic.new(rollback, session, indata=deepcopy(indata), kwargs=kwargs_, headers=headers)
 230             if expect_code:
 231                 self.assertTrue(e.exception.http_code == expect_code)
 232             if expect_text_list:
 233                 for expect_text in expect_text_list:
 234                     self.assertIn(expect_text, str(e.exception).lower(),
 235                                   "Expected '{}' at exception text".format(expect_text))