osm_nbi/tests/test_descriptor_topics.py

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