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 import unittest
  21 from unittest import TestCase
  22 from unittest.mock import Mock
  23 from uuid import uuid4
  24 from http import HTTPStatus
  25 from copy import deepcopy
  26 from time import time
  27 from osm_common import dbbase, fsbase, msgbase
  28 from osm_nbi import authconn
  29 from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text
  30 from osm_nbi.descriptor_topics import VnfdTopic, NsdTopic
  31 from osm_nbi.engine import EngineException
  32 from osm_common.dbbase import DbException
  33 import yaml
  34
  35 test_name = "test-user"
  36 db_vnfd_content = yaml.load(db_vnfds_text, Loader=yaml.Loader)[0]
  37 db_nsd_content = yaml.load(db_nsds_text, Loader=yaml.Loader)[0]
  38 test_pid = db_vnfd_content["_admin"]["projects_read"][0]
  39 fake_session = {"username": test_name, "project_id": (test_pid,), "method": None,
  40                 "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}
  41
  42
  43 def norm(str):
  44     """Normalize string for checking"""
  45     return ' '.join(str.strip().split()).lower()
  46
  47
  48 def compare_desc(tc, d1, d2, k):
  49     """
  50     Compare two descriptors
  51     We need this function because some methods are adding/removing items to/from the descriptors
  52     before they are stored in the database, so the original and stored versions will differ
  53     What we check is that COMMON LEAF ITEMS are equal
  54     Lists of different length are not compared
  55     :param tc: Test Case wich provides context (in particular the assert* methods)
  56     :param d1,d2: Descriptors to be compared
  57     :param key/item being compared
  58     :return: Nothing
  59     """
  60     if isinstance(d1, dict) and isinstance(d2, dict):
  61         for key in d1.keys():
  62             if key in d2:
  63                 compare_desc(tc, d1[key], d2[key], k + "[{}]".format(key))
  64     elif isinstance(d1, list) and isinstance(d2, list) and len(d1) == len(d2):
  65         for i in range(len(d1)):
  66             compare_desc(tc, d1[i], d2[i], k + "[{}]".format(i))
  67     else:
  68         tc.assertEqual(d1, d2, "Wrong descriptor content: {}".format(k))
  69
  70
  71 class Test_VnfdTopic(TestCase):
  72
  73     @classmethod
  74     def setUpClass(cls):
  75         cls.test_name = "test-vnfd-topic"
  76
  77     @classmethod
  78     def tearDownClass(cls):
  79         pass
  80
  81     def setUp(self):
  82         self.db = Mock(dbbase.DbBase())
  83         self.fs = Mock(fsbase.FsBase())
  84         self.msg = Mock(msgbase.MsgBase())
  85         self.auth = Mock(authconn.Authconn(None, None, None))
  86         self.topic = VnfdTopic(self.db, self.fs, self.msg, self.auth)
  87         self.topic.check_quota = Mock(return_value=None)  # skip quota
  88
  89     def test_new_vnfd(self):
  90         did = db_vnfd_content["_id"]
  91         self.fs.get_params.return_value = {}
  92         self.fs.file_exists.return_value = False
  93         self.fs.file_open.side_effect = lambda path, mode: open("/tmp/" + str(uuid4()), "a+b")
  94         test_vnfd = deepcopy(db_vnfd_content)
  95         del test_vnfd["_id"]
  96         del test_vnfd["_admin"]
  97         with self.subTest(i=1, t='Normal Creation'):
  98             self.db.create.return_value = did
  99             rollback = []
 100             did2, oid = self.topic.new(rollback, fake_session, {})
 101             db_args = self.db.create.call_args[0]
 102             msg_args = self.msg.write.call_args[0]
 103             self.assertEqual(len(rollback), 1, "Wrong rollback length")
 104             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 105             self.assertEqual(msg_args[1], "created", "Wrong message action")
 106             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 107             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 108             self.assertEqual(did2, did, "Wrong DB VNFD id")
 109             self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
 110             self.assertEqual(db_args[1]["_admin"]["modified"], db_args[1]["_admin"]["created"],
 111                              "Wrong modification time")
 112             self.assertEqual(db_args[1]["_admin"]["projects_read"], [test_pid], "Wrong read-only project list")
 113             self.assertEqual(db_args[1]["_admin"]["projects_write"], [test_pid], "Wrong read-write project list")
 114             tmp1 = test_vnfd["vdu"][0]["cloud-init-file"]
 115             tmp2 = test_vnfd["df"][
 116                 0
 117             ]["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][
 118                 0
 119             ]["execution-environment-list"][
 120                 0
 121             ]["juju"]
 122             del test_vnfd["vdu"][0]["cloud-init-file"]
 123             del test_vnfd["df"][
 124                 0
 125             ]["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][
 126                 0
 127             ]["execution-environment-list"][
 128                 0
 129             ]["juju"]
 130             try:
 131                 self.db.get_one.side_effect = [{"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}, None]
 132                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 133                 msg_args = self.msg.write.call_args[0]
 134                 test_vnfd["_id"] = did
 135                 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 136                 self.assertEqual(msg_args[1], "edited", "Wrong message action")
 137                 self.assertEqual(msg_args[2], test_vnfd, "Wrong message content")
 138                 db_args = self.db.get_one.mock_calls[0][1]
 139                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 140                 self.assertEqual(db_args[1]["_id"], did, "Wrong DB VNFD id")
 141                 db_args = self.db.replace.call_args[0]
 142                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 143                 self.assertEqual(db_args[1], did, "Wrong DB VNFD id")
 144                 admin = db_args[2]["_admin"]
 145                 db_admin = deepcopy(db_vnfd_content["_admin"])
 146                 self.assertEqual(admin["type"], "vnfd", "Wrong descriptor type")
 147                 self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time")
 148                 self.assertGreater(admin["modified"], db_admin["created"], "Wrong modification time")
 149                 self.assertEqual(admin["projects_read"], db_admin["projects_read"], "Wrong read-only project list")
 150                 self.assertEqual(admin["projects_write"], db_admin["projects_write"], "Wrong read-write project list")
 151                 self.assertEqual(admin["onboardingState"], "ONBOARDED", "Wrong onboarding state")
 152                 self.assertEqual(admin["operationalState"], "ENABLED", "Wrong operational state")
 153                 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
 154                 storage = admin["storage"]
 155                 self.assertEqual(storage["folder"], did, "Wrong storage folder")
 156                 self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor")
 157                 compare_desc(self, test_vnfd, db_args[2], "VNFD")
 158             finally:
 159                 test_vnfd["vdu"][0]["cloud-init-file"] = tmp1
 160                 test_vnfd["df"][
 161                     0
 162                 ]["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][
 163                     0
 164                 ]["execution-environment-list"][
 165                     0
 166                 ]["juju"] = tmp2
 167         self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None: \
 168             {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}
 169         with self.subTest(i=2, t='Check Pyangbind Validation: additional properties'):
 170             test_vnfd["extra-property"] = 0
 171             try:
 172                 with self.assertRaises(EngineException, msg="Accepted VNFD with an additional property") as e:
 173                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 174                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 175                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 176                                    .format("json object contained a key that did not exist", "extra-property")),
 177                               norm(str(e.exception)), "Wrong exception text")
 178             finally:
 179                 del test_vnfd["extra-property"]
 180         with self.subTest(i=3, t='Check Pyangbind Validation: property types'):
 181             tmp = test_vnfd["product-name"]
 182             test_vnfd["product-name"] = {"key": 0}
 183             try:
 184                 with self.assertRaises(EngineException, msg="Accepted VNFD with a wrongly typed property") as e:
 185                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 186                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 187                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 188                                    .format("json object contained a key that did not exist", "key")),
 189                               norm(str(e.exception)), "Wrong exception text")
 190             finally:
 191                 test_vnfd["product-name"] = tmp
 192         with self.subTest(i=4, t='Check Input Validation: cloud-init'):
 193             with self.assertRaises(EngineException, msg="Accepted non-existent cloud_init file") as e:
 194                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 195             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 196             self.assertIn(norm("{} defined in vnf[id={}]:vdu[id={}] but not present in package"
 197                                .format("cloud-init", test_vnfd["id"], test_vnfd["vdu"][0]["id"])),
 198                           norm(str(e.exception)), "Wrong exception text")
 199         with self.subTest(i=5, t='Check Input Validation: day1-2 configuration[juju]'):
 200             del test_vnfd["vdu"][0]["cloud-init-file"]
 201             with self.assertRaises(EngineException, msg="Accepted non-existent charm in VNF configuration") as e:
 202                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 203             print(str(e.exception))
 204             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 205             self.assertIn(norm("{} defined in vnf[id={}] but not present in package".format("charm", test_vnfd["id"])),
 206                           norm(str(e.exception)), "Wrong exception text")
 207             del test_vnfd["df"][
 208                 0
 209             ]["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][
 210                 0
 211             ]["execution-environment-list"][
 212                 0
 213             ]["juju"]
 214         with self.subTest(i=6, t='Check Input Validation: mgmt-cp'):
 215             tmp = test_vnfd["mgmt-cp"]
 216             del test_vnfd["mgmt-cp"]
 217             try:
 218                 with self.assertRaises(EngineException, msg="Accepted VNFD without management interface") as e:
 219                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 220                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 221                 self.assertIn(norm("'{}' is a mandatory field and it is not defined".format("mgmt-cp")),
 222                               norm(str(e.exception)), "Wrong exception text")
 223             finally:
 224                 test_vnfd["mgmt-cp"] = tmp
 225         with self.subTest(i=7, t='Check Input Validation: mgmt-cp connection point'):
 226             tmp = test_vnfd["mgmt-cp"]
 227             test_vnfd["mgmt-cp"] = "wrong-cp"
 228             try:
 229                 with self.assertRaises(EngineException, msg="Accepted wrong mgmt-cp connection point") as e:
 230                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 231                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 232                 self.assertIn(norm("mgmt-cp='{}' must match an existing ext-cpd".format(test_vnfd["mgmt-cp"])),
 233                               norm(str(e.exception)), "Wrong exception text")
 234             finally:
 235                 test_vnfd["mgmt-cp"] = tmp
 236         with self.subTest(i=8, t='Check Input Validation: vdu int-cpd'):
 237             ext_cpd = test_vnfd["ext-cpd"][1]
 238             tmp = ext_cpd["int-cpd"]["cpd"]
 239             ext_cpd["int-cpd"]["cpd"] = "wrong-cpd"
 240             try:
 241                 with self.assertRaises(EngineException, msg="Accepted wrong ext-cpd internal connection point") as e:
 242                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 243                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 244                 self.assertIn(norm("ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(ext_cpd["id"])),
 245                               norm(str(e.exception)), "Wrong exception text")
 246             finally:
 247                 ext_cpd["int-cpd"]["cpd"] = tmp
 248         with self.subTest(i=9, t='Check Input Validation: Duplicated VLD'):
 249             test_vnfd['int-virtual-link-desc'].insert(0, {'id': 'internal'})
 250             try:
 251                 with self.assertRaises(EngineException, msg="Accepted duplicated VLD name") as e:
 252                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 253                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 254                 self.assertIn(
 255                     norm("identifier id '{}' is not unique".format(test_vnfd['int-virtual-link-desc'][0]["id"])),
 256                     norm(str(e.exception)), "Wrong exception text")
 257             finally:
 258                 del test_vnfd['int-virtual-link-desc'][0]
 259         with self.subTest(i=10, t='Check Input Validation: vdu int-virtual-link-desc'):
 260             vdu = test_vnfd['vdu'][0]
 261             int_cpd = vdu['int-cpd'][1]
 262             tmp = int_cpd['int-virtual-link-desc']
 263             int_cpd['int-virtual-link-desc'] = 'non-existing-int-virtual-link-desc'
 264             try:
 265                 with self.assertRaises(EngineException, msg="Accepted int-virtual-link-desc") as e:
 266                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 267                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 268                 self.assertIn(norm("vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
 269                                    "int-virtual-link-desc".format(vdu["id"], int_cpd["id"],
 270                                                                   int_cpd['int-virtual-link-desc'])),
 271                               norm(str(e.exception)), "Wrong exception text")
 272             finally:
 273                 int_cpd['int-virtual-link-desc'] = tmp
 274         with self.subTest(i=11, t='Check Input Validation: virtual-link-profile)'):
 275             fake_ivld_profile = {'id': 'fake-profile-ref', 'flavour': 'fake-flavour'}
 276             df = test_vnfd['df'][0]
 277             df['virtual-link-profile'] = [fake_ivld_profile]
 278             try:
 279                 with self.assertRaises(EngineException, msg="Accepted non-existent Profile Ref") as e:
 280                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 281                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 282                 self.assertIn(norm("df[id='{}']:virtual-link-profile='{}' must match an existing "
 283                                    "int-virtual-link-desc".format(df["id"], fake_ivld_profile["id"])),
 284                               norm(str(e.exception)), "Wrong exception text")
 285             finally:
 286                 del df['virtual-link-profile']
 287         with self.subTest(i=12, t='Check Input Validation: scaling-criteria monitoring-param-ref'):
 288             vdu = test_vnfd['vdu'][1]
 289             affected_df = test_vnfd['df'][0]
 290             sa = affected_df['scaling-aspect'][0]
 291             sp = sa['scaling-policy'][0]
 292             sc = sp['scaling-criteria'][0]
 293             tmp = vdu.pop('monitoring-parameter')
 294             try:
 295                 with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group Policy Criteria") as e:
 296                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 297                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 298                 self.assertIn(norm("df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
 299                                    "[name='{}']:scaling-criteria[name='{}']: "
 300                                    "vnf-monitoring-param-ref='{}' not defined in any monitoring-param"
 301                                    .format(affected_df["id"], sa["id"], sp["name"],
 302                                            sc["name"], sc["vnf-monitoring-param-ref"])),
 303                               norm(str(e.exception)), "Wrong exception text")
 304             finally:
 305                 vdu['monitoring-parameter'] = tmp
 306         with self.subTest(i=13, t='Check Input Validation: scaling-aspect vnf-configuration'):
 307             df = test_vnfd['df'][0]
 308             tmp = test_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"].pop()
 309             try:
 310                 with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference") \
 311                         as e:
 312                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 313                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 314                 self.assertIn(norm("'day1-2 configuration' not defined in the descriptor but it is referenced "
 315                                    "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action"
 316                                    .format(df["id"], df['scaling-aspect'][0]["id"])),
 317                               norm(str(e.exception)), "Wrong exception text")
 318             finally:
 319                 test_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"].append(tmp)
 320         with self.subTest(i=14, t='Check Input Validation: scaling-config-action'):
 321             df = test_vnfd['df'][0]
 322             tmp = test_vnfd["df"][0].get(
 323                 "lcm-operations-configuration"
 324             ).get(
 325                 "operate-vnf-op-config"
 326             )["day1-2"][0]['config-primitive']
 327             test_vnfd["df"][0].get(
 328                 "lcm-operations-configuration"
 329             ).get(
 330                 "operate-vnf-op-config"
 331             )["day1-2"][0]['config-primitive'] = [{'name': 'wrong-primitive'}]
 332             try:
 333                 with self.assertRaises(EngineException,
 334                                        msg="Accepted non-existent Scaling Group VDU ID Reference") as e:
 335                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 336                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 337                 self.assertIn(norm("df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
 338                                    "config-primitive-name-ref='{}' does not match any "
 339                                    "day1-2 configuration:config-primitive:name"
 340                                    .format(df["id"], df['scaling-aspect'][0]["id"],
 341                                            sa['scaling-config-action'][0]['vnf-config-primitive-name-ref'])),
 342                               norm(str(e.exception)), "Wrong exception text")
 343             finally:
 344                 test_vnfd["df"][0].get(
 345                     "lcm-operations-configuration"
 346                 ).get(
 347                     "operate-vnf-op-config"
 348                 )["day1-2"][0]['config-primitive'] = tmp
 349         with self.subTest(i=15, t='Check Input Validation: everything right'):
 350             test_vnfd["id"] = "fake-vnfd-id"
 351             test_vnfd["df"][0].get(
 352                 "lcm-operations-configuration"
 353             ).get(
 354                 "operate-vnf-op-config"
 355             )["day1-2"][0]["id"] = "fake-vnfd-id"
 356             self.db.get_one.side_effect = [{"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}, None]
 357             rc = self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 358             self.assertTrue(rc, "Input Validation: Unexpected failure")
 359         return
 360
 361     def test_edit_vnfd(self):
 362         vnfd_content = deepcopy(db_vnfd_content)
 363         did = vnfd_content["_id"]
 364         self.fs.file_exists.return_value = True
 365         self.fs.dir_ls.return_value = True
 366         with self.subTest(i=1, t='Normal Edition'):
 367             now = time()
 368             self.db.get_one.side_effect = [deepcopy(vnfd_content), None]
 369             data = {
 370                 "product-name": "new-vnfd-name"
 371             }
 372             self.topic.edit(fake_session, did, data)
 373             db_args = self.db.replace.call_args[0]
 374             msg_args = self.msg.write.call_args[0]
 375             data["_id"] = did
 376             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 377             self.assertEqual(msg_args[1], "edited", "Wrong message action")
 378             self.assertEqual(msg_args[2], data, "Wrong message content")
 379             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 380             self.assertEqual(db_args[1], did, "Wrong DB ID")
 381             self.assertEqual(db_args[2]["_admin"]["created"], vnfd_content["_admin"]["created"],
 382                              "Wrong creation time")
 383             self.assertGreater(db_args[2]["_admin"]["modified"], now,
 384                                "Wrong modification time")
 385             self.assertEqual(db_args[2]["_admin"]["projects_read"], vnfd_content["_admin"]["projects_read"],
 386                              "Wrong read-only project list")
 387             self.assertEqual(db_args[2]["_admin"]["projects_write"], vnfd_content["_admin"]["projects_write"],
 388                              "Wrong read-write project list")
 389             self.assertEqual(db_args[2]["product-name"], data["product-name"], "Wrong VNFD Name")
 390         with self.subTest(i=2, t='Conflict on Edit'):
 391             data = {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
 392             self.db.get_one.side_effect = [deepcopy(vnfd_content), {"_id": str(uuid4()), "id": data["id"]}]
 393             with self.assertRaises(EngineException, msg="Accepted existing VNFD ID") as e:
 394                 self.topic.edit(fake_session, did, data)
 395             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 396             self.assertIn(norm("{} with id '{}' already exists for this project".format("vnfd", data["id"])),
 397                           norm(str(e.exception)), "Wrong exception text")
 398         with self.subTest(i=3, t='Check Envelope'):
 399             data = {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
 400             with self.assertRaises(EngineException, msg="Accepted VNFD with wrong envelope") as e:
 401                 self.topic.edit(fake_session, did, data, content=vnfd_content)
 402             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 403             self.assertIn("'vnfd' must be dict", norm(str(e.exception)), "Wrong exception text")
 404         return
 405
 406     def test_delete_vnfd(self):
 407         did = db_vnfd_content["_id"]
 408         self.db.get_one.return_value = db_vnfd_content
 409         p_id = db_vnfd_content["_admin"]["projects_read"][0]
 410         with self.subTest(i=1, t='Normal Deletion'):
 411             self.db.get_list.return_value = []
 412             self.db.del_one.return_value = {"deleted": 1}
 413             self.topic.delete(fake_session, did)
 414             db_args = self.db.del_one.call_args[0]
 415             msg_args = self.msg.write.call_args[0]
 416             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 417             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 418             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 419             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 420             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
 421             self.assertEqual(db_args[1]["_admin.projects_write.cont"], [p_id, 'ANY'], "Wrong DB filter")
 422             db_g1_args = self.db.get_one.call_args[0]
 423             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 424             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 425             db_gl_calls = self.db.get_list.call_args_list
 426             self.assertEqual(db_gl_calls[0][0][0], "vnfrs", "Wrong DB topic")
 427             # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID")   # Filter changed after call
 428             self.assertEqual(db_gl_calls[1][0][0], "nsds", "Wrong DB topic")
 429             self.assertEqual(db_gl_calls[1][0][1]["vnfd-id"], db_vnfd_content["id"],
 430                              "Wrong DB NSD vnfd-id")
 431
 432             self.db.set_one.assert_not_called()
 433             fs_del_calls = self.fs.file_delete.call_args_list
 434             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
 435             self.assertEqual(fs_del_calls[1][0][0], did + '_', "Wrong FS folder id")
 436         with self.subTest(i=2, t='Conflict on Delete - VNFD in use by VNFR'):
 437             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-vnfr"}]
 438             with self.assertRaises(EngineException, msg="Accepted VNFD in use by VNFR") as e:
 439                 self.topic.delete(fake_session, did)
 440             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 441             self.assertIn("there is at least one vnf instance using this descriptor", norm(str(e.exception)),
 442                           "Wrong exception text")
 443         with self.subTest(i=3, t='Conflict on Delete - VNFD in use by NSD'):
 444             self.db.get_list.side_effect = [[], [{"_id": str(uuid4()), "name": "fake-nsd"}]]
 445             with self.assertRaises(EngineException, msg="Accepted VNFD in use by NSD") as e:
 446                 self.topic.delete(fake_session, did)
 447             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 448             self.assertIn("there is at least one ns package referencing this descriptor", norm(str(e.exception)),
 449                           "Wrong exception text")
 450         with self.subTest(i=4, t='Non-existent VNFD'):
 451             excp_msg = "Not found any {} with filter='{}'".format("VNFD", {"_id": did})
 452             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
 453             with self.assertRaises(DbException, msg="Accepted non-existent VNFD ID") as e:
 454                 self.topic.delete(fake_session, did)
 455             self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
 456             self.assertIn(norm(excp_msg), norm(str(e.exception)), "Wrong exception text")
 457         with self.subTest(i=5, t='No delete because referenced by other project'):
 458             db_vnfd_content["_admin"]["projects_read"].append("other_project")
 459             self.db.get_one = Mock(return_value=db_vnfd_content)
 460             self.db.get_list = Mock(return_value=[])
 461             self.msg.write.reset_mock()
 462             self.db.del_one.reset_mock()
 463             self.fs.file_delete.reset_mock()
 464
 465             self.topic.delete(fake_session, did)
 466             self.db.del_one.assert_not_called()
 467             self.msg.write.assert_not_called()
 468             db_g1_args = self.db.get_one.call_args[0]
 469             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 470             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 471             db_s1_args = self.db.set_one.call_args
 472             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
 473             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
 474             self.assertIn(p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter")
 475             self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary")
 476             self.assertEqual(db_s1_args[1]["pull_list"],
 477                              {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
 478                              "Wrong DB pull_list dictionary")
 479             self.fs.file_delete.assert_not_called()
 480         return
 481
 482     def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self):
 483         indata = deepcopy(db_vnfd_content)
 484         self.topic.validate_mgmt_interface_connection_point(indata)
 485
 486     def test_validate_mgmt_interface_connection_point_when_missing_connection_point(self):
 487         indata = deepcopy(db_vnfd_content)
 488         indata['ext-cpd'] = []
 489         with self.assertRaises(EngineException) as e:
 490             self.topic.validate_mgmt_interface_connection_point(indata)
 491         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 492         self.assertIn(norm("mgmt-cp='{}' must match an existing ext-cpd"
 493                            .format(indata["mgmt-cp"])),
 494                       norm(str(e.exception)), "Wrong exception text")
 495
 496     def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self):
 497         indata = deepcopy(db_vnfd_content)
 498         indata.pop('mgmt-cp')
 499         with self.assertRaises(EngineException) as e:
 500             self.topic.validate_mgmt_interface_connection_point(indata)
 501         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 502         self.assertIn(norm("'mgmt-cp' is a mandatory field and it is not defined"),
 503                       norm(str(e.exception)), "Wrong exception text")
 504
 505     def test_validate_vdu_internal_connection_points_on_valid_descriptor(self):
 506         indata = db_vnfd_content
 507         vdu = indata['vdu'][0]
 508         self.topic.validate_vdu_internal_connection_points(vdu)
 509
 510     def test_validate_external_connection_points_on_valid_descriptor(self):
 511         indata = db_vnfd_content
 512         self.topic.validate_external_connection_points(indata)
 513
 514     def test_validate_external_connection_points_when_missing_internal_connection_point(self):
 515         indata = deepcopy(db_vnfd_content)
 516         vdu = indata['vdu'][0]
 517         vdu.pop('int-cpd')
 518         affected_ext_cpd = indata["ext-cpd"][0]
 519         with self.assertRaises(EngineException) as e:
 520             self.topic.validate_external_connection_points(indata)
 521         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 522         self.assertIn(norm("ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd"
 523                            .format(affected_ext_cpd["id"])),
 524                       norm(str(e.exception)), "Wrong exception text")
 525
 526     def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(self):
 527         indata = deepcopy(db_vnfd_content)
 528         vdu = indata['vdu'][0]
 529         duplicated_cpd = {'id': 'vnf-mgmt', 'order': 3,
 530                           'virtual-network-interface-requirement': [{'name': 'duplicated'}]}
 531         vdu['int-cpd'].insert(0, duplicated_cpd)
 532         with self.assertRaises(EngineException) as e:
 533             self.topic.validate_vdu_internal_connection_points(vdu)
 534         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 535         self.assertIn(norm("vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd"
 536                            .format(vdu["id"], duplicated_cpd["id"])),
 537                       norm(str(e.exception)), "Wrong exception text")
 538
 539     def test_validate_external_connection_points_on_duplicated_external_connection_point(self):
 540         indata = deepcopy(db_vnfd_content)
 541         duplicated_cpd = {'id': 'vnf-mgmt-ext', 'int-cpd': {'vdu-id': 'dataVM', 'cpd': 'vnf-data'}}
 542         indata['ext-cpd'].insert(0, duplicated_cpd)
 543         with self.assertRaises(EngineException) as e:
 544             self.topic.validate_external_connection_points(indata)
 545         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 546         self.assertIn(norm("ext-cpd[id='{}'] is already used by other ext-cpd"
 547                            .format(duplicated_cpd["id"])),
 548                       norm(str(e.exception)), "Wrong exception text")
 549
 550     def test_validate_internal_virtual_links_on_valid_descriptor(self):
 551         indata = db_vnfd_content
 552         self.topic.validate_internal_virtual_links(indata)
 553
 554     def test_validate_internal_virtual_links_on_duplicated_ivld(self):
 555         indata = deepcopy(db_vnfd_content)
 556         duplicated_vld = {'id': 'internal'}
 557         indata['int-virtual-link-desc'].insert(0, duplicated_vld)
 558         with self.assertRaises(EngineException) as e:
 559             self.topic.validate_internal_virtual_links(indata)
 560         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 561         self.assertIn(norm("Duplicated VLD id in int-virtual-link-desc[id={}]"
 562                            .format(duplicated_vld["id"])),
 563                       norm(str(e.exception)), "Wrong exception text")
 564
 565     def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(self):
 566         indata = deepcopy(db_vnfd_content)
 567         vdu = indata['vdu'][0]
 568         affected_int_cpd = vdu['int-cpd'][0]
 569         affected_int_cpd['int-virtual-link-desc'] = 'non-existing-int-virtual-link-desc'
 570         with self.assertRaises(EngineException) as e:
 571             self.topic.validate_internal_virtual_links(indata)
 572         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 573         self.assertIn(norm("vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
 574                            "int-virtual-link-desc".format(vdu["id"], affected_int_cpd["id"],
 575                                                           affected_int_cpd['int-virtual-link-desc'])),
 576                       norm(str(e.exception)), "Wrong exception text")
 577
 578     def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self):
 579         indata = deepcopy(db_vnfd_content)
 580         affected_ivld_profile = {'id': 'non-existing-int-virtual-link-desc'}
 581         df = indata['df'][0]
 582         df['virtual-link-profile'] = [affected_ivld_profile]
 583         with self.assertRaises(EngineException) as e:
 584             self.topic.validate_internal_virtual_links(indata)
 585         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 586         self.assertIn(norm("df[id='{}']:virtual-link-profile='{}' must match an existing "
 587                            "int-virtual-link-desc".format(df["id"], affected_ivld_profile["id"])),
 588                       norm(str(e.exception)), "Wrong exception text")
 589
 590     def test_validate_monitoring_params_on_valid_descriptor(self):
 591         indata = db_vnfd_content
 592         self.topic.validate_monitoring_params(indata)
 593
 594     def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self):
 595         indata = deepcopy(db_vnfd_content)
 596         duplicated_mp = {'id': 'cpu', 'name': 'cpu', 'performance_metric': 'cpu'}
 597         affected_ivld = indata['int-virtual-link-desc'][0]
 598         affected_ivld['monitoring-parameters'] = [duplicated_mp, duplicated_mp]
 599         with self.assertRaises(EngineException) as e:
 600             self.topic.validate_monitoring_params(indata)
 601         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 602         self.assertIn(norm("Duplicated monitoring-parameter id in "
 603                            "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']"
 604                            .format(affected_ivld["id"], duplicated_mp["id"])),
 605                       norm(str(e.exception)), "Wrong exception text")
 606
 607     def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self):
 608         indata = deepcopy(db_vnfd_content)
 609         duplicated_mp = {'id': 'dataVM_cpu_util', 'name': 'dataVM_cpu_util', 'performance_metric': 'cpu'}
 610         affected_vdu = indata['vdu'][1]
 611         affected_vdu['monitoring-parameter'].insert(0, duplicated_mp)
 612         with self.assertRaises(EngineException) as e:
 613             self.topic.validate_monitoring_params(indata)
 614         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 615         self.assertIn(norm("Duplicated monitoring-parameter id in "
 616                            "vdu[id='{}']:monitoring-parameter[id='{}']"
 617                            .format(affected_vdu["id"], duplicated_mp["id"])),
 618                       norm(str(e.exception)), "Wrong exception text")
 619
 620     def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self):
 621         indata = deepcopy(db_vnfd_content)
 622         duplicated_mp = {'id': 'memory', 'name': 'memory', 'performance_metric': 'memory'}
 623         affected_df = indata['df'][0]
 624         affected_df['monitoring-parameter'] = [duplicated_mp, duplicated_mp]
 625         with self.assertRaises(EngineException) as e:
 626             self.topic.validate_monitoring_params(indata)
 627         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 628         self.assertIn(norm("Duplicated monitoring-parameter id in "
 629                            "df[id='{}']:monitoring-parameter[id='{}']"
 630                            .format(affected_df["id"], duplicated_mp["id"])),
 631                       norm(str(e.exception)), "Wrong exception text")
 632
 633     def test_validate_scaling_group_descriptor_on_valid_descriptor(self):
 634         indata = db_vnfd_content
 635         self.topic.validate_scaling_group_descriptor(indata)
 636
 637     def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self):
 638         indata = deepcopy(db_vnfd_content)
 639         vdu = indata['vdu'][1]
 640         affected_df = indata['df'][0]
 641         affected_sa = affected_df['scaling-aspect'][0]
 642         affected_sp = affected_sa['scaling-policy'][0]
 643         affected_sc = affected_sp['scaling-criteria'][0]
 644         vdu.pop('monitoring-parameter')
 645         with self.assertRaises(EngineException) as e:
 646             self.topic.validate_scaling_group_descriptor(indata)
 647         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 648         self.assertIn(norm("df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
 649                            "[name='{}']:scaling-criteria[name='{}']: "
 650                            "vnf-monitoring-param-ref='{}' not defined in any monitoring-param"
 651                            .format(affected_df["id"], affected_sa["id"], affected_sp["name"], affected_sc["name"],
 652                                    affected_sc["vnf-monitoring-param-ref"])),
 653                       norm(str(e.exception)), "Wrong exception text")
 654
 655     def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self):
 656         indata = deepcopy(db_vnfd_content)
 657         df = indata['df'][0]
 658         affected_sa = df['scaling-aspect'][0]
 659         indata["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"].pop()
 660         with self.assertRaises(EngineException) as e:
 661             self.topic.validate_scaling_group_descriptor(indata)
 662         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 663         self.assertIn(norm("'day1-2 configuration' not defined in the descriptor but it is referenced "
 664                            "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action"
 665                            .format(df["id"], affected_sa["id"])),
 666                       norm(str(e.exception)), "Wrong exception text")
 667
 668     def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(self):
 669         indata = deepcopy(db_vnfd_content)
 670         df = indata['df'][0]
 671         affected_sa = df['scaling-aspect'][0]
 672         affected_sca_primitive = affected_sa['scaling-config-action'][0]['vnf-config-primitive-name-ref']
 673         df["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0]['config-primitive'] = []
 674         with self.assertRaises(EngineException) as e:
 675             self.topic.validate_scaling_group_descriptor(indata)
 676         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 677         self.assertIn(norm("df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
 678                            "config-primitive-name-ref='{}' does not match any "
 679                            "day1-2 configuration:config-primitive:name"
 680                            .format(df["id"], affected_sa["id"], affected_sca_primitive)),
 681                       norm(str(e.exception)), "Wrong exception text")
 682
 683
 684 class Test_NsdTopic(TestCase):
 685
 686     @classmethod
 687     def setUpClass(cls):
 688         cls.test_name = "test-nsd-topic"
 689
 690     @classmethod
 691     def tearDownClass(cls):
 692         pass
 693
 694     def setUp(self):
 695         self.db = Mock(dbbase.DbBase())
 696         self.fs = Mock(fsbase.FsBase())
 697         self.msg = Mock(msgbase.MsgBase())
 698         self.auth = Mock(authconn.Authconn(None, None, None))
 699         self.topic = NsdTopic(self.db, self.fs, self.msg, self.auth)
 700         self.topic.check_quota = Mock(return_value=None)  # skip quota
 701
 702     def test_new_nsd(self):
 703         did = db_nsd_content["_id"]
 704         self.fs.get_params.return_value = {}
 705         self.fs.file_exists.return_value = False
 706         self.fs.file_open.side_effect = lambda path, mode: open("/tmp/" + str(uuid4()), "a+b")
 707         test_nsd = deepcopy(db_nsd_content)
 708         del test_nsd["_id"]
 709         del test_nsd["_admin"]
 710         with self.subTest(i=1, t='Normal Creation'):
 711             self.db.create.return_value = did
 712             rollback = []
 713             did2, oid = self.topic.new(rollback, fake_session, {})
 714             db_args = self.db.create.call_args[0]
 715             msg_args = self.msg.write.call_args[0]
 716             self.assertEqual(len(rollback), 1, "Wrong rollback length")
 717             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 718             self.assertEqual(msg_args[1], "created", "Wrong message action")
 719             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 720             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 721             self.assertEqual(did2, did, "Wrong DB NSD id")
 722             self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
 723             self.assertEqual(db_args[1]["_admin"]["modified"], db_args[1]["_admin"]["created"],
 724                              "Wrong modification time")
 725             self.assertEqual(db_args[1]["_admin"]["projects_read"], [test_pid], "Wrong read-only project list")
 726             self.assertEqual(db_args[1]["_admin"]["projects_write"], [test_pid], "Wrong read-write project list")
 727             try:
 728                 self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 729                 self.db.get_list.return_value = [db_vnfd_content]
 730                 self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 731                 msg_args = self.msg.write.call_args[0]
 732                 test_nsd["_id"] = did
 733                 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 734                 self.assertEqual(msg_args[1], "edited", "Wrong message action")
 735                 self.assertEqual(msg_args[2], test_nsd, "Wrong message content")
 736                 db_args = self.db.get_one.mock_calls[0][1]
 737                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 738                 self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id")
 739                 db_args = self.db.replace.call_args[0]
 740                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 741                 self.assertEqual(db_args[1], did, "Wrong DB NSD id")
 742                 admin = db_args[2]["_admin"]
 743                 db_admin = db_nsd_content["_admin"]
 744                 self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time")
 745                 self.assertGreater(admin["modified"], db_admin["created"], "Wrong modification time")
 746                 self.assertEqual(admin["projects_read"], db_admin["projects_read"], "Wrong read-only project list")
 747                 self.assertEqual(admin["projects_write"], db_admin["projects_write"], "Wrong read-write project list")
 748                 self.assertEqual(admin["onboardingState"], "ONBOARDED", "Wrong onboarding state")
 749                 self.assertEqual(admin["operationalState"], "ENABLED", "Wrong operational state")
 750                 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
 751                 storage = admin["storage"]
 752                 self.assertEqual(storage["folder"], did, "Wrong storage folder")
 753                 self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor")
 754                 compare_desc(self, test_nsd, db_args[2], "NSD")
 755             finally:
 756                 pass
 757         self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None: \
 758             {"_id": did, "_admin": db_nsd_content["_admin"]}
 759         with self.subTest(i=2, t='Check Pyangbind Validation: required properties'):
 760             tmp = test_nsd["id"]
 761             del test_nsd["id"]
 762             try:
 763                 with self.assertRaises(EngineException, msg="Accepted NSD with a missing required property") as e:
 764                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 765                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 766                 self.assertIn(norm("Error in pyangbind validation: '{}'".format("id")),
 767                               norm(str(e.exception)), "Wrong exception text")
 768             finally:
 769                 test_nsd["id"] = tmp
 770         with self.subTest(i=3, t='Check Pyangbind Validation: additional properties'):
 771             test_nsd["extra-property"] = 0
 772             try:
 773                 with self.assertRaises(EngineException, msg="Accepted NSD with an additional property") as e:
 774                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 775                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 776                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 777                                    .format("json object contained a key that did not exist", "extra-property")),
 778                               norm(str(e.exception)), "Wrong exception text")
 779             finally:
 780                 del test_nsd["extra-property"]
 781         with self.subTest(i=4, t='Check Pyangbind Validation: property types'):
 782             tmp = test_nsd["designer"]
 783             test_nsd["designer"] = {"key": 0}
 784             try:
 785                 with self.assertRaises(EngineException, msg="Accepted NSD with a wrongly typed property") as e:
 786                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 787                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 788                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 789                                    .format("json object contained a key that did not exist", "key")),
 790                               norm(str(e.exception)), "Wrong exception text")
 791             finally:
 792                 test_nsd["designer"] = tmp
 793         with self.subTest(i=5, t='Check Input Validation: mgmt-network+virtual-link-protocol-data'):
 794             df = test_nsd['df'][0]
 795             mgmt_profile = {'id': 'id', 'virtual-link-desc-id': 'mgmt',
 796                             'virtual-link-protocol-data': {'associated-layer-protocol': 'ipv4'}}
 797             df['virtual-link-profile'] = [mgmt_profile]
 798             try:
 799                 with self.assertRaises(EngineException, msg="Accepted VLD with mgmt-network+ip-profile") as e:
 800                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 801                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 802                 self.assertIn(norm("Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
 803                                    " You cannot set a virtual-link-protocol-data when mgmt-network is True"
 804                                    .format(df["id"], mgmt_profile["id"])),
 805                               norm(str(e.exception)), "Wrong exception text")
 806             finally:
 807                 del df['virtual-link-profile']
 808         with self.subTest(i=6, t='Check Descriptor Dependencies: vnfd-id[]'):
 809             self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 810             self.db.get_list.return_value = []
 811             try:
 812                 with self.assertRaises(EngineException, msg="Accepted wrong VNFD ID reference") as e:
 813                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 814                 self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 815                 self.assertIn(norm("'vnfd-id'='{}' references a non existing vnfd".format(test_nsd['vnfd-id'][0])),
 816                               norm(str(e.exception)), "Wrong exception text")
 817             finally:
 818                 pass
 819         with self.subTest(i=7, t='Check Descriptor Dependencies: '
 820                                  'vld[vnfd-connection-point-ref][vnfd-connection-point-ref]'):
 821             vnfd_descriptor = deepcopy(db_vnfd_content)
 822             df = test_nsd['df'][0]
 823             affected_vnf_profile = df['vnf-profile'][0]
 824             affected_virtual_link = affected_vnf_profile['virtual-link-connectivity'][1]
 825             affected_cpd = vnfd_descriptor['ext-cpd'].pop()
 826             self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 827             self.db.get_list.return_value = [vnfd_descriptor]
 828             try:
 829                 with self.assertRaises(EngineException, msg="Accepted wrong VLD CP reference") as e:
 830                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 831                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 832                 self.assertIn(norm("Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
 833                                    "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
 834                                    "non existing ext-cpd:id inside vnfd '{}'"
 835                                    .format(df["id"], affected_vnf_profile["id"],
 836                                            affected_virtual_link["virtual-link-profile-id"], affected_cpd["id"],
 837                                            vnfd_descriptor["id"])),
 838                               norm(str(e.exception)), "Wrong exception text")
 839             finally:
 840                 pass
 841         return
 842
 843     def test_edit_nsd(self):
 844         nsd_content = deepcopy(db_nsd_content)
 845         did = nsd_content["_id"]
 846         self.fs.file_exists.return_value = True
 847         self.fs.dir_ls.return_value = True
 848         with self.subTest(i=1, t='Normal Edition'):
 849             now = time()
 850             self.db.get_one.side_effect = [deepcopy(nsd_content), None]
 851             self.db.get_list.return_value = [db_vnfd_content]
 852             data = {"id": "new-nsd-id", "name": "new-nsd-name"}
 853             self.topic.edit(fake_session, did, data)
 854             db_args = self.db.replace.call_args[0]
 855             msg_args = self.msg.write.call_args[0]
 856             data["_id"] = did
 857             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 858             self.assertEqual(msg_args[1], "edited", "Wrong message action")
 859             self.assertEqual(msg_args[2], data, "Wrong message content")
 860             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 861             self.assertEqual(db_args[1], did, "Wrong DB ID")
 862             self.assertEqual(db_args[2]["_admin"]["created"], nsd_content["_admin"]["created"],
 863                              "Wrong creation time")
 864             self.assertGreater(db_args[2]["_admin"]["modified"], now, "Wrong modification time")
 865             self.assertEqual(db_args[2]["_admin"]["projects_read"], nsd_content["_admin"]["projects_read"],
 866                              "Wrong read-only project list")
 867             self.assertEqual(db_args[2]["_admin"]["projects_write"], nsd_content["_admin"]["projects_write"],
 868                              "Wrong read-write project list")
 869             self.assertEqual(db_args[2]["id"], data["id"], "Wrong NSD ID")
 870             self.assertEqual(db_args[2]["name"], data["name"], "Wrong NSD Name")
 871         with self.subTest(i=2, t='Conflict on Edit'):
 872             data = {"id": "fake-nsd-id", "name": "new-nsd-name"}
 873             self.db.get_one.side_effect = [nsd_content, {"_id": str(uuid4()), "id": data["id"]}]
 874             with self.assertRaises(EngineException, msg="Accepted existing NSD ID") as e:
 875                 self.topic.edit(fake_session, did, data)
 876             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 877             self.assertIn(norm("{} with id '{}' already exists for this project".format("nsd", data["id"])),
 878                           norm(str(e.exception)), "Wrong exception text")
 879         with self.subTest(i=3, t='Check Envelope'):
 880             data = {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
 881             self.db.get_one.side_effect = [nsd_content, None]
 882             with self.assertRaises(EngineException, msg="Accepted NSD with wrong envelope") as e:
 883                 self.topic.edit(fake_session, did, data, content=nsd_content)
 884             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 885             self.assertIn("'nsd' must be a list of only one element", norm(str(e.exception)), "Wrong exception text")
 886         return
 887
 888     def test_delete_nsd(self):
 889         did = db_nsd_content["_id"]
 890         self.db.get_one.return_value = db_nsd_content
 891         p_id = db_nsd_content["_admin"]["projects_read"][0]
 892         with self.subTest(i=1, t='Normal Deletion'):
 893             self.db.get_list.return_value = []
 894             self.db.del_one.return_value = {"deleted": 1}
 895             self.topic.delete(fake_session, did)
 896             db_args = self.db.del_one.call_args[0]
 897             msg_args = self.msg.write.call_args[0]
 898             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 899             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 900             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 901             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 902             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
 903             self.assertEqual(db_args[1]["_admin.projects_write.cont"], [p_id, 'ANY'], "Wrong DB filter")
 904             db_g1_args = self.db.get_one.call_args[0]
 905             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 906             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB NSD ID")
 907             db_gl_calls = self.db.get_list.call_args_list
 908             self.assertEqual(db_gl_calls[0][0][0], "nsrs", "Wrong DB topic")
 909             # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID")   # Filter changed after call
 910             self.assertEqual(db_gl_calls[1][0][0], "nsts", "Wrong DB topic")
 911             self.assertEqual(db_gl_calls[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"], db_nsd_content["id"],
 912                              "Wrong DB NSD netslice-subnet nsd-ref")
 913             self.db.set_one.assert_not_called()
 914             fs_del_calls = self.fs.file_delete.call_args_list
 915             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
 916             self.assertEqual(fs_del_calls[1][0][0], did + '_', "Wrong FS folder id")
 917         with self.subTest(i=2, t='Conflict on Delete - NSD in use by nsr'):
 918             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-nsr"}]
 919             with self.assertRaises(EngineException, msg="Accepted NSD in use by NSR") as e:
 920                 self.topic.delete(fake_session, did)
 921             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 922             self.assertIn("there is at least one ns instance using this descriptor", norm(str(e.exception)),
 923                           "Wrong exception text")
 924         with self.subTest(i=3, t='Conflict on Delete - NSD in use by NST'):
 925             self.db.get_list.side_effect = [[], [{"_id": str(uuid4()), "name": "fake-nst"}]]
 926             with self.assertRaises(EngineException, msg="Accepted NSD in use by NST") as e:
 927                 self.topic.delete(fake_session, did)
 928             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 929             self.assertIn("there is at least one netslice template referencing this descriptor", norm(str(e.exception)),
 930                           "Wrong exception text")
 931         with self.subTest(i=4, t='Non-existent NSD'):
 932             excp_msg = "Not found any {} with filter='{}'".format("NSD", {"_id": did})
 933             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
 934             with self.assertRaises(DbException, msg="Accepted non-existent NSD ID") as e:
 935                 self.topic.delete(fake_session, did)
 936             self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
 937             self.assertIn(norm(excp_msg), norm(str(e.exception)), "Wrong exception text")
 938         with self.subTest(i=5, t='No delete because referenced by other project'):
 939             db_nsd_content["_admin"]["projects_read"].append("other_project")
 940             self.db.get_one = Mock(return_value=db_nsd_content)
 941             self.db.get_list = Mock(return_value=[])
 942             self.msg.write.reset_mock()
 943             self.db.del_one.reset_mock()
 944             self.fs.file_delete.reset_mock()
 945
 946             self.topic.delete(fake_session, did)
 947             self.db.del_one.assert_not_called()
 948             self.msg.write.assert_not_called()
 949             db_g1_args = self.db.get_one.call_args[0]
 950             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 951             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 952             db_s1_args = self.db.set_one.call_args
 953             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
 954             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
 955             self.assertIn(p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter")
 956             self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary")
 957             self.assertEqual(db_s1_args[1]["pull_list"],
 958                              {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
 959                              "Wrong DB pull_list dictionary")
 960             self.fs.file_delete.assert_not_called()
 961         return
 962
 963     def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(self):
 964         indata = deepcopy(db_nsd_content)
 965         vld = indata['virtual-link-desc'][0]
 966         self.topic.validate_vld_mgmt_network_with_virtual_link_protocol_data(vld, indata)
 967
 968     def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(self):
 969         indata = deepcopy(db_nsd_content)
 970         vld = indata['virtual-link-desc'][0]
 971         df = indata['df'][0]
 972         affected_vlp = {'id': 'id', 'virtual-link-desc-id': 'mgmt',
 973                         'virtual-link-protocol-data': {'associated-layer-protocol': 'ipv4'}}
 974         df['virtual-link-profile'] = [affected_vlp]
 975         with self.assertRaises(EngineException) as e:
 976             self.topic.validate_vld_mgmt_network_with_virtual_link_protocol_data(vld, indata)
 977         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 978         self.assertIn(norm("Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
 979                            " You cannot set a virtual-link-protocol-data when mgmt-network is True"
 980                            .format(df["id"], affected_vlp["id"])),
 981                       norm(str(e.exception)), "Wrong exception text")
 982
 983     def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self):
 984         indata = deepcopy(db_nsd_content)
 985         self.topic.validate_vnf_profiles_vnfd_id(indata)
 986
 987     def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self):
 988         indata = deepcopy(db_nsd_content)
 989         df = indata['df'][0]
 990         affected_vnf_profile = df['vnf-profile'][0]
 991         indata['vnfd-id'] = ['non-existing-vnfd']
 992         with self.assertRaises(EngineException) as e:
 993             self.topic.validate_vnf_profiles_vnfd_id(indata)
 994         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 995         self.assertIn(norm("Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
 996                            "does not match any vnfd-id"
 997                            .format(df["id"], affected_vnf_profile["id"], affected_vnf_profile['vnfd-id'])),
 998                       norm(str(e.exception)), "Wrong exception text")
 999
1000     def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(self):
1001         nsd_descriptor = deepcopy(db_nsd_content)
1002         vnfd_descriptor = deepcopy(db_vnfd_content)
1003         df = nsd_descriptor['df'][0]
1004         vnfds_index = {vnfd_descriptor['id']: vnfd_descriptor}
1005         self.topic.validate_df_vnf_profiles_constituent_connection_points(df, vnfds_index)
1006
1007     def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(self):
1008         nsd_descriptor = deepcopy(db_nsd_content)
1009         vnfd_descriptor = deepcopy(db_vnfd_content)
1010         df = nsd_descriptor['df'][0]
1011         affected_vnf_profile = df['vnf-profile'][0]
1012         affected_virtual_link = affected_vnf_profile['virtual-link-connectivity'][1]
1013         vnfds_index = {vnfd_descriptor['id']: vnfd_descriptor}
1014         affected_cpd = vnfd_descriptor['ext-cpd'].pop()
1015         with self.assertRaises(EngineException) as e:
1016             self.topic.validate_df_vnf_profiles_constituent_connection_points(df, vnfds_index)
1017         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
1018         self.assertIn(norm("Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1019                            "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1020                            "non existing ext-cpd:id inside vnfd '{}'"
1021                            .format(df["id"], affected_vnf_profile["id"],
1022                                    affected_virtual_link["virtual-link-profile-id"], affected_cpd["id"],
1023                                    vnfd_descriptor["id"])),
1024                       norm(str(e.exception)), "Wrong exception text")
1025
1026     def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self):
1027         nsd_descriptor = deepcopy(db_nsd_content)
1028         invalid_vnfd_id = 'invalid-vnfd-id'
1029         nsd_descriptor['id'] = 'invalid-vnfd-id-ns'
1030         nsd_descriptor['vnfd-id'][0] = invalid_vnfd_id
1031         nsd_descriptor['df'][0]['vnf-profile'][0]['vnfd-id'] = invalid_vnfd_id
1032         nsd_descriptor['df'][0]['vnf-profile'][1]['vnfd-id'] = invalid_vnfd_id
1033         with self.assertRaises(EngineException) as e:
1034             self.db.get_list.return_value = []
1035             nsd_descriptor = self.topic.check_conflict_on_edit(fake_session, nsd_descriptor, [], 'id')
1036         self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
1037         self.assertIn(norm("Descriptor error at 'vnfd-id'='{}' references a non "
1038                            "existing vnfd".format(invalid_vnfd_id)),
1039                       norm(str(e.exception)), "Wrong exception text")
1040
1041
1042 if __name__ == '__main__':
1043     unittest.main()