osm_nbi/tests/test_descriptor_topics.py

   1 #! /usr/bin/python3
   2 # -*- coding: utf-8 -*-
   3
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #    http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  13 # implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17 __author__ = "Pedro de la Cruz Ramos, pedro.delacruzramos@altran.com"
  18 __date__ = "2019-11-20"
  19
  20 import unittest
  21 from unittest import TestCase
  22 from unittest.mock import Mock, patch
  23 from uuid import uuid4
  24 from http import HTTPStatus
  25 from copy import deepcopy
  26 from time import time
  27 from osm_common import dbbase, fsbase, msgbase
  28 from osm_nbi import authconn
  29 from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text
  30 from osm_nbi.descriptor_topics import VnfdTopic, NsdTopic
  31 from osm_nbi.engine import EngineException
  32 from osm_common.dbbase import DbException
  33 import yaml
  34
  35 test_name = "test-user"
  36 db_vnfd_content = yaml.load(db_vnfds_text, Loader=yaml.Loader)[0]
  37 db_nsd_content = yaml.load(db_nsds_text, Loader=yaml.Loader)[0]
  38 test_pid = db_vnfd_content["_admin"]["projects_read"][0]
  39 fake_session = {
  40     "username": test_name,
  41     "project_id": (test_pid,),
  42     "method": None,
  43     "admin": True,
  44     "force": False,
  45     "public": False,
  46     "allow_show_user_project_role": True,
  47 }
  48
  49
  50 def norm(str):
  51     """Normalize string for checking"""
  52     return " ".join(str.strip().split()).lower()
  53
  54
  55 def compare_desc(tc, d1, d2, k):
  56     """
  57     Compare two descriptors
  58     We need this function because some methods are adding/removing items to/from the descriptors
  59     before they are stored in the database, so the original and stored versions will differ
  60     What we check is that COMMON LEAF ITEMS are equal
  61     Lists of different length are not compared
  62     :param tc: Test Case wich provides context (in particular the assert* methods)
  63     :param d1,d2: Descriptors to be compared
  64     :param key/item being compared
  65     :return: Nothing
  66     """
  67     if isinstance(d1, dict) and isinstance(d2, dict):
  68         for key in d1.keys():
  69             if key in d2:
  70                 compare_desc(tc, d1[key], d2[key], k + "[{}]".format(key))
  71     elif isinstance(d1, list) and isinstance(d2, list) and len(d1) == len(d2):
  72         for i in range(len(d1)):
  73             compare_desc(tc, d1[i], d2[i], k + "[{}]".format(i))
  74     else:
  75         tc.assertEqual(d1, d2, "Wrong descriptor content: {}".format(k))
  76
  77
  78 class Test_VnfdTopic(TestCase):
  79     @classmethod
  80     def setUpClass(cls):
  81         cls.test_name = "test-vnfd-topic"
  82
  83     @classmethod
  84     def tearDownClass(cls):
  85         pass
  86
  87     def setUp(self):
  88         self.db = Mock(dbbase.DbBase())
  89         self.fs = Mock(fsbase.FsBase())
  90         self.msg = Mock(msgbase.MsgBase())
  91         self.auth = Mock(authconn.Authconn(None, None, None))
  92         self.topic = VnfdTopic(self.db, self.fs, self.msg, self.auth)
  93         self.topic.check_quota = Mock(return_value=None)  # skip quota
  94
  95     @patch("osm_nbi.descriptor_topics.shutil")
  96     @patch("osm_nbi.descriptor_topics.os.rename")
  97     def test_new_vnfd(self, mock_rename, mock_shutil):
  98         did = db_vnfd_content["_id"]
  99         self.fs.path = ""
 100         self.fs.get_params.return_value = {}
 101         self.fs.file_exists.return_value = False
 102         self.fs.file_open.side_effect = lambda path, mode: open(
 103             "/tmp/" + str(uuid4()), "a+b"
 104         )
 105         test_vnfd = deepcopy(db_vnfd_content)
 106         del test_vnfd["_id"]
 107         del test_vnfd["_admin"]
 108         with self.subTest(i=1, t="Normal Creation"):
 109             self.db.create.return_value = did
 110             rollback = []
 111             did2, oid = self.topic.new(rollback, fake_session, {})
 112             db_args = self.db.create.call_args[0]
 113             msg_args = self.msg.write.call_args[0]
 114             self.assertEqual(len(rollback), 1, "Wrong rollback length")
 115             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 116             self.assertEqual(msg_args[1], "created", "Wrong message action")
 117             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 118             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 119             self.assertEqual(did2, did, "Wrong DB VNFD id")
 120             self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
 121             self.assertEqual(
 122                 db_args[1]["_admin"]["modified"],
 123                 db_args[1]["_admin"]["created"],
 124                 "Wrong modification time",
 125             )
 126             self.assertEqual(
 127                 db_args[1]["_admin"]["projects_read"],
 128                 [test_pid],
 129                 "Wrong read-only project list",
 130             )
 131             self.assertEqual(
 132                 db_args[1]["_admin"]["projects_write"],
 133                 [test_pid],
 134                 "Wrong read-write project list",
 135             )
 136             tmp1 = test_vnfd["vdu"][0]["cloud-init-file"]
 137             tmp2 = test_vnfd["df"][0]["lcm-operations-configuration"][
 138                 "operate-vnf-op-config"
 139             ]["day1-2"][0]["execution-environment-list"][0]["juju"]
 140             del test_vnfd["vdu"][0]["cloud-init-file"]
 141             del test_vnfd["df"][0]["lcm-operations-configuration"][
 142                 "operate-vnf-op-config"
 143             ]["day1-2"][0]["execution-environment-list"][0]["juju"]
 144             try:
 145                 self.db.get_one.side_effect = [
 146                     {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])},
 147                     None,
 148                 ]
 149                 self.topic.upload_content(
 150                     fake_session, did, test_vnfd, {}, {"Content-Type": []}
 151                 )
 152                 msg_args = self.msg.write.call_args[0]
 153                 test_vnfd["_id"] = did
 154                 self.assertEqual(
 155                     msg_args[0], self.topic.topic_msg, "Wrong message topic"
 156                 )
 157                 self.assertEqual(msg_args[1], "edited", "Wrong message action")
 158                 self.assertEqual(msg_args[2], test_vnfd, "Wrong message content")
 159                 db_args = self.db.get_one.mock_calls[0][1]
 160                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 161                 self.assertEqual(db_args[1]["_id"], did, "Wrong DB VNFD id")
 162                 db_args = self.db.replace.call_args[0]
 163                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 164                 self.assertEqual(db_args[1], did, "Wrong DB VNFD id")
 165                 admin = db_args[2]["_admin"]
 166                 db_admin = deepcopy(db_vnfd_content["_admin"])
 167                 self.assertEqual(admin["type"], "vnfd", "Wrong descriptor type")
 168                 self.assertEqual(
 169                     admin["created"], db_admin["created"], "Wrong creation time"
 170                 )
 171                 self.assertGreater(
 172                     admin["modified"], db_admin["created"], "Wrong modification time"
 173                 )
 174                 self.assertEqual(
 175                     admin["projects_read"],
 176                     db_admin["projects_read"],
 177                     "Wrong read-only project list",
 178                 )
 179                 self.assertEqual(
 180                     admin["projects_write"],
 181                     db_admin["projects_write"],
 182                     "Wrong read-write project list",
 183                 )
 184                 self.assertEqual(
 185                     admin["onboardingState"], "ONBOARDED", "Wrong onboarding state"
 186                 )
 187                 self.assertEqual(
 188                     admin["operationalState"], "ENABLED", "Wrong operational state"
 189                 )
 190                 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
 191                 storage = admin["storage"]
 192                 self.assertEqual(storage["folder"], did, "Wrong storage folder")
 193                 self.assertEqual(
 194                     storage["descriptor"], "package", "Wrong storage descriptor"
 195                 )
 196                 self.assertEqual(
 197                     admin["revision"], 1, "Wrong revision number"
 198                 )
 199
 200                 compare_desc(self, test_vnfd, db_args[2], "VNFD")
 201             finally:
 202                 test_vnfd["vdu"][0]["cloud-init-file"] = tmp1
 203                 test_vnfd["df"][0]["lcm-operations-configuration"][
 204                     "operate-vnf-op-config"
 205                 ]["day1-2"][0]["execution-environment-list"][0]["juju"] = tmp2
 206         self.db.get_one.side_effect = (
 207             lambda table, filter, fail_on_empty=None, fail_on_more=None: {
 208                 "_id": did,
 209                 "_admin": deepcopy(db_vnfd_content["_admin"]),
 210             }
 211         )
 212         with self.subTest(i=2, t="Check Pyangbind Validation: additional properties"):
 213             test_vnfd["extra-property"] = 0
 214             try:
 215                 with self.assertRaises(
 216                     EngineException, msg="Accepted VNFD with an additional property"
 217                 ) as e:
 218                     self.topic.upload_content(
 219                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 220                     )
 221                 self.assertEqual(
 222                     e.exception.http_code,
 223                     HTTPStatus.UNPROCESSABLE_ENTITY,
 224                     "Wrong HTTP status code",
 225                 )
 226                 self.assertIn(
 227                     norm(
 228                         "Error in pyangbind validation: {} ({})".format(
 229                             "json object contained a key that did not exist",
 230                             "extra-property",
 231                         )
 232                     ),
 233                     norm(str(e.exception)),
 234                     "Wrong exception text",
 235                 )
 236                 db_args = self.db.replace.call_args[0]
 237                 admin = db_args[2]["_admin"]
 238                 self.assertEqual(
 239                     admin["revision"], 1, "Wrong revision number"
 240                 )
 241
 242             finally:
 243                 del test_vnfd["extra-property"]
 244         with self.subTest(i=3, t="Check Pyangbind Validation: property types"):
 245             tmp = test_vnfd["product-name"]
 246             test_vnfd["product-name"] = {"key": 0}
 247             try:
 248                 with self.assertRaises(
 249                     EngineException, msg="Accepted VNFD with a wrongly typed property"
 250                 ) as e:
 251                     self.topic.upload_content(
 252                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 253                     )
 254                 self.assertEqual(
 255                     e.exception.http_code,
 256                     HTTPStatus.UNPROCESSABLE_ENTITY,
 257                     "Wrong HTTP status code",
 258                 )
 259                 self.assertIn(
 260                     norm(
 261                         "Error in pyangbind validation: {} ({})".format(
 262                             "json object contained a key that did not exist", "key"
 263                         )
 264                     ),
 265                     norm(str(e.exception)),
 266                     "Wrong exception text",
 267                 )
 268             finally:
 269                 test_vnfd["product-name"] = tmp
 270         with self.subTest(i=4, t="Check Input Validation: cloud-init"):
 271             with self.assertRaises(
 272                 EngineException, msg="Accepted non-existent cloud_init file"
 273             ) as e:
 274                 self.topic.upload_content(
 275                     fake_session, did, test_vnfd, {}, {"Content-Type": []}
 276                 )
 277             self.assertEqual(
 278                 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
 279             )
 280             self.assertIn(
 281                 norm(
 282                     "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format(
 283                         "cloud-init", test_vnfd["id"], test_vnfd["vdu"][0]["id"]
 284                     )
 285                 ),
 286                 norm(str(e.exception)),
 287                 "Wrong exception text",
 288             )
 289         with self.subTest(i=5, t="Check Input Validation: day1-2 configuration[juju]"):
 290             del test_vnfd["vdu"][0]["cloud-init-file"]
 291             with self.assertRaises(
 292                 EngineException, msg="Accepted non-existent charm in VNF configuration"
 293             ) as e:
 294                 self.topic.upload_content(
 295                     fake_session, did, test_vnfd, {}, {"Content-Type": []}
 296                 )
 297             print(str(e.exception))
 298             self.assertEqual(
 299                 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
 300             )
 301             self.assertIn(
 302                 norm(
 303                     "{} defined in vnf[id={}] but not present in package".format(
 304                         "charm", test_vnfd["id"]
 305                     )
 306                 ),
 307                 norm(str(e.exception)),
 308                 "Wrong exception text",
 309             )
 310             del test_vnfd["df"][0]["lcm-operations-configuration"][
 311                 "operate-vnf-op-config"
 312             ]["day1-2"][0]["execution-environment-list"][0]["juju"]
 313         with self.subTest(i=6, t="Check Input Validation: mgmt-cp"):
 314             tmp = test_vnfd["mgmt-cp"]
 315             del test_vnfd["mgmt-cp"]
 316             try:
 317                 with self.assertRaises(
 318                     EngineException, msg="Accepted VNFD without management interface"
 319                 ) as e:
 320                     self.topic.upload_content(
 321                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 322                     )
 323                 self.assertEqual(
 324                     e.exception.http_code,
 325                     HTTPStatus.UNPROCESSABLE_ENTITY,
 326                     "Wrong HTTP status code",
 327                 )
 328                 self.assertIn(
 329                     norm(
 330                         "'{}' is a mandatory field and it is not defined".format(
 331                             "mgmt-cp"
 332                         )
 333                     ),
 334                     norm(str(e.exception)),
 335                     "Wrong exception text",
 336                 )
 337             finally:
 338                 test_vnfd["mgmt-cp"] = tmp
 339         with self.subTest(i=7, t="Check Input Validation: mgmt-cp connection point"):
 340             tmp = test_vnfd["mgmt-cp"]
 341             test_vnfd["mgmt-cp"] = "wrong-cp"
 342             try:
 343                 with self.assertRaises(
 344                     EngineException, msg="Accepted wrong mgmt-cp connection point"
 345                 ) as e:
 346                     self.topic.upload_content(
 347                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 348                     )
 349                 self.assertEqual(
 350                     e.exception.http_code,
 351                     HTTPStatus.UNPROCESSABLE_ENTITY,
 352                     "Wrong HTTP status code",
 353                 )
 354                 self.assertIn(
 355                     norm(
 356                         "mgmt-cp='{}' must match an existing ext-cpd".format(
 357                             test_vnfd["mgmt-cp"]
 358                         )
 359                     ),
 360                     norm(str(e.exception)),
 361                     "Wrong exception text",
 362                 )
 363             finally:
 364                 test_vnfd["mgmt-cp"] = tmp
 365         with self.subTest(i=8, t="Check Input Validation: vdu int-cpd"):
 366             ext_cpd = test_vnfd["ext-cpd"][1]
 367             tmp = ext_cpd["int-cpd"]["cpd"]
 368             ext_cpd["int-cpd"]["cpd"] = "wrong-cpd"
 369             try:
 370                 with self.assertRaises(
 371                     EngineException,
 372                     msg="Accepted wrong ext-cpd internal connection point",
 373                 ) as e:
 374                     self.topic.upload_content(
 375                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 376                     )
 377                 self.assertEqual(
 378                     e.exception.http_code,
 379                     HTTPStatus.UNPROCESSABLE_ENTITY,
 380                     "Wrong HTTP status code",
 381                 )
 382                 self.assertIn(
 383                     norm(
 384                         "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
 385                             ext_cpd["id"]
 386                         )
 387                     ),
 388                     norm(str(e.exception)),
 389                     "Wrong exception text",
 390                 )
 391             finally:
 392                 ext_cpd["int-cpd"]["cpd"] = tmp
 393         with self.subTest(i=9, t="Check Input Validation: Duplicated VLD"):
 394             test_vnfd["int-virtual-link-desc"].insert(0, {"id": "internal"})
 395             try:
 396                 with self.assertRaises(
 397                     EngineException, msg="Accepted duplicated VLD name"
 398                 ) as e:
 399                     self.topic.upload_content(
 400                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 401                     )
 402                 self.assertEqual(
 403                     e.exception.http_code,
 404                     HTTPStatus.UNPROCESSABLE_ENTITY,
 405                     "Wrong HTTP status code",
 406                 )
 407                 self.assertIn(
 408                     norm(
 409                         "identifier id '{}' is not unique".format(
 410                             test_vnfd["int-virtual-link-desc"][0]["id"]
 411                         )
 412                     ),
 413                     norm(str(e.exception)),
 414                     "Wrong exception text",
 415                 )
 416             finally:
 417                 del test_vnfd["int-virtual-link-desc"][0]
 418         with self.subTest(i=10, t="Check Input Validation: vdu int-virtual-link-desc"):
 419             vdu = test_vnfd["vdu"][0]
 420             int_cpd = vdu["int-cpd"][1]
 421             tmp = int_cpd["int-virtual-link-desc"]
 422             int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
 423             try:
 424                 with self.assertRaises(
 425                     EngineException, msg="Accepted int-virtual-link-desc"
 426                 ) as e:
 427                     self.topic.upload_content(
 428                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 429                     )
 430                 self.assertEqual(
 431                     e.exception.http_code,
 432                     HTTPStatus.UNPROCESSABLE_ENTITY,
 433                     "Wrong HTTP status code",
 434                 )
 435                 self.assertIn(
 436                     norm(
 437                         "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
 438                         "int-virtual-link-desc".format(
 439                             vdu["id"], int_cpd["id"], int_cpd["int-virtual-link-desc"]
 440                         )
 441                     ),
 442                     norm(str(e.exception)),
 443                     "Wrong exception text",
 444                 )
 445             finally:
 446                 int_cpd["int-virtual-link-desc"] = tmp
 447         with self.subTest(i=11, t="Check Input Validation: virtual-link-profile)"):
 448             fake_ivld_profile = {"id": "fake-profile-ref", "flavour": "fake-flavour"}
 449             df = test_vnfd["df"][0]
 450             df["virtual-link-profile"] = [fake_ivld_profile]
 451             try:
 452                 with self.assertRaises(
 453                     EngineException, msg="Accepted non-existent Profile Ref"
 454                 ) as e:
 455                     self.topic.upload_content(
 456                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 457                     )
 458                 self.assertEqual(
 459                     e.exception.http_code,
 460                     HTTPStatus.UNPROCESSABLE_ENTITY,
 461                     "Wrong HTTP status code",
 462                 )
 463                 self.assertIn(
 464                     norm(
 465                         "df[id='{}']:virtual-link-profile='{}' must match an existing "
 466                         "int-virtual-link-desc".format(
 467                             df["id"], fake_ivld_profile["id"]
 468                         )
 469                     ),
 470                     norm(str(e.exception)),
 471                     "Wrong exception text",
 472                 )
 473             finally:
 474                 del df["virtual-link-profile"]
 475         with self.subTest(
 476             i=12, t="Check Input Validation: scaling-criteria monitoring-param-ref"
 477         ):
 478             vdu = test_vnfd["vdu"][1]
 479             affected_df = test_vnfd["df"][0]
 480             sa = affected_df["scaling-aspect"][0]
 481             sp = sa["scaling-policy"][0]
 482             sc = sp["scaling-criteria"][0]
 483             tmp = vdu.pop("monitoring-parameter")
 484             try:
 485                 with self.assertRaises(
 486                     EngineException,
 487                     msg="Accepted non-existent Scaling Group Policy Criteria",
 488                 ) as e:
 489                     self.topic.upload_content(
 490                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 491                     )
 492                 self.assertEqual(
 493                     e.exception.http_code,
 494                     HTTPStatus.UNPROCESSABLE_ENTITY,
 495                     "Wrong HTTP status code",
 496                 )
 497                 self.assertIn(
 498                     norm(
 499                         "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
 500                         "[name='{}']:scaling-criteria[name='{}']: "
 501                         "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
 502                             affected_df["id"],
 503                             sa["id"],
 504                             sp["name"],
 505                             sc["name"],
 506                             sc["vnf-monitoring-param-ref"],
 507                         )
 508                     ),
 509                     norm(str(e.exception)),
 510                     "Wrong exception text",
 511                 )
 512             finally:
 513                 vdu["monitoring-parameter"] = tmp
 514         with self.subTest(
 515             i=13, t="Check Input Validation: scaling-aspect vnf-configuration"
 516         ):
 517             df = test_vnfd["df"][0]
 518             tmp = test_vnfd["df"][0]["lcm-operations-configuration"][
 519                 "operate-vnf-op-config"
 520             ]["day1-2"].pop()
 521             try:
 522                 with self.assertRaises(
 523                     EngineException,
 524                     msg="Accepted non-existent Scaling Group VDU ID Reference",
 525                 ) as e:
 526                     self.topic.upload_content(
 527                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 528                     )
 529                 self.assertEqual(
 530                     e.exception.http_code,
 531                     HTTPStatus.UNPROCESSABLE_ENTITY,
 532                     "Wrong HTTP status code",
 533                 )
 534                 self.assertIn(
 535                     norm(
 536                         "'day1-2 configuration' not defined in the descriptor but it is referenced "
 537                         "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
 538                             df["id"], df["scaling-aspect"][0]["id"]
 539                         )
 540                     ),
 541                     norm(str(e.exception)),
 542                     "Wrong exception text",
 543                 )
 544             finally:
 545                 test_vnfd["df"][0]["lcm-operations-configuration"][
 546                     "operate-vnf-op-config"
 547                 ]["day1-2"].append(tmp)
 548         with self.subTest(i=14, t="Check Input Validation: scaling-config-action"):
 549             df = test_vnfd["df"][0]
 550             tmp = (
 551                 test_vnfd["df"][0]
 552                 .get("lcm-operations-configuration")
 553                 .get("operate-vnf-op-config")["day1-2"][0]["config-primitive"]
 554             )
 555             test_vnfd["df"][0].get("lcm-operations-configuration").get(
 556                 "operate-vnf-op-config"
 557             )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
 558             try:
 559                 with self.assertRaises(
 560                     EngineException,
 561                     msg="Accepted non-existent Scaling Group VDU ID Reference",
 562                 ) as e:
 563                     self.topic.upload_content(
 564                         fake_session, did, test_vnfd, {}, {"Content-Type": []}
 565                     )
 566                 self.assertEqual(
 567                     e.exception.http_code,
 568                     HTTPStatus.UNPROCESSABLE_ENTITY,
 569                     "Wrong HTTP status code",
 570                 )
 571                 self.assertIn(
 572                     norm(
 573                         "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
 574                         "config-primitive-name-ref='{}' does not match any "
 575                         "day1-2 configuration:config-primitive:name".format(
 576                             df["id"],
 577                             df["scaling-aspect"][0]["id"],
 578                             sa["scaling-config-action"][0][
 579                                 "vnf-config-primitive-name-ref"
 580                             ],
 581                         )
 582                     ),
 583                     norm(str(e.exception)),
 584                     "Wrong exception text",
 585                 )
 586             finally:
 587                 test_vnfd["df"][0].get("lcm-operations-configuration").get(
 588                     "operate-vnf-op-config"
 589                 )["day1-2"][0]["config-primitive"] = tmp
 590         with self.subTest(i=15, t="Check Input Validation: everything right"):
 591             test_vnfd["id"] = "fake-vnfd-id"
 592             test_vnfd["df"][0].get("lcm-operations-configuration").get(
 593                 "operate-vnf-op-config"
 594             )["day1-2"][0]["id"] = "fake-vnfd-id"
 595             self.db.get_one.side_effect = [
 596                 {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])},
 597                 None,
 598             ]
 599             rc = self.topic.upload_content(
 600                 fake_session, did, test_vnfd, {}, {"Content-Type": []}
 601             )
 602             self.assertTrue(rc, "Input Validation: Unexpected failure")
 603         return
 604
 605     def test_edit_vnfd(self):
 606         vnfd_content = deepcopy(db_vnfd_content)
 607         did = vnfd_content["_id"]
 608         self.fs.file_exists.return_value = True
 609         self.fs.dir_ls.return_value = True
 610         with self.subTest(i=1, t="Normal Edition"):
 611             now = time()
 612             self.db.get_one.side_effect = [deepcopy(vnfd_content), None]
 613             data = {"product-name": "new-vnfd-name"}
 614             self.topic.edit(fake_session, did, data)
 615             db_args = self.db.replace.call_args[0]
 616             msg_args = self.msg.write.call_args[0]
 617             data["_id"] = did
 618             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 619             self.assertEqual(msg_args[1], "edited", "Wrong message action")
 620             self.assertEqual(msg_args[2], data, "Wrong message content")
 621             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 622             self.assertEqual(db_args[1], did, "Wrong DB ID")
 623             self.assertEqual(
 624                 db_args[2]["_admin"]["created"],
 625                 vnfd_content["_admin"]["created"],
 626                 "Wrong creation time",
 627             )
 628             self.assertGreater(
 629                 db_args[2]["_admin"]["modified"], now, "Wrong modification time"
 630             )
 631             self.assertEqual(
 632                 db_args[2]["_admin"]["projects_read"],
 633                 vnfd_content["_admin"]["projects_read"],
 634                 "Wrong read-only project list",
 635             )
 636             self.assertEqual(
 637                 db_args[2]["_admin"]["projects_write"],
 638                 vnfd_content["_admin"]["projects_write"],
 639                 "Wrong read-write project list",
 640             )
 641             self.assertEqual(
 642                 db_args[2]["product-name"], data["product-name"], "Wrong VNFD Name"
 643             )
 644         with self.subTest(i=2, t="Conflict on Edit"):
 645             data = {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
 646             self.db.get_one.side_effect = [
 647                 deepcopy(vnfd_content),
 648                 {"_id": str(uuid4()), "id": data["id"]},
 649             ]
 650             with self.assertRaises(
 651                 EngineException, msg="Accepted existing VNFD ID"
 652             ) as e:
 653                 self.topic.edit(fake_session, did, data)
 654             self.assertEqual(
 655                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
 656             )
 657             self.assertIn(
 658                 norm(
 659                     "{} with id '{}' already exists for this project".format(
 660                         "vnfd", data["id"]
 661                     )
 662                 ),
 663                 norm(str(e.exception)),
 664                 "Wrong exception text",
 665             )
 666         with self.subTest(i=3, t="Check Envelope"):
 667             data = {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
 668             with self.assertRaises(
 669                 EngineException, msg="Accepted VNFD with wrong envelope"
 670             ) as e:
 671                 self.topic.edit(fake_session, did, data, content=vnfd_content)
 672             self.assertEqual(
 673                 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
 674             )
 675             self.assertIn(
 676                 "'vnfd' must be dict", norm(str(e.exception)), "Wrong exception text"
 677             )
 678         return
 679
 680     def test_delete_vnfd(self):
 681         did = db_vnfd_content["_id"]
 682         self.db.get_one.return_value = db_vnfd_content
 683         p_id = db_vnfd_content["_admin"]["projects_read"][0]
 684         with self.subTest(i=1, t="Normal Deletion"):
 685             self.db.get_list.return_value = []
 686             self.db.del_one.return_value = {"deleted": 1}
 687             self.topic.delete(fake_session, did)
 688             db_args = self.db.del_one.call_args[0]
 689             msg_args = self.msg.write.call_args[0]
 690             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 691             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 692             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 693             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 694             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
 695             self.assertEqual(
 696                 db_args[1]["_admin.projects_write.cont"],
 697                 [p_id, "ANY"],
 698                 "Wrong DB filter",
 699             )
 700             db_g1_args = self.db.get_one.call_args[0]
 701             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 702             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 703             db_gl_calls = self.db.get_list.call_args_list
 704             self.assertEqual(db_gl_calls[0][0][0], "vnfrs", "Wrong DB topic")
 705             # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID")   # Filter changed after call
 706             self.assertEqual(db_gl_calls[1][0][0], "nsds", "Wrong DB topic")
 707             self.assertEqual(
 708                 db_gl_calls[1][0][1]["vnfd-id"],
 709                 db_vnfd_content["id"],
 710                 "Wrong DB NSD vnfd-id",
 711             )
 712
 713             db_del_args = self.db.del_list.call_args[0]
 714             self.assertEqual(
 715                 self.db.del_list.call_args[0][0],
 716                 self.topic.topic+"_revisions",
 717                 "Wrong DB topic",
 718             )
 719
 720             self.assertEqual(
 721                 self.db.del_list.call_args[0][1]['_id']['$regex'],
 722                 did,
 723                 "Wrong ID for rexep delete",
 724             )
 725
 726             self.db.set_one.assert_not_called()
 727             fs_del_calls = self.fs.file_delete.call_args_list
 728             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
 729             self.assertEqual(fs_del_calls[1][0][0], did + "_", "Wrong FS folder id")
 730         with self.subTest(i=2, t="Conflict on Delete - VNFD in use by VNFR"):
 731             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-vnfr"}]
 732             with self.assertRaises(
 733                 EngineException, msg="Accepted VNFD in use by VNFR"
 734             ) as e:
 735                 self.topic.delete(fake_session, did)
 736             self.assertEqual(
 737                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
 738             )
 739             self.assertIn(
 740                 "there is at least one vnf instance using this descriptor",
 741                 norm(str(e.exception)),
 742                 "Wrong exception text",
 743             )
 744         with self.subTest(i=3, t="Conflict on Delete - VNFD in use by NSD"):
 745             self.db.get_list.side_effect = [
 746                 [],
 747                 [{"_id": str(uuid4()), "name": "fake-nsd"}],
 748             ]
 749             with self.assertRaises(
 750                 EngineException, msg="Accepted VNFD in use by NSD"
 751             ) as e:
 752                 self.topic.delete(fake_session, did)
 753             self.assertEqual(
 754                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
 755             )
 756             self.assertIn(
 757                 "there is at least one ns package referencing this descriptor",
 758                 norm(str(e.exception)),
 759                 "Wrong exception text",
 760             )
 761         with self.subTest(i=4, t="Non-existent VNFD"):
 762             excp_msg = "Not found any {} with filter='{}'".format("VNFD", {"_id": did})
 763             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
 764             with self.assertRaises(
 765                 DbException, msg="Accepted non-existent VNFD ID"
 766             ) as e:
 767                 self.topic.delete(fake_session, did)
 768             self.assertEqual(
 769                 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
 770             )
 771             self.assertIn(
 772                 norm(excp_msg), norm(str(e.exception)), "Wrong exception text"
 773             )
 774         with self.subTest(i=5, t="No delete because referenced by other project"):
 775             db_vnfd_content["_admin"]["projects_read"].append("other_project")
 776             self.db.get_one = Mock(return_value=db_vnfd_content)
 777             self.db.get_list = Mock(return_value=[])
 778             self.msg.write.reset_mock()
 779             self.db.del_one.reset_mock()
 780             self.fs.file_delete.reset_mock()
 781
 782             self.topic.delete(fake_session, did)
 783             self.db.del_one.assert_not_called()
 784             self.msg.write.assert_not_called()
 785             db_g1_args = self.db.get_one.call_args[0]
 786             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 787             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 788             db_s1_args = self.db.set_one.call_args
 789             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
 790             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
 791             self.assertIn(
 792                 p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
 793             )
 794             self.assertIsNone(
 795                 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
 796             )
 797             self.assertEqual(
 798                 db_s1_args[1]["pull_list"],
 799                 {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
 800                 "Wrong DB pull_list dictionary",
 801             )
 802             self.fs.file_delete.assert_not_called()
 803         return
 804
 805     def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self):
 806         indata = deepcopy(db_vnfd_content)
 807         self.topic.validate_mgmt_interface_connection_point(indata)
 808
 809     def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
 810         self,
 811     ):
 812         indata = deepcopy(db_vnfd_content)
 813         indata["ext-cpd"] = []
 814         with self.assertRaises(EngineException) as e:
 815             self.topic.validate_mgmt_interface_connection_point(indata)
 816         self.assertEqual(
 817             e.exception.http_code,
 818             HTTPStatus.UNPROCESSABLE_ENTITY,
 819             "Wrong HTTP status code",
 820         )
 821         self.assertIn(
 822             norm(
 823                 "mgmt-cp='{}' must match an existing ext-cpd".format(indata["mgmt-cp"])
 824             ),
 825             norm(str(e.exception)),
 826             "Wrong exception text",
 827         )
 828
 829     def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self):
 830         indata = deepcopy(db_vnfd_content)
 831         indata.pop("mgmt-cp")
 832         with self.assertRaises(EngineException) as e:
 833             self.topic.validate_mgmt_interface_connection_point(indata)
 834         self.assertEqual(
 835             e.exception.http_code,
 836             HTTPStatus.UNPROCESSABLE_ENTITY,
 837             "Wrong HTTP status code",
 838         )
 839         self.assertIn(
 840             norm("'mgmt-cp' is a mandatory field and it is not defined"),
 841             norm(str(e.exception)),
 842             "Wrong exception text",
 843         )
 844
 845     def test_validate_vdu_internal_connection_points_on_valid_descriptor(self):
 846         indata = db_vnfd_content
 847         vdu = indata["vdu"][0]
 848         self.topic.validate_vdu_internal_connection_points(vdu)
 849
 850     def test_validate_external_connection_points_on_valid_descriptor(self):
 851         indata = db_vnfd_content
 852         self.topic.validate_external_connection_points(indata)
 853
 854     def test_validate_external_connection_points_when_missing_internal_connection_point(
 855         self,
 856     ):
 857         indata = deepcopy(db_vnfd_content)
 858         vdu = indata["vdu"][0]
 859         vdu.pop("int-cpd")
 860         affected_ext_cpd = indata["ext-cpd"][0]
 861         with self.assertRaises(EngineException) as e:
 862             self.topic.validate_external_connection_points(indata)
 863         self.assertEqual(
 864             e.exception.http_code,
 865             HTTPStatus.UNPROCESSABLE_ENTITY,
 866             "Wrong HTTP status code",
 867         )
 868         self.assertIn(
 869             norm(
 870                 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
 871                     affected_ext_cpd["id"]
 872                 )
 873             ),
 874             norm(str(e.exception)),
 875             "Wrong exception text",
 876         )
 877
 878     def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
 879         self,
 880     ):
 881         indata = deepcopy(db_vnfd_content)
 882         vdu = indata["vdu"][0]
 883         duplicated_cpd = {
 884             "id": "vnf-mgmt",
 885             "order": 3,
 886             "virtual-network-interface-requirement": [{"name": "duplicated"}],
 887         }
 888         vdu["int-cpd"].insert(0, duplicated_cpd)
 889         with self.assertRaises(EngineException) as e:
 890             self.topic.validate_vdu_internal_connection_points(vdu)
 891         self.assertEqual(
 892             e.exception.http_code,
 893             HTTPStatus.UNPROCESSABLE_ENTITY,
 894             "Wrong HTTP status code",
 895         )
 896         self.assertIn(
 897             norm(
 898                 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
 899                     vdu["id"], duplicated_cpd["id"]
 900                 )
 901             ),
 902             norm(str(e.exception)),
 903             "Wrong exception text",
 904         )
 905
 906     def test_validate_external_connection_points_on_duplicated_external_connection_point(
 907         self,
 908     ):
 909         indata = deepcopy(db_vnfd_content)
 910         duplicated_cpd = {
 911             "id": "vnf-mgmt-ext",
 912             "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
 913         }
 914         indata["ext-cpd"].insert(0, duplicated_cpd)
 915         with self.assertRaises(EngineException) as e:
 916             self.topic.validate_external_connection_points(indata)
 917         self.assertEqual(
 918             e.exception.http_code,
 919             HTTPStatus.UNPROCESSABLE_ENTITY,
 920             "Wrong HTTP status code",
 921         )
 922         self.assertIn(
 923             norm(
 924                 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
 925                     duplicated_cpd["id"]
 926                 )
 927             ),
 928             norm(str(e.exception)),
 929             "Wrong exception text",
 930         )
 931
 932     def test_validate_internal_virtual_links_on_valid_descriptor(self):
 933         indata = db_vnfd_content
 934         self.topic.validate_internal_virtual_links(indata)
 935
 936     def test_validate_internal_virtual_links_on_duplicated_ivld(self):
 937         indata = deepcopy(db_vnfd_content)
 938         duplicated_vld = {"id": "internal"}
 939         indata["int-virtual-link-desc"].insert(0, duplicated_vld)
 940         with self.assertRaises(EngineException) as e:
 941             self.topic.validate_internal_virtual_links(indata)
 942         self.assertEqual(
 943             e.exception.http_code,
 944             HTTPStatus.UNPROCESSABLE_ENTITY,
 945             "Wrong HTTP status code",
 946         )
 947         self.assertIn(
 948             norm(
 949                 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
 950                     duplicated_vld["id"]
 951                 )
 952             ),
 953             norm(str(e.exception)),
 954             "Wrong exception text",
 955         )
 956
 957     def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
 958         self,
 959     ):
 960         indata = deepcopy(db_vnfd_content)
 961         vdu = indata["vdu"][0]
 962         affected_int_cpd = vdu["int-cpd"][0]
 963         affected_int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
 964         with self.assertRaises(EngineException) as e:
 965             self.topic.validate_internal_virtual_links(indata)
 966         self.assertEqual(
 967             e.exception.http_code,
 968             HTTPStatus.UNPROCESSABLE_ENTITY,
 969             "Wrong HTTP status code",
 970         )
 971         self.assertIn(
 972             norm(
 973                 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
 974                 "int-virtual-link-desc".format(
 975                     vdu["id"],
 976                     affected_int_cpd["id"],
 977                     affected_int_cpd["int-virtual-link-desc"],
 978                 )
 979             ),
 980             norm(str(e.exception)),
 981             "Wrong exception text",
 982         )
 983
 984     def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self):
 985         indata = deepcopy(db_vnfd_content)
 986         affected_ivld_profile = {"id": "non-existing-int-virtual-link-desc"}
 987         df = indata["df"][0]
 988         df["virtual-link-profile"] = [affected_ivld_profile]
 989         with self.assertRaises(EngineException) as e:
 990             self.topic.validate_internal_virtual_links(indata)
 991         self.assertEqual(
 992             e.exception.http_code,
 993             HTTPStatus.UNPROCESSABLE_ENTITY,
 994             "Wrong HTTP status code",
 995         )
 996         self.assertIn(
 997             norm(
 998                 "df[id='{}']:virtual-link-profile='{}' must match an existing "
 999                 "int-virtual-link-desc".format(df["id"], affected_ivld_profile["id"])
1000             ),
1001             norm(str(e.exception)),
1002             "Wrong exception text",
1003         )
1004
1005     def test_validate_monitoring_params_on_valid_descriptor(self):
1006         indata = db_vnfd_content
1007         self.topic.validate_monitoring_params(indata)
1008
1009     def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self):
1010         indata = deepcopy(db_vnfd_content)
1011         duplicated_mp = {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1012         affected_ivld = indata["int-virtual-link-desc"][0]
1013         affected_ivld["monitoring-parameters"] = [duplicated_mp, duplicated_mp]
1014         with self.assertRaises(EngineException) as e:
1015             self.topic.validate_monitoring_params(indata)
1016         self.assertEqual(
1017             e.exception.http_code,
1018             HTTPStatus.UNPROCESSABLE_ENTITY,
1019             "Wrong HTTP status code",
1020         )
1021         self.assertIn(
1022             norm(
1023                 "Duplicated monitoring-parameter id in "
1024                 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1025                     affected_ivld["id"], duplicated_mp["id"]
1026                 )
1027             ),
1028             norm(str(e.exception)),
1029             "Wrong exception text",
1030         )
1031
1032     def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self):
1033         indata = deepcopy(db_vnfd_content)
1034         duplicated_mp = {
1035             "id": "dataVM_cpu_util",
1036             "name": "dataVM_cpu_util",
1037             "performance_metric": "cpu",
1038         }
1039         affected_vdu = indata["vdu"][1]
1040         affected_vdu["monitoring-parameter"].insert(0, duplicated_mp)
1041         with self.assertRaises(EngineException) as e:
1042             self.topic.validate_monitoring_params(indata)
1043         self.assertEqual(
1044             e.exception.http_code,
1045             HTTPStatus.UNPROCESSABLE_ENTITY,
1046             "Wrong HTTP status code",
1047         )
1048         self.assertIn(
1049             norm(
1050                 "Duplicated monitoring-parameter id in "
1051                 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1052                     affected_vdu["id"], duplicated_mp["id"]
1053                 )
1054             ),
1055             norm(str(e.exception)),
1056             "Wrong exception text",
1057         )
1058
1059     def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self):
1060         indata = deepcopy(db_vnfd_content)
1061         duplicated_mp = {
1062             "id": "memory",
1063             "name": "memory",
1064             "performance_metric": "memory",
1065         }
1066         affected_df = indata["df"][0]
1067         affected_df["monitoring-parameter"] = [duplicated_mp, duplicated_mp]
1068         with self.assertRaises(EngineException) as e:
1069             self.topic.validate_monitoring_params(indata)
1070         self.assertEqual(
1071             e.exception.http_code,
1072             HTTPStatus.UNPROCESSABLE_ENTITY,
1073             "Wrong HTTP status code",
1074         )
1075         self.assertIn(
1076             norm(
1077                 "Duplicated monitoring-parameter id in "
1078                 "df[id='{}']:monitoring-parameter[id='{}']".format(
1079                     affected_df["id"], duplicated_mp["id"]
1080                 )
1081             ),
1082             norm(str(e.exception)),
1083             "Wrong exception text",
1084         )
1085
1086     def test_validate_scaling_group_descriptor_on_valid_descriptor(self):
1087         indata = db_vnfd_content
1088         self.topic.validate_scaling_group_descriptor(indata)
1089
1090     def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self):
1091         indata = deepcopy(db_vnfd_content)
1092         vdu = indata["vdu"][1]
1093         affected_df = indata["df"][0]
1094         affected_sa = affected_df["scaling-aspect"][0]
1095         affected_sp = affected_sa["scaling-policy"][0]
1096         affected_sc = affected_sp["scaling-criteria"][0]
1097         vdu.pop("monitoring-parameter")
1098         with self.assertRaises(EngineException) as e:
1099             self.topic.validate_scaling_group_descriptor(indata)
1100         self.assertEqual(
1101             e.exception.http_code,
1102             HTTPStatus.UNPROCESSABLE_ENTITY,
1103             "Wrong HTTP status code",
1104         )
1105         self.assertIn(
1106             norm(
1107                 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1108                 "[name='{}']:scaling-criteria[name='{}']: "
1109                 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1110                     affected_df["id"],
1111                     affected_sa["id"],
1112                     affected_sp["name"],
1113                     affected_sc["name"],
1114                     affected_sc["vnf-monitoring-param-ref"],
1115                 )
1116             ),
1117             norm(str(e.exception)),
1118             "Wrong exception text",
1119         )
1120
1121     def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self):
1122         indata = deepcopy(db_vnfd_content)
1123         df = indata["df"][0]
1124         affected_sa = df["scaling-aspect"][0]
1125         indata["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1126             "day1-2"
1127         ].pop()
1128         with self.assertRaises(EngineException) as e:
1129             self.topic.validate_scaling_group_descriptor(indata)
1130         self.assertEqual(
1131             e.exception.http_code,
1132             HTTPStatus.UNPROCESSABLE_ENTITY,
1133             "Wrong HTTP status code",
1134         )
1135         self.assertIn(
1136             norm(
1137                 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1138                 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1139                     df["id"], affected_sa["id"]
1140                 )
1141             ),
1142             norm(str(e.exception)),
1143             "Wrong exception text",
1144         )
1145
1146     def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1147         self,
1148     ):
1149         indata = deepcopy(db_vnfd_content)
1150         df = indata["df"][0]
1151         affected_sa = df["scaling-aspect"][0]
1152         affected_sca_primitive = affected_sa["scaling-config-action"][0][
1153             "vnf-config-primitive-name-ref"
1154         ]
1155         df["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1156             "config-primitive"
1157         ] = []
1158         with self.assertRaises(EngineException) as e:
1159             self.topic.validate_scaling_group_descriptor(indata)
1160         self.assertEqual(
1161             e.exception.http_code,
1162             HTTPStatus.UNPROCESSABLE_ENTITY,
1163             "Wrong HTTP status code",
1164         )
1165         self.assertIn(
1166             norm(
1167                 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1168                 "config-primitive-name-ref='{}' does not match any "
1169                 "day1-2 configuration:config-primitive:name".format(
1170                     df["id"], affected_sa["id"], affected_sca_primitive
1171                 )
1172             ),
1173             norm(str(e.exception)),
1174             "Wrong exception text",
1175         )
1176
1177     def test_new_vnfd_revision(self):
1178         did = db_vnfd_content["_id"]
1179         self.fs.get_params.return_value = {}
1180         self.fs.file_exists.return_value = False
1181         self.fs.file_open.side_effect = lambda path, mode: open(
1182             "/tmp/" + str(uuid4()), "a+b"
1183         )
1184         test_vnfd = deepcopy(db_vnfd_content)
1185         del test_vnfd["_id"]
1186         del test_vnfd["_admin"]
1187         self.db.create.return_value = did
1188         rollback = []
1189         did2, oid = self.topic.new(rollback, fake_session, {})
1190         db_args = self.db.create.call_args[0]
1191         self.assertEqual(db_args[1]['_admin']['revision'], 0,
1192             "New package should be at revision 0")
1193
1194     @patch("osm_nbi.descriptor_topics.shutil")
1195     @patch("osm_nbi.descriptor_topics.os.rename")
1196     def test_update_vnfd(self, mock_rename, mock_shutil):
1197         old_revision = 5
1198         did = db_vnfd_content["_id"]
1199         self.fs.path = ""
1200         self.fs.get_params.return_value = {}
1201         self.fs.file_exists.return_value = False
1202         self.fs.file_open.side_effect = lambda path, mode: open(
1203             "/tmp/" + str(uuid4()), "a+b"
1204         )
1205         new_vnfd = deepcopy(db_vnfd_content)
1206         del new_vnfd["_id"]
1207         self.db.create.return_value = did
1208         rollback = []
1209         did2, oid = self.topic.new(rollback, fake_session, {})
1210         del new_vnfd["vdu"][0]["cloud-init-file"]
1211         del new_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1212             "day1-2"][0]["execution-environment-list"][0]["juju"]
1213
1214
1215         old_vnfd = {
1216             "_id": did,
1217             "_admin": deepcopy(db_vnfd_content["_admin"])
1218         }
1219         old_vnfd["_admin"]["revision"] = old_revision
1220         self.db.get_one.side_effect = [
1221             old_vnfd,
1222             None,
1223         ]
1224         self.topic.upload_content(
1225             fake_session, did, new_vnfd, {}, {"Content-Type": []}
1226         )
1227
1228         db_args = self.db.replace.call_args[0]
1229         self.assertEqual(db_args[2]['_admin']['revision'], old_revision + 1,
1230             "Revision should increment")
1231
1232
1233 class Test_NsdTopic(TestCase):
1234     @classmethod
1235     def setUpClass(cls):
1236         cls.test_name = "test-nsd-topic"
1237
1238     @classmethod
1239     def tearDownClass(cls):
1240         pass
1241
1242     def setUp(self):
1243         self.db = Mock(dbbase.DbBase())
1244         self.fs = Mock(fsbase.FsBase())
1245         self.msg = Mock(msgbase.MsgBase())
1246         self.auth = Mock(authconn.Authconn(None, None, None))
1247         self.topic = NsdTopic(self.db, self.fs, self.msg, self.auth)
1248         self.topic.check_quota = Mock(return_value=None)  # skip quota
1249
1250     @patch("osm_nbi.descriptor_topics.shutil")
1251     @patch("osm_nbi.descriptor_topics.os.rename")
1252     def test_new_nsd(self, mock_rename, mock_shutil):
1253         did = db_nsd_content["_id"]
1254         self.fs.get_params.return_value = {}
1255         self.fs.file_exists.return_value = False
1256         self.fs.file_open.side_effect = lambda path, mode: open(
1257             "/tmp/" + str(uuid4()), "a+b"
1258         )
1259         test_nsd = deepcopy(db_nsd_content)
1260         del test_nsd["_id"]
1261         del test_nsd["_admin"]
1262         with self.subTest(i=1, t="Normal Creation"):
1263             self.db.create.return_value = did
1264             rollback = []
1265             did2, oid = self.topic.new(rollback, fake_session, {})
1266             db_args = self.db.create.call_args[0]
1267             msg_args = self.msg.write.call_args[0]
1268             self.assertEqual(len(rollback), 1, "Wrong rollback length")
1269             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1270             self.assertEqual(msg_args[1], "created", "Wrong message action")
1271             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
1272             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1273             self.assertEqual(did2, did, "Wrong DB NSD id")
1274             self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
1275             self.assertEqual(
1276                 db_args[1]["_admin"]["modified"],
1277                 db_args[1]["_admin"]["created"],
1278                 "Wrong modification time",
1279             )
1280             self.assertEqual(
1281                 db_args[1]["_admin"]["projects_read"],
1282                 [test_pid],
1283                 "Wrong read-only project list",
1284             )
1285             self.assertEqual(
1286                 db_args[1]["_admin"]["projects_write"],
1287                 [test_pid],
1288                 "Wrong read-write project list",
1289             )
1290             try:
1291                 self.fs.path = ""
1292                 self.db.get_one.side_effect = [
1293                     {"_id": did, "_admin": db_nsd_content["_admin"]},
1294                     None,
1295                 ]
1296                 self.db.get_list.return_value = [db_vnfd_content]
1297                 self.topic.upload_content(
1298                     fake_session, did, test_nsd, {}, {"Content-Type": []}
1299                 )
1300                 msg_args = self.msg.write.call_args[0]
1301                 test_nsd["_id"] = did
1302                 self.assertEqual(
1303                     msg_args[0], self.topic.topic_msg, "Wrong message topic"
1304                 )
1305                 self.assertEqual(msg_args[1], "edited", "Wrong message action")
1306                 self.assertEqual(msg_args[2], test_nsd, "Wrong message content")
1307                 db_args = self.db.get_one.mock_calls[0][1]
1308                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1309                 self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id")
1310                 db_args = self.db.replace.call_args[0]
1311                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1312                 self.assertEqual(db_args[1], did, "Wrong DB NSD id")
1313                 admin = db_args[2]["_admin"]
1314                 db_admin = db_nsd_content["_admin"]
1315                 self.assertEqual(
1316                     admin["created"], db_admin["created"], "Wrong creation time"
1317                 )
1318                 self.assertGreater(
1319                     admin["modified"], db_admin["created"], "Wrong modification time"
1320                 )
1321                 self.assertEqual(
1322                     admin["projects_read"],
1323                     db_admin["projects_read"],
1324                     "Wrong read-only project list",
1325                 )
1326                 self.assertEqual(
1327                     admin["projects_write"],
1328                     db_admin["projects_write"],
1329                     "Wrong read-write project list",
1330                 )
1331                 self.assertEqual(
1332                     admin["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1333                 )
1334                 self.assertEqual(
1335                     admin["operationalState"], "ENABLED", "Wrong operational state"
1336                 )
1337                 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
1338                 storage = admin["storage"]
1339                 self.assertEqual(storage["folder"], did, "Wrong storage folder")
1340                 self.assertEqual(
1341                     storage["descriptor"], "package", "Wrong storage descriptor"
1342                 )
1343                 compare_desc(self, test_nsd, db_args[2], "NSD")
1344                 revision_args = self.db.create.call_args[0]
1345                 self.assertEqual(
1346                     revision_args[0], self.topic.topic + "_revisions", "Wrong topic"
1347                 )
1348                 self.assertEqual(
1349                     revision_args[1]["id"], db_args[2]["id"], "Wrong revision id"
1350                 )
1351                 self.assertEqual(
1352                     revision_args[1]["_id"],
1353                     db_args[2]["_id"] + ":1",
1354                     "Wrong revision _id"
1355                 )
1356
1357             finally:
1358                 pass
1359         self.db.get_one.side_effect = (
1360             lambda table, filter, fail_on_empty=None, fail_on_more=None: {
1361                 "_id": did,
1362                 "_admin": db_nsd_content["_admin"],
1363             }
1364         )
1365         with self.subTest(i=2, t="Check Pyangbind Validation: required properties"):
1366             tmp = test_nsd["id"]
1367             del test_nsd["id"]
1368             try:
1369                 with self.assertRaises(
1370                     EngineException, msg="Accepted NSD with a missing required property"
1371                 ) as e:
1372                     self.topic.upload_content(
1373                         fake_session, did, test_nsd, {}, {"Content-Type": []}
1374                     )
1375                 self.assertEqual(
1376                     e.exception.http_code,
1377                     HTTPStatus.UNPROCESSABLE_ENTITY,
1378                     "Wrong HTTP status code",
1379                 )
1380                 self.assertIn(
1381                     norm("Error in pyangbind validation: '{}'".format("id")),
1382                     norm(str(e.exception)),
1383                     "Wrong exception text",
1384                 )
1385             finally:
1386                 test_nsd["id"] = tmp
1387         with self.subTest(i=3, t="Check Pyangbind Validation: additional properties"):
1388             test_nsd["extra-property"] = 0
1389             try:
1390                 with self.assertRaises(
1391                     EngineException, msg="Accepted NSD with an additional property"
1392                 ) as e:
1393                     self.topic.upload_content(
1394                         fake_session, did, test_nsd, {}, {"Content-Type": []}
1395                     )
1396                 self.assertEqual(
1397                     e.exception.http_code,
1398                     HTTPStatus.UNPROCESSABLE_ENTITY,
1399                     "Wrong HTTP status code",
1400                 )
1401                 self.assertIn(
1402                     norm(
1403                         "Error in pyangbind validation: {} ({})".format(
1404                             "json object contained a key that did not exist",
1405                             "extra-property",
1406                         )
1407                     ),
1408                     norm(str(e.exception)),
1409                     "Wrong exception text",
1410                 )
1411             finally:
1412                 del test_nsd["extra-property"]
1413         with self.subTest(i=4, t="Check Pyangbind Validation: property types"):
1414             tmp = test_nsd["designer"]
1415             test_nsd["designer"] = {"key": 0}
1416             try:
1417                 with self.assertRaises(
1418                     EngineException, msg="Accepted NSD with a wrongly typed property"
1419                 ) as e:
1420                     self.topic.upload_content(
1421                         fake_session, did, test_nsd, {}, {"Content-Type": []}
1422                     )
1423                 self.assertEqual(
1424                     e.exception.http_code,
1425                     HTTPStatus.UNPROCESSABLE_ENTITY,
1426                     "Wrong HTTP status code",
1427                 )
1428                 self.assertIn(
1429                     norm(
1430                         "Error in pyangbind validation: {} ({})".format(
1431                             "json object contained a key that did not exist", "key"
1432                         )
1433                     ),
1434                     norm(str(e.exception)),
1435                     "Wrong exception text",
1436                 )
1437             finally:
1438                 test_nsd["designer"] = tmp
1439         with self.subTest(
1440             i=5, t="Check Input Validation: mgmt-network+virtual-link-protocol-data"
1441         ):
1442             df = test_nsd["df"][0]
1443             mgmt_profile = {
1444                 "id": "id",
1445                 "virtual-link-desc-id": "mgmt",
1446                 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1447             }
1448             df["virtual-link-profile"] = [mgmt_profile]
1449             try:
1450                 with self.assertRaises(
1451                     EngineException, msg="Accepted VLD with mgmt-network+ip-profile"
1452                 ) as e:
1453                     self.topic.upload_content(
1454                         fake_session, did, test_nsd, {}, {"Content-Type": []}
1455                     )
1456                 self.assertEqual(
1457                     e.exception.http_code,
1458                     HTTPStatus.UNPROCESSABLE_ENTITY,
1459                     "Wrong HTTP status code",
1460                 )
1461                 self.assertIn(
1462                     norm(
1463                         "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1464                         " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1465                             df["id"], mgmt_profile["id"]
1466                         )
1467                     ),
1468                     norm(str(e.exception)),
1469                     "Wrong exception text",
1470                 )
1471             finally:
1472                 del df["virtual-link-profile"]
1473         with self.subTest(i=6, t="Check Descriptor Dependencies: vnfd-id[]"):
1474             self.db.get_one.side_effect = [
1475                 {"_id": did, "_admin": db_nsd_content["_admin"]},
1476                 None,
1477             ]
1478             self.db.get_list.return_value = []
1479             try:
1480                 with self.assertRaises(
1481                     EngineException, msg="Accepted wrong VNFD ID reference"
1482                 ) as e:
1483                     self.topic.upload_content(
1484                         fake_session, did, test_nsd, {}, {"Content-Type": []}
1485                     )
1486                 self.assertEqual(
1487                     e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1488                 )
1489                 self.assertIn(
1490                     norm(
1491                         "'vnfd-id'='{}' references a non existing vnfd".format(
1492                             test_nsd["vnfd-id"][0]
1493                         )
1494                     ),
1495                     norm(str(e.exception)),
1496                     "Wrong exception text",
1497                 )
1498             finally:
1499                 pass
1500         with self.subTest(
1501             i=7,
1502             t="Check Descriptor Dependencies: "
1503             "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]",
1504         ):
1505             vnfd_descriptor = deepcopy(db_vnfd_content)
1506             df = test_nsd["df"][0]
1507             affected_vnf_profile = df["vnf-profile"][0]
1508             affected_virtual_link = affected_vnf_profile["virtual-link-connectivity"][1]
1509             affected_cpd = vnfd_descriptor["ext-cpd"].pop()
1510             self.db.get_one.side_effect = [
1511                 {"_id": did, "_admin": db_nsd_content["_admin"]},
1512                 None,
1513             ]
1514             self.db.get_list.return_value = [vnfd_descriptor]
1515             try:
1516                 with self.assertRaises(
1517                     EngineException, msg="Accepted wrong VLD CP reference"
1518                 ) as e:
1519                     self.topic.upload_content(
1520                         fake_session, did, test_nsd, {}, {"Content-Type": []}
1521                     )
1522                 self.assertEqual(
1523                     e.exception.http_code,
1524                     HTTPStatus.UNPROCESSABLE_ENTITY,
1525                     "Wrong HTTP status code",
1526                 )
1527                 self.assertIn(
1528                     norm(
1529                         "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1530                         "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1531                         "non existing ext-cpd:id inside vnfd '{}'".format(
1532                             df["id"],
1533                             affected_vnf_profile["id"],
1534                             affected_virtual_link["virtual-link-profile-id"],
1535                             affected_cpd["id"],
1536                             vnfd_descriptor["id"],
1537                         )
1538                     ),
1539                     norm(str(e.exception)),
1540                     "Wrong exception text",
1541                 )
1542             finally:
1543                 pass
1544         return
1545
1546     def test_edit_nsd(self):
1547         nsd_content = deepcopy(db_nsd_content)
1548         did = nsd_content["_id"]
1549         self.fs.file_exists.return_value = True
1550         self.fs.dir_ls.return_value = True
1551         with self.subTest(i=1, t="Normal Edition"):
1552             now = time()
1553             self.db.get_one.side_effect = [deepcopy(nsd_content), None]
1554             self.db.get_list.return_value = [db_vnfd_content]
1555             data = {"id": "new-nsd-id", "name": "new-nsd-name"}
1556             self.topic.edit(fake_session, did, data)
1557             db_args = self.db.replace.call_args[0]
1558             msg_args = self.msg.write.call_args[0]
1559             data["_id"] = did
1560             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1561             self.assertEqual(msg_args[1], "edited", "Wrong message action")
1562             self.assertEqual(msg_args[2], data, "Wrong message content")
1563             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1564             self.assertEqual(db_args[1], did, "Wrong DB ID")
1565             self.assertEqual(
1566                 db_args[2]["_admin"]["created"],
1567                 nsd_content["_admin"]["created"],
1568                 "Wrong creation time",
1569             )
1570             self.assertGreater(
1571                 db_args[2]["_admin"]["modified"], now, "Wrong modification time"
1572             )
1573             self.assertEqual(
1574                 db_args[2]["_admin"]["projects_read"],
1575                 nsd_content["_admin"]["projects_read"],
1576                 "Wrong read-only project list",
1577             )
1578             self.assertEqual(
1579                 db_args[2]["_admin"]["projects_write"],
1580                 nsd_content["_admin"]["projects_write"],
1581                 "Wrong read-write project list",
1582             )
1583             self.assertEqual(db_args[2]["id"], data["id"], "Wrong NSD ID")
1584             self.assertEqual(db_args[2]["name"], data["name"], "Wrong NSD Name")
1585         with self.subTest(i=2, t="Conflict on Edit"):
1586             data = {"id": "fake-nsd-id", "name": "new-nsd-name"}
1587             self.db.get_one.side_effect = [
1588                 nsd_content,
1589                 {"_id": str(uuid4()), "id": data["id"]},
1590             ]
1591             with self.assertRaises(
1592                 EngineException, msg="Accepted existing NSD ID"
1593             ) as e:
1594                 self.topic.edit(fake_session, did, data)
1595             self.assertEqual(
1596                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1597             )
1598             self.assertIn(
1599                 norm(
1600                     "{} with id '{}' already exists for this project".format(
1601                         "nsd", data["id"]
1602                     )
1603                 ),
1604                 norm(str(e.exception)),
1605                 "Wrong exception text",
1606             )
1607         with self.subTest(i=3, t="Check Envelope"):
1608             data = {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
1609             self.db.get_one.side_effect = [nsd_content, None]
1610             with self.assertRaises(
1611                 EngineException, msg="Accepted NSD with wrong envelope"
1612             ) as e:
1613                 self.topic.edit(fake_session, did, data, content=nsd_content)
1614             self.assertEqual(
1615                 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
1616             )
1617             self.assertIn(
1618                 "'nsd' must be a list of only one element",
1619                 norm(str(e.exception)),
1620                 "Wrong exception text",
1621             )
1622         return
1623
1624     def test_delete_nsd(self):
1625         did = db_nsd_content["_id"]
1626         self.db.get_one.return_value = db_nsd_content
1627         p_id = db_nsd_content["_admin"]["projects_read"][0]
1628         with self.subTest(i=1, t="Normal Deletion"):
1629             self.db.get_list.return_value = []
1630             self.db.del_one.return_value = {"deleted": 1}
1631             self.topic.delete(fake_session, did)
1632             db_args = self.db.del_one.call_args[0]
1633             msg_args = self.msg.write.call_args[0]
1634             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1635             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
1636             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
1637             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1638             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
1639             self.assertEqual(
1640                 db_args[1]["_admin.projects_write.cont"],
1641                 [p_id, "ANY"],
1642                 "Wrong DB filter",
1643             )
1644             db_g1_args = self.db.get_one.call_args[0]
1645             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
1646             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB NSD ID")
1647             db_gl_calls = self.db.get_list.call_args_list
1648             self.assertEqual(db_gl_calls[0][0][0], "nsrs", "Wrong DB topic")
1649             # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID")   # Filter changed after call
1650             self.assertEqual(db_gl_calls[1][0][0], "nsts", "Wrong DB topic")
1651             self.assertEqual(
1652                 db_gl_calls[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
1653                 db_nsd_content["id"],
1654                 "Wrong DB NSD netslice-subnet nsd-ref",
1655             )
1656             self.db.set_one.assert_not_called()
1657             fs_del_calls = self.fs.file_delete.call_args_list
1658             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
1659             self.assertEqual(fs_del_calls[1][0][0], did + "_", "Wrong FS folder id")
1660         with self.subTest(i=2, t="Conflict on Delete - NSD in use by nsr"):
1661             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-nsr"}]
1662             with self.assertRaises(
1663                 EngineException, msg="Accepted NSD in use by NSR"
1664             ) as e:
1665                 self.topic.delete(fake_session, did)
1666             self.assertEqual(
1667                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1668             )
1669             self.assertIn(
1670                 "there is at least one ns instance using this descriptor",
1671                 norm(str(e.exception)),
1672                 "Wrong exception text",
1673             )
1674         with self.subTest(i=3, t="Conflict on Delete - NSD in use by NST"):
1675             self.db.get_list.side_effect = [
1676                 [],
1677                 [{"_id": str(uuid4()), "name": "fake-nst"}],
1678             ]
1679             with self.assertRaises(
1680                 EngineException, msg="Accepted NSD in use by NST"
1681             ) as e:
1682                 self.topic.delete(fake_session, did)
1683             self.assertEqual(
1684                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1685             )
1686             self.assertIn(
1687                 "there is at least one netslice template referencing this descriptor",
1688                 norm(str(e.exception)),
1689                 "Wrong exception text",
1690             )
1691         with self.subTest(i=4, t="Non-existent NSD"):
1692             excp_msg = "Not found any {} with filter='{}'".format("NSD", {"_id": did})
1693             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
1694             with self.assertRaises(
1695                 DbException, msg="Accepted non-existent NSD ID"
1696             ) as e:
1697                 self.topic.delete(fake_session, did)
1698             self.assertEqual(
1699                 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
1700             )
1701             self.assertIn(
1702                 norm(excp_msg), norm(str(e.exception)), "Wrong exception text"
1703             )
1704         with self.subTest(i=5, t="No delete because referenced by other project"):
1705             db_nsd_content["_admin"]["projects_read"].append("other_project")
1706             self.db.get_one = Mock(return_value=db_nsd_content)
1707             self.db.get_list = Mock(return_value=[])
1708             self.msg.write.reset_mock()
1709             self.db.del_one.reset_mock()
1710             self.fs.file_delete.reset_mock()
1711
1712             self.topic.delete(fake_session, did)
1713             self.db.del_one.assert_not_called()
1714             self.msg.write.assert_not_called()
1715             db_g1_args = self.db.get_one.call_args[0]
1716             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
1717             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
1718             db_s1_args = self.db.set_one.call_args
1719             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
1720             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
1721             self.assertIn(
1722                 p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1723             )
1724             self.assertIsNone(
1725                 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
1726             )
1727             self.assertEqual(
1728                 db_s1_args[1]["pull_list"],
1729                 {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
1730                 "Wrong DB pull_list dictionary",
1731             )
1732             self.fs.file_delete.assert_not_called()
1733         return
1734
1735     def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
1736         self,
1737     ):
1738         indata = deepcopy(db_nsd_content)
1739         vld = indata["virtual-link-desc"][0]
1740         self.topic.validate_vld_mgmt_network_with_virtual_link_protocol_data(
1741             vld, indata
1742         )
1743
1744     def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
1745         self,
1746     ):
1747         indata = deepcopy(db_nsd_content)
1748         vld = indata["virtual-link-desc"][0]
1749         df = indata["df"][0]
1750         affected_vlp = {
1751             "id": "id",
1752             "virtual-link-desc-id": "mgmt",
1753             "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1754         }
1755         df["virtual-link-profile"] = [affected_vlp]
1756         with self.assertRaises(EngineException) as e:
1757             self.topic.validate_vld_mgmt_network_with_virtual_link_protocol_data(
1758                 vld, indata
1759             )
1760         self.assertEqual(
1761             e.exception.http_code,
1762             HTTPStatus.UNPROCESSABLE_ENTITY,
1763             "Wrong HTTP status code",
1764         )
1765         self.assertIn(
1766             norm(
1767                 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1768                 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1769                     df["id"], affected_vlp["id"]
1770                 )
1771             ),
1772             norm(str(e.exception)),
1773             "Wrong exception text",
1774         )
1775
1776     def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self):
1777         indata = deepcopy(db_nsd_content)
1778         self.topic.validate_vnf_profiles_vnfd_id(indata)
1779
1780     def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self):
1781         indata = deepcopy(db_nsd_content)
1782         df = indata["df"][0]
1783         affected_vnf_profile = df["vnf-profile"][0]
1784         indata["vnfd-id"] = ["non-existing-vnfd"]
1785         with self.assertRaises(EngineException) as e:
1786             self.topic.validate_vnf_profiles_vnfd_id(indata)
1787         self.assertEqual(
1788             e.exception.http_code,
1789             HTTPStatus.UNPROCESSABLE_ENTITY,
1790             "Wrong HTTP status code",
1791         )
1792         self.assertIn(
1793             norm(
1794                 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
1795                 "does not match any vnfd-id".format(
1796                     df["id"],
1797                     affected_vnf_profile["id"],
1798                     affected_vnf_profile["vnfd-id"],
1799                 )
1800             ),
1801             norm(str(e.exception)),
1802             "Wrong exception text",
1803         )
1804
1805     def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
1806         self,
1807     ):
1808         nsd_descriptor = deepcopy(db_nsd_content)
1809         vnfd_descriptor = deepcopy(db_vnfd_content)
1810         df = nsd_descriptor["df"][0]
1811         vnfds_index = {vnfd_descriptor["id"]: vnfd_descriptor}
1812         self.topic.validate_df_vnf_profiles_constituent_connection_points(
1813             df, vnfds_index
1814         )
1815
1816     def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
1817         self,
1818     ):
1819         nsd_descriptor = deepcopy(db_nsd_content)
1820         vnfd_descriptor = deepcopy(db_vnfd_content)
1821         df = nsd_descriptor["df"][0]
1822         affected_vnf_profile = df["vnf-profile"][0]
1823         affected_virtual_link = affected_vnf_profile["virtual-link-connectivity"][1]
1824         vnfds_index = {vnfd_descriptor["id"]: vnfd_descriptor}
1825         affected_cpd = vnfd_descriptor["ext-cpd"].pop()
1826         with self.assertRaises(EngineException) as e:
1827             self.topic.validate_df_vnf_profiles_constituent_connection_points(
1828                 df, vnfds_index
1829             )
1830         self.assertEqual(
1831             e.exception.http_code,
1832             HTTPStatus.UNPROCESSABLE_ENTITY,
1833             "Wrong HTTP status code",
1834         )
1835         self.assertIn(
1836             norm(
1837                 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1838                 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1839                 "non existing ext-cpd:id inside vnfd '{}'".format(
1840                     df["id"],
1841                     affected_vnf_profile["id"],
1842                     affected_virtual_link["virtual-link-profile-id"],
1843                     affected_cpd["id"],
1844                     vnfd_descriptor["id"],
1845                 )
1846             ),
1847             norm(str(e.exception)),
1848             "Wrong exception text",
1849         )
1850
1851     def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self):
1852         nsd_descriptor = deepcopy(db_nsd_content)
1853         invalid_vnfd_id = "invalid-vnfd-id"
1854         nsd_descriptor["id"] = "invalid-vnfd-id-ns"
1855         nsd_descriptor["vnfd-id"][0] = invalid_vnfd_id
1856         nsd_descriptor["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
1857         nsd_descriptor["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
1858         with self.assertRaises(EngineException) as e:
1859             self.db.get_list.return_value = []
1860             nsd_descriptor = self.topic.check_conflict_on_edit(
1861                 fake_session, nsd_descriptor, [], "id"
1862             )
1863         self.assertEqual(
1864             e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1865         )
1866         self.assertIn(
1867             norm(
1868                 "Descriptor error at 'vnfd-id'='{}' references a non "
1869                 "existing vnfd".format(invalid_vnfd_id)
1870             ),
1871             norm(str(e.exception)),
1872             "Wrong exception text",
1873         )
1874
1875
1876 if __name__ == "__main__":
1877     unittest.main()