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