osm_nbi/tests/test_descriptor_topics.py

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