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.safe_load(db_vnfds_text)[0]
  38 db_nsd_content = yaml.safe_load(db_nsds_text)[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         """Testing input validation during new vnfd creation
 405         for vdu internal connection point"""
 406         did, test_vnfd = self.prepare_vnfd_creation()
 407         test_vnfd = self.prepare_test_vnfd(test_vnfd)
 408         ext_cpd = test_vnfd["ext-cpd"][1]
 409         ext_cpd["int-cpd"]["cpd"] = "wrong-cpd"
 410
 411         with self.assertRaises(
 412             EngineException, msg="Accepted wrong ext-cpd internal connection point"
 413         ) as e:
 414             self.topic.upload_content(
 415                 fake_session, did, test_vnfd, {}, {"Content-Type": []}
 416             )
 417         self.assertEqual(
 418             e.exception.http_code,
 419             HTTPStatus.UNPROCESSABLE_ENTITY,
 420             "Wrong HTTP status code",
 421         )
 422         self.assertIn(
 423             norm(
 424                 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
 425                     ext_cpd["id"]
 426                 )
 427             ),
 428             norm(str(e.exception)),
 429             "Wrong exception text",
 430         )
 431
 432     @patch("osm_nbi.descriptor_topics.shutil")
 433     @patch("osm_nbi.descriptor_topics.os.rename")
 434     def test_new_vnfd_check_input_validation_duplicated_vld(
 435         self, mock_rename, mock_shutil
 436     ):
 437         """Testing input validation during new vnfd creation
 438         for dublicated virtual link description"""
 439         did, test_vnfd = self.prepare_vnfd_creation()
 440         test_vnfd = self.prepare_test_vnfd(test_vnfd)
 441         test_vnfd["int-virtual-link-desc"].insert(0, {"id": "internal"})
 442
 443         with self.assertRaises(
 444             EngineException, msg="Accepted duplicated VLD name"
 445         ) as e:
 446             self.topic.upload_content(
 447                 fake_session, did, test_vnfd, {}, {"Content-Type": []}
 448             )
 449         self.assertEqual(
 450             e.exception.http_code,
 451             HTTPStatus.UNPROCESSABLE_ENTITY,
 452             "Wrong HTTP status code",
 453         )
 454         self.assertIn(
 455             norm(
 456                 "identifier id '{}' is not unique".format(
 457                     test_vnfd["int-virtual-link-desc"][0]["id"]
 458                 )
 459             ),
 460             norm(str(e.exception)),
 461             "Wrong exception text",
 462         )
 463
 464     @patch("osm_nbi.descriptor_topics.shutil")
 465     @patch("osm_nbi.descriptor_topics.os.rename")
 466     def test_new_vnfd_check_input_validation_vdu_int_virtual_link_desc(
 467         self, mock_rename, mock_shutil
 468     ):
 469         """Testing input validation during new vnfd creation
 470         for vdu internal virtual link description"""
 471         did, test_vnfd = self.prepare_vnfd_creation()
 472         test_vnfd = self.prepare_test_vnfd(test_vnfd)
 473         vdu = test_vnfd["vdu"][0]
 474         int_cpd = vdu["int-cpd"][1]
 475         int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
 476
 477         with self.assertRaises(
 478             EngineException, msg="Accepted int-virtual-link-desc"
 479         ) as e:
 480             self.topic.upload_content(
 481                 fake_session, did, test_vnfd, {}, {"Content-Type": []}
 482             )
 483         self.assertEqual(
 484             e.exception.http_code,
 485             HTTPStatus.UNPROCESSABLE_ENTITY,
 486             "Wrong HTTP status code",
 487         )
 488         self.assertIn(
 489             norm(
 490                 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
 491                 "int-virtual-link-desc".format(
 492                     vdu["id"], int_cpd["id"], int_cpd["int-virtual-link-desc"]
 493                 )
 494             ),
 495             norm(str(e.exception)),
 496             "Wrong exception text",
 497         )
 498
 499     @patch("osm_nbi.descriptor_topics.shutil")
 500     @patch("osm_nbi.descriptor_topics.os.rename")
 501     def test_new_vnfd_check_input_validation_virtual_link_profile(
 502         self, mock_rename, mock_shutil
 503     ):
 504         """Testing input validation during new vnfd creation
 505         for virtual link profile"""
 506         did, test_vnfd = self.prepare_vnfd_creation()
 507         test_vnfd = self.prepare_test_vnfd(test_vnfd)
 508         fake_ivld_profile = {"id": "fake-profile-ref", "flavour": "fake-flavour"}
 509         df = test_vnfd["df"][0]
 510         df["virtual-link-profile"] = [fake_ivld_profile]
 511
 512         with self.assertRaises(
 513             EngineException, msg="Accepted non-existent Profile Ref"
 514         ) as e:
 515             self.topic.upload_content(
 516                 fake_session, did, test_vnfd, {}, {"Content-Type": []}
 517             )
 518         self.assertEqual(
 519             e.exception.http_code,
 520             HTTPStatus.UNPROCESSABLE_ENTITY,
 521             "Wrong HTTP status code",
 522         )
 523         self.assertIn(
 524             norm(
 525                 "df[id='{}']:virtual-link-profile='{}' must match an existing "
 526                 "int-virtual-link-desc".format(df["id"], fake_ivld_profile["id"])
 527             ),
 528             norm(str(e.exception)),
 529             "Wrong exception text",
 530         )
 531
 532     @patch("osm_nbi.descriptor_topics.shutil")
 533     @patch("osm_nbi.descriptor_topics.os.rename")
 534     def test_new_vnfd_check_input_validation_scaling_criteria_monitoring_param_ref(
 535         self, mock_rename, mock_shutil
 536     ):
 537         """Testing input validation during new vnfd creation
 538         for scaling criteria without monitoring parameter"""
 539         did, test_vnfd = self.prepare_vnfd_creation()
 540         test_vnfd = self.prepare_test_vnfd(test_vnfd)
 541         vdu = test_vnfd["vdu"][1]
 542         affected_df = test_vnfd["df"][0]
 543         sa = affected_df["scaling-aspect"][0]
 544         sp = sa["scaling-policy"][0]
 545         sc = sp["scaling-criteria"][0]
 546         vdu.pop("monitoring-parameter")
 547
 548         with self.assertRaises(
 549             EngineException, msg="Accepted non-existent Scaling Group Policy Criteria"
 550         ) as e:
 551             self.topic.upload_content(
 552                 fake_session, did, test_vnfd, {}, {"Content-Type": []}
 553             )
 554         self.assertEqual(
 555             e.exception.http_code,
 556             HTTPStatus.UNPROCESSABLE_ENTITY,
 557             "Wrong HTTP status code",
 558         )
 559         self.assertIn(
 560             norm(
 561                 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
 562                 "[name='{}']:scaling-criteria[name='{}']: "
 563                 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
 564                     affected_df["id"],
 565                     sa["id"],
 566                     sp["name"],
 567                     sc["name"],
 568                     sc["vnf-monitoring-param-ref"],
 569                 )
 570             ),
 571             norm(str(e.exception)),
 572             "Wrong exception text",
 573         )
 574
 575     @patch("osm_nbi.descriptor_topics.shutil")
 576     @patch("osm_nbi.descriptor_topics.os.rename")
 577     def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration(
 578         self, mock_rename, mock_shutil
 579     ):
 580         """Testing input validation during new vnfd creation
 581         for scaling criteria without day12 configuration"""
 582         did, test_vnfd = self.prepare_vnfd_creation()
 583         test_vnfd = self.prepare_test_vnfd(test_vnfd)
 584         test_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
 585             "day1-2"
 586         ].pop()
 587         df = test_vnfd["df"][0]
 588
 589         with self.assertRaises(
 590             EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference"
 591         ) as e:
 592             self.topic.upload_content(
 593                 fake_session, did, test_vnfd, {}, {"Content-Type": []}
 594             )
 595         self.assertEqual(
 596             e.exception.http_code,
 597             HTTPStatus.UNPROCESSABLE_ENTITY,
 598             "Wrong HTTP status code",
 599         )
 600         self.assertIn(
 601             norm(
 602                 "'day1-2 configuration' not defined in the descriptor but it is referenced "
 603                 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
 604                     df["id"], df["scaling-aspect"][0]["id"]
 605                 )
 606             ),
 607             norm(str(e.exception)),
 608             "Wrong exception text",
 609         )
 610
 611     @patch("osm_nbi.descriptor_topics.shutil")
 612     @patch("osm_nbi.descriptor_topics.os.rename")
 613     def test_new_vnfd_check_input_validation_scaling_config_action(
 614         self, mock_rename, mock_shutil
 615     ):
 616         """Testing input validation during new vnfd creation
 617         for scaling criteria wrong config primitive"""
 618         did, test_vnfd = self.prepare_vnfd_creation()
 619         test_vnfd = self.prepare_test_vnfd(test_vnfd)
 620         df = test_vnfd["df"][0]
 621         affected_df = test_vnfd["df"][0]
 622         sa = affected_df["scaling-aspect"][0]
 623         test_vnfd["df"][0].get("lcm-operations-configuration").get(
 624             "operate-vnf-op-config"
 625         )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
 626
 627         with self.assertRaises(
 628             EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference"
 629         ) as e:
 630             self.topic.upload_content(
 631                 fake_session, did, test_vnfd, {}, {"Content-Type": []}
 632             )
 633         self.assertEqual(
 634             e.exception.http_code,
 635             HTTPStatus.UNPROCESSABLE_ENTITY,
 636             "Wrong HTTP status code",
 637         )
 638         self.assertIn(
 639             norm(
 640                 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
 641                 "config-primitive-name-ref='{}' does not match any "
 642                 "day1-2 configuration:config-primitive:name".format(
 643                     df["id"],
 644                     df["scaling-aspect"][0]["id"],
 645                     sa["scaling-config-action"][0]["vnf-config-primitive-name-ref"],
 646                 )
 647             ),
 648             norm(str(e.exception)),
 649             "Wrong exception text",
 650         )
 651
 652     @patch("osm_nbi.descriptor_topics.shutil")
 653     @patch("osm_nbi.descriptor_topics.os.rename")
 654     def test_new_vnfd_check_input_validation_everything_right(
 655         self, mock_rename, mock_shutil
 656     ):
 657         """Testing input validation during new vnfd creation
 658         everything correct"""
 659         did, test_vnfd = self.prepare_vnfd_creation()
 660         test_vnfd = self.prepare_test_vnfd(test_vnfd)
 661         test_vnfd["id"] = "fake-vnfd-id"
 662         test_vnfd["df"][0].get("lcm-operations-configuration").get(
 663             "operate-vnf-op-config"
 664         )["day1-2"][0]["id"] = "fake-vnfd-id"
 665         self.db.get_one.side_effect = [
 666             {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])},
 667             None,
 668         ]
 669         rc = self.topic.upload_content(
 670             fake_session, did, test_vnfd, {}, {"Content-Type": []}
 671         )
 672         self.assertTrue(rc, "Input Validation: Unexpected failure")
 673
 674     def test_edit_vnfd(self):
 675         vnfd_content = deepcopy(db_vnfd_content)
 676         did = vnfd_content["_id"]
 677         self.fs.file_exists.return_value = True
 678         self.fs.dir_ls.return_value = True
 679         with self.subTest(i=1, t="Normal Edition"):
 680             now = time()
 681             self.db.get_one.side_effect = [deepcopy(vnfd_content), None]
 682             data = {"product-name": "new-vnfd-name"}
 683             self.topic.edit(fake_session, did, data)
 684             db_args = self.db.replace.call_args[0]
 685             msg_args = self.msg.write.call_args[0]
 686             data["_id"] = did
 687             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 688             self.assertEqual(msg_args[1], "edited", "Wrong message action")
 689             self.assertEqual(msg_args[2], data, "Wrong message content")
 690             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 691             self.assertEqual(db_args[1], did, "Wrong DB ID")
 692             self.assertEqual(
 693                 db_args[2]["_admin"]["created"],
 694                 vnfd_content["_admin"]["created"],
 695                 "Wrong creation time",
 696             )
 697             self.assertGreater(
 698                 db_args[2]["_admin"]["modified"], now, "Wrong modification time"
 699             )
 700             self.assertEqual(
 701                 db_args[2]["_admin"]["projects_read"],
 702                 vnfd_content["_admin"]["projects_read"],
 703                 "Wrong read-only project list",
 704             )
 705             self.assertEqual(
 706                 db_args[2]["_admin"]["projects_write"],
 707                 vnfd_content["_admin"]["projects_write"],
 708                 "Wrong read-write project list",
 709             )
 710             self.assertEqual(
 711                 db_args[2]["product-name"], data["product-name"], "Wrong VNFD Name"
 712             )
 713         with self.subTest(i=2, t="Conflict on Edit"):
 714             data = {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
 715             self.db.get_one.side_effect = [
 716                 deepcopy(vnfd_content),
 717                 {"_id": str(uuid4()), "id": data["id"]},
 718             ]
 719             with self.assertRaises(
 720                 EngineException, msg="Accepted existing VNFD ID"
 721             ) as e:
 722                 self.topic.edit(fake_session, did, data)
 723             self.assertEqual(
 724                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
 725             )
 726             self.assertIn(
 727                 norm(
 728                     "{} with id '{}' already exists for this project".format(
 729                         "vnfd", data["id"]
 730                     )
 731                 ),
 732                 norm(str(e.exception)),
 733                 "Wrong exception text",
 734             )
 735         with self.subTest(i=3, t="Check Envelope"):
 736             data = {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
 737             with self.assertRaises(
 738                 EngineException, msg="Accepted VNFD with wrong envelope"
 739             ) as e:
 740                 self.topic.edit(fake_session, did, data, content=vnfd_content)
 741             self.assertEqual(
 742                 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
 743             )
 744             self.assertIn(
 745                 "'vnfd' must be dict", norm(str(e.exception)), "Wrong exception text"
 746             )
 747         return
 748
 749     def test_delete_vnfd(self):
 750         did = db_vnfd_content["_id"]
 751         self.db.get_one.return_value = db_vnfd_content
 752         p_id = db_vnfd_content["_admin"]["projects_read"][0]
 753         with self.subTest(i=1, t="Normal Deletion"):
 754             self.db.get_list.return_value = []
 755             self.db.del_one.return_value = {"deleted": 1}
 756             self.topic.delete(fake_session, did)
 757             db_args = self.db.del_one.call_args[0]
 758             msg_args = self.msg.write.call_args[0]
 759             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 760             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 761             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 762             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 763             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
 764             self.assertEqual(
 765                 db_args[1]["_admin.projects_write.cont"],
 766                 [p_id, "ANY"],
 767                 "Wrong DB filter",
 768             )
 769             db_g1_args = self.db.get_one.call_args[0]
 770             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 771             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 772             db_gl_calls = self.db.get_list.call_args_list
 773             self.assertEqual(db_gl_calls[0][0][0], "vnfrs", "Wrong DB topic")
 774             # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID")   # Filter changed after call
 775             self.assertEqual(db_gl_calls[1][0][0], "nsds", "Wrong DB topic")
 776             self.assertEqual(
 777                 db_gl_calls[1][0][1]["vnfd-id"],
 778                 db_vnfd_content["id"],
 779                 "Wrong DB NSD vnfd-id",
 780             )
 781
 782             self.assertEqual(
 783                 self.db.del_list.call_args[0][0],
 784                 self.topic.topic + "_revisions",
 785                 "Wrong DB topic",
 786             )
 787
 788             self.assertEqual(
 789                 self.db.del_list.call_args[0][1]["_id"]["$regex"],
 790                 did,
 791                 "Wrong ID for rexep delete",
 792             )
 793
 794             self.db.set_one.assert_not_called()
 795             fs_del_calls = self.fs.file_delete.call_args_list
 796             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
 797             self.assertEqual(fs_del_calls[1][0][0], did + "_", "Wrong FS folder id")
 798         with self.subTest(i=2, t="Conflict on Delete - VNFD in use by VNFR"):
 799             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-vnfr"}]
 800             with self.assertRaises(
 801                 EngineException, msg="Accepted VNFD in use by VNFR"
 802             ) as e:
 803                 self.topic.delete(fake_session, did)
 804             self.assertEqual(
 805                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
 806             )
 807             self.assertIn(
 808                 "there is at least one vnf instance using this descriptor",
 809                 norm(str(e.exception)),
 810                 "Wrong exception text",
 811             )
 812         with self.subTest(i=3, t="Conflict on Delete - VNFD in use by NSD"):
 813             self.db.get_list.side_effect = [
 814                 [],
 815                 [{"_id": str(uuid4()), "name": "fake-nsd"}],
 816             ]
 817             with self.assertRaises(
 818                 EngineException, msg="Accepted VNFD in use by NSD"
 819             ) as e:
 820                 self.topic.delete(fake_session, did)
 821             self.assertEqual(
 822                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
 823             )
 824             self.assertIn(
 825                 "there is at least one ns package referencing this descriptor",
 826                 norm(str(e.exception)),
 827                 "Wrong exception text",
 828             )
 829         with self.subTest(i=4, t="Non-existent VNFD"):
 830             excp_msg = "Not found any {} with filter='{}'".format("VNFD", {"_id": did})
 831             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
 832             with self.assertRaises(
 833                 DbException, msg="Accepted non-existent VNFD ID"
 834             ) as e:
 835                 self.topic.delete(fake_session, did)
 836             self.assertEqual(
 837                 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
 838             )
 839             self.assertIn(
 840                 norm(excp_msg), norm(str(e.exception)), "Wrong exception text"
 841             )
 842         with self.subTest(i=5, t="No delete because referenced by other project"):
 843             db_vnfd_content["_admin"]["projects_read"].append("other_project")
 844             self.db.get_one = Mock(return_value=db_vnfd_content)
 845             self.db.get_list = Mock(return_value=[])
 846             self.msg.write.reset_mock()
 847             self.db.del_one.reset_mock()
 848             self.fs.file_delete.reset_mock()
 849
 850             self.topic.delete(fake_session, did)
 851             self.db.del_one.assert_not_called()
 852             self.msg.write.assert_not_called()
 853             db_g1_args = self.db.get_one.call_args[0]
 854             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 855             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 856             db_s1_args = self.db.set_one.call_args
 857             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
 858             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
 859             self.assertIn(
 860                 p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
 861             )
 862             self.assertIsNone(
 863                 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
 864             )
 865             self.assertEqual(
 866                 db_s1_args[1]["pull_list"],
 867                 {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
 868                 "Wrong DB pull_list dictionary",
 869             )
 870             self.fs.file_delete.assert_not_called()
 871         return
 872
 873     def prepare_vnfd_validation(self):
 874         descriptor_name = "test_descriptor"
 875         self.fs.file_open.side_effect = lambda path, mode: open(
 876             "/tmp/" + str(uuid4()), "a+b"
 877         )
 878         old_vnfd, new_vnfd = self.create_desc_temp(db_vnfd_content)
 879         return descriptor_name, old_vnfd, new_vnfd
 880
 881     @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
 882     @patch("osm_nbi.descriptor_topics.yaml.safe_load")
 883     def test_validate_vnfd_changes_day12_config_primitive_changed(
 884         self, mock_safe_load, mock_detect_usage
 885     ):
 886         """Validating VNFD for VNFD updates, day1-2 config primitive has changed"""
 887         descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
 888         did = old_vnfd["_id"]
 889         new_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
 890             "day1-2"
 891         ][0]["config-primitive"][0]["name"] = "new_action"
 892         mock_safe_load.side_effect = [old_vnfd, new_vnfd]
 893         mock_detect_usage.return_value = True
 894         self.db.get_one.return_value = old_vnfd
 895
 896         with self.assertNotRaises(EngineException):
 897             self.topic._validate_descriptor_changes(
 898                 did, descriptor_name, "/tmp/", "/tmp:1/"
 899             )
 900         self.db.get_one.assert_called_once()
 901         mock_detect_usage.assert_called_once()
 902         self.assertEqual(mock_safe_load.call_count, 2)
 903
 904     @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
 905     @patch("osm_nbi.descriptor_topics.yaml.safe_load")
 906     def test_validate_vnfd_changes_sw_version_changed(
 907         self, mock_safe_load, mock_detect_usage
 908     ):
 909         """Validating VNFD for updates, software version has changed"""
 910         # old vnfd uses the default software version: 1.0
 911         descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
 912         did = old_vnfd["_id"]
 913         new_vnfd["software-version"] = "1.3"
 914         new_vnfd["sw-image-desc"][0]["name"] = "new-image"
 915         mock_safe_load.side_effect = [old_vnfd, new_vnfd]
 916         mock_detect_usage.return_value = True
 917         self.db.get_one.return_value = old_vnfd
 918
 919         with self.assertNotRaises(EngineException):
 920             self.topic._validate_descriptor_changes(
 921                 did, descriptor_name, "/tmp/", "/tmp:1/"
 922             )
 923         self.db.get_one.assert_called_once()
 924         mock_detect_usage.assert_called_once()
 925         self.assertEqual(mock_safe_load.call_count, 2)
 926
 927     @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
 928     @patch("osm_nbi.descriptor_topics.yaml.safe_load")
 929     def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed(
 930         self, mock_safe_load, mock_detect_usage
 931     ):
 932         """Validating VNFD for updates, software version has not
 933         changed, mgmt-cp has changed."""
 934         descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
 935         new_vnfd["mgmt-cp"] = "new-mgmt-cp"
 936         mock_safe_load.side_effect = [old_vnfd, new_vnfd]
 937         did = old_vnfd["_id"]
 938         mock_detect_usage.return_value = True
 939         self.db.get_one.return_value = old_vnfd
 940
 941         with self.assertRaises(
 942             EngineException, msg="there are disallowed changes in the vnf descriptor"
 943         ) as e:
 944             self.topic._validate_descriptor_changes(
 945                 did, descriptor_name, "/tmp/", "/tmp:1/"
 946             )
 947
 948         self.assertEqual(
 949             e.exception.http_code,
 950             HTTPStatus.UNPROCESSABLE_ENTITY,
 951             "Wrong HTTP status code",
 952         )
 953         self.assertIn(
 954             norm("there are disallowed changes in the vnf descriptor"),
 955             norm(str(e.exception)),
 956             "Wrong exception text",
 957         )
 958         self.db.get_one.assert_called_once()
 959         mock_detect_usage.assert_called_once()
 960         self.assertEqual(mock_safe_load.call_count, 2)
 961
 962     @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
 963     @patch("osm_nbi.descriptor_topics.yaml.safe_load")
 964     def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed_vnfd_not_in_use(
 965         self, mock_safe_load, mock_detect_usage
 966     ):
 967         """Validating VNFD for updates, software version has not
 968         changed, mgmt-cp has changed, vnfd is not in use."""
 969         descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
 970         new_vnfd["mgmt-cp"] = "new-mgmt-cp"
 971         mock_safe_load.side_effect = [old_vnfd, new_vnfd]
 972         did = old_vnfd["_id"]
 973         mock_detect_usage.return_value = None
 974         self.db.get_one.return_value = old_vnfd
 975
 976         with self.assertNotRaises(EngineException):
 977             self.topic._validate_descriptor_changes(
 978                 did, descriptor_name, "/tmp/", "/tmp:1/"
 979             )
 980
 981         self.db.get_one.assert_called_once()
 982         mock_detect_usage.assert_called_once()
 983         mock_safe_load.assert_not_called()
 984
 985     def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self):
 986         indata = deepcopy(db_vnfd_content)
 987         self.topic.validate_mgmt_interface_connection_point(indata)
 988
 989     def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
 990         self,
 991     ):
 992         indata = deepcopy(db_vnfd_content)
 993         indata["ext-cpd"] = []
 994         with self.assertRaises(EngineException) as e:
 995             self.topic.validate_mgmt_interface_connection_point(indata)
 996         self.assertEqual(
 997             e.exception.http_code,
 998             HTTPStatus.UNPROCESSABLE_ENTITY,
 999             "Wrong HTTP status code",
1000         )
1001         self.assertIn(
1002             norm(
1003                 "mgmt-cp='{}' must match an existing ext-cpd".format(indata["mgmt-cp"])
1004             ),
1005             norm(str(e.exception)),
1006             "Wrong exception text",
1007         )
1008
1009     def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self):
1010         indata = deepcopy(db_vnfd_content)
1011         indata.pop("mgmt-cp")
1012         with self.assertRaises(EngineException) as e:
1013             self.topic.validate_mgmt_interface_connection_point(indata)
1014         self.assertEqual(
1015             e.exception.http_code,
1016             HTTPStatus.UNPROCESSABLE_ENTITY,
1017             "Wrong HTTP status code",
1018         )
1019         self.assertIn(
1020             norm("'mgmt-cp' is a mandatory field and it is not defined"),
1021             norm(str(e.exception)),
1022             "Wrong exception text",
1023         )
1024
1025     def test_validate_vdu_internal_connection_points_on_valid_descriptor(self):
1026         indata = db_vnfd_content
1027         vdu = indata["vdu"][0]
1028         self.topic.validate_vdu_internal_connection_points(vdu)
1029
1030     def test_validate_external_connection_points_on_valid_descriptor(self):
1031         indata = db_vnfd_content
1032         self.topic.validate_external_connection_points(indata)
1033
1034     def test_validate_external_connection_points_when_missing_internal_connection_point(
1035         self,
1036     ):
1037         indata = deepcopy(db_vnfd_content)
1038         vdu = indata["vdu"][0]
1039         vdu.pop("int-cpd")
1040         affected_ext_cpd = indata["ext-cpd"][0]
1041         with self.assertRaises(EngineException) as e:
1042             self.topic.validate_external_connection_points(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                 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
1051                     affected_ext_cpd["id"]
1052                 )
1053             ),
1054             norm(str(e.exception)),
1055             "Wrong exception text",
1056         )
1057
1058     def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
1059         self,
1060     ):
1061         indata = deepcopy(db_vnfd_content)
1062         vdu = indata["vdu"][0]
1063         duplicated_cpd = {
1064             "id": "vnf-mgmt",
1065             "order": 3,
1066             "virtual-network-interface-requirement": [{"name": "duplicated"}],
1067         }
1068         vdu["int-cpd"].insert(0, duplicated_cpd)
1069         with self.assertRaises(EngineException) as e:
1070             self.topic.validate_vdu_internal_connection_points(vdu)
1071         self.assertEqual(
1072             e.exception.http_code,
1073             HTTPStatus.UNPROCESSABLE_ENTITY,
1074             "Wrong HTTP status code",
1075         )
1076         self.assertIn(
1077             norm(
1078                 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
1079                     vdu["id"], duplicated_cpd["id"]
1080                 )
1081             ),
1082             norm(str(e.exception)),
1083             "Wrong exception text",
1084         )
1085
1086     def test_validate_external_connection_points_on_duplicated_external_connection_point(
1087         self,
1088     ):
1089         indata = deepcopy(db_vnfd_content)
1090         duplicated_cpd = {
1091             "id": "vnf-mgmt-ext",
1092             "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
1093         }
1094         indata["ext-cpd"].insert(0, duplicated_cpd)
1095         with self.assertRaises(EngineException) as e:
1096             self.topic.validate_external_connection_points(indata)
1097         self.assertEqual(
1098             e.exception.http_code,
1099             HTTPStatus.UNPROCESSABLE_ENTITY,
1100             "Wrong HTTP status code",
1101         )
1102         self.assertIn(
1103             norm(
1104                 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
1105                     duplicated_cpd["id"]
1106                 )
1107             ),
1108             norm(str(e.exception)),
1109             "Wrong exception text",
1110         )
1111
1112     def test_validate_internal_virtual_links_on_valid_descriptor(self):
1113         indata = db_vnfd_content
1114         self.topic.validate_internal_virtual_links(indata)
1115
1116     def test_validate_internal_virtual_links_on_duplicated_ivld(self):
1117         indata = deepcopy(db_vnfd_content)
1118         duplicated_vld = {"id": "internal"}
1119         indata["int-virtual-link-desc"].insert(0, duplicated_vld)
1120         with self.assertRaises(EngineException) as e:
1121             self.topic.validate_internal_virtual_links(indata)
1122         self.assertEqual(
1123             e.exception.http_code,
1124             HTTPStatus.UNPROCESSABLE_ENTITY,
1125             "Wrong HTTP status code",
1126         )
1127         self.assertIn(
1128             norm(
1129                 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
1130                     duplicated_vld["id"]
1131                 )
1132             ),
1133             norm(str(e.exception)),
1134             "Wrong exception text",
1135         )
1136
1137     def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
1138         self,
1139     ):
1140         indata = deepcopy(db_vnfd_content)
1141         vdu = indata["vdu"][0]
1142         affected_int_cpd = vdu["int-cpd"][0]
1143         affected_int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
1144         with self.assertRaises(EngineException) as e:
1145             self.topic.validate_internal_virtual_links(indata)
1146         self.assertEqual(
1147             e.exception.http_code,
1148             HTTPStatus.UNPROCESSABLE_ENTITY,
1149             "Wrong HTTP status code",
1150         )
1151         self.assertIn(
1152             norm(
1153                 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
1154                 "int-virtual-link-desc".format(
1155                     vdu["id"],
1156                     affected_int_cpd["id"],
1157                     affected_int_cpd["int-virtual-link-desc"],
1158                 )
1159             ),
1160             norm(str(e.exception)),
1161             "Wrong exception text",
1162         )
1163
1164     def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self):
1165         indata = deepcopy(db_vnfd_content)
1166         affected_ivld_profile = {"id": "non-existing-int-virtual-link-desc"}
1167         df = indata["df"][0]
1168         df["virtual-link-profile"] = [affected_ivld_profile]
1169         with self.assertRaises(EngineException) as e:
1170             self.topic.validate_internal_virtual_links(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                 "df[id='{}']:virtual-link-profile='{}' must match an existing "
1179                 "int-virtual-link-desc".format(df["id"], affected_ivld_profile["id"])
1180             ),
1181             norm(str(e.exception)),
1182             "Wrong exception text",
1183         )
1184
1185     def test_validate_monitoring_params_on_valid_descriptor(self):
1186         indata = db_vnfd_content
1187         self.topic.validate_monitoring_params(indata)
1188
1189     def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self):
1190         indata = deepcopy(db_vnfd_content)
1191         duplicated_mp = {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1192         affected_ivld = indata["int-virtual-link-desc"][0]
1193         affected_ivld["monitoring-parameters"] = [duplicated_mp, duplicated_mp]
1194         with self.assertRaises(EngineException) as e:
1195             self.topic.validate_monitoring_params(indata)
1196         self.assertEqual(
1197             e.exception.http_code,
1198             HTTPStatus.UNPROCESSABLE_ENTITY,
1199             "Wrong HTTP status code",
1200         )
1201         self.assertIn(
1202             norm(
1203                 "Duplicated monitoring-parameter id in "
1204                 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1205                     affected_ivld["id"], duplicated_mp["id"]
1206                 )
1207             ),
1208             norm(str(e.exception)),
1209             "Wrong exception text",
1210         )
1211
1212     def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self):
1213         indata = deepcopy(db_vnfd_content)
1214         duplicated_mp = {
1215             "id": "dataVM_cpu_util",
1216             "name": "dataVM_cpu_util",
1217             "performance_metric": "cpu",
1218         }
1219         affected_vdu = indata["vdu"][1]
1220         affected_vdu["monitoring-parameter"].insert(0, duplicated_mp)
1221         with self.assertRaises(EngineException) as e:
1222             self.topic.validate_monitoring_params(indata)
1223         self.assertEqual(
1224             e.exception.http_code,
1225             HTTPStatus.UNPROCESSABLE_ENTITY,
1226             "Wrong HTTP status code",
1227         )
1228         self.assertIn(
1229             norm(
1230                 "Duplicated monitoring-parameter id in "
1231                 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1232                     affected_vdu["id"], duplicated_mp["id"]
1233                 )
1234             ),
1235             norm(str(e.exception)),
1236             "Wrong exception text",
1237         )
1238
1239     def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self):
1240         indata = deepcopy(db_vnfd_content)
1241         duplicated_mp = {
1242             "id": "memory",
1243             "name": "memory",
1244             "performance_metric": "memory",
1245         }
1246         affected_df = indata["df"][0]
1247         affected_df["monitoring-parameter"] = [duplicated_mp, duplicated_mp]
1248         with self.assertRaises(EngineException) as e:
1249             self.topic.validate_monitoring_params(indata)
1250         self.assertEqual(
1251             e.exception.http_code,
1252             HTTPStatus.UNPROCESSABLE_ENTITY,
1253             "Wrong HTTP status code",
1254         )
1255         self.assertIn(
1256             norm(
1257                 "Duplicated monitoring-parameter id in "
1258                 "df[id='{}']:monitoring-parameter[id='{}']".format(
1259                     affected_df["id"], duplicated_mp["id"]
1260                 )
1261             ),
1262             norm(str(e.exception)),
1263             "Wrong exception text",
1264         )
1265
1266     def test_validate_scaling_group_descriptor_on_valid_descriptor(self):
1267         indata = db_vnfd_content
1268         self.topic.validate_scaling_group_descriptor(indata)
1269
1270     def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self):
1271         indata = deepcopy(db_vnfd_content)
1272         vdu = indata["vdu"][1]
1273         affected_df = indata["df"][0]
1274         affected_sa = affected_df["scaling-aspect"][0]
1275         affected_sp = affected_sa["scaling-policy"][0]
1276         affected_sc = affected_sp["scaling-criteria"][0]
1277         vdu.pop("monitoring-parameter")
1278         with self.assertRaises(EngineException) as e:
1279             self.topic.validate_scaling_group_descriptor(indata)
1280         self.assertEqual(
1281             e.exception.http_code,
1282             HTTPStatus.UNPROCESSABLE_ENTITY,
1283             "Wrong HTTP status code",
1284         )
1285         self.assertIn(
1286             norm(
1287                 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1288                 "[name='{}']:scaling-criteria[name='{}']: "
1289                 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1290                     affected_df["id"],
1291                     affected_sa["id"],
1292                     affected_sp["name"],
1293                     affected_sc["name"],
1294                     affected_sc["vnf-monitoring-param-ref"],
1295                 )
1296             ),
1297             norm(str(e.exception)),
1298             "Wrong exception text",
1299         )
1300
1301     def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self):
1302         indata = deepcopy(db_vnfd_content)
1303         df = indata["df"][0]
1304         affected_sa = df["scaling-aspect"][0]
1305         indata["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1306             "day1-2"
1307         ].pop()
1308         with self.assertRaises(EngineException) as e:
1309             self.topic.validate_scaling_group_descriptor(indata)
1310         self.assertEqual(
1311             e.exception.http_code,
1312             HTTPStatus.UNPROCESSABLE_ENTITY,
1313             "Wrong HTTP status code",
1314         )
1315         self.assertIn(
1316             norm(
1317                 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1318                 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1319                     df["id"], affected_sa["id"]
1320                 )
1321             ),
1322             norm(str(e.exception)),
1323             "Wrong exception text",
1324         )
1325
1326     def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1327         self,
1328     ):
1329         indata = deepcopy(db_vnfd_content)
1330         df = indata["df"][0]
1331         affected_sa = df["scaling-aspect"][0]
1332         affected_sca_primitive = affected_sa["scaling-config-action"][0][
1333             "vnf-config-primitive-name-ref"
1334         ]
1335         df["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1336             "config-primitive"
1337         ] = []
1338         with self.assertRaises(EngineException) as e:
1339             self.topic.validate_scaling_group_descriptor(indata)
1340         self.assertEqual(
1341             e.exception.http_code,
1342             HTTPStatus.UNPROCESSABLE_ENTITY,
1343             "Wrong HTTP status code",
1344         )
1345         self.assertIn(
1346             norm(
1347                 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1348                 "config-primitive-name-ref='{}' does not match any "
1349                 "day1-2 configuration:config-primitive:name".format(
1350                     df["id"], affected_sa["id"], affected_sca_primitive
1351                 )
1352             ),
1353             norm(str(e.exception)),
1354             "Wrong exception text",
1355         )
1356
1357     def test_new_vnfd_revision(self):
1358         did = db_vnfd_content["_id"]
1359         self.fs.get_params.return_value = {}
1360         self.fs.file_exists.return_value = False
1361         self.fs.file_open.side_effect = lambda path, mode: open(
1362             "/tmp/" + str(uuid4()), "a+b"
1363         )
1364         test_vnfd = deepcopy(db_vnfd_content)
1365         del test_vnfd["_id"]
1366         del test_vnfd["_admin"]
1367         self.db.create.return_value = did
1368         rollback = []
1369         did2, oid = self.topic.new(rollback, fake_session, {})
1370         db_args = self.db.create.call_args[0]
1371         self.assertEqual(
1372             db_args[1]["_admin"]["revision"], 0, "New package should be at revision 0"
1373         )
1374
1375     @patch("osm_nbi.descriptor_topics.shutil")
1376     @patch("osm_nbi.descriptor_topics.os.rename")
1377     def test_update_vnfd(self, mock_rename, mock_shutil):
1378         old_revision = 5
1379         did = db_vnfd_content["_id"]
1380         self.fs.path = ""
1381         self.fs.get_params.return_value = {}
1382         self.fs.file_exists.return_value = False
1383         self.fs.file_open.side_effect = lambda path, mode: open(
1384             "/tmp/" + str(uuid4()), "a+b"
1385         )
1386         new_vnfd = deepcopy(db_vnfd_content)
1387         del new_vnfd["_id"]
1388         self.db.create.return_value = did
1389         rollback = []
1390         did2, oid = self.topic.new(rollback, fake_session, {})
1391         del new_vnfd["vdu"][0]["cloud-init-file"]
1392         del new_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1393             "day1-2"
1394         ][0]["execution-environment-list"][0]["juju"]
1395
1396         old_vnfd = {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}
1397         old_vnfd["_admin"]["revision"] = old_revision
1398
1399         self.db.get_one.side_effect = [old_vnfd, old_vnfd, None]
1400         self.topic.upload_content(fake_session, did, new_vnfd, {}, {"Content-Type": []})
1401
1402         db_args = self.db.replace.call_args[0]
1403         self.assertEqual(
1404             db_args[2]["_admin"]["revision"],
1405             old_revision + 1,
1406             "Revision should increment",
1407         )
1408
1409
1410 class Test_NsdTopic(TestCase):
1411     @classmethod
1412     def setUpClass(cls):
1413         cls.test_name = "test-nsd-topic"
1414
1415     @classmethod
1416     def tearDownClass(cls):
1417         pass
1418
1419     def setUp(self):
1420         self.db = Mock(dbbase.DbBase())
1421         self.fs = Mock(fsbase.FsBase())
1422         self.msg = Mock(msgbase.MsgBase())
1423         self.auth = Mock(authconn.Authconn(None, None, None))
1424         self.topic = NsdTopic(self.db, self.fs, self.msg, self.auth)
1425         self.topic.check_quota = Mock(return_value=None)  # skip quota
1426
1427     @contextmanager
1428     def assertNotRaises(self, exception_type):
1429         try:
1430             yield None
1431         except exception_type:
1432             raise self.failureException("{} raised".format(exception_type.__name__))
1433
1434     def create_desc_temp(self, template):
1435         old_desc = deepcopy(template)
1436         new_desc = deepcopy(template)
1437         return old_desc, new_desc
1438
1439     def prepare_nsd_creation(self):
1440         self.fs.path = ""
1441         did = db_nsd_content["_id"]
1442         self.fs.get_params.return_value = {}
1443         self.fs.file_exists.return_value = False
1444         self.fs.file_open.side_effect = lambda path, mode: open(
1445             "/tmp/" + str(uuid4()), "a+b"
1446         )
1447         self.db.get_one.side_effect = [
1448             {"_id": did, "_admin": deepcopy(db_nsd_content["_admin"])},
1449             None,
1450         ]
1451         test_nsd = deepcopy(db_nsd_content)
1452         del test_nsd["_id"]
1453         del test_nsd["_admin"]
1454         return did, test_nsd
1455
1456     @patch("osm_nbi.descriptor_topics.shutil")
1457     @patch("osm_nbi.descriptor_topics.os.rename")
1458     def test_new_nsd_normal_creation(self, mock_rename, mock_shutil):
1459         did, test_nsd = self.prepare_nsd_creation()
1460         self.db.create.return_value = did
1461         rollback = []
1462
1463         did2, oid = self.topic.new(rollback, fake_session, {})
1464         db_args = self.db.create.call_args[0]
1465         msg_args = self.msg.write.call_args[0]
1466         self.assertEqual(len(rollback), 1, "Wrong rollback length")
1467         self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1468         self.assertEqual(msg_args[1], "created", "Wrong message action")
1469         self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
1470         self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1471         self.assertEqual(did2, did, "Wrong DB NSD id")
1472         self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
1473         self.assertEqual(
1474             db_args[1]["_admin"]["modified"],
1475             db_args[1]["_admin"]["created"],
1476             "Wrong modification time",
1477         )
1478         self.assertEqual(
1479             db_args[1]["_admin"]["projects_read"],
1480             [test_pid],
1481             "Wrong read-only project list",
1482         )
1483         self.assertEqual(
1484             db_args[1]["_admin"]["projects_write"],
1485             [test_pid],
1486             "Wrong read-write project list",
1487         )
1488
1489         self.db.get_list.return_value = [db_vnfd_content]
1490
1491         self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
1492         msg_args = self.msg.write.call_args[0]
1493         test_nsd["_id"] = did
1494         self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1495         self.assertEqual(msg_args[1], "edited", "Wrong message action")
1496         self.assertEqual(msg_args[2], test_nsd, "Wrong message content")
1497
1498         db_args = self.db.get_one.mock_calls[0][1]
1499         self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1500         self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id")
1501
1502         db_args = self.db.replace.call_args[0]
1503         self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1504         self.assertEqual(db_args[1], did, "Wrong DB NSD id")
1505
1506         admin = db_args[2]["_admin"]
1507         db_admin = db_nsd_content["_admin"]
1508         self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time")
1509         self.assertGreater(
1510             admin["modified"], db_admin["created"], "Wrong modification time"
1511         )
1512         self.assertEqual(
1513             admin["projects_read"],
1514             db_admin["projects_read"],
1515             "Wrong read-only project list",
1516         )
1517         self.assertEqual(
1518             admin["projects_write"],
1519             db_admin["projects_write"],
1520             "Wrong read-write project list",
1521         )
1522         self.assertEqual(
1523             admin["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1524         )
1525         self.assertEqual(
1526             admin["operationalState"], "ENABLED", "Wrong operational state"
1527         )
1528         self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
1529
1530         storage = admin["storage"]
1531         self.assertEqual(storage["folder"], did + ":1", "Wrong storage folder")
1532         self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor")
1533
1534         compare_desc(self, test_nsd, db_args[2], "NSD")
1535         revision_args = self.db.create.call_args[0]
1536         self.assertEqual(
1537             revision_args[0], self.topic.topic + "_revisions", "Wrong topic"
1538         )
1539         self.assertEqual(revision_args[1]["id"], db_args[2]["id"], "Wrong revision id")
1540         self.assertEqual(
1541             revision_args[1]["_id"], db_args[2]["_id"] + ":1", "Wrong revision _id"
1542         )
1543
1544     @patch("osm_nbi.descriptor_topics.shutil")
1545     @patch("osm_nbi.descriptor_topics.os.rename")
1546     def test_new_nsd_check_pyangbind_validation_required_properties(
1547         self, mock_rename, mock_shutil
1548     ):
1549         did, test_nsd = self.prepare_nsd_creation()
1550         del test_nsd["id"]
1551
1552         with self.assertRaises(
1553             EngineException, msg="Accepted NSD with a missing required property"
1554         ) as e:
1555             self.topic.upload_content(
1556                 fake_session, did, test_nsd, {}, {"Content-Type": []}
1557             )
1558         self.assertEqual(
1559             e.exception.http_code,
1560             HTTPStatus.UNPROCESSABLE_ENTITY,
1561             "Wrong HTTP status code",
1562         )
1563         self.assertIn(
1564             norm("Error in pyangbind validation: '{}'".format("id")),
1565             norm(str(e.exception)),
1566             "Wrong exception text",
1567         )
1568
1569     @patch("osm_nbi.descriptor_topics.shutil")
1570     @patch("osm_nbi.descriptor_topics.os.rename")
1571     def test_new_nsd_check_pyangbind_validation_additional_properties(
1572         self, mock_rename, mock_shutil
1573     ):
1574         did, test_nsd = self.prepare_nsd_creation()
1575         test_nsd["extra-property"] = 0
1576
1577         with self.assertRaises(
1578             EngineException, msg="Accepted NSD with an additional property"
1579         ) as e:
1580             self.topic.upload_content(
1581                 fake_session, did, test_nsd, {}, {"Content-Type": []}
1582             )
1583         self.assertEqual(
1584             e.exception.http_code,
1585             HTTPStatus.UNPROCESSABLE_ENTITY,
1586             "Wrong HTTP status code",
1587         )
1588         self.assertIn(
1589             norm(
1590                 "Error in pyangbind validation: {} ({})".format(
1591                     "json object contained a key that did not exist", "extra-property"
1592                 )
1593             ),
1594             norm(str(e.exception)),
1595             "Wrong exception text",
1596         )
1597
1598     @patch("osm_nbi.descriptor_topics.shutil")
1599     @patch("osm_nbi.descriptor_topics.os.rename")
1600     def test_new_nsd_check_pyangbind_validation_property_types(
1601         self, mock_rename, mock_shutil
1602     ):
1603         did, test_nsd = self.prepare_nsd_creation()
1604         test_nsd["designer"] = {"key": 0}
1605
1606         with self.assertRaises(
1607             EngineException, msg="Accepted NSD with a wrongly typed property"
1608         ) as e:
1609             self.topic.upload_content(
1610                 fake_session, did, test_nsd, {}, {"Content-Type": []}
1611             )
1612         self.assertEqual(
1613             e.exception.http_code,
1614             HTTPStatus.UNPROCESSABLE_ENTITY,
1615             "Wrong HTTP status code",
1616         )
1617         self.assertIn(
1618             norm(
1619                 "Error in pyangbind validation: {} ({})".format(
1620                     "json object contained a key that did not exist", "key"
1621                 )
1622             ),
1623             norm(str(e.exception)),
1624             "Wrong exception text",
1625         )
1626
1627     @patch("osm_nbi.descriptor_topics.shutil")
1628     @patch("osm_nbi.descriptor_topics.os.rename")
1629     def test_new_nsd_check_input_validation_mgmt_network_virtual_link_protocol_data(
1630         self, mock_rename, mock_shutil
1631     ):
1632         did, test_nsd = self.prepare_nsd_creation()
1633         df = test_nsd["df"][0]
1634         mgmt_profile = {
1635             "id": "id",
1636             "virtual-link-desc-id": "mgmt",
1637             "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1638         }
1639         df["virtual-link-profile"] = [mgmt_profile]
1640
1641         with self.assertRaises(
1642             EngineException, msg="Accepted VLD with mgmt-network+ip-profile"
1643         ) as e:
1644             self.topic.upload_content(
1645                 fake_session, did, test_nsd, {}, {"Content-Type": []}
1646             )
1647         self.assertEqual(
1648             e.exception.http_code,
1649             HTTPStatus.UNPROCESSABLE_ENTITY,
1650             "Wrong HTTP status code",
1651         )
1652         self.assertIn(
1653             norm(
1654                 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1655                 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1656                     df["id"], mgmt_profile["id"]
1657                 )
1658             ),
1659             norm(str(e.exception)),
1660             "Wrong exception text",
1661         )
1662
1663     @patch("osm_nbi.descriptor_topics.shutil")
1664     @patch("osm_nbi.descriptor_topics.os.rename")
1665     def test_new_nsd_check_descriptor_dependencies_vnfd_id(
1666         self, mock_rename, mock_shutil
1667     ):
1668         did, test_nsd = self.prepare_nsd_creation()
1669         self.db.get_list.return_value = []
1670
1671         with self.assertRaises(
1672             EngineException, msg="Accepted wrong VNFD ID reference"
1673         ) as e:
1674             self.topic.upload_content(
1675                 fake_session, did, test_nsd, {}, {"Content-Type": []}
1676             )
1677         self.assertEqual(
1678             e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1679         )
1680         self.assertIn(
1681             norm(
1682                 "'vnfd-id'='{}' references a non existing vnfd".format(
1683                     test_nsd["vnfd-id"][0]
1684                 )
1685             ),
1686             norm(str(e.exception)),
1687             "Wrong exception text",
1688         )
1689
1690     @patch("osm_nbi.descriptor_topics.shutil")
1691     @patch("osm_nbi.descriptor_topics.os.rename")
1692     def test_new_nsd_check_descriptor_dependencies_vld_vnfd_connection_point_ref(
1693         self, mock_rename, mock_shutil
1694     ):
1695         # Check Descriptor Dependencies: "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]
1696         did, test_nsd = self.prepare_nsd_creation()
1697         vnfd_descriptor = deepcopy(db_vnfd_content)
1698         df = test_nsd["df"][0]
1699         affected_vnf_profile = df["vnf-profile"][0]
1700         affected_virtual_link = affected_vnf_profile["virtual-link-connectivity"][1]
1701         affected_cpd = vnfd_descriptor["ext-cpd"].pop()
1702         self.db.get_list.return_value = [vnfd_descriptor]
1703
1704         with self.assertRaises(
1705             EngineException, msg="Accepted wrong VLD CP reference"
1706         ) as e:
1707             self.topic.upload_content(
1708                 fake_session, did, test_nsd, {}, {"Content-Type": []}
1709             )
1710         self.assertEqual(
1711             e.exception.http_code,
1712             HTTPStatus.UNPROCESSABLE_ENTITY,
1713             "Wrong HTTP status code",
1714         )
1715         self.assertIn(
1716             norm(
1717                 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1718                 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1719                 "non existing ext-cpd:id inside vnfd '{}'".format(
1720                     df["id"],
1721                     affected_vnf_profile["id"],
1722                     affected_virtual_link["virtual-link-profile-id"],
1723                     affected_cpd["id"],
1724                     vnfd_descriptor["id"],
1725                 )
1726             ),
1727             norm(str(e.exception)),
1728             "Wrong exception text",
1729         )
1730
1731     def test_edit_nsd(self):
1732         nsd_content = deepcopy(db_nsd_content)
1733         did = nsd_content["_id"]
1734         self.fs.file_exists.return_value = True
1735         self.fs.dir_ls.return_value = True
1736         with self.subTest(i=1, t="Normal Edition"):
1737             now = time()
1738             self.db.get_one.side_effect = [deepcopy(nsd_content), None]
1739             self.db.get_list.return_value = [db_vnfd_content]
1740             data = {"id": "new-nsd-id", "name": "new-nsd-name"}
1741             self.topic.edit(fake_session, did, data)
1742             db_args = self.db.replace.call_args[0]
1743             msg_args = self.msg.write.call_args[0]
1744             data["_id"] = did
1745             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1746             self.assertEqual(msg_args[1], "edited", "Wrong message action")
1747             self.assertEqual(msg_args[2], data, "Wrong message content")
1748             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1749             self.assertEqual(db_args[1], did, "Wrong DB ID")
1750             self.assertEqual(
1751                 db_args[2]["_admin"]["created"],
1752                 nsd_content["_admin"]["created"],
1753                 "Wrong creation time",
1754             )
1755             self.assertGreater(
1756                 db_args[2]["_admin"]["modified"], now, "Wrong modification time"
1757             )
1758             self.assertEqual(
1759                 db_args[2]["_admin"]["projects_read"],
1760                 nsd_content["_admin"]["projects_read"],
1761                 "Wrong read-only project list",
1762             )
1763             self.assertEqual(
1764                 db_args[2]["_admin"]["projects_write"],
1765                 nsd_content["_admin"]["projects_write"],
1766                 "Wrong read-write project list",
1767             )
1768             self.assertEqual(db_args[2]["id"], data["id"], "Wrong NSD ID")
1769             self.assertEqual(db_args[2]["name"], data["name"], "Wrong NSD Name")
1770         with self.subTest(i=2, t="Conflict on Edit"):
1771             data = {"id": "fake-nsd-id", "name": "new-nsd-name"}
1772             self.db.get_one.side_effect = [
1773                 nsd_content,
1774                 {"_id": str(uuid4()), "id": data["id"]},
1775             ]
1776             with self.assertRaises(
1777                 EngineException, msg="Accepted existing NSD ID"
1778             ) as e:
1779                 self.topic.edit(fake_session, did, data)
1780             self.assertEqual(
1781                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1782             )
1783             self.assertIn(
1784                 norm(
1785                     "{} with id '{}' already exists for this project".format(
1786                         "nsd", data["id"]
1787                     )
1788                 ),
1789                 norm(str(e.exception)),
1790                 "Wrong exception text",
1791             )
1792         with self.subTest(i=3, t="Check Envelope"):
1793             data = {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
1794             self.db.get_one.side_effect = [nsd_content, None]
1795             with self.assertRaises(
1796                 EngineException, msg="Accepted NSD with wrong envelope"
1797             ) as e:
1798                 self.topic.edit(fake_session, did, data, content=nsd_content)
1799             self.assertEqual(
1800                 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
1801             )
1802             self.assertIn(
1803                 "'nsd' must be a list of only one element",
1804                 norm(str(e.exception)),
1805                 "Wrong exception text",
1806             )
1807         self.db.reset_mock()
1808         return
1809
1810     def test_delete_nsd(self):
1811         did = db_nsd_content["_id"]
1812         self.db.get_one.return_value = db_nsd_content
1813         p_id = db_nsd_content["_admin"]["projects_read"][0]
1814         with self.subTest(i=1, t="Normal Deletion"):
1815             self.db.get_list.return_value = []
1816             self.db.del_one.return_value = {"deleted": 1}
1817             self.topic.delete(fake_session, did)
1818             db_args = self.db.del_one.call_args[0]
1819             msg_args = self.msg.write.call_args[0]
1820             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1821             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
1822             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
1823             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1824             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
1825             self.assertEqual(
1826                 db_args[1]["_admin.projects_write.cont"],
1827                 [p_id, "ANY"],
1828                 "Wrong DB filter",
1829             )
1830             db_g1_args = self.db.get_one.call_args[0]
1831             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
1832             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB NSD ID")
1833             db_gl_calls = self.db.get_list.call_args_list
1834             self.assertEqual(db_gl_calls[0][0][0], "nsrs", "Wrong DB topic")
1835             # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID")   # Filter changed after call
1836             self.assertEqual(db_gl_calls[1][0][0], "nsts", "Wrong DB topic")
1837             self.assertEqual(
1838                 db_gl_calls[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
1839                 db_nsd_content["id"],
1840                 "Wrong DB NSD netslice-subnet nsd-ref",
1841             )
1842             self.db.set_one.assert_not_called()
1843             fs_del_calls = self.fs.file_delete.call_args_list
1844             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
1845             self.assertEqual(fs_del_calls[1][0][0], did + "_", "Wrong FS folder id")
1846         with self.subTest(i=2, t="Conflict on Delete - NSD in use by nsr"):
1847             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-nsr"}]
1848             with self.assertRaises(
1849                 EngineException, msg="Accepted NSD in use by NSR"
1850             ) as e:
1851                 self.topic.delete(fake_session, did)
1852             self.assertEqual(
1853                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1854             )
1855             self.assertIn(
1856                 "there is at least one ns instance using this descriptor",
1857                 norm(str(e.exception)),
1858                 "Wrong exception text",
1859             )
1860         with self.subTest(i=3, t="Conflict on Delete - NSD in use by NST"):
1861             self.db.get_list.side_effect = [
1862                 [],
1863                 [{"_id": str(uuid4()), "name": "fake-nst"}],
1864             ]
1865             with self.assertRaises(
1866                 EngineException, msg="Accepted NSD in use by NST"
1867             ) as e:
1868                 self.topic.delete(fake_session, did)
1869             self.assertEqual(
1870                 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1871             )
1872             self.assertIn(
1873                 "there is at least one netslice template referencing this descriptor",
1874                 norm(str(e.exception)),
1875                 "Wrong exception text",
1876             )
1877         with self.subTest(i=4, t="Non-existent NSD"):
1878             excp_msg = "Not found any {} with filter='{}'".format("NSD", {"_id": did})
1879             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
1880             with self.assertRaises(
1881                 DbException, msg="Accepted non-existent NSD ID"
1882             ) as e:
1883                 self.topic.delete(fake_session, did)
1884             self.assertEqual(
1885                 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
1886             )
1887             self.assertIn(
1888                 norm(excp_msg), norm(str(e.exception)), "Wrong exception text"
1889             )
1890         with self.subTest(i=5, t="No delete because referenced by other project"):
1891             db_nsd_content["_admin"]["projects_read"].append("other_project")
1892             self.db.get_one = Mock(return_value=db_nsd_content)
1893             self.db.get_list = Mock(return_value=[])
1894             self.msg.write.reset_mock()
1895             self.db.del_one.reset_mock()
1896             self.fs.file_delete.reset_mock()
1897
1898             self.topic.delete(fake_session, did)
1899             self.db.del_one.assert_not_called()
1900             self.msg.write.assert_not_called()
1901             db_g1_args = self.db.get_one.call_args[0]
1902             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
1903             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
1904             db_s1_args = self.db.set_one.call_args
1905             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
1906             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
1907             self.assertIn(
1908                 p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1909             )
1910             self.assertIsNone(
1911                 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
1912             )
1913             self.assertEqual(
1914                 db_s1_args[1]["pull_list"],
1915                 {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
1916                 "Wrong DB pull_list dictionary",
1917             )
1918             self.fs.file_delete.assert_not_called()
1919         self.db.reset_mock()
1920         return
1921
1922     def prepare_nsd_validation(self):
1923         descriptor_name = "test_ns_descriptor"
1924         self.fs.file_open.side_effect = lambda path, mode: open(
1925             "/tmp/" + str(uuid4()), "a+b"
1926         )
1927         old_nsd, new_nsd = self.create_desc_temp(db_nsd_content)
1928         return descriptor_name, old_nsd, new_nsd
1929
1930     @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1931     @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1932     def test_validate_descriptor_ns_configuration_changed(
1933         self, mock_safe_load, mock_detect_usage
1934     ):
1935         """Validating NSD and NSD has changes in ns-configuration:config-primitive"""
1936         descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
1937         mock_safe_load.side_effect = [old_nsd, new_nsd]
1938         mock_detect_usage.return_value = True
1939         self.db.get_one.return_value = old_nsd
1940         old_nsd.update(
1941             {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
1942         )
1943         new_nsd.update(
1944             {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
1945         )
1946
1947
1948         with self.assertNotRaises(EngineException):
1949             self.topic._validate_descriptor_changes(
1950                 old_nsd["_id"], descriptor_name, "/tmp", "/tmp:1"
1951             )
1952         self.db.get_one.assert_called_once()
1953         mock_detect_usage.assert_called_once()
1954         self.assertEqual(mock_safe_load.call_count, 2)
1955
1956     @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1957     @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1958     def test_validate_descriptor_nsd_name_changed(
1959         self, mock_safe_load, mock_detect_usage
1960     ):
1961         """Validating NSD, NSD name has changed."""
1962         descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
1963         did = old_nsd["_id"]
1964         new_nsd["name"] = "nscharm-ns2"
1965         mock_safe_load.side_effect = [old_nsd, new_nsd]
1966         mock_detect_usage.return_value = True
1967         self.db.get_one.return_value = old_nsd
1968
1969         with self.assertRaises(
1970             EngineException, msg="there are disallowed changes in the ns descriptor"
1971         ) as e:
1972             self.topic._validate_descriptor_changes(
1973                 did, descriptor_name, "/tmp", "/tmp:1"
1974             )
1975         self.assertEqual(
1976             e.exception.http_code,
1977             HTTPStatus.UNPROCESSABLE_ENTITY,
1978             "Wrong HTTP status code",
1979         )
1980         self.assertIn(
1981             norm("there are disallowed changes in the ns descriptor"),
1982             norm(str(e.exception)),
1983             "Wrong exception text",
1984         )
1985
1986         self.db.get_one.assert_called_once()
1987         mock_detect_usage.assert_called_once()
1988         self.assertEqual(mock_safe_load.call_count, 2)
1989
1990     @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1991     @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1992     def test_validate_descriptor_nsd_name_changed_nsd_not_in_use(
1993         self, mock_safe_load, mock_detect_usage
1994     ):
1995         """Validating NSD, NSD name has changed, NSD is not in use."""
1996         descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
1997         did = old_nsd["_id"]
1998         new_nsd["name"] = "nscharm-ns2"
1999         mock_safe_load.side_effect = [old_nsd, new_nsd]
2000         mock_detect_usage.return_value = None
2001         self.db.get_one.return_value = old_nsd
2002
2003         with self.assertNotRaises(Exception):
2004             self.topic._validate_descriptor_changes(
2005                 did, descriptor_name, "/tmp", "/tmp:1"
2006             )
2007
2008         self.db.get_one.assert_called_once()
2009         mock_detect_usage.assert_called_once()
2010         mock_safe_load.assert_not_called()
2011
2012     def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
2013         self,
2014     ):
2015         indata = deepcopy(db_nsd_content)
2016         vld = indata["virtual-link-desc"][0]
2017         self.topic.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2018             vld, indata
2019         )
2020
2021     def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
2022         self,
2023     ):
2024         indata = deepcopy(db_nsd_content)
2025         vld = indata["virtual-link-desc"][0]
2026         df = indata["df"][0]
2027         affected_vlp = {
2028             "id": "id",
2029             "virtual-link-desc-id": "mgmt",
2030             "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
2031         }
2032         df["virtual-link-profile"] = [affected_vlp]
2033         with self.assertRaises(EngineException) as e:
2034             self.topic.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2035                 vld, indata
2036             )
2037         self.assertEqual(
2038             e.exception.http_code,
2039             HTTPStatus.UNPROCESSABLE_ENTITY,
2040             "Wrong HTTP status code",
2041         )
2042         self.assertIn(
2043             norm(
2044                 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
2045                 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
2046                     df["id"], affected_vlp["id"]
2047                 )
2048             ),
2049             norm(str(e.exception)),
2050             "Wrong exception text",
2051         )
2052
2053     def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self):
2054         indata = deepcopy(db_nsd_content)
2055         self.topic.validate_vnf_profiles_vnfd_id(indata)
2056
2057     def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self):
2058         indata = deepcopy(db_nsd_content)
2059         df = indata["df"][0]
2060         affected_vnf_profile = df["vnf-profile"][0]
2061         indata["vnfd-id"] = ["non-existing-vnfd"]
2062         with self.assertRaises(EngineException) as e:
2063             self.topic.validate_vnf_profiles_vnfd_id(indata)
2064         self.assertEqual(
2065             e.exception.http_code,
2066             HTTPStatus.UNPROCESSABLE_ENTITY,
2067             "Wrong HTTP status code",
2068         )
2069         self.assertIn(
2070             norm(
2071                 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
2072                 "does not match any vnfd-id".format(
2073                     df["id"],
2074                     affected_vnf_profile["id"],
2075                     affected_vnf_profile["vnfd-id"],
2076                 )
2077             ),
2078             norm(str(e.exception)),
2079             "Wrong exception text",
2080         )
2081
2082     def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
2083         self,
2084     ):
2085         nsd_descriptor = deepcopy(db_nsd_content)
2086         vnfd_descriptor = deepcopy(db_vnfd_content)
2087         df = nsd_descriptor["df"][0]
2088         vnfds_index = {vnfd_descriptor["id"]: vnfd_descriptor}
2089         self.topic.validate_df_vnf_profiles_constituent_connection_points(
2090             df, vnfds_index
2091         )
2092
2093     def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
2094         self,
2095     ):
2096         nsd_descriptor = deepcopy(db_nsd_content)
2097         vnfd_descriptor = deepcopy(db_vnfd_content)
2098         df = nsd_descriptor["df"][0]
2099         affected_vnf_profile = df["vnf-profile"][0]
2100         affected_virtual_link = affected_vnf_profile["virtual-link-connectivity"][1]
2101         vnfds_index = {vnfd_descriptor["id"]: vnfd_descriptor}
2102         affected_cpd = vnfd_descriptor["ext-cpd"].pop()
2103         with self.assertRaises(EngineException) as e:
2104             self.topic.validate_df_vnf_profiles_constituent_connection_points(
2105                 df, vnfds_index
2106             )
2107         self.assertEqual(
2108             e.exception.http_code,
2109             HTTPStatus.UNPROCESSABLE_ENTITY,
2110             "Wrong HTTP status code",
2111         )
2112         self.assertIn(
2113             norm(
2114                 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
2115                 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
2116                 "non existing ext-cpd:id inside vnfd '{}'".format(
2117                     df["id"],
2118                     affected_vnf_profile["id"],
2119                     affected_virtual_link["virtual-link-profile-id"],
2120                     affected_cpd["id"],
2121                     vnfd_descriptor["id"],
2122                 )
2123             ),
2124             norm(str(e.exception)),
2125             "Wrong exception text",
2126         )
2127
2128     def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self):
2129         nsd_descriptor = deepcopy(db_nsd_content)
2130         invalid_vnfd_id = "invalid-vnfd-id"
2131         nsd_descriptor["id"] = "invalid-vnfd-id-ns"
2132         nsd_descriptor["vnfd-id"][0] = invalid_vnfd_id
2133         nsd_descriptor["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
2134         nsd_descriptor["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
2135         with self.assertRaises(EngineException) as e:
2136             self.db.get_list.return_value = []
2137             nsd_descriptor = self.topic.check_conflict_on_edit(
2138                 fake_session, nsd_descriptor, [], "id"
2139             )
2140         self.assertEqual(
2141             e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
2142         )
2143         self.assertIn(
2144             norm(
2145                 "Descriptor error at 'vnfd-id'='{}' references a non "
2146                 "existing vnfd".format(invalid_vnfd_id)
2147             ),
2148             norm(str(e.exception)),
2149             "Wrong exception text",
2150         )
2151
2152
2153 if __name__ == "__main__":
2154     unittest.main()