osm_nbi/tests/test_descriptor_topics.py

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