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