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 (
  28     db_vim_accounts_text,
  29     db_nsds_text,
  30     db_vnfds_text,
  31     db_nsrs_text,
  32     db_vnfrs_text,
  33 )
  34 from copy import deepcopy
  35 import yaml
  36
  37
  38 class TestNsLcmOpTopic(unittest.TestCase):
  39     def setUp(self):
  40         self.db = DbMemory()
  41         self.fs = Mock(FsBase())
  42         self.fs.get_params.return_value = {"./fake/folder"}
  43         self.fs.file_open = mock_open()
  44         self.msg = Mock(MsgBase())
  45         # create class
  46         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
  47         self.nslcmop_topic.check_quota = Mock(return_value=None)  # skip quota
  48
  49         self.db.create_list(
  50             "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
  51         )
  52         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
  53         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
  54         self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
  55         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
  56         self.db.create = Mock(return_value="created_id")
  57         self.db.set_one = Mock(return_value={"updated": 1})
  58         self.nsd = self.db.get_list("nsds")[0]
  59         self.nsd_id = self.nsd["_id"]
  60         self.nsr = self.db.get_list("nsrs")[0]
  61         self.nsr_id = self.nsr["_id"]
  62         self.nsr_project = self.nsr["_admin"]["projects_read"][0]
  63
  64         self.vim = self.db.get_list("vim_accounts")[0]
  65         self.vim_id = self.vim["_id"]
  66
  67     def test_create_instantiate(self):
  68         session = {
  69             "force": False,
  70             "admin": False,
  71             "public": False,
  72             "project_id": [self.nsr_project],
  73             "method": "write",
  74         }
  75         indata = {
  76             "nsdId": self.nsd_id,
  77             "nsInstanceId": self.nsr_id,
  78             "nsName": "name",
  79             "vimAccountId": self.vim_id,
  80             "additionalParamsForVnf": [
  81                 {
  82                     "member-vnf-index": "1",
  83                     "additionalParams": {"touch_filename": "file"},
  84                 },
  85                 {
  86                     "member-vnf-index": "2",
  87                     "additionalParams": {"touch_filename": "file"},
  88                 },
  89             ],
  90             "vnf": [
  91                 {
  92                     "member-vnf-index": "1",
  93                     "vdu": [
  94                         {
  95                             "id": "dataVM",
  96                             "interface": [
  97                                 {
  98                                     "name": "dataVM-eth0",
  99                                     "ip-address": "10.11.12.13",
 100                                     "floating-ip-required": True,
 101                                 }
 102                             ],
 103                         }
 104                     ],
 105                     "internal-vld": [
 106                         {"name": "internal", "vim-network-id": "vim-net-id"}
 107                     ],
 108                 }
 109             ],
 110             "lcmOperationType": "instantiate",
 111         }
 112         rollback = []
 113         headers = {}
 114
 115         nslcmop_id, _ = self.nslcmop_topic.new(
 116             rollback, session, indata=deepcopy(indata), kwargs=None, headers=headers
 117         )
 118
 119         # check nslcmop is created at database
 120         self.assertEqual(
 121             self.db.create.call_count,
 122             1,
 123             "database create not called, or called more than once",
 124         )
 125         _call = self.db.create.call_args_list[0]
 126         self.assertEqual(
 127             _call[0][0], "nslcmops", "must be create a nslcmops entry at database"
 128         )
 129
 130         created_nslcmop = _call[0][1]
 131         self.assertEqual(
 132             nslcmop_id,
 133             created_nslcmop["_id"],
 134             "mismatch between return id and database '_id'",
 135         )
 136         self.assertEqual(
 137             self.nsr_id,
 138             created_nslcmop["nsInstanceId"],
 139             "bad reference id from nslcmop to nsr",
 140         )
 141         self.assertTrue(
 142             created_nslcmop["_admin"].get("projects_read"),
 143             "Database record must contain '_amdin.projects_read'",
 144         )
 145         self.assertIn(
 146             "created",
 147             created_nslcmop["_admin"],
 148             "Database record must contain '_admin.created'",
 149         )
 150         self.assertTrue(
 151             created_nslcmop["lcmOperationType"] == "instantiate",
 152             "Database record must contain 'lcmOperationType=instantiate'",
 153         )
 154
 155         self.assertEqual(
 156             len(rollback),
 157             len(self.db.set_one.call_args_list) + 1,
 158             "rollback mismatch with created/set items at database",
 159         )
 160
 161         # test parameters with error
 162         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
 163         test_set = (
 164             (
 165                 "nsr not found",
 166                 {"nsInstanceId": bad_id},
 167                 DbException,
 168                 HTTPStatus.NOT_FOUND,
 169                 ("not found", bad_id),
 170             ),
 171             # TODO add "nsd"
 172             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
 173             (
 174                 "bad member-vnf-index",
 175                 {"vnf.0.member-vnf-index": "k"},
 176                 EngineException,
 177                 HTTPStatus.BAD_REQUEST,
 178                 ("k",),
 179             ),
 180         )
 181         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
 182             with self.assertRaises(expect_exc, msg=message) as e:
 183                 self.nslcmop_topic.new(
 184                     rollback,
 185                     session,
 186                     indata=deepcopy(indata),
 187                     kwargs=kwargs_,
 188                     headers=headers,
 189                 )
 190             if expect_code:
 191                 self.assertTrue(e.exception.http_code == expect_code)
 192             if expect_text_list:
 193                 for expect_text in expect_text_list:
 194                     self.assertIn(
 195                         expect_text,
 196                         str(e.exception).lower(),
 197                         "Expected '{}' at exception text".format(expect_text),
 198                     )
 199
 200     def test_check_ns_operation_action(self):
 201         nsrs = self.db.get_list("nsrs")[0]
 202         session = {}
 203
 204         indata = {
 205             "member_vnf_index": "1",
 206             "vdu_id": None,
 207             "primitive": "touch",
 208             "primitive_params": {"filename": "file"},
 209         }
 210
 211         self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata)
 212         for k in indata:
 213             indata_copy = indata.copy()
 214             if k == "primitive_params":
 215                 continue
 216             indata_copy[k] = "non_existing"
 217             with self.assertRaises(EngineException) as exc_manager:
 218                 self.nslcmop_topic._check_ns_operation(
 219                     session, nsrs, "action", indata_copy
 220                 )
 221             exc = exc_manager.exception
 222             self.assertEqual(
 223                 exc.http_code,
 224                 HTTPStatus.BAD_REQUEST,
 225                 "Engine exception bad http_code with {}".format(indata_copy),
 226             )
 227
 228
 229 class TestNsrTopic(unittest.TestCase):
 230     def setUp(self):
 231         self.db = DbMemory()
 232         self.fs = Mock(FsBase())
 233         self.fs.get_params.return_value = {"./fake/folder"}
 234         self.fs.file_open = mock_open()
 235         self.msg = Mock(MsgBase())
 236         # create class
 237         self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None)
 238         self.nsr_topic.check_quota = Mock(return_value=None)  # skip quota
 239
 240         self.db.create_list(
 241             "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
 242         )
 243         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
 244         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
 245         self.db.create = Mock(return_value="created_id")
 246         self.nsd = self.db.get_list("nsds")[0]
 247         self.nsd_id = self.nsd["_id"]
 248         self.nsd_project = self.nsd["_admin"]["projects_read"][0]
 249
 250         self.vim = self.db.get_list("vim_accounts")[0]
 251         self.vim_id = self.vim["_id"]
 252
 253     def test_create(self):
 254         session = {
 255             "force": False,
 256             "admin": False,
 257             "public": False,
 258             "project_id": [self.nsd_project],
 259             "method": "write",
 260         }
 261         indata = {
 262             "nsdId": self.nsd_id,
 263             "nsName": "name",
 264             "vimAccountId": self.vim_id,
 265             "additionalParamsForVnf": [
 266                 {
 267                     "member-vnf-index": "hackfest_vnf1",
 268                     "additionalParams": {"touch_filename": "file"},
 269                 },
 270                 {
 271                     "member-vnf-index": "hackfest_vnf2",
 272                     "additionalParams": {"touch_filename": "file"},
 273                 },
 274             ],
 275         }
 276         rollback = []
 277         headers = {}
 278
 279         self.nsr_topic.new(
 280             rollback, session, indata=indata, kwargs=None, headers=headers
 281         )
 282
 283         # check vnfrs and nsrs created in whatever order
 284         created_vnfrs = []
 285         created_nsrs = []
 286         nsr_id = None
 287         for _call in self.db.create.call_args_list:
 288             assert len(_call[0]) >= 2, "called db.create with few parameters"
 289             created_item = _call[0][1]
 290             if _call[0][0] == "vnfrs":
 291                 created_vnfrs.append(created_item)
 292                 self.assertIn(
 293                     "member-vnf-index-ref",
 294                     created_item,
 295                     "Created item must contain member-vnf-index-ref section",
 296                 )
 297                 if nsr_id:
 298                     self.assertEqual(
 299                         nsr_id,
 300                         created_item["nsr-id-ref"],
 301                         "bad reference id from vnfr to nsr",
 302                     )
 303                 else:
 304                     nsr_id = created_item["nsr-id-ref"]
 305
 306             elif _call[0][0] == "nsrs":
 307                 created_nsrs.append(created_item)
 308                 if nsr_id:
 309                     self.assertEqual(
 310                         nsr_id, created_item["_id"], "bad reference id from vnfr to nsr"
 311                     )
 312                 else:
 313                     nsr_id = created_item["_id"]
 314             else:
 315                 assert True, "created an unknown record {} at database".format(
 316                     _call[0][0]
 317                 )
 318
 319             self.assertTrue(
 320                 created_item["_admin"].get("projects_read"),
 321                 "Database record must contain '_amdin.projects_read'",
 322             )
 323             self.assertIn(
 324                 "created",
 325                 created_item["_admin"],
 326                 "Database record must contain '_admin.created'",
 327             )
 328             self.assertTrue(
 329                 created_item["_admin"]["nsState"] == "NOT_INSTANTIATED",
 330                 "Database record must contain '_admin.nstate=NOT INSTANTIATE'",
 331             )
 332
 333         self.assertEqual(
 334             len(created_vnfrs), 2, "created a mismatch number of vnfr at database"
 335         )
 336         self.assertEqual(
 337             len(created_nsrs), 1, "Only one nsrs must be created at database"
 338         )
 339         self.assertEqual(
 340             len(rollback),
 341             len(created_vnfrs) + 1,
 342             "rollback mismatch with created items at database",
 343         )
 344
 345         # test parameters with error
 346         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
 347         test_set = (
 348             # TODO add "nsd"
 349             (
 350                 "nsd not found",
 351                 {"nsdId": bad_id},
 352                 DbException,
 353                 HTTPStatus.NOT_FOUND,
 354                 ("not found", bad_id),
 355             ),
 356             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
 357             (
 358                 "additional params not supply",
 359                 {"additionalParamsForVnf.0.member-vnf-index": "k"},
 360                 EngineException,
 361                 HTTPStatus.BAD_REQUEST,
 362                 None,
 363             ),
 364         )
 365         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
 366             with self.assertRaises(expect_exc, msg=message) as e:
 367                 self.nsr_topic.new(
 368                     rollback,
 369                     session,
 370                     indata=deepcopy(indata),
 371                     kwargs=kwargs_,
 372                     headers=headers,
 373                 )
 374             if expect_code:
 375                 self.assertTrue(e.exception.http_code == expect_code)
 376             if expect_text_list:
 377                 for expect_text in expect_text_list:
 378                     self.assertIn(
 379                         expect_text,
 380                         str(e.exception).lower(),
 381                         "Expected '{}' at exception text".format(expect_text),
 382                     )
 383
 384     def test_delete_ns(self):
 385         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 386         self.nsr = self.db.get_list("nsrs")[0]
 387         self.nsr_id = self.nsr["_id"]
 388         self.db_set_one = self.db.set_one
 389         p_id = self.nsd_project
 390         p_other = "other_p"
 391
 392         session = {
 393             "force": False,
 394             "admin": False,
 395             "public": None,
 396             "project_id": [p_id],
 397             "method": "delete",
 398         }
 399         session2 = {
 400             "force": False,
 401             "admin": False,
 402             "public": None,
 403             "project_id": [p_other],
 404             "method": "delete",
 405         }
 406         session_force = {
 407             "force": True,
 408             "admin": True,
 409             "public": None,
 410             "project_id": [],
 411             "method": "delete",
 412         }
 413         with self.subTest(i=1, t="Normal Deletion"):
 414             self.db.del_one = Mock()
 415             self.db.set_one = Mock()
 416             self.nsr_topic.delete(session, self.nsr_id)
 417
 418             db_args_ro_nsrs = self.db.del_one.call_args_list[1][0]
 419             db_args = self.db.del_one.call_args_list[0][0]
 420             msg_args = self.msg.write.call_args[0]
 421             self.assertEqual(
 422                 msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic"
 423             )
 424             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 425             self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
 426             self.assertEqual(db_args_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
 427             self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
 428             self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
 429             self.assertEqual(
 430                 db_args[1]["_admin.projects_read.cont"], [p_id], "Wrong DB filter"
 431             )
 432             self.db.set_one.assert_not_called()
 433             fs_del_calls = self.fs.file_delete.call_args_list
 434             self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
 435         with self.subTest(i=2, t="No delete because referenced by other project"):
 436             self.db_set_one(
 437                 "nsrs",
 438                 {"_id": self.nsr_id},
 439                 update_dict=None,
 440                 push={
 441                     "_admin.projects_read": p_other,
 442                     "_admin.projects_write": p_other,
 443                 },
 444             )
 445             self.db.del_one.reset_mock()
 446             self.db.set_one.reset_mock()
 447             self.msg.write.reset_mock()
 448             self.fs.file_delete.reset_mock()
 449
 450             self.nsr_topic.delete(session2, self.nsr_id)
 451             self.db.del_one.assert_not_called()
 452             self.msg.write.assert_not_called()
 453             db_s1_args = self.db.set_one.call_args
 454             self.assertEqual(db_s1_args[0][0], self.nsr_topic.topic, "Wrong DB topic")
 455             self.assertEqual(db_s1_args[0][1]["_id"], self.nsr_id, "Wrong DB ID")
 456             self.assertIsNone(
 457                 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
 458             )
 459             self.assertEqual(
 460                 db_s1_args[1]["pull_list"],
 461                 {"_admin.projects_read": [p_other], "_admin.projects_write": [p_other]},
 462                 "Wrong DB pull_list dictionary",
 463             )
 464             self.fs.file_delete.assert_not_called()
 465         with self.subTest(i=4, t="Delete with force and admin"):
 466             self.db.del_one.reset_mock()
 467             self.db.set_one.reset_mock()
 468             self.msg.write.reset_mock()
 469             self.fs.file_delete.reset_mock()
 470             self.nsr_topic.delete(session_force, self.nsr_id)
 471
 472             db_args_ro_nsrs = self.db.del_one.call_args_list[1][0]
 473             db_args = self.db.del_one.call_args_list[0][0]
 474             msg_args = self.msg.write.call_args[0]
 475             self.assertEqual(
 476                 msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic"
 477             )
 478             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 479             self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
 480             self.assertEqual(db_args_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
 481             self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
 482             self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
 483             self.db.set_one.assert_not_called()
 484             fs_del_calls = self.fs.file_delete.call_args_list
 485             self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
 486         with self.subTest(i=3, t="Conflict on Delete - NS in INSTANTIATED state"):
 487             self.db_set_one(
 488                 "nsrs",
 489                 {"_id": self.nsr_id},
 490                 {"_admin.nsState": "INSTANTIATED"},
 491                 pull={
 492                     "_admin.projects_read": p_other,
 493                     "_admin.projects_write": p_other,
 494                 },
 495             )
 496             self.db.del_one.reset_mock()
 497             self.db.set_one.reset_mock()
 498             self.msg.write.reset_mock()
 499             self.fs.file_delete.reset_mock()
 500
 501             with self.assertRaises(
 502                 EngineException, msg="Accepted NSR with nsState INSTANTIATED"
 503             ) as e:
 504                 self.nsr_topic.delete(session, self.nsr_id)
 505             self.assertEqual(
 506                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
 507             )
 508             self.assertIn("INSTANTIATED", str(e.exception), "Wrong exception text")
 509         # TODOD with self.subTest(i=3, t='Conflict on Delete - NS in use by NSI'):
 510
 511         with self.subTest(i=4, t="Non-existent NS"):
 512             self.db.del_one.reset_mock()
 513             self.db.set_one.reset_mock()
 514             self.msg.write.reset_mock()
 515             self.fs.file_delete.reset_mock()
 516             excp_msg = "Not found"
 517             with self.assertRaises(
 518                 DbException, msg="Accepted non-existent NSD ID"
 519             ) as e:
 520                 self.nsr_topic.delete(session2, "other_id")
 521             self.assertEqual(
 522                 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
 523             )
 524             self.assertIn(excp_msg, str(e.exception), "Wrong exception text")
 525             self.assertIn("other_id", str(e.exception), "Wrong exception text")
 526         return