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