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         _ = 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": {"vduCountIndex": 0, "vduId": "mgmtVM"},
 308                 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
 309             }
 310             nslcmop_id, _ = self.nslcmop_topic.new(
 311                 rollback, session, indata, kwargs=None, headers=headers
 312             )
 313
 314             self.assertEqual(
 315                 self.db.create.call_count,
 316                 1,
 317                 "database create not called, or called more than once",
 318             )
 319             _call = self.db.create.call_args_list[0]
 320             self.assertEqual(
 321                 _call[0][0], "nslcmops", "nslcmops entry must be created at database"
 322             )
 323             created_nslcmop = _call[0][1]
 324             self.assertEqual(
 325                 self.nsr_id,
 326                 created_nslcmop["nsInstanceId"],
 327                 "mismatch between nsId '_id' in created nslcmop and database nsr",
 328             )
 329             self.assertTrue(
 330                 created_nslcmop["lcmOperationType"] == "migrate",
 331                 "Database record must contain 'lcmOperationType=migrate'",
 332             )
 333         with self.subTest(i=2, t="Migration of all VDUs in a VNF"):
 334             indata = {
 335                 "lcmOperationType": "migrate",
 336                 "nsInstanceId": self.nsr_id,
 337                 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
 338             }
 339             nslcmop_id, _ = self.nslcmop_topic.new(
 340                 rollback, session, indata, kwargs=None, headers=headers
 341             )
 342
 343             self.assertEqual(
 344                 self.db.create.call_count,
 345                 2,
 346                 "database create not called, or called more than once",
 347             )
 348             _call = self.db.create.call_args_list[0]
 349             self.assertEqual(
 350                 _call[0][0], "nslcmops", "nslcmops entry must be created at database"
 351             )
 352             created_nslcmop = _call[0][1]
 353             self.assertEqual(
 354                 self.nsr_id,
 355                 created_nslcmop["nsInstanceId"],
 356                 "mismatch between nsId '_id' in created nslcmop and database nsr",
 357             )
 358             self.assertTrue(
 359                 created_nslcmop["lcmOperationType"] == "migrate",
 360                 "Database record must contain 'lcmOperationType=migrate'",
 361             )
 362         with self.subTest(i=3, t="Migration failure - vduId not provided in vdu "):
 363             indata = {
 364                 "lcmOperationType": "migrate",
 365                 "nsInstanceId": self.nsr_id,
 366                 "migrateToHost": "sample02",
 367                 "vdu": {"vduCountIndex": 0},
 368                 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
 369             }
 370
 371             with self.assertRaises(Exception) as e:
 372                 nslcmop_id, _ = self.nslcmop_topic.new(
 373                     rollback, session, indata, kwargs=None, headers=headers
 374                 )
 375             self.assertTrue(
 376                 "Format error at 'vdu' ''vduId' is a required property'"
 377                 in str(e.exception)
 378             )
 379
 380
 381 class TestNsLcmOpTopicWithMock(unittest.TestCase):
 382     def setUp(self):
 383         self.db = Mock(dbbase.DbBase())
 384         self.fs = Mock(FsBase())
 385         self.fs.get_params.return_value = {"./fake/folder"}
 386         self.fs.file_open = mock_open()
 387         self.msg = Mock(MsgBase())
 388         # create class
 389         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
 390
 391     def test_get_vnfd_from_vnf_member_revision(self):
 392         test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
 393         test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
 394         self.db.get_one.side_effect = [test_vnfr, test_vnfd]
 395         _ = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"])
 396         self.assertEqual(
 397             self.db.get_one.call_args_list[0][0][0],
 398             "vnfrs",
 399             "Incorrect first DB lookup",
 400         )
 401         self.assertEqual(
 402             self.db.get_one.call_args_list[1][0][0],
 403             "vnfds",
 404             "Incorrect second DB lookup",
 405         )
 406
 407     def test_get_vnfd_from_vnf_member_no_revision(self):
 408         test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
 409         test_vnfr["revision"] = 3
 410         test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
 411         self.db.get_one.side_effect = [test_vnfr, test_vnfd]
 412         _ = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"])
 413         self.assertEqual(
 414             self.db.get_one.call_args_list[0][0][0],
 415             "vnfrs",
 416             "Incorrect first DB lookup",
 417         )
 418         self.assertEqual(
 419             self.db.get_one.call_args_list[1][0][0],
 420             "vnfds_revisions",
 421             "Incorrect second DB lookup",
 422         )
 423
 424     @contextmanager
 425     def assertNotRaises(self, exception_type):
 426         try:
 427             yield None
 428         except exception_type:
 429             raise self.failureException("{} raised".format(exception_type.__name__))
 430
 431     def test_check_ns_update_operation(self):
 432         self.db = DbMemory()
 433         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
 434         session = {}
 435
 436         with self.subTest(i=1, t="VNF instance does not belong to NS"):
 437             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 438             test_vnfr[0]["revision"] = 2
 439             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 440             test_nsr[0]["constituent-vnfr-ref"][
 441                 0
 442             ] = "99d90b0c-faff-4b9f-bccd-017f33985984"
 443             self.db.create_list("vnfrs", test_vnfr)
 444             self.db.create_list("nsrs", test_nsr)
 445             nsrs = self.db.get_list("nsrs")[0]
 446             indata = {
 447                 "updateType": "CHANGE_VNFPKG",
 448                 "changeVnfPackageData": {
 449                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 450                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 451                 },
 452                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 453             }
 454             with self.assertRaises(EngineException) as expected_exception:
 455                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 456             self.assertEqual(
 457                 str(expected_exception.exception),
 458                 "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984"
 459                 " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85",
 460             )
 461
 462         with self.subTest(i=2, t="Ns update request validated with no exception"):
 463             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 464             test_vnfr[0]["revision"] = 2
 465             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 466             self.db.create_list("vnfrs", test_vnfr)
 467             self.db.create_list("nsrs", test_nsr)
 468             nsrs = self.db.get_list("nsrs")[1]
 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             }
 477             with self.assertNotRaises(EngineException):
 478                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 479
 480         with self.subTest(
 481             i=3, t="Ns update request rejected because of too small timeout"
 482         ):
 483             indata = {
 484                 "updateType": "CHANGE_VNFPKG",
 485                 "changeVnfPackageData": {
 486                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 487                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 488                 },
 489                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 490                 "timeout_ns_update": 50,
 491             }
 492             with self.assertRaises(EngineException) as expected_exception:
 493                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 494             self.assertEqual(
 495                 str(expected_exception.exception),
 496                 "Error in validating ns-update request: 50 second is not enough "
 497                 "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
 498             )
 499
 500         with self.subTest(i=4, t="wrong vnfdid is given as an update parameter"):
 501             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 502             test_vnfr[0]["revision"] = 2
 503             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 504             self.db.create_list("vnfrs", test_vnfr)
 505             self.db.create_list("nsrs", test_nsr)
 506             nsrs = self.db.get_list("nsrs")[2]
 507             indata = {
 508                 "updateType": "CHANGE_VNFPKG",
 509                 "changeVnfPackageData": {
 510                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 511                     "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 512                 },
 513                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 514             }
 515             with self.assertRaises(EngineException) as expected_exception:
 516                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 517             self.assertEqual(
 518                 str(expected_exception.exception),
 519                 "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not "
 520                 "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of "
 521                 "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
 522             )
 523
 524         with self.subTest(
 525             i=5, t="Ns update REMOVE_VNF request validated with no exception"
 526         ):
 527             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 528             test_vnfr[0]["revision"] = 2
 529             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 530             self.db.create_list("vnfrs", test_vnfr)
 531             self.db.create_list("nsrs", test_nsr)
 532             nsrs = self.db.get_list("nsrs")[1]
 533             indata = {
 534                 "updateType": "REMOVE_VNF",
 535                 "removeVnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 536                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 537             }
 538             with self.assertNotRaises(EngineException):
 539                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 540
 541
 542 class TestNsrTopic(unittest.TestCase):
 543     def setUp(self):
 544         self.db = DbMemory()
 545         self.fs = Mock(FsBase())
 546         self.fs.get_params.return_value = {"./fake/folder"}
 547         self.fs.file_open = mock_open()
 548         self.msg = Mock(MsgBase())
 549         # create class
 550         self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None)
 551         self.nsr_topic.check_quota = Mock(return_value=None)  # skip quota
 552
 553         self.db.create_list(
 554             "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
 555         )
 556         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
 557         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
 558         self.db.create = Mock(return_value="created_id")
 559         self.nsd = self.db.get_list("nsds")[0]
 560         self.nsd_id = self.nsd["_id"]
 561         self.nsd_project = self.nsd["_admin"]["projects_read"][0]
 562
 563         self.vim = self.db.get_list("vim_accounts")[0]
 564         self.vim_id = self.vim["_id"]
 565
 566     def test_create(self):
 567         session = {
 568             "force": False,
 569             "admin": False,
 570             "public": False,
 571             "project_id": [self.nsd_project],
 572             "method": "write",
 573         }
 574         indata = {
 575             "nsdId": self.nsd_id,
 576             "nsName": "name",
 577             "vimAccountId": self.vim_id,
 578             "additionalParamsForVnf": [
 579                 {
 580                     "member-vnf-index": "hackfest_vnf1",
 581                     "additionalParams": {"touch_filename": "file"},
 582                 },
 583                 {
 584                     "member-vnf-index": "hackfest_vnf2",
 585                     "additionalParams": {"touch_filename": "file"},
 586                 },
 587             ],
 588         }
 589         rollback = []
 590         headers = {}
 591
 592         self.nsr_topic.new(
 593             rollback, session, indata=indata, kwargs=None, headers=headers
 594         )
 595
 596         # check vnfrs and nsrs created in whatever order
 597         created_vnfrs = []
 598         created_nsrs = []
 599         nsr_id = None
 600         for _call in self.db.create.call_args_list:
 601             assert len(_call[0]) >= 2, "called db.create with few parameters"
 602             created_item = _call[0][1]
 603             if _call[0][0] == "vnfrs":
 604                 created_vnfrs.append(created_item)
 605                 self.assertIn(
 606                     "member-vnf-index-ref",
 607                     created_item,
 608                     "Created item must contain member-vnf-index-ref section",
 609                 )
 610                 if nsr_id:
 611                     self.assertEqual(
 612                         nsr_id,
 613                         created_item["nsr-id-ref"],
 614                         "bad reference id from vnfr to nsr",
 615                     )
 616                 else:
 617                     nsr_id = created_item["nsr-id-ref"]
 618                 self.assertTrue(
 619                     "nsState" not in created_item["_admin"],
 620                     "Database VNF record must not contain '_admin.nsState'",
 621                 )
 622
 623             elif _call[0][0] == "nsrs":
 624                 created_nsrs.append(created_item)
 625                 if nsr_id:
 626                     self.assertEqual(
 627                         nsr_id, created_item["_id"], "bad reference id from vnfr to nsr"
 628                     )
 629                 else:
 630                     nsr_id = created_item["_id"]
 631
 632                 self.assertTrue(
 633                     created_item["_admin"]["nsState"] == "NOT_INSTANTIATED",
 634                     "Database record must contain '_admin.nsState=NOT INSTANTIATE'",
 635                 )
 636             else:
 637                 assert True, "created an unknown record {} at database".format(
 638                     _call[0][0]
 639                 )
 640
 641             self.assertTrue(
 642                 created_item["_admin"].get("projects_read"),
 643                 "Database record must contain '_amdin.projects_read'",
 644             )
 645             self.assertIn(
 646                 "created",
 647                 created_item["_admin"],
 648                 "Database record must contain '_admin.created'",
 649             )
 650
 651         self.assertEqual(
 652             len(created_vnfrs), 2, "created a mismatch number of vnfr at database"
 653         )
 654
 655         self.assertEqual(
 656             created_vnfrs[0]["vdur"][0]["interfaces"][0]["position"],
 657             1,
 658             "vdur first interface position does not match",
 659         )
 660
 661         self.assertEqual(
 662             created_vnfrs[0]["vdur"][0]["interfaces"][1]["position"],
 663             2,
 664             "vdur second interface position does not match",
 665         )
 666
 667         self.assertEqual(
 668             len(created_nsrs), 1, "Only one nsrs must be created at database"
 669         )
 670         self.assertEqual(
 671             len(rollback),
 672             len(created_vnfrs) + 1,
 673             "rollback mismatch with created items at database",
 674         )
 675
 676         # test parameters with error
 677         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
 678         test_set = (
 679             # TODO add "nsd"
 680             (
 681                 "nsd not found",
 682                 {"nsdId": bad_id},
 683                 DbException,
 684                 HTTPStatus.NOT_FOUND,
 685                 ("not found", bad_id),
 686             ),
 687             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
 688             (
 689                 "additional params not supply",
 690                 {"additionalParamsForVnf.0.member-vnf-index": "k"},
 691                 EngineException,
 692                 HTTPStatus.BAD_REQUEST,
 693                 None,
 694             ),
 695         )
 696         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
 697             with self.assertRaises(expect_exc, msg=message) as e:
 698                 self.nsr_topic.new(
 699                     rollback,
 700                     session,
 701                     indata=deepcopy(indata),
 702                     kwargs=kwargs_,
 703                     headers=headers,
 704                 )
 705             if expect_code:
 706                 self.assertTrue(e.exception.http_code == expect_code)
 707             if expect_text_list:
 708                 for expect_text in expect_text_list:
 709                     self.assertIn(
 710                         expect_text,
 711                         str(e.exception).lower(),
 712                         "Expected '{}' at exception text".format(expect_text),
 713                     )
 714
 715     def test_show_instance(self):
 716         session = {
 717             "force": False,
 718             "admin": False,
 719             "public": False,
 720             "project_id": [self.nsd_project],
 721             "method": "write",
 722         }
 723         filter_q = {}
 724         for refresh_status in ("true", "false"):
 725             self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 726             actual_nsr = self.db.get_list("nsrs")[0]
 727             nsr_id = actual_nsr["_id"]
 728             filter_q["vcaStatus-refresh"] = refresh_status
 729             expected_nsr = self.nsr_topic.show(session, nsr_id, filter_q=filter_q)
 730             self.nsr_topic.delete(session, nsr_id)
 731             actual_nsr.pop("_admin")
 732             expected_nsr.pop("_admin")
 733             self.assertEqual(
 734                 expected_nsr, actual_nsr, "Database nsr and show() nsr do not match."
 735             )
 736
 737     def test_vca_status_refresh(self):
 738         session = {
 739             "force": False,
 740             "admin": False,
 741             "public": False,
 742             "project_id": [self.nsd_project],
 743             "method": "write",
 744         }
 745         filter_q = {"vcaStatus-refresh": "true"}
 746         time_delta = 120
 747         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 748         nsr = self.db.get_list("nsrs")[0]
 749
 750         # When vcaStatus-refresh is true
 751         filter_q["vcaStatus-refresh"] = "true"
 752         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
 753         msg_args = self.msg.write.call_args[0]
 754         self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
 755         self.assertGreater(nsr["_admin"]["modified"], time() - time_delta)
 756
 757         # When vcaStatus-refresh is false but modified time is within threshold
 758         filter_q["vcaStatus-refresh"] = "false"
 759         time_now = time()
 760         nsr["_admin"]["modified"] = time_now
 761         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
 762         msg_args = self.msg.write.call_args[1]
 763         self.assertEqual(msg_args, {}, "Message should not be sent.")
 764         self.assertEqual(
 765             nsr["_admin"]["modified"], time_now, "Modified time should not be changed."
 766         )
 767
 768         # When vcaStatus-refresh is false but modified time is less than threshold
 769         filter_q["vcaStatus-refresh"] = "false"
 770         nsr["_admin"]["modified"] = time() - (2 * time_delta)
 771         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
 772         msg_args = self.msg.write.call_args[0]
 773         self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
 774         self.nsr_topic.delete(session, nsr["_id"])
 775         self.assertGreater(
 776             nsr["_admin"]["modified"],
 777             time() - time_delta,
 778             "Modified time is not changed.",
 779         )
 780
 781     def test_delete_ns(self):
 782         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 783         self.nsr = self.db.get_list("nsrs")[0]
 784         self.nsr_id = self.nsr["_id"]
 785         self.db_set_one = self.db.set_one
 786         p_id = self.nsd_project
 787         p_other = "other_p"
 788
 789         session = {
 790             "force": False,
 791             "admin": False,
 792             "public": None,
 793             "project_id": [p_id],
 794             "method": "delete",
 795         }
 796         session2 = {
 797             "force": False,
 798             "admin": False,
 799             "public": None,
 800             "project_id": [p_other],
 801             "method": "delete",
 802         }
 803         session_force = {
 804             "force": True,
 805             "admin": True,
 806             "public": None,
 807             "project_id": [],
 808             "method": "delete",
 809         }
 810         with self.subTest(i=1, t="Normal Deletion"):
 811             self.db.del_one = Mock()
 812             self.db.set_one = Mock()
 813             self.nsr_topic.delete(session, self.nsr_id)
 814
 815             db_args_ro_nsrs = self.db.del_one.call_args_list[1][0]
 816             db_args = self.db.del_one.call_args_list[0][0]
 817             msg_args = self.msg.write.call_args[0]
 818             self.assertEqual(
 819                 msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic"
 820             )
 821             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 822             self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
 823             self.assertEqual(db_args_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
 824             self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
 825             self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
 826             self.assertEqual(
 827                 db_args[1]["_admin.projects_read.cont"], [p_id], "Wrong DB filter"
 828             )
 829             self.db.set_one.assert_not_called()
 830             fs_del_calls = self.fs.file_delete.call_args_list
 831             self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
 832         with self.subTest(i=2, t="No delete because referenced by other project"):
 833             self.db_set_one(
 834                 "nsrs",
 835                 {"_id": self.nsr_id},
 836                 update_dict=None,
 837                 push={
 838                     "_admin.projects_read": p_other,
 839                     "_admin.projects_write": p_other,
 840                 },
 841             )
 842             self.db.del_one.reset_mock()
 843             self.db.set_one.reset_mock()
 844             self.msg.write.reset_mock()
 845             self.fs.file_delete.reset_mock()
 846
 847             self.nsr_topic.delete(session2, self.nsr_id)
 848             self.db.del_one.assert_not_called()
 849             self.msg.write.assert_not_called()
 850             db_s1_args = self.db.set_one.call_args
 851             self.assertEqual(db_s1_args[0][0], self.nsr_topic.topic, "Wrong DB topic")
 852             self.assertEqual(db_s1_args[0][1]["_id"], self.nsr_id, "Wrong DB ID")
 853             self.assertIsNone(
 854                 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
 855             )
 856             self.assertEqual(
 857                 db_s1_args[1]["pull_list"],
 858                 {"_admin.projects_read": [p_other], "_admin.projects_write": [p_other]},
 859                 "Wrong DB pull_list dictionary",
 860             )
 861             self.fs.file_delete.assert_not_called()
 862         with self.subTest(i=4, t="Delete with force and admin"):
 863             self.db.del_one.reset_mock()
 864             self.db.set_one.reset_mock()
 865             self.msg.write.reset_mock()
 866             self.fs.file_delete.reset_mock()
 867             self.nsr_topic.delete(session_force, self.nsr_id)
 868
 869             db_args_ro_nsrs = self.db.del_one.call_args_list[1][0]
 870             db_args = self.db.del_one.call_args_list[0][0]
 871             msg_args = self.msg.write.call_args[0]
 872             self.assertEqual(
 873                 msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic"
 874             )
 875             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 876             self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
 877             self.assertEqual(db_args_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
 878             self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
 879             self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
 880             self.db.set_one.assert_not_called()
 881             fs_del_calls = self.fs.file_delete.call_args_list
 882             self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
 883         with self.subTest(i=3, t="Conflict on Delete - NS in INSTANTIATED state"):
 884             self.db_set_one(
 885                 "nsrs",
 886                 {"_id": self.nsr_id},
 887                 {"_admin.nsState": "INSTANTIATED"},
 888                 pull={
 889                     "_admin.projects_read": p_other,
 890                     "_admin.projects_write": p_other,
 891                 },
 892             )
 893             self.db.del_one.reset_mock()
 894             self.db.set_one.reset_mock()
 895             self.msg.write.reset_mock()
 896             self.fs.file_delete.reset_mock()
 897
 898             with self.assertRaises(
 899                 EngineException, msg="Accepted NSR with nsState INSTANTIATED"
 900             ) as e:
 901                 self.nsr_topic.delete(session, self.nsr_id)
 902             self.assertEqual(
 903                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
 904             )
 905             self.assertIn("INSTANTIATED", str(e.exception), "Wrong exception text")
 906         # TODOD with self.subTest(i=3, t='Conflict on Delete - NS in use by NSI'):
 907
 908         with self.subTest(i=4, t="Non-existent NS"):
 909             self.db.del_one.reset_mock()
 910             self.db.set_one.reset_mock()
 911             self.msg.write.reset_mock()
 912             self.fs.file_delete.reset_mock()
 913             excp_msg = "Not found"
 914             with self.assertRaises(
 915                 DbException, msg="Accepted non-existent NSD ID"
 916             ) as e:
 917                 self.nsr_topic.delete(session2, "other_id")
 918             self.assertEqual(
 919                 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
 920             )
 921             self.assertIn(excp_msg, str(e.exception), "Wrong exception text")
 922             self.assertIn("other_id", str(e.exception), "Wrong exception text")
 923         return