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