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