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