osm_nbi/tests/test_descriptor_topics.py

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