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