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 from contextlib import contextmanager
  19 import unittest
  20 from time import time
  21 from unittest.mock import Mock, mock_open   # patch, MagicMock
  22 from osm_common.dbbase import DbException
  23 from osm_nbi.engine import EngineException
  24 from osm_common.dbmemory import DbMemory
  25 from osm_common.fsbase import FsBase
  26 from osm_common.msgbase import MsgBase
  27 from osm_common import dbbase
  28 from http import HTTPStatus
  29 from osm_nbi.instance_topics import NsLcmOpTopic, NsrTopic
  30 from osm_nbi.tests.test_db_descriptors import (
  31     db_vim_accounts_text,
  32     db_nsds_text,
  33     db_vnfds_text,
  34     db_nsrs_text,
  35     db_vnfrs_text,
  36 )
  37 from copy import deepcopy
  38 import yaml
  39
  40
  41 class TestNsLcmOpTopic(unittest.TestCase):
  42     def setUp(self):
  43         self.db = DbMemory()
  44         self.fs = Mock(FsBase())
  45         self.fs.get_params.return_value = {"./fake/folder"}
  46         self.fs.file_open = mock_open()
  47         self.msg = Mock(MsgBase())
  48         # create class
  49         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
  50         self.nslcmop_topic.check_quota = Mock(return_value=None)  # skip quota
  51
  52         self.db.create_list(
  53             "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
  54         )
  55         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
  56         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
  57         self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
  58         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
  59         self.db.create = Mock(return_value="created_id")
  60         self.nsd = self.db.get_list("nsds")[0]
  61         self.nsd_id = self.nsd["_id"]
  62         self.nsr = self.db.get_list("nsrs")[0]
  63         self.nsr_id = self.nsr["_id"]
  64         self.nsr_project = self.nsr["_admin"]["projects_read"][0]
  65
  66         self.vim = self.db.get_list("vim_accounts")[0]
  67         self.vim_id = self.vim["_id"]
  68
  69     def test_create_instantiate(self):
  70         self.db.set_one = Mock(return_value={"updated": 1})
  71         session = {
  72             "force": False,
  73             "admin": False,
  74             "public": False,
  75             "project_id": [self.nsr_project],
  76             "method": "write",
  77         }
  78         indata = {
  79             "nsdId": self.nsd_id,
  80             "nsInstanceId": self.nsr_id,
  81             "nsName": "name",
  82             "vimAccountId": self.vim_id,
  83             "additionalParamsForVnf": [
  84                 {
  85                     "member-vnf-index": "1",
  86                     "additionalParams": {"touch_filename": "file"},
  87                 },
  88                 {
  89                     "member-vnf-index": "2",
  90                     "additionalParams": {"touch_filename": "file"},
  91                 },
  92             ],
  93             "vnf": [
  94                 {
  95                     "member-vnf-index": "1",
  96                     "vdu": [
  97                         {
  98                             "id": "dataVM",
  99                             "interface": [
 100                                 {
 101                                     "name": "dataVM-eth0",
 102                                     "ip-address": "10.11.12.13",
 103                                     "floating-ip-required": True,
 104                                 }
 105                             ],
 106                         }
 107                     ],
 108                     "internal-vld": [
 109                         {"name": "internal", "vim-network-id": "vim-net-id"}
 110                     ],
 111                 }
 112             ],
 113             "lcmOperationType": "instantiate",
 114         }
 115         rollback = []
 116         headers = {}
 117
 118         nslcmop_id, _ = self.nslcmop_topic.new(
 119             rollback, session, indata=deepcopy(indata), kwargs=None, headers=headers
 120         )
 121
 122         # check nslcmop is created at database
 123         self.assertEqual(
 124             self.db.create.call_count,
 125             1,
 126             "database create not called, or called more than once",
 127         )
 128         _call = self.db.create.call_args_list[0]
 129         self.assertEqual(
 130             _call[0][0], "nslcmops", "must be create a nslcmops entry at database"
 131         )
 132
 133         created_nslcmop = _call[0][1]
 134         self.assertEqual(
 135             nslcmop_id,
 136             created_nslcmop["_id"],
 137             "mismatch between return id and database '_id'",
 138         )
 139         self.assertEqual(
 140             self.nsr_id,
 141             created_nslcmop["nsInstanceId"],
 142             "bad reference id from nslcmop to nsr",
 143         )
 144         self.assertTrue(
 145             created_nslcmop["_admin"].get("projects_read"),
 146             "Database record must contain '_amdin.projects_read'",
 147         )
 148         self.assertIn(
 149             "created",
 150             created_nslcmop["_admin"],
 151             "Database record must contain '_admin.created'",
 152         )
 153         self.assertTrue(
 154             created_nslcmop["lcmOperationType"] == "instantiate",
 155             "Database record must contain 'lcmOperationType=instantiate'",
 156         )
 157
 158         self.assertEqual(
 159             len(rollback),
 160             len(self.db.set_one.call_args_list) + 1,
 161             "rollback mismatch with created/set items at database",
 162         )
 163
 164         # test parameters with error
 165         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
 166         test_set = (
 167             (
 168                 "nsr not found",
 169                 {"nsInstanceId": bad_id},
 170                 DbException,
 171                 HTTPStatus.NOT_FOUND,
 172                 ("not found", bad_id),
 173             ),
 174             # TODO add "nsd"
 175             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
 176             (
 177                 "bad member-vnf-index",
 178                 {"vnf.0.member-vnf-index": "k"},
 179                 EngineException,
 180                 HTTPStatus.BAD_REQUEST,
 181                 ("k",),
 182             ),
 183         )
 184         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
 185             with self.assertRaises(expect_exc, msg=message) as e:
 186                 self.nslcmop_topic.new(
 187                     rollback,
 188                     session,
 189                     indata=deepcopy(indata),
 190                     kwargs=kwargs_,
 191                     headers=headers,
 192                 )
 193             if expect_code:
 194                 self.assertTrue(e.exception.http_code == expect_code)
 195             if expect_text_list:
 196                 for expect_text in expect_text_list:
 197                     self.assertIn(
 198                         expect_text,
 199                         str(e.exception).lower(),
 200                         "Expected '{}' at exception text".format(expect_text),
 201                     )
 202
 203     def test_check_ns_operation_action(self):
 204         nsrs = self.db.get_list("nsrs")[0]
 205         session = {}
 206
 207         indata = {
 208             "member_vnf_index": "1",
 209             "vdu_id": None,
 210             "primitive": "touch",
 211             "primitive_params": {"filename": "file"},
 212         }
 213
 214         self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata)
 215         for k in indata:
 216             indata_copy = indata.copy()
 217             if k == "primitive_params":
 218                 continue
 219             indata_copy[k] = "non_existing"
 220             with self.assertRaises(EngineException) as exc_manager:
 221                 self.nslcmop_topic._check_ns_operation(
 222                     session, nsrs, "action", indata_copy
 223                 )
 224             exc = exc_manager.exception
 225             self.assertEqual(
 226                 exc.http_code,
 227                 HTTPStatus.BAD_REQUEST,
 228                 "Engine exception bad http_code with {}".format(indata_copy),
 229             )
 230
 231     def test_update_remove_vnf(self):
 232         vnfr_id = self.db.get_list("vnfrs")[0]["_id"]
 233         session = {}
 234         self.db.set_one(
 235             "nsrs",
 236             {"_id": self.nsr_id},
 237             {"_admin.nsState": "INSTANTIATED"},
 238         )
 239         indata = {
 240             "lcmOperationType": "update",
 241             "updateType": "REMOVE_VNF",
 242             "nsInstanceId": self.nsr_id,
 243             "removeVnfInstanceId": vnfr_id
 244         }
 245
 246         session = {
 247             "force": False,
 248             "admin": False,
 249             "public": False,
 250             "project_id": [self.nsr_project],
 251             "method": "write",
 252         }
 253         rollback = []
 254         headers = {}
 255
 256         nslcmop_id, _ = self.nslcmop_topic.new(
 257             rollback, session, indata, kwargs=None, headers=headers
 258         )
 259
 260         self.assertEqual(
 261             self.db.create.call_count,
 262             1,
 263             "database create not called, or called more than once",
 264         )
 265         _call = self.db.create.call_args_list[0]
 266         self.assertEqual(
 267             _call[0][0], "nslcmops", "nslcmops entry must be created at database"
 268         )
 269         created_nslcmop = _call[0][1]
 270         self.assertEqual(
 271             self.nsr_id,
 272             created_nslcmop["nsInstanceId"],
 273             "mismatch between nsId '_id' in created nslcmop and database nsr",
 274         )
 275         self.assertTrue(
 276             created_nslcmop["lcmOperationType"] == "update",
 277             "Database record must contain 'lcmOperationType=update'",
 278         )
 279         self.assertTrue(
 280             created_nslcmop["operationParams"]["updateType"] == "REMOVE_VNF",
 281             "Database record must contain 'updateType=REMOVE_VNF'",
 282         )
 283
 284     def test_migrate(self):
 285         vnfr_id = self.db.get_list("vnfrs")[0]["_id"]
 286         session = {}
 287         self.db.set_one(
 288             "nsrs",
 289             {"_id": self.nsr_id},
 290             {"_admin.nsState": "INSTANTIATED"},
 291         )
 292         session = {
 293             "force": False,
 294             "admin": False,
 295             "public": False,
 296             "project_id": [self.nsr_project],
 297             "method": "write",
 298         }
 299         rollback = []
 300         headers = {}
 301
 302         with self.subTest(i=1, t="Migration for Specific VM"):
 303             indata = {
 304                 "lcmOperationType": "migrate",
 305                 "nsInstanceId": self.nsr_id,
 306                 "migrateToHost":"sample02",
 307                 "vdu": {
 308                     "vduCountIndex": 0,
 309                     "vduId": "mgmtVM"
 310                 },
 311                 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f"
 312             }
 313             nslcmop_id, _ = self.nslcmop_topic.new(
 314                 rollback, session, indata, kwargs=None, headers=headers
 315             )
 316
 317             self.assertEqual(
 318                 self.db.create.call_count,
 319                 1,
 320                 "database create not called, or called more than once",
 321             )
 322             _call = self.db.create.call_args_list[0]
 323             self.assertEqual(
 324                 _call[0][0], "nslcmops", "nslcmops entry must be created at database"
 325             )
 326             created_nslcmop = _call[0][1]
 327             self.assertEqual(
 328                 self.nsr_id,
 329                 created_nslcmop["nsInstanceId"],
 330                 "mismatch between nsId '_id' in created nslcmop and database nsr",
 331             )
 332             self.assertTrue(
 333                 created_nslcmop["lcmOperationType"] == "migrate",
 334                 "Database record must contain 'lcmOperationType=migrate'",
 335             )
 336         with self.subTest(i=2, t="Migration of all VDUs in a VNF"):
 337             indata = {
 338                 "lcmOperationType": "migrate",
 339                 "nsInstanceId": self.nsr_id,
 340                 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f"
 341             }
 342             nslcmop_id, _ = self.nslcmop_topic.new(
 343                 rollback, session, indata, kwargs=None, headers=headers
 344             )
 345
 346             self.assertEqual(
 347                 self.db.create.call_count,
 348                 2,
 349                 "database create not called, or called more than once",
 350             )
 351             _call = self.db.create.call_args_list[0]
 352             self.assertEqual(
 353                 _call[0][0], "nslcmops", "nslcmops entry must be created at database"
 354             )
 355             created_nslcmop = _call[0][1]
 356             self.assertEqual(
 357                 self.nsr_id,
 358                 created_nslcmop["nsInstanceId"],
 359                 "mismatch between nsId '_id' in created nslcmop and database nsr",
 360             )
 361             self.assertTrue(
 362                 created_nslcmop["lcmOperationType"] == "migrate",
 363                 "Database record must contain 'lcmOperationType=migrate'",
 364             )
 365         with self.subTest(i=3, t="Migration failure - vduId not provided in vdu "):
 366             indata = {
 367                 "lcmOperationType": "migrate",
 368                 "nsInstanceId": self.nsr_id,
 369                 "migrateToHost":"sample02",
 370                 "vdu": {
 371                     "vduCountIndex": 0
 372                 },
 373                 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f"
 374             }
 375
 376             with self.assertRaises(Exception) as e:
 377                 nslcmop_id, _ = self.nslcmop_topic.new(
 378                 rollback, session, indata, kwargs=None, headers=headers
 379             )
 380             self.assertTrue("Format error at 'vdu' ''vduId' is a required property'" in str(e.exception))
 381
 382
 383 class TestNsLcmOpTopicWithMock(unittest.TestCase):
 384     def setUp(self):
 385         self.db = Mock(dbbase.DbBase())
 386         self.fs = Mock(FsBase())
 387         self.fs.get_params.return_value = {"./fake/folder"}
 388         self.fs.file_open = mock_open()
 389         self.msg = Mock(MsgBase())
 390         # create class
 391         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
 392
 393     def test_get_vnfd_from_vnf_member_revision(self):
 394         test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
 395         test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
 396         self.db.get_one.side_effect = [test_vnfr, test_vnfd]
 397         vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id'])
 398         self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup")
 399         self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds', "Incorrect second DB lookup")
 400
 401     def test_get_vnfd_from_vnf_member_no_revision(self):
 402         test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
 403         test_vnfr['revision'] = 3
 404         test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
 405         self.db.get_one.side_effect = [test_vnfr, test_vnfd]
 406         vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id'])
 407         self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup")
 408         self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds_revisions', "Incorrect second DB lookup")
 409
 410     @contextmanager
 411     def assertNotRaises(self, exception_type):
 412         try:
 413             yield None
 414         except exception_type:
 415             raise self.failureException("{} raised".format(exception_type.__name__))
 416
 417     def test_check_ns_update_operation(self):
 418         self.db = DbMemory()
 419         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
 420         session = {}
 421
 422         with self.subTest(i=1, t="VNF instance does not belong to NS"):
 423             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 424             test_vnfr[0]["revision"] = 2
 425             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 426             test_nsr[0]["constituent-vnfr-ref"][
 427                 0
 428             ] = "99d90b0c-faff-4b9f-bccd-017f33985984"
 429             self.db.create_list("vnfrs", test_vnfr)
 430             self.db.create_list("nsrs", test_nsr)
 431             nsrs = self.db.get_list("nsrs")[0]
 432             indata = {
 433                 "updateType": "CHANGE_VNFPKG",
 434                 "changeVnfPackageData": {
 435                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 436                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 437                 },
 438                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 439             }
 440             with self.assertRaises(EngineException) as expected_exception:
 441                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 442             self.assertEqual(
 443                 str(expected_exception.exception),
 444                 "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984"
 445                 " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85",
 446             )
 447
 448         with self.subTest(i=2, t="Ns update request validated with no exception"):
 449             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 450             test_vnfr[0]["revision"] = 2
 451             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 452             self.db.create_list("vnfrs", test_vnfr)
 453             self.db.create_list("nsrs", test_nsr)
 454             nsrs = self.db.get_list("nsrs")[1]
 455             indata = {
 456                 "updateType": "CHANGE_VNFPKG",
 457                 "changeVnfPackageData": {
 458                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 459                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 460                 },
 461                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 462             }
 463             with self.assertNotRaises(EngineException):
 464                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 465
 466         with self.subTest(
 467             i=3, t="Ns update request rejected because of too small timeout"
 468         ):
 469             indata = {
 470                 "updateType": "CHANGE_VNFPKG",
 471                 "changeVnfPackageData": {
 472                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 473                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 474                 },
 475                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 476                 "timeout_ns_update": 50,
 477             }
 478             with self.assertRaises(EngineException) as expected_exception:
 479                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 480             self.assertEqual(
 481                 str(expected_exception.exception),
 482                 "Error in validating ns-update request: 50 second is not enough "
 483                 "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
 484             )
 485
 486         with self.subTest(i=4, t="wrong vnfdid is given as an update parameter"):
 487             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 488             test_vnfr[0]["revision"] = 2
 489             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 490             self.db.create_list("vnfrs", test_vnfr)
 491             self.db.create_list("nsrs", test_nsr)
 492             nsrs = self.db.get_list("nsrs")[2]
 493             indata = {
 494                 "updateType": "CHANGE_VNFPKG",
 495                 "changeVnfPackageData": {
 496                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 497                     "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 498                 },
 499                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 500             }
 501             with self.assertRaises(EngineException) as expected_exception:
 502                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 503             self.assertEqual(
 504                 str(expected_exception.exception),
 505                 "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not "
 506                 "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of "
 507                 "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
 508             )
 509
 510         with self.subTest(i=5, t="Ns update REMOVE_VNF request validated with no exception"):
 511             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 512             test_vnfr[0]["revision"] = 2
 513             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 514             self.db.create_list("vnfrs", test_vnfr)
 515             self.db.create_list("nsrs", test_nsr)
 516             nsrs = self.db.get_list("nsrs")[1]
 517             indata = {
 518                 "updateType": "REMOVE_VNF",
 519                 "removeVnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 520                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 521             }
 522             with self.assertNotRaises(EngineException):
 523                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 524
 525 class TestNsrTopic(unittest.TestCase):
 526     def setUp(self):
 527         self.db = DbMemory()
 528         self.fs = Mock(FsBase())
 529         self.fs.get_params.return_value = {"./fake/folder"}
 530         self.fs.file_open = mock_open()
 531         self.msg = Mock(MsgBase())
 532         # create class
 533         self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None)
 534         self.nsr_topic.check_quota = Mock(return_value=None)  # skip quota
 535
 536         self.db.create_list(
 537             "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
 538         )
 539         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
 540         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
 541         self.db.create = Mock(return_value="created_id")
 542         self.nsd = self.db.get_list("nsds")[0]
 543         self.nsd_id = self.nsd["_id"]
 544         self.nsd_project = self.nsd["_admin"]["projects_read"][0]
 545
 546         self.vim = self.db.get_list("vim_accounts")[0]
 547         self.vim_id = self.vim["_id"]
 548
 549     def test_create(self):
 550         session = {
 551             "force": False,
 552             "admin": False,
 553             "public": False,
 554             "project_id": [self.nsd_project],
 555             "method": "write",
 556         }
 557         indata = {
 558             "nsdId": self.nsd_id,
 559             "nsName": "name",
 560             "vimAccountId": self.vim_id,
 561             "additionalParamsForVnf": [
 562                 {
 563                     "member-vnf-index": "hackfest_vnf1",
 564                     "additionalParams": {"touch_filename": "file"},
 565                 },
 566                 {
 567                     "member-vnf-index": "hackfest_vnf2",
 568                     "additionalParams": {"touch_filename": "file"},
 569                 },
 570             ],
 571         }
 572         rollback = []
 573         headers = {}
 574
 575         self.nsr_topic.new(
 576             rollback, session, indata=indata, kwargs=None, headers=headers
 577         )
 578
 579         # check vnfrs and nsrs created in whatever order
 580         created_vnfrs = []
 581         created_nsrs = []
 582         nsr_id = None
 583         for _call in self.db.create.call_args_list:
 584             assert len(_call[0]) >= 2, "called db.create with few parameters"
 585             created_item = _call[0][1]
 586             if _call[0][0] == "vnfrs":
 587                 created_vnfrs.append(created_item)
 588                 self.assertIn(
 589                     "member-vnf-index-ref",
 590                     created_item,
 591                     "Created item must contain member-vnf-index-ref section",
 592                 )
 593                 if nsr_id:
 594                     self.assertEqual(
 595                         nsr_id,
 596                         created_item["nsr-id-ref"],
 597                         "bad reference id from vnfr to nsr",
 598                     )
 599                 else:
 600                     nsr_id = created_item["nsr-id-ref"]
 601
 602             elif _call[0][0] == "nsrs":
 603                 created_nsrs.append(created_item)
 604                 if nsr_id:
 605                     self.assertEqual(
 606                         nsr_id, created_item["_id"], "bad reference id from vnfr to nsr"
 607                     )
 608                 else:
 609                     nsr_id = created_item["_id"]
 610             else:
 611                 assert True, "created an unknown record {} at database".format(
 612                     _call[0][0]
 613                 )
 614
 615             self.assertTrue(
 616                 created_item["_admin"].get("projects_read"),
 617                 "Database record must contain '_amdin.projects_read'",
 618             )
 619             self.assertIn(
 620                 "created",
 621                 created_item["_admin"],
 622                 "Database record must contain '_admin.created'",
 623             )
 624             self.assertTrue(
 625                 created_item["_admin"]["nsState"] == "NOT_INSTANTIATED",
 626                 "Database record must contain '_admin.nstate=NOT INSTANTIATE'",
 627             )
 628
 629         self.assertEqual(
 630             len(created_vnfrs), 2, "created a mismatch number of vnfr at database"
 631         )
 632
 633         self.assertEqual(
 634             created_vnfrs[0]["vdur"][0]["interfaces"][0]["position"],
 635             1,
 636             "vdur first interface position does not match",
 637         )
 638
 639         self.assertEqual(
 640             created_vnfrs[0]["vdur"][0]["interfaces"][1]["position"],
 641             2,
 642             "vdur second interface position does not match",
 643         )
 644
 645         self.assertEqual(
 646             len(created_nsrs), 1, "Only one nsrs must be created at database"
 647         )
 648         self.assertEqual(
 649             len(rollback),
 650             len(created_vnfrs) + 1,
 651             "rollback mismatch with created items at database",
 652         )
 653
 654         # test parameters with error
 655         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
 656         test_set = (
 657             # TODO add "nsd"
 658             (
 659                 "nsd not found",
 660                 {"nsdId": bad_id},
 661                 DbException,
 662                 HTTPStatus.NOT_FOUND,
 663                 ("not found", bad_id),
 664             ),
 665             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
 666             (
 667                 "additional params not supply",
 668                 {"additionalParamsForVnf.0.member-vnf-index": "k"},
 669                 EngineException,
 670                 HTTPStatus.BAD_REQUEST,
 671                 None,
 672             ),
 673         )
 674         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
 675             with self.assertRaises(expect_exc, msg=message) as e:
 676                 self.nsr_topic.new(
 677                     rollback,
 678                     session,
 679                     indata=deepcopy(indata),
 680                     kwargs=kwargs_,
 681                     headers=headers,
 682                 )
 683             if expect_code:
 684                 self.assertTrue(e.exception.http_code == expect_code)
 685             if expect_text_list:
 686                 for expect_text in expect_text_list:
 687                     self.assertIn(expect_text, str(e.exception).lower(),
 688                                   "Expected '{}' at exception text".format(expect_text))
 689
 690     def test_show_instance(self):
 691         session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
 692         filter_q = {}
 693         for refresh_status in ("true", "false"):
 694             self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 695             actual_nsr = self.db.get_list("nsrs")[0]
 696             nsr_id = actual_nsr["_id"]
 697             filter_q['vcaStatus-refresh'] = refresh_status
 698             expected_nsr = self.nsr_topic.show(session, nsr_id, filter_q=filter_q)
 699             self.nsr_topic.delete(session, nsr_id)
 700             actual_nsr.pop("_admin")
 701             expected_nsr.pop("_admin")
 702             self.assertEqual(expected_nsr, actual_nsr, "Database nsr and show() nsr do not match.")
 703
 704     def test_vca_status_refresh(self):
 705         session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
 706         filter_q = {'vcaStatus-refresh': 'true'}
 707         time_delta = 120
 708         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 709         nsr = self.db.get_list("nsrs")[0]
 710
 711         # When vcaStatus-refresh is true
 712         filter_q['vcaStatus-refresh'] = "true"
 713         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
 714         msg_args = self.msg.write.call_args[0]
 715         self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
 716         self.assertGreater(nsr["_admin"]["modified"], time() - time_delta)
 717
 718         # When vcaStatus-refresh is false but modified time is within threshold
 719         filter_q['vcaStatus-refresh'] = "false"
 720         time_now = time()
 721         nsr["_admin"]["modified"] = time_now
 722         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
 723         msg_args = self.msg.write.call_args[1]
 724         self.assertEqual(msg_args, {}, "Message should not be sent.")
 725         self.assertEqual(nsr["_admin"]["modified"], time_now, "Modified time should not be changed.")
 726
 727         # When vcaStatus-refresh is false but modified time is less than threshold
 728         filter_q['vcaStatus-refresh'] = "false"
 729         nsr["_admin"]["modified"] = time() - (2*time_delta)
 730         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
 731         msg_args = self.msg.write.call_args[0]
 732         self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
 733         self.nsr_topic.delete(session, nsr["_id"])
 734         self.assertGreater(nsr["_admin"]["modified"], time() - time_delta, "Modified time is not changed.")
 735
 736     def test_delete_ns(self):
 737         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 738         self.nsr = self.db.get_list("nsrs")[0]
 739         self.nsr_id = self.nsr["_id"]
 740         self.db_set_one = self.db.set_one
 741         p_id = self.nsd_project
 742         p_other = "other_p"
 743
 744         session = {
 745             "force": False,
 746             "admin": False,
 747             "public": None,
 748             "project_id": [p_id],
 749             "method": "delete",
 750         }
 751         session2 = {
 752             "force": False,
 753             "admin": False,
 754             "public": None,
 755             "project_id": [p_other],
 756             "method": "delete",
 757         }
 758         session_force = {
 759             "force": True,
 760             "admin": True,
 761             "public": None,
 762             "project_id": [],
 763             "method": "delete",
 764         }
 765         with self.subTest(i=1, t="Normal Deletion"):
 766             self.db.del_one = Mock()
 767             self.db.set_one = Mock()
 768             self.nsr_topic.delete(session, self.nsr_id)
 769
 770             db_args_ro_nsrs = self.db.del_one.call_args_list[1][0]
 771             db_args = self.db.del_one.call_args_list[0][0]
 772             msg_args = self.msg.write.call_args[0]
 773             self.assertEqual(
 774                 msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic"
 775             )
 776             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 777             self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
 778             self.assertEqual(db_args_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
 779             self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
 780             self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
 781             self.assertEqual(
 782                 db_args[1]["_admin.projects_read.cont"], [p_id], "Wrong DB filter"
 783             )
 784             self.db.set_one.assert_not_called()
 785             fs_del_calls = self.fs.file_delete.call_args_list
 786             self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
 787         with self.subTest(i=2, t="No delete because referenced by other project"):
 788             self.db_set_one(
 789                 "nsrs",
 790                 {"_id": self.nsr_id},
 791                 update_dict=None,
 792                 push={
 793                     "_admin.projects_read": p_other,
 794                     "_admin.projects_write": p_other,
 795                 },
 796             )
 797             self.db.del_one.reset_mock()
 798             self.db.set_one.reset_mock()
 799             self.msg.write.reset_mock()
 800             self.fs.file_delete.reset_mock()
 801
 802             self.nsr_topic.delete(session2, self.nsr_id)
 803             self.db.del_one.assert_not_called()
 804             self.msg.write.assert_not_called()
 805             db_s1_args = self.db.set_one.call_args
 806             self.assertEqual(db_s1_args[0][0], self.nsr_topic.topic, "Wrong DB topic")
 807             self.assertEqual(db_s1_args[0][1]["_id"], self.nsr_id, "Wrong DB ID")
 808             self.assertIsNone(
 809                 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
 810             )
 811             self.assertEqual(
 812                 db_s1_args[1]["pull_list"],
 813                 {"_admin.projects_read": [p_other], "_admin.projects_write": [p_other]},
 814                 "Wrong DB pull_list dictionary",
 815             )
 816             self.fs.file_delete.assert_not_called()
 817         with self.subTest(i=4, t="Delete with force and admin"):
 818             self.db.del_one.reset_mock()
 819             self.db.set_one.reset_mock()
 820             self.msg.write.reset_mock()
 821             self.fs.file_delete.reset_mock()
 822             self.nsr_topic.delete(session_force, self.nsr_id)
 823
 824             db_args_ro_nsrs = self.db.del_one.call_args_list[1][0]
 825             db_args = self.db.del_one.call_args_list[0][0]
 826             msg_args = self.msg.write.call_args[0]
 827             self.assertEqual(
 828                 msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic"
 829             )
 830             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 831             self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
 832             self.assertEqual(db_args_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
 833             self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
 834             self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
 835             self.db.set_one.assert_not_called()
 836             fs_del_calls = self.fs.file_delete.call_args_list
 837             self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
 838         with self.subTest(i=3, t="Conflict on Delete - NS in INSTANTIATED state"):
 839             self.db_set_one(
 840                 "nsrs",
 841                 {"_id": self.nsr_id},
 842                 {"_admin.nsState": "INSTANTIATED"},
 843                 pull={
 844                     "_admin.projects_read": p_other,
 845                     "_admin.projects_write": p_other,
 846                 },
 847             )
 848             self.db.del_one.reset_mock()
 849             self.db.set_one.reset_mock()
 850             self.msg.write.reset_mock()
 851             self.fs.file_delete.reset_mock()
 852
 853             with self.assertRaises(
 854                 EngineException, msg="Accepted NSR with nsState INSTANTIATED"
 855             ) as e:
 856                 self.nsr_topic.delete(session, self.nsr_id)
 857             self.assertEqual(
 858                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
 859             )
 860             self.assertIn("INSTANTIATED", str(e.exception), "Wrong exception text")
 861         # TODOD with self.subTest(i=3, t='Conflict on Delete - NS in use by NSI'):
 862
 863         with self.subTest(i=4, t="Non-existent NS"):
 864             self.db.del_one.reset_mock()
 865             self.db.set_one.reset_mock()
 866             self.msg.write.reset_mock()
 867             self.fs.file_delete.reset_mock()
 868             excp_msg = "Not found"
 869             with self.assertRaises(
 870                 DbException, msg="Accepted non-existent NSD ID"
 871             ) as e:
 872                 self.nsr_topic.delete(session2, "other_id")
 873             self.assertEqual(
 874                 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
 875             )
 876             self.assertIn(excp_msg, str(e.exception), "Wrong exception text")
 877             self.assertIn("other_id", str(e.exception), "Wrong exception text")
 878         return