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.db.set_one = Mock(return_value={"updated": 1})
  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
  70     def test_create_instantiate(self):
  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
 232 class TestNsLcmOpTopicWithMock(unittest.TestCase):
 233     def setUp(self):
 234         self.db = Mock(dbbase.DbBase())
 235         self.fs = Mock(FsBase())
 236         self.fs.get_params.return_value = {"./fake/folder"}
 237         self.fs.file_open = mock_open()
 238         self.msg = Mock(MsgBase())
 239         # create class
 240         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
 241
 242     def test_get_vnfd_from_vnf_member_revision(self):
 243         test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
 244         test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
 245         self.db.get_one.side_effect = [test_vnfr, test_vnfd]
 246         vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id'])
 247         self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup")
 248         self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds', "Incorrect second DB lookup")
 249
 250     def test_get_vnfd_from_vnf_member_no_revision(self):
 251         test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0]
 252         test_vnfr['revision'] = 3
 253         test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader)
 254         self.db.get_one.side_effect = [test_vnfr, test_vnfd]
 255         vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id'])
 256         self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup")
 257         self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds_revisions', "Incorrect second DB lookup")
 258
 259     @contextmanager
 260     def assertNotRaises(self, exception_type):
 261         try:
 262             yield None
 263         except exception_type:
 264             raise self.failureException("{} raised".format(exception_type.__name__))
 265
 266     def test_check_ns_update_operation(self):
 267         self.db = DbMemory()
 268         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
 269         session = {}
 270
 271         with self.subTest(i=1, t="VNF instance does not belong to NS"):
 272             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 273             test_vnfr[0]["revision"] = 2
 274             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 275             test_nsr[0]["constituent-vnfr-ref"][
 276                 0
 277             ] = "99d90b0c-faff-4b9f-bccd-017f33985984"
 278             self.db.create_list("vnfrs", test_vnfr)
 279             self.db.create_list("nsrs", test_nsr)
 280             nsrs = self.db.get_list("nsrs")[0]
 281             indata = {
 282                 "updateType": "CHANGE_VNFPKG",
 283                 "changeVnfPackageData": {
 284                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 285                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 286                 },
 287                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 288             }
 289             with self.assertRaises(EngineException) as expected_exception:
 290                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 291             self.assertEqual(
 292                 str(expected_exception.exception),
 293                 "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984"
 294                 " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85",
 295             )
 296
 297         with self.subTest(i=2, t="Ns update request validated with no exception"):
 298             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 299             test_vnfr[0]["revision"] = 2
 300             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 301             self.db.create_list("vnfrs", test_vnfr)
 302             self.db.create_list("nsrs", test_nsr)
 303             nsrs = self.db.get_list("nsrs")[1]
 304             indata = {
 305                 "updateType": "CHANGE_VNFPKG",
 306                 "changeVnfPackageData": {
 307                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 308                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 309                 },
 310                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 311             }
 312             with self.assertNotRaises(EngineException):
 313                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 314
 315         with self.subTest(
 316             i=3, t="Ns update request rejected because of too small timeout"
 317         ):
 318             indata = {
 319                 "updateType": "CHANGE_VNFPKG",
 320                 "changeVnfPackageData": {
 321                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 322                     "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 323                 },
 324                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 325                 "timeout_ns_update": 50,
 326             }
 327             with self.assertRaises(EngineException) as expected_exception:
 328                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 329             self.assertEqual(
 330                 str(expected_exception.exception),
 331                 "Error in validating ns-update request: 50 second is not enough "
 332                 "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
 333             )
 334
 335         with self.subTest(i=4, t="wrong vnfdid is given as an update parameter"):
 336             test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)
 337             test_vnfr[0]["revision"] = 2
 338             test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader)
 339             self.db.create_list("vnfrs", test_vnfr)
 340             self.db.create_list("nsrs", test_nsr)
 341             nsrs = self.db.get_list("nsrs")[2]
 342             indata = {
 343                 "updateType": "CHANGE_VNFPKG",
 344                 "changeVnfPackageData": {
 345                     "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
 346                     "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
 347                 },
 348                 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
 349             }
 350             with self.assertRaises(EngineException) as expected_exception:
 351                 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata)
 352             self.assertEqual(
 353                 str(expected_exception.exception),
 354                 "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not "
 355                 "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of "
 356                 "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
 357             )
 358
 359
 360 class TestNsrTopic(unittest.TestCase):
 361     def setUp(self):
 362         self.db = DbMemory()
 363         self.fs = Mock(FsBase())
 364         self.fs.get_params.return_value = {"./fake/folder"}
 365         self.fs.file_open = mock_open()
 366         self.msg = Mock(MsgBase())
 367         # create class
 368         self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None)
 369         self.nsr_topic.check_quota = Mock(return_value=None)  # skip quota
 370
 371         self.db.create_list(
 372             "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
 373         )
 374         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
 375         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
 376         self.db.create = Mock(return_value="created_id")
 377         self.nsd = self.db.get_list("nsds")[0]
 378         self.nsd_id = self.nsd["_id"]
 379         self.nsd_project = self.nsd["_admin"]["projects_read"][0]
 380
 381         self.vim = self.db.get_list("vim_accounts")[0]
 382         self.vim_id = self.vim["_id"]
 383
 384     def test_create(self):
 385         session = {
 386             "force": False,
 387             "admin": False,
 388             "public": False,
 389             "project_id": [self.nsd_project],
 390             "method": "write",
 391         }
 392         indata = {
 393             "nsdId": self.nsd_id,
 394             "nsName": "name",
 395             "vimAccountId": self.vim_id,
 396             "additionalParamsForVnf": [
 397                 {
 398                     "member-vnf-index": "hackfest_vnf1",
 399                     "additionalParams": {"touch_filename": "file"},
 400                 },
 401                 {
 402                     "member-vnf-index": "hackfest_vnf2",
 403                     "additionalParams": {"touch_filename": "file"},
 404                 },
 405             ],
 406         }
 407         rollback = []
 408         headers = {}
 409
 410         self.nsr_topic.new(
 411             rollback, session, indata=indata, kwargs=None, headers=headers
 412         )
 413
 414         # check vnfrs and nsrs created in whatever order
 415         created_vnfrs = []
 416         created_nsrs = []
 417         nsr_id = None
 418         for _call in self.db.create.call_args_list:
 419             assert len(_call[0]) >= 2, "called db.create with few parameters"
 420             created_item = _call[0][1]
 421             if _call[0][0] == "vnfrs":
 422                 created_vnfrs.append(created_item)
 423                 self.assertIn(
 424                     "member-vnf-index-ref",
 425                     created_item,
 426                     "Created item must contain member-vnf-index-ref section",
 427                 )
 428                 if nsr_id:
 429                     self.assertEqual(
 430                         nsr_id,
 431                         created_item["nsr-id-ref"],
 432                         "bad reference id from vnfr to nsr",
 433                     )
 434                 else:
 435                     nsr_id = created_item["nsr-id-ref"]
 436
 437             elif _call[0][0] == "nsrs":
 438                 created_nsrs.append(created_item)
 439                 if nsr_id:
 440                     self.assertEqual(
 441                         nsr_id, created_item["_id"], "bad reference id from vnfr to nsr"
 442                     )
 443                 else:
 444                     nsr_id = created_item["_id"]
 445             else:
 446                 assert True, "created an unknown record {} at database".format(
 447                     _call[0][0]
 448                 )
 449
 450             self.assertTrue(
 451                 created_item["_admin"].get("projects_read"),
 452                 "Database record must contain '_amdin.projects_read'",
 453             )
 454             self.assertIn(
 455                 "created",
 456                 created_item["_admin"],
 457                 "Database record must contain '_admin.created'",
 458             )
 459             self.assertTrue(
 460                 created_item["_admin"]["nsState"] == "NOT_INSTANTIATED",
 461                 "Database record must contain '_admin.nstate=NOT INSTANTIATE'",
 462             )
 463
 464         self.assertEqual(
 465             len(created_vnfrs), 2, "created a mismatch number of vnfr at database"
 466         )
 467         self.assertEqual(
 468             len(created_nsrs), 1, "Only one nsrs must be created at database"
 469         )
 470         self.assertEqual(
 471             len(rollback),
 472             len(created_vnfrs) + 1,
 473             "rollback mismatch with created items at database",
 474         )
 475
 476         # test parameters with error
 477         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
 478         test_set = (
 479             # TODO add "nsd"
 480             (
 481                 "nsd not found",
 482                 {"nsdId": bad_id},
 483                 DbException,
 484                 HTTPStatus.NOT_FOUND,
 485                 ("not found", bad_id),
 486             ),
 487             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
 488             (
 489                 "additional params not supply",
 490                 {"additionalParamsForVnf.0.member-vnf-index": "k"},
 491                 EngineException,
 492                 HTTPStatus.BAD_REQUEST,
 493                 None,
 494             ),
 495         )
 496         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
 497             with self.assertRaises(expect_exc, msg=message) as e:
 498                 self.nsr_topic.new(
 499                     rollback,
 500                     session,
 501                     indata=deepcopy(indata),
 502                     kwargs=kwargs_,
 503                     headers=headers,
 504                 )
 505             if expect_code:
 506                 self.assertTrue(e.exception.http_code == expect_code)
 507             if expect_text_list:
 508                 for expect_text in expect_text_list:
 509                     self.assertIn(expect_text, str(e.exception).lower(),
 510                                   "Expected '{}' at exception text".format(expect_text))
 511
 512     def test_show_instance(self):
 513         session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
 514         filter_q = {}
 515         for refresh_status in ("true", "false"):
 516             self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 517             actual_nsr = self.db.get_list("nsrs")[0]
 518             nsr_id = actual_nsr["_id"]
 519             filter_q['vcaStatus-refresh'] = refresh_status
 520             expected_nsr = self.nsr_topic.show(session, nsr_id, filter_q=filter_q)
 521             self.nsr_topic.delete(session, nsr_id)
 522             actual_nsr.pop("_admin")
 523             expected_nsr.pop("_admin")
 524             self.assertEqual(expected_nsr, actual_nsr, "Database nsr and show() nsr do not match.")
 525
 526     def test_vca_status_refresh(self):
 527         session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
 528         filter_q = {'vcaStatus-refresh': 'true'}
 529         time_delta = 120
 530         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 531         nsr = self.db.get_list("nsrs")[0]
 532
 533         # When vcaStatus-refresh is true
 534         filter_q['vcaStatus-refresh'] = "true"
 535         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
 536         msg_args = self.msg.write.call_args[0]
 537         self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
 538         self.assertGreater(nsr["_admin"]["modified"], time() - time_delta)
 539
 540         # When vcaStatus-refresh is false but modified time is within threshold
 541         filter_q['vcaStatus-refresh'] = "false"
 542         time_now = time()
 543         nsr["_admin"]["modified"] = time_now
 544         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
 545         msg_args = self.msg.write.call_args[1]
 546         self.assertEqual(msg_args, {}, "Message should not be sent.")
 547         self.assertEqual(nsr["_admin"]["modified"], time_now, "Modified time should not be changed.")
 548
 549         # When vcaStatus-refresh is false but modified time is less than threshold
 550         filter_q['vcaStatus-refresh'] = "false"
 551         nsr["_admin"]["modified"] = time() - (2*time_delta)
 552         self.nsr_topic.vca_status_refresh(session, nsr, filter_q)
 553         msg_args = self.msg.write.call_args[0]
 554         self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action")
 555         self.nsr_topic.delete(session, nsr["_id"])
 556         self.assertGreater(nsr["_admin"]["modified"], time() - time_delta, "Modified time is not changed.")
 557
 558     def test_delete_ns(self):
 559         self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
 560         self.nsr = self.db.get_list("nsrs")[0]
 561         self.nsr_id = self.nsr["_id"]
 562         self.db_set_one = self.db.set_one
 563         p_id = self.nsd_project
 564         p_other = "other_p"
 565
 566         session = {
 567             "force": False,
 568             "admin": False,
 569             "public": None,
 570             "project_id": [p_id],
 571             "method": "delete",
 572         }
 573         session2 = {
 574             "force": False,
 575             "admin": False,
 576             "public": None,
 577             "project_id": [p_other],
 578             "method": "delete",
 579         }
 580         session_force = {
 581             "force": True,
 582             "admin": True,
 583             "public": None,
 584             "project_id": [],
 585             "method": "delete",
 586         }
 587         with self.subTest(i=1, t="Normal Deletion"):
 588             self.db.del_one = Mock()
 589             self.db.set_one = Mock()
 590             self.nsr_topic.delete(session, self.nsr_id)
 591
 592             db_args_ro_nsrs = self.db.del_one.call_args_list[1][0]
 593             db_args = self.db.del_one.call_args_list[0][0]
 594             msg_args = self.msg.write.call_args[0]
 595             self.assertEqual(
 596                 msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic"
 597             )
 598             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 599             self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
 600             self.assertEqual(db_args_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
 601             self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
 602             self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
 603             self.assertEqual(
 604                 db_args[1]["_admin.projects_read.cont"], [p_id], "Wrong DB filter"
 605             )
 606             self.db.set_one.assert_not_called()
 607             fs_del_calls = self.fs.file_delete.call_args_list
 608             self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
 609         with self.subTest(i=2, t="No delete because referenced by other project"):
 610             self.db_set_one(
 611                 "nsrs",
 612                 {"_id": self.nsr_id},
 613                 update_dict=None,
 614                 push={
 615                     "_admin.projects_read": p_other,
 616                     "_admin.projects_write": p_other,
 617                 },
 618             )
 619             self.db.del_one.reset_mock()
 620             self.db.set_one.reset_mock()
 621             self.msg.write.reset_mock()
 622             self.fs.file_delete.reset_mock()
 623
 624             self.nsr_topic.delete(session2, self.nsr_id)
 625             self.db.del_one.assert_not_called()
 626             self.msg.write.assert_not_called()
 627             db_s1_args = self.db.set_one.call_args
 628             self.assertEqual(db_s1_args[0][0], self.nsr_topic.topic, "Wrong DB topic")
 629             self.assertEqual(db_s1_args[0][1]["_id"], self.nsr_id, "Wrong DB ID")
 630             self.assertIsNone(
 631                 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
 632             )
 633             self.assertEqual(
 634                 db_s1_args[1]["pull_list"],
 635                 {"_admin.projects_read": [p_other], "_admin.projects_write": [p_other]},
 636                 "Wrong DB pull_list dictionary",
 637             )
 638             self.fs.file_delete.assert_not_called()
 639         with self.subTest(i=4, t="Delete with force and admin"):
 640             self.db.del_one.reset_mock()
 641             self.db.set_one.reset_mock()
 642             self.msg.write.reset_mock()
 643             self.fs.file_delete.reset_mock()
 644             self.nsr_topic.delete(session_force, self.nsr_id)
 645
 646             db_args_ro_nsrs = self.db.del_one.call_args_list[1][0]
 647             db_args = self.db.del_one.call_args_list[0][0]
 648             msg_args = self.msg.write.call_args[0]
 649             self.assertEqual(
 650                 msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic"
 651             )
 652             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 653             self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content")
 654             self.assertEqual(db_args_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
 655             self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic")
 656             self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID")
 657             self.db.set_one.assert_not_called()
 658             fs_del_calls = self.fs.file_delete.call_args_list
 659             self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id")
 660         with self.subTest(i=3, t="Conflict on Delete - NS in INSTANTIATED state"):
 661             self.db_set_one(
 662                 "nsrs",
 663                 {"_id": self.nsr_id},
 664                 {"_admin.nsState": "INSTANTIATED"},
 665                 pull={
 666                     "_admin.projects_read": p_other,
 667                     "_admin.projects_write": p_other,
 668                 },
 669             )
 670             self.db.del_one.reset_mock()
 671             self.db.set_one.reset_mock()
 672             self.msg.write.reset_mock()
 673             self.fs.file_delete.reset_mock()
 674
 675             with self.assertRaises(
 676                 EngineException, msg="Accepted NSR with nsState INSTANTIATED"
 677             ) as e:
 678                 self.nsr_topic.delete(session, self.nsr_id)
 679             self.assertEqual(
 680                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
 681             )
 682             self.assertIn("INSTANTIATED", str(e.exception), "Wrong exception text")
 683         # TODOD with self.subTest(i=3, t='Conflict on Delete - NS in use by NSI'):
 684
 685         with self.subTest(i=4, t="Non-existent NS"):
 686             self.db.del_one.reset_mock()
 687             self.db.set_one.reset_mock()
 688             self.msg.write.reset_mock()
 689             self.fs.file_delete.reset_mock()
 690             excp_msg = "Not found"
 691             with self.assertRaises(
 692                 DbException, msg="Accepted non-existent NSD ID"
 693             ) as e:
 694                 self.nsr_topic.delete(session2, "other_id")
 695             self.assertEqual(
 696                 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
 697             )
 698             self.assertIn(excp_msg, str(e.exception), "Wrong exception text")
 699             self.assertIn("other_id", str(e.exception), "Wrong exception text")
 700         return