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
  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_paas_accounts_text,
  33     db_nsds_text,
  34     db_vnfds_text,
  35     db_nsrs_text,
  36     db_vnfrs_text,
  37 )
  38 from copy import deepcopy
  39 import yaml
  40
  41
  42 class TestNsLcmOpTopic(unittest.TestCase):
  43     def setUp(self):
  44         self.db = DbMemory()
  45         self.fs = Mock(FsBase())
  46         self.fs.get_params.return_value = {"./fake/folder"}
  47         self.fs.file_open = mock_open()
  48         self.msg = Mock(MsgBase())
  49         # create class
  50         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
  51         self.nslcmop_topic.check_quota = Mock(return_value=None)  # skip quota
  52
  53         self.db.create_list(
  54             "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
  55         )
  56         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
  57         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
  58         self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
  59         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
  60         self.db.create = Mock(return_value="created_id")
  61         self.nsd = self.db.get_list("nsds")[0]
  62         self.nsd_id = self.nsd["_id"]
  63         self.nsr = self.db.get_list("nsrs")[0]
  64         self.nsr_id = self.nsr["_id"]
  65         self.nsr_project = self.nsr["_admin"]["projects_read"][0]
  66
  67         self.vim = self.db.get_list("vim_accounts")[0]
  68         self.vim_id = self.vim["_id"]
  69         self.db.create_list(
  70             "paas", yaml.load(db_paas_accounts_text, Loader=yaml.Loader)
  71         )
  72         self.paas_ids = [
  73             paas_element["_id"] for paas_element in self.db.get_list("paas")
  74         ]
  75         self.paas_id = self.paas_ids[0]
  76         self.vnfrs = {
  77             vnfr["_id"]: vnfr["member-vnf-index-ref"]
  78             for vnfr in self.db.get_list("vnfrs")
  79         }
  80         self.number_of_vnfr = len(self.vnfrs)
  81         self.new_session = {
  82             "force": False,
  83             "admin": False,
  84             "public": False,
  85             "project_id": [self.nsr_project],
  86             "method": "write",
  87         }
  88
  89     def check_operation_params(self, operation_params):
  90         self.assertEqual(self.nsd_id, operation_params["nsdId"])
  91         self.assertEqual(self.nsr_id, operation_params["nsInstanceId"])
  92         self.assertEqual("name", operation_params["nsName"])
  93
  94     def test_create_instantiate_with_vim_account(self):
  95         self.db.set_one = Mock(return_value={"updated": 1})
  96         indata = {
  97             "nsdId": self.nsd_id,
  98             "nsInstanceId": self.nsr_id,
  99             "nsName": "name",
 100             "vimAccountId": self.vim_id,
 101             "additionalParamsForVnf": [
 102                 {
 103                     "member-vnf-index": "1",
 104                     "additionalParams": {"touch_filename": "file"},
 105                 },
 106                 {
 107                     "member-vnf-index": "2",
 108                     "additionalParams": {"touch_filename": "file"},
 109                 },
 110             ],
 111             "vnf": [
 112                 {
 113                     "member-vnf-index": "1",
 114                     "vdu": [
 115                         {
 116                             "id": "dataVM",
 117                             "interface": [
 118                                 {
 119                                     "name": "dataVM-eth0",
 120                                     "ip-address": "10.11.12.13",
 121                                     "floating-ip-required": True,
 122                                 }
 123                             ],
 124                         }
 125                     ],
 126                     "internal-vld": [
 127                         {"name": "internal", "vim-network-id": "vim-net-id"}
 128                     ],
 129                 }
 130             ],
 131             "lcmOperationType": "instantiate",
 132         }
 133         rollback = []
 134         headers = {}
 135
 136         nslcmop_id, _ = self.nslcmop_topic.new(
 137             rollback,
 138             self.new_session,
 139             indata=deepcopy(indata),
 140             kwargs=None,
 141             headers=headers,
 142         )
 143
 144         # check nslcmop is created at database
 145         self.assertEqual(
 146             self.db.create.call_count,
 147             1,
 148             "database create not called, or called more than once",
 149         )
 150         _call = self.db.create.call_args_list[0]
 151         self.assertEqual(
 152             _call[0][0], "nslcmops", "must be create a nslcmops entry at database"
 153         )
 154
 155         created_nslcmop = _call[0][1]
 156         self.assertEqual(
 157             nslcmop_id,
 158             created_nslcmop["_id"],
 159             "mismatch between return id and database '_id'",
 160         )
 161         self.assertEqual(
 162             self.nsr_id,
 163             created_nslcmop["nsInstanceId"],
 164             "bad reference id from nslcmop to nsr",
 165         )
 166         self.assertTrue(
 167             created_nslcmop["_admin"].get("projects_read"),
 168             "Database record must contain '_amdin.projects_read'",
 169         )
 170         self.assertIn(
 171             "created",
 172             created_nslcmop["_admin"],
 173             "Database record must contain '_admin.created'",
 174         )
 175         self.assertTrue(
 176             created_nslcmop["lcmOperationType"] == "instantiate",
 177             "Database record must contain 'lcmOperationType=instantiate'",
 178         )
 179
 180         self.assertEqual(
 181             len(rollback),
 182             len(self.db.set_one.call_args_list) + 1,
 183             "rollback mismatch with created/set items at database",
 184         )
 185
 186         operation_params = created_nslcmop["operationParams"]
 187         self.check_operation_params(operation_params)
 188         self.assertEqual(self.vim_id, operation_params["vimAccountId"])
 189         self.assertNotIn("paasAccountId", operation_params)
 190
 191         self.assertEqual(len(self.db.set_one.call_args_list), self.number_of_vnfr)
 192
 193         for call in self.db.set_one.call_args_list:
 194             topic, vnfr_id, update = call[0]
 195             self.assertEqual(topic, "vnfrs")
 196             self.assertIn(vnfr_id["_id"], self.vnfrs)
 197             self.assertEqual(self.vim_id, update["vim-account-id"])
 198             self.assertNotIn("paas-account-id", update)
 199
 200         # test parameters with error
 201         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
 202         test_set = (
 203             (
 204                 "nsr not found",
 205                 {"nsInstanceId": bad_id},
 206                 DbException,
 207                 HTTPStatus.NOT_FOUND,
 208                 ("not found", bad_id),
 209             ),
 210             # TODO add "nsd"
 211             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
 212             (
 213                 "bad member-vnf-index",
 214                 {"vnf.0.member-vnf-index": "k"},
 215                 EngineException,
 216                 HTTPStatus.BAD_REQUEST,
 217                 ("k",),
 218             ),
 219         )
 220         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
 221             with self.assertRaises(expect_exc, msg=message) as e:
 222                 self.nslcmop_topic.new(
 223                     rollback,
 224                     self.new_session,
 225                     indata=deepcopy(indata),
 226                     kwargs=kwargs_,
 227                     headers=headers,
 228                 )
 229             if expect_code:
 230                 self.assertTrue(e.exception.http_code == expect_code)
 231             if expect_text_list:
 232                 for expect_text in expect_text_list:
 233                     self.assertIn(
 234                         expect_text,
 235                         str(e.exception).lower(),
 236                         "Expected '{}' at exception text".format(expect_text),
 237                     )
 238
 239     def test_create_instantiate_with_paas_account(self):
 240         self.db.set_one = Mock(return_value={"updated": 1})
 241         indata = {
 242             "nsdId": self.nsd_id,
 243             "nsInstanceId": self.nsr_id,
 244             "nsName": "name",
 245             "paasAccountId": self.paas_id,
 246             "lcmOperationType": "instantiate",
 247         }
 248
 249         nslcmop_id, _ = self.nslcmop_topic.new(
 250             [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={}
 251         )
 252
 253         # check nslcmop is created at database
 254         self.assertEqual(
 255             self.db.create.call_count,
 256             1,
 257             "database create not called, or called more than once",
 258         )
 259         _call = self.db.create.call_args_list[0]
 260         self.assertEqual(
 261             _call[0][0], "nslcmops", "must be create a nslcmops entry at database"
 262         )
 263
 264         created_nslcmop = _call[0][1]
 265         self.assertEqual(
 266             nslcmop_id,
 267             created_nslcmop["_id"],
 268             "mismatch between return id and database '_id'",
 269         )
 270
 271         operation_params = created_nslcmop["operationParams"]
 272         self.check_operation_params(operation_params)
 273         self.assertEqual(self.paas_id, operation_params["paasAccountId"])
 274         self.assertNotIn("vimAccountId", operation_params)
 275
 276         self.assertEqual(len(self.db.set_one.call_args_list), self.number_of_vnfr)
 277
 278         for call in self.db.set_one.call_args_list:
 279             topic, vnfr_id, update = call[0]
 280             self.assertEqual(topic, "vnfrs")
 281             self.assertIn(vnfr_id["_id"], self.vnfrs)
 282             self.assertEqual(self.paas_id, update["paas-account-id"])
 283             self.assertNotIn("vim-account-id", update)
 284
 285     def test_create_instantiate_invalid_paas_account_raises_exception(self):
 286         self.db.set_one = Mock(return_value={"updated": 1})
 287         invalid_paas_id = "88d90b0c-faff-4b9f-bccd-017f33985984"
 288         indata = {
 289             "nsdId": self.nsd_id,
 290             "nsInstanceId": self.nsr_id,
 291             "nsName": "name",
 292             "paasAccountId": invalid_paas_id,
 293             "lcmOperationType": "instantiate",
 294         }
 295
 296         with self.assertRaises(EngineException):
 297             nslcmop_id, _ = self.nslcmop_topic.new(
 298                 [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={}
 299             )
 300         self.db.set_one.assert_not_called()
 301
 302     def test_create_instantiate_with_paas_account_in_vnf(self):
 303         self.db.set_one = Mock(return_value={"updated": 1})
 304         indata = {
 305             "nsdId": self.nsd_id,
 306             "nsInstanceId": self.nsr_id,
 307             "nsName": "name",
 308             "paasAccountId": self.paas_ids[0],
 309             "lcmOperationType": "instantiate",
 310             "vnf": [{"member-vnf-index": "1", "paasAccountId": self.paas_ids[1]}],
 311         }
 312
 313         nslcmop_id, _ = self.nslcmop_topic.new(
 314             [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={}
 315         )
 316
 317         # check nslcmop is created at database
 318         self.assertEqual(
 319             self.db.create.call_count,
 320             1,
 321             "database create not called, or called more than once",
 322         )
 323         _call = self.db.create.call_args_list[0]
 324         self.assertEqual(
 325             _call[0][0], "nslcmops", "must be create a nslcmops entry at database"
 326         )
 327
 328         created_nslcmop = _call[0][1]
 329         self.assertEqual(
 330             nslcmop_id,
 331             created_nslcmop["_id"],
 332             "mismatch between return id and database '_id'",
 333         )
 334
 335         operation_params = created_nslcmop["operationParams"]
 336         self.check_operation_params(operation_params)
 337         self.assertEqual(self.paas_id, operation_params["paasAccountId"])
 338         self.assertNotIn("vimAccountId", operation_params)
 339         expected_paas_id = ""
 340
 341         self.assertEqual(len(self.db.set_one.call_args_list), self.number_of_vnfr)
 342         for call in self.db.set_one.call_args_list:
 343             topic, vnfr_id, update = call[0]
 344             self.assertEqual(topic, "vnfrs")
 345             self.assertIn(vnfr_id["_id"], self.vnfrs)
 346             vnf_index_ref = self.vnfrs[vnfr_id["_id"]]
 347             if vnf_index_ref == "1":
 348                 expected_paas_id = self.paas_ids[1]
 349             elif vnf_index_ref == "2":
 350                 expected_paas_id = self.paas_ids[0]
 351             else:
 352                 assert False
 353             self.assertEqual(expected_paas_id, update["paas-account-id"])
 354             self.assertNotIn("vim-account-id", update)
 355
 356     def test_create_instantiate_invalid_paas_account_in_vnf_raises_exception(self):
 357         self.db.set_one = Mock(return_value={"updated": 1})
 358         invalid_paas_id = "88d90b0c-faff-4b9f-bccd-017f33985984"
 359         indata = {
 360             "nsdId": self.nsd_id,
 361             "nsInstanceId": self.nsr_id,
 362             "nsName": "name",
 363             "paasAccountId": self.paas_ids[0],
 364             "lcmOperationType": "instantiate",
 365             "vnf": [{"member-vnf-index": "1", "paasAccountId": invalid_paas_id}],
 366         }
 367
 368         with self.assertRaises(EngineException):
 369             self.nslcmop_topic.new(
 370                 [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={}
 371             )
 372         self.db.set_one.assert_not_called()
 373
 374     def test_check_ns_operation_action(self):
 375         nsrs = self.db.get_list("nsrs")[0]
 376         session = {}
 377
 378         indata = {
 379             "member_vnf_index": "1",
 380             "vdu_id": None,
 381             "primitive": "touch",
 382             "primitive_params": {"filename": "file"},
 383         }
 384
 385         self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata)
 386         for k in indata:
 387             indata_copy = indata.copy()
 388             if k == "primitive_params":
 389                 continue
 390             indata_copy[k] = "non_existing"
 391             with self.assertRaises(EngineException) as exc_manager:
 392                 self.nslcmop_topic._check_ns_operation(
 393                     session, nsrs, "action", indata_copy
 394                 )
 395             exc = exc_manager.exception
 396             self.assertEqual(
 397                 exc.http_code,
 398                 HTTPStatus.BAD_REQUEST,
 399                 "Engine exception bad http_code with {}".format(indata_copy),
 400             )
 401
 402     def test_update_remove_vnf(self):
 403         vnfr_id = self.db.get_list("vnfrs")[0]["_id"]
 404         session = {}
 405         self.db.set_one(
 406             "nsrs", {"_id": self.nsr_id}, {"_admin.nsState": "INSTANTIATED"}
 407         )
 408         indata = {
 409             "lcmOperationType": "update",
 410             "updateType": "REMOVE_VNF",
 411             "nsInstanceId": self.nsr_id,
 412             "removeVnfInstanceId": vnfr_id,
 413         }
 414
 415         session = {
 416             "force": False,
 417             "admin": False,
 418             "public": False,
 419             "project_id": [self.nsr_project],
 420             "method": "write",
 421         }
 422         rollback = []
 423         headers = {}
 424
 425         nslcmop_id, _ = self.nslcmop_topic.new(
 426             rollback, session, indata, kwargs=None, headers=headers
 427         )
 428
 429         self.assertEqual(
 430             self.db.create.call_count,
 431             1,
 432             "database create not called, or called more than once",
 433         )
 434         _call = self.db.create.call_args_list[0]
 435         self.assertEqual(
 436             _call[0][0], "nslcmops", "nslcmops entry must be created at database"
 437         )
 438         created_nslcmop = _call[0][1]
 439         self.assertEqual(
 440             self.nsr_id,
 441             created_nslcmop["nsInstanceId"],
 442             "mismatch between nsId '_id' in created nslcmop and database nsr",
 443         )
 444         self.assertTrue(
 445             created_nslcmop["lcmOperationType"] == "update",
 446             "Database record must contain 'lcmOperationType=update'",
 447         )
 448         self.assertTrue(
 449             created_nslcmop["operationParams"]["updateType"] == "REMOVE_VNF",
 450             "Database record must contain 'updateType=REMOVE_VNF'",
 451         )
 452
 453     def test_migrate(self):
 454         session = {}
 455         self.db.set_one(
 456             "nsrs", {"_id": self.nsr_id}, {"_admin.nsState": "INSTANTIATED"}
 457         )
 458         session = {
 459             "force": False,
 460             "admin": False,
 461             "public": False,
 462             "project_id": [self.nsr_project],
 463             "method": "write",
 464         }
 465         rollback = []
 466         headers = {}
 467
 468         with self.subTest(i=1, t="Migration for Specific VM"):
 469             indata = {
 470                 "lcmOperationType": "migrate",
 471                 "nsInstanceId": self.nsr_id,
 472                 "migrateToHost": "sample02",
 473                 "vdu": {"vduCountIndex": 0, "vduId": "mgmtVM"},
 474                 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
 475             }
 476             nslcmop_id, _ = self.nslcmop_topic.new(
 477                 rollback, session, indata, kwargs=None, headers=headers
 478             )
 479
 480             self.assertEqual(
 481                 self.db.create.call_count,
 482                 1,
 483                 "database create not called, or called more than once",
 484             )
 485             _call = self.db.create.call_args_list[0]
 486             self.assertEqual(
 487                 _call[0][0], "nslcmops", "nslcmops entry must be created at database"
 488             )
 489             created_nslcmop = _call[0][1]
 490             self.assertEqual(
 491                 self.nsr_id,
 492                 created_nslcmop["nsInstanceId"],
 493                 "mismatch between nsId '_id' in created nslcmop and database nsr",
 494             )
 495             self.assertTrue(
 496                 created_nslcmop["lcmOperationType"] == "migrate",
 497                 "Database record must contain 'lcmOperationType=migrate'",
 498             )
 499         with self.subTest(i=2, t="Migration of all VDUs in a VNF"):
 500             indata = {
 501                 "lcmOperationType": "migrate",
 502                 "nsInstanceId": self.nsr_id,
 503                 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
 504             }
 505             nslcmop_id, _ = self.nslcmop_topic.new(
 506                 rollback, session, indata, kwargs=None, headers=headers
 507             )
 508
 509             self.assertEqual(
 510                 self.db.create.call_count,
 511                 2,
 512                 "database create not called, or called more than once",
 513             )
 514             _call = self.db.create.call_args_list[0]
 515             self.assertEqual(
 516                 _call[0][0], "nslcmops", "nslcmops entry must be created at database"
 517             )
 518             created_nslcmop = _call[0][1]
 519             self.assertEqual(
 520                 self.nsr_id,
 521                 created_nslcmop["nsInstanceId"],
 522                 "mismatch between nsId '_id' in created nslcmop and database nsr",
 523             )
 524             self.assertTrue(
 525                 created_nslcmop["lcmOperationType"] == "migrate",
 526                 "Database record must contain 'lcmOperationType=migrate'",
 527             )
 528         with self.subTest(i=3, t="Migration failure - vduId not provided in vdu "):
 529             indata = {
 530                 "lcmOperationType": "migrate",
 531                 "nsInstanceId": self.nsr_id,
 532                 "migrateToHost": "sample02",
 533                 "vdu": {"vduCountIndex": 0},
 534                 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
 535             }
 536
 537             with self.assertRaises(Exception) as e:
 538                 nslcmop_id, _ = self.nslcmop_topic.new(
 539                     rollback, session, indata, kwargs=None, headers=headers
 540                 )
 541             self.assertTrue(
 542                 "Format error at 'vdu' ''vduId' is a required property'"
 543                 in str(e.exception)
 544             )
 545
 546
 547 class TestNsLcmOpTopicWithMock(unittest.TestCase):
 548     def setUp(self):
 549         self.db = Mock(dbbase.DbBase())
 550         self.fs = Mock(FsBase())
 551         self.fs.get_params.return_value = {"./fake/folder"}
 552         self.fs.file_open = mock_open()
 553         self.msg = Mock(MsgBase())
 554         # create class
 555         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
 556
 557     def test_get_vnfd_from_vnf_member_revision(self):
 558         test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
 559         test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
 560         self.db.get_one.side_effect = [test_vnfr, test_vnfd]
 561         self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"])
 562         self.assertEqual(
 563             self.db.get_one.call_args_list[0][0][0],
 564             "vnfrs",
 565             "Incorrect first DB lookup",
 566         )
 567         self.assertEqual(
 568             self.db.get_one.call_args_list[1][0][0],
 569             "vnfds",
 570             "Incorrect second DB lookup",
 571         )
 572
 573     def test_get_vnfd_from_vnf_member_no_revision(self):
 574         test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
 575         test_vnfr["revision"] = 3
 576         test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
 577         self.db.get_one.side_effect = [test_vnfr, test_vnfd]
 578         self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"])
 579         self.assertEqual(
 580             self.db.get_one.call_args_list[0][0][0],
 581             "vnfrs",
 582             "Incorrect first DB lookup",
 583         )
 584         self.assertEqual(
 585             self.db.get_one.call_args_list[1][0][0],
 586             "vnfds_revisions",
 587             "Incorrect second DB lookup",
 588         )
 589
 590     @contextmanager
 591     def assertNotRaises(self, exception_type):
 592         try:
 593             yield None
 594         except exception_type:
 595             raise self.failureException("{} raised".format(exception_type.__name__))
 596
 597     def test_check_ns_update_operation(self):
 598         self.db = DbMemory()
 599         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
 600         session = {}
 601
 602         with self.subTest(i=1, t="VNF instance does not belong to NS"):
 603             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 604             test_vnfr[0]["revision"] = 2
 605             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 606             test_nsr[0]["constituent-vnfr-ref"][
 607                 0
 608             ] = "99d90b0c-faff-4b9f-bccd-017f33985984"
 609             self.db.create_list("vnfrs", test_vnfr)
 610             self.db.create_list("nsrs", test_nsr)
 611             nsrs = self.db.get_list("nsrs")[0]
 612             indata = {
 613                 "updateType": "CHANGE_VNFPKG",
 614                 "changeVnfPackageData": {
 615                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 616                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 617                 },
 618                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 619             }
 620             with self.assertRaises(EngineException) as expected_exception:
 621                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 622             self.assertEqual(
 623                 str(expected_exception.exception),
 624                 "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984"
 625                 " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85",
 626             )
 627
 628         with self.subTest(i=2, t="Ns update request validated with no exception"):
 629             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 630             test_vnfr[0]["revision"] = 2
 631             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 632             self.db.create_list("vnfrs", test_vnfr)
 633             self.db.create_list("nsrs", test_nsr)
 634             nsrs = self.db.get_list("nsrs")[1]
 635             indata = {
 636                 "updateType": "CHANGE_VNFPKG",
 637                 "changeVnfPackageData": {
 638                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 639                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 640                 },
 641                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 642             }
 643             with self.assertNotRaises(EngineException):
 644                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 645
 646         with self.subTest(
 647             i=3, t="Ns update request rejected because of too small timeout"
 648         ):
 649             indata = {
 650                 "updateType": "CHANGE_VNFPKG",
 651                 "changeVnfPackageData": {
 652                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 653                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 654                 },
 655                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 656                 "timeout_ns_update": 50,
 657             }
 658             with self.assertRaises(EngineException) as expected_exception:
 659                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 660             self.assertEqual(
 661                 str(expected_exception.exception),
 662                 "Error in validating ns-update request: 50 second is not enough "
 663                 "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
 664             )
 665
 666         with self.subTest(i=4, t="wrong vnfdid is given as an update parameter"):
 667             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 668             test_vnfr[0]["revision"] = 2
 669             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 670             self.db.create_list("vnfrs", test_vnfr)
 671             self.db.create_list("nsrs", test_nsr)
 672             nsrs = self.db.get_list("nsrs")[2]
 673             indata = {
 674                 "updateType": "CHANGE_VNFPKG",
 675                 "changeVnfPackageData": {
 676                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 677                     "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 678                 },
 679                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 680             }
 681             with self.assertRaises(EngineException) as expected_exception:
 682                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 683             self.assertEqual(
 684                 str(expected_exception.exception),
 685                 "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not "
 686                 "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of "
 687                 "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
 688             )
 689
 690         with self.subTest(
 691             i=5, t="Ns update REMOVE_VNF request validated with no exception"
 692         ):
 693             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 694             test_vnfr[0]["revision"] = 2
 695             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 696             self.db.create_list("vnfrs", test_vnfr)
 697             self.db.create_list("nsrs", test_nsr)
 698             nsrs = self.db.get_list("nsrs")[1]
 699             indata = {
 700                 "updateType": "REMOVE_VNF",
 701                 "removeVnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 702                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 703             }
 704             with self.assertNotRaises(EngineException):
 705                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 706
 707
 708 class TestNsrTopic(unittest.TestCase):
 709     def setUp(self):
 710         self.db = DbMemory()
 711         self.fs = Mock(FsBase())
 712         self.fs.get_params.return_value = {"./fake/folder"}
 713         self.fs.file_open = mock_open()
 714         self.msg = Mock(MsgBase())
 715         # create class
 716         self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None)
 717         self.nsr_topic.check_quota = Mock(return_value=None)  # skip quota
 718
 719         self.db.create_list(
 720             "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
 721         )
 722         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
 723         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
 724         self.db.create = Mock(return_value="created_id")
 725         self.nsd = self.db.get_list("nsds")[0]
 726         self.nsd_id = self.nsd["_id"]
 727         self.nsd_project = self.nsd["_admin"]["projects_read"][0]
 728
 729         self.vim = self.db.get_list("vim_accounts")[0]
 730         self.vim_id = self.vim["_id"]
 731
 732         self.paas_id = "3d93ca4e-b007-4d94-bbdc-61911078b864"
 733
 734     def stock_db_creations(self, created_vnfrs, created_nsrs):
 735         for _call in self.db.create.call_args_list:
 736             assert len(_call[0]) >= 2, "called db.create with few parameters"
 737             created_item = _call[0][1]
 738             if _call[0][0] == "vnfrs":
 739                 created_vnfrs.append(created_item)
 740             elif _call[0][0] == "nsrs":
 741                 created_nsrs.append(created_item)
 742             else:
 743                 assert False, "created an unknown record {} at database".format(
 744                     _call[0][0]
 745                 )
 746             self.check_admin_field_in_db_call(created_item)
 747
 748     def check_admin_field_in_db_call(self, created_item):
 749         self.assertTrue(
 750             created_item["_admin"].get("projects_read"),
 751             "Database record must contain '_amdin.projects_read'",
 752         )
 753         self.assertIn(
 754             "created",
 755             created_item["_admin"],
 756             "Database record must contain '_admin.created'",
 757         )
 758         self.assertTrue(
 759             created_item["_admin"]["nsState"] == "NOT_INSTANTIATED",
 760             "Database record must contain '_admin.nstate=NOT INSTANTIATE'",
 761         )
 762
 763     def check_vnfr_content(self, vnfr, nsr_id):
 764         self.assertEqual(vnfr["vim-account-id"], None)
 765         self.assertEqual(vnfr["paas-account-id"], None)
 766         self.assertIn(
 767             "member-vnf-index-ref",
 768             vnfr,
 769             "Created item must contain member-vnf-index-ref section",
 770         )
 771         self.assertEqual(
 772             nsr_id, vnfr["nsr-id-ref"], "bad reference id from vnfr to nsr"
 773         )
 774         self.check_vdur_interfaces(vnfr)
 775
 776     def check_vdur_interfaces(self, vnfr):
 777         self.assertEqual(
 778             vnfr["vdur"][0]["interfaces"][0]["position"],
 779             1,
 780             "vdur first interface position does not match",
 781         )
 782         self.assertEqual(
 783             vnfr["vdur"][0]["interfaces"][1]["position"],
 784             2,
 785             "vdur second interface position does not match",
 786         )
 787
 788     def check_vnfrs(self, created_vnfrs, nsrs_id):
 789         self.assertEqual(
 790             len(created_vnfrs), 2, "created a mismatch number of vnfr at database"
 791         )
 792         for vnfr in created_vnfrs:
 793             self.check_vnfr_content(vnfr, nsrs_id)
 794
 795     def check_vnfrs_ref_in_nsr(self, nsrs, created_vnfrs):
 796         self.assertEqual(len(nsrs["constituent-vnfr-ref"]), len(created_vnfrs))
 797         self.assertNotEqual(created_vnfrs[0]["_id"], created_vnfrs[1]["_id"])
 798         for vnfr in created_vnfrs:
 799             vnfr_id = vnfr["_id"]
 800             self.assertIn(vnfr_id, nsrs["constituent-vnfr-ref"])
 801
 802     def check_nsrs(self, created_nsrs, created_vnfrs, vim_id, paas_id):
 803         self.assertEqual(
 804             len(created_nsrs), 1, "Only one nsrs must be created at database"
 805         )
 806         nsrs = created_nsrs[0]
 807         self.assertEqual(nsrs["vimdatacenter"], vim_id)
 808         self.assertEqual(nsrs["paasdatacenter"], paas_id)
 809         self.check_vnfrs_ref_in_nsr(nsrs, created_vnfrs)
 810
 811     def check_created_nsrs_and_vnfrs(
 812         self, created_nsrs, created_vnfrs, rollback, vim_id, paas_id
 813     ):
 814
 815         nsrs_id = created_nsrs[0]["_id"]
 816
 817         self.assertEqual(
 818             len(rollback),
 819             len(created_vnfrs) + 1,
 820             "rollback mismatch with created items at database",
 821         )
 822
 823         self.check_nsrs(created_nsrs, created_vnfrs, vim_id, paas_id)
 824         self.check_vnfrs(created_vnfrs, nsrs_id)
 825
 826     def test_create_with_vim_account(self):
 827         session = {
 828             "force": False,
 829             "admin": False,
 830             "public": False,
 831             "project_id": [self.nsd_project],
 832             "method": "write",
 833         }
 834         indata = {
 835             "nsdId": self.nsd_id,
 836             "nsName": "name",
 837             "vimAccountId": self.vim_id,
 838             "additionalParamsForVnf": [
 839                 {
 840                     "member-vnf-index": "hackfest_vnf1",
 841                     "additionalParams": {"touch_filename": "file"},
 842                 },
 843                 {
 844                     "member-vnf-index": "hackfest_vnf2",
 845                     "additionalParams": {"touch_filename": "file"},
 846                 },
 847             ],
 848         }
 849         rollback = []
 850         headers = {}
 851
 852         self.nsr_topic.new(
 853             rollback, session, indata=indata, kwargs=None, headers=headers
 854         )
 855
 856         # check vnfrs and nsrs created in whatever order
 857         created_vnfrs = []
 858         created_nsrs = []
 859         self.stock_db_creations(created_vnfrs, created_nsrs)
 860         self.check_created_nsrs_and_vnfrs(
 861             created_nsrs, created_vnfrs, rollback, self.vim_id, None
 862         )
 863
 864     def test_create_with_vim_account_raise_exception(self):
 865         # test parameters with error
 866         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
 867         session = {
 868             "force": False,
 869             "admin": False,
 870             "public": False,
 871             "project_id": [self.nsd_project],
 872             "method": "write",
 873         }
 874         indata = {
 875             "nsdId": self.nsd_id,
 876             "nsName": "name",
 877             "vimAccountId": self.vim_id,
 878             "additionalParamsForVnf": [
 879                 {
 880                     "member-vnf-index": "hackfest_vnf1",
 881                     "additionalParams": {"touch_filename": "file"},
 882                 },
 883                 {
 884                     "member-vnf-index": "hackfest_vnf2",
 885                     "additionalParams": {"touch_filename": "file"},
 886                 },
 887             ],
 888         }
 889         rollback = []
 890         headers = {}
 891         test_set = (
 892             # TODO add "nsd"
 893             (
 894                 "nsd not found",
 895                 {"nsdId": bad_id},
 896                 DbException,
 897                 HTTPStatus.NOT_FOUND,
 898                 ("not found", bad_id),
 899             ),
 900             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
 901             (
 902                 "additional params not supply",
 903                 {"additionalParamsForVnf.0.member-vnf-index": "k"},
 904                 EngineException,
 905                 HTTPStatus.BAD_REQUEST,
 906                 None,
 907             ),
 908         )
 909         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
 910             with self.assertRaises(expect_exc, msg=message) as e:
 911                 self.nsr_topic.new(
 912                     rollback,
 913                     session,
 914                     indata=deepcopy(indata),
 915                     kwargs=kwargs_,
 916                     headers=headers,
 917                 )
 918             if expect_code:
 919                 self.assertTrue(e.exception.http_code == expect_code)
 920             if expect_text_list:
 921                 for expect_text in expect_text_list:
 922                     self.assertIn(
 923                         expect_text,
 924                         str(e.exception).lower(),
 925                         "Expected '{}' at exception text".format(expect_text),
 926                     )
 927
 928     def test_create_with_paas_account(self):
 929         session = {
 930             "force": False,
 931             "admin": False,
 932             "public": False,
 933             "project_id": [self.nsd_project],
 934             "method": "write",
 935         }
 936         indata = {
 937             "nsdId": self.nsd_id,
 938             "nsName": "name",
 939             "paasAccountId": self.paas_id,
 940             "additionalParamsForVnf": [
 941                 {
 942                     "member-vnf-index": "hackfest_vnf1",
 943                     "additionalParams": {"touch_filename": "file"},
 944                 },
 945                 {
 946                     "member-vnf-index": "hackfest_vnf2",
 947                     "additionalParams": {"touch_filename": "file"},
 948                 },
 949             ],
 950         }
 951         rollback = []
 952         headers = {}
 953
 954         self.nsr_topic.new(
 955             rollback, session, indata=indata, kwargs=None, headers=headers
 956         )
 957
 958         created_vnfrs = []
 959         created_nsrs = []
 960         self.stock_db_creations(created_vnfrs, created_nsrs)
 961         self.check_created_nsrs_and_vnfrs(
 962             created_nsrs, created_vnfrs, rollback, None, self.paas_id
 963         )
 964
 965     def test_show_instance(self):
 966         session = {
 967             "force": False,
 968             "admin": False,
 969             "public": False,
 970             "project_id": [self.nsd_project],
 971             "method": "write",
 972         }
 973         filter_q = {}
 974         for refresh_status in ("true", "false"):
 975             self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 976             actual_nsr = self.db.get_list("nsrs")[0]
 977             nsr_id = actual_nsr["_id"]
 978             filter_q["vcaStatus-refresh"] = refresh_status
 979             expected_nsr = self.nsr_topic.show(session, nsr_id, filter_q=filter_q)
 980             self.nsr_topic.delete(session, nsr_id)
 981             actual_nsr.pop("_admin")
 982             expected_nsr.pop("_admin")
 983             self.assertEqual(
 984                 expected_nsr, actual_nsr, "Database nsr and show() nsr do not match."
 985             )
 986
 987     def test_vca_status_refresh(self):
 988         session = {
 989             "force": False,
 990             "admin": False,
 991             "public": False,
 992             "project_id": [self.nsd_project],
 993             "method": "write",
 994         }
 995         filter_q = {"vcaStatus-refresh": "true"}
 996         time_delta = 120
 997         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 998         nsr = self.db.get_list("nsrs")[0]
 999
1000         # When vcaStatus-refresh is true
1001         filter_q["vcaStatus-refresh"] = "true"
1002         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
1003         msg_args = self.msg.write.call_args[0]
1004         self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
1005         self.assertGreater(nsr["_admin"]["modified"], time() - time_delta)
1006
1007         # When vcaStatus-refresh is false but modified time is within threshold
1008         filter_q["vcaStatus-refresh"] = "false"
1009         time_now = time()
1010         nsr["_admin"]["modified"] = time_now
1011         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
1012         msg_args = self.msg.write.call_args[1]
1013         self.assertEqual(msg_args, {}, "Message should not be sent.")
1014         self.assertEqual(
1015             nsr["_admin"]["modified"], time_now, "Modified time should not be changed."
1016         )
1017
1018         # When vcaStatus-refresh is false but modified time is less than threshold
1019         filter_q["vcaStatus-refresh"] = "false"
1020         nsr["_admin"]["modified"] = time() - (2 * time_delta)
1021         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
1022         msg_args = self.msg.write.call_args[0]
1023         self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
1024         self.nsr_topic.delete(session, nsr["_id"])
1025         self.assertGreater(
1026             nsr["_admin"]["modified"],
1027             time() - time_delta,
1028             "Modified time is not changed.",
1029         )
1030
1031     def test_delete_ns(self):
1032         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
1033         self.nsr = self.db.get_list("nsrs")[0]
1034         self.nsr_id = self.nsr["_id"]
1035         self.db_set_one = self.db.set_one
1036         p_id = self.nsd_project
1037         p_other = "other_p"
1038
1039         session = {
1040             "force": False,
1041             "admin": False,
1042             "public": None,
1043             "project_id": [p_id],
1044             "method": "delete",
1045         }
1046         session2 = {
1047             "force": False,
1048             "admin": False,
1049             "public": None,
1050             "project_id": [p_other],
1051             "method": "delete",
1052         }
1053         session_force = {
1054             "force": True,
1055             "admin": True,
1056             "public": None,
1057             "project_id": [],
1058             "method": "delete",
1059         }
1060         with self.subTest(i=1, t="Normal Deletion"):
1061             self.db.del_one = Mock()
1062             self.db.set_one = Mock()
1063             self.nsr_topic.delete(session, self.nsr_id)
1064
1065             db_args_ro_nsrs = self.db.del_one.call_args_list[1][0]
1066             db_args = self.db.del_one.call_args_list[0][0]
1067             msg_args = self.msg.write.call_args[0]
1068             self.assertEqual(
1069                 msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic"
1070             )
1071             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
1072             self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
1073             self.assertEqual(db_args_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
1074             self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
1075             self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
1076             self.assertEqual(
1077                 db_args[1]["_admin.projects_read.cont"], [p_id], "Wrong DB filter"
1078             )
1079             self.db.set_one.assert_not_called()
1080             fs_del_calls = self.fs.file_delete.call_args_list
1081             self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
1082         with self.subTest(i=2, t="No delete because referenced by other project"):
1083             self.db_set_one(
1084                 "nsrs",
1085                 {"_id": self.nsr_id},
1086                 update_dict=None,
1087                 push={
1088                     "_admin.projects_read": p_other,
1089                     "_admin.projects_write": p_other,
1090                 },
1091             )
1092             self.db.del_one.reset_mock()
1093             self.db.set_one.reset_mock()
1094             self.msg.write.reset_mock()
1095             self.fs.file_delete.reset_mock()
1096
1097             self.nsr_topic.delete(session2, self.nsr_id)
1098             self.db.del_one.assert_not_called()
1099             self.msg.write.assert_not_called()
1100             db_s1_args = self.db.set_one.call_args
1101             self.assertEqual(db_s1_args[0][0], self.nsr_topic.topic, "Wrong DB topic")
1102             self.assertEqual(db_s1_args[0][1]["_id"], self.nsr_id, "Wrong DB ID")
1103             self.assertIsNone(
1104                 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
1105             )
1106             self.assertEqual(
1107                 db_s1_args[1]["pull_list"],
1108                 {"_admin.projects_read": [p_other], "_admin.projects_write": [p_other]},
1109                 "Wrong DB pull_list dictionary",
1110             )
1111             self.fs.file_delete.assert_not_called()
1112         with self.subTest(i=4, t="Delete with force and admin"):
1113             self.db.del_one.reset_mock()
1114             self.db.set_one.reset_mock()
1115             self.msg.write.reset_mock()
1116             self.fs.file_delete.reset_mock()
1117             self.nsr_topic.delete(session_force, self.nsr_id)
1118
1119             db_args_ro_nsrs = self.db.del_one.call_args_list[1][0]
1120             db_args = self.db.del_one.call_args_list[0][0]
1121             msg_args = self.msg.write.call_args[0]
1122             self.assertEqual(
1123                 msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic"
1124             )
1125             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
1126             self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
1127             self.assertEqual(db_args_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
1128             self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
1129             self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
1130             self.db.set_one.assert_not_called()
1131             fs_del_calls = self.fs.file_delete.call_args_list
1132             self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
1133         with self.subTest(i=3, t="Conflict on Delete - NS in INSTANTIATED state"):
1134             self.db_set_one(
1135                 "nsrs",
1136                 {"_id": self.nsr_id},
1137                 {"_admin.nsState": "INSTANTIATED"},
1138                 pull={
1139                     "_admin.projects_read": p_other,
1140                     "_admin.projects_write": p_other,
1141                 },
1142             )
1143             self.db.del_one.reset_mock()
1144             self.db.set_one.reset_mock()
1145             self.msg.write.reset_mock()
1146             self.fs.file_delete.reset_mock()
1147
1148             with self.assertRaises(
1149                 EngineException, msg="Accepted NSR with nsState INSTANTIATED"
1150             ) as e:
1151                 self.nsr_topic.delete(session, self.nsr_id)
1152             self.assertEqual(
1153                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1154             )
1155             self.assertIn("INSTANTIATED", str(e.exception), "Wrong exception text")
1156         # TODOD with self.subTest(i=3, t='Conflict on Delete - NS in use by NSI'):
1157
1158         with self.subTest(i=4, t="Non-existent NS"):
1159             self.db.del_one.reset_mock()
1160             self.db.set_one.reset_mock()
1161             self.msg.write.reset_mock()
1162             self.fs.file_delete.reset_mock()
1163             excp_msg = "Not found"
1164             with self.assertRaises(
1165                 DbException, msg="Accepted non-existent NSD ID"
1166             ) as e:
1167                 self.nsr_topic.delete(session2, "other_id")
1168             self.assertEqual(
1169                 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
1170             )
1171             self.assertIn(excp_msg, str(e.exception), "Wrong exception text")
1172             self.assertIn("other_id", str(e.exception), "Wrong exception text")
1173         return