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"]["juju"]
 116             del test_vnfd["vdu"][0]["cloud-init-file"]
 117             del test_vnfd["vnf-configuration"]["juju"]
 118             try:
 119                 self.db.get_one.side_effect = [{"_id": did, "_admin": 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 = 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"]["juju"] = tmp2
 149         self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None: \
 150             {"_id": did, "_admin": db_vnfd_content["_admin"]}
 151         with self.subTest(i=2, t='Check Pyangbind Validation: required properties'):
 152             tmp = test_vnfd["id"]
 153             del test_vnfd["id"]
 154             try:
 155                 with self.assertRaises(EngineException, msg="Accepted VNFD with a missing required property") as e:
 156                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 157                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 158                 self.assertIn(norm("Error in pyangbind validation: '{}'".format("id")),
 159                               norm(str(e.exception)), "Wrong exception text")
 160             finally:
 161                 test_vnfd["id"] = tmp
 162         with self.subTest(i=3, t='Check Pyangbind Validation: additional properties'):
 163             test_vnfd["extra-property"] = 0
 164             try:
 165                 with self.assertRaises(EngineException, msg="Accepted VNFD with an additional property") as e:
 166                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 167                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 168                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 169                                    .format("json object contained a key that did not exist", "extra-property")),
 170                               norm(str(e.exception)), "Wrong exception text")
 171             finally:
 172                 del test_vnfd["extra-property"]
 173         with self.subTest(i=4, t='Check Pyangbind Validation: property types'):
 174             tmp = test_vnfd["short-name"]
 175             test_vnfd["short-name"] = {"key": 0}
 176             try:
 177                 with self.assertRaises(EngineException, msg="Accepted VNFD with a wrongly typed property") as e:
 178                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 179                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 180                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 181                                    .format("json object contained a key that did not exist", "key")),
 182                               norm(str(e.exception)), "Wrong exception text")
 183             finally:
 184                 test_vnfd["short-name"] = tmp
 185         with self.subTest(i=5, t='Check Input Validation: cloud-init'):
 186             with self.assertRaises(EngineException, msg="Accepted non-existent cloud_init file") as e:
 187                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 188             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 189             self.assertIn(norm("{} defined in vnf[id={}]:vdu[id={}] but not present in package"
 190                                .format("cloud-init", test_vnfd["id"], test_vnfd["vdu"][0]["id"])),
 191                           norm(str(e.exception)), "Wrong exception text")
 192         with self.subTest(i=6, t='Check Input Validation: vnf-configuration[juju]'):
 193             del test_vnfd["vdu"][0]["cloud-init-file"]
 194             with self.assertRaises(EngineException, msg="Accepted non-existent charm in VNF configuration") as e:
 195                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 196             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 197             self.assertIn(norm("{} defined in vnf[id={}] but not present in package".format("charm", test_vnfd["id"])),
 198                           norm(str(e.exception)), "Wrong exception text")
 199         with self.subTest(i=7, t='Check Input Validation: mgmt-interface'):
 200             del test_vnfd["vnf-configuration"]["juju"]
 201             tmp = test_vnfd["mgmt-interface"]
 202             del test_vnfd["mgmt-interface"]
 203             try:
 204                 with self.assertRaises(EngineException, msg="Accepted VNFD without management interface") 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("'{}' is a mandatory field and it is not defined".format("mgmt-interface")),
 208                               norm(str(e.exception)), "Wrong exception text")
 209             finally:
 210                 test_vnfd["mgmt-interface"] = tmp
 211         with self.subTest(i=8, t='Check Input Validation: mgmt-interface[cp]'):
 212             tmp = test_vnfd["mgmt-interface"]["cp"]
 213             test_vnfd["mgmt-interface"]["cp"] = "wrong-cp"
 214             try:
 215                 with self.assertRaises(EngineException,
 216                                        msg="Accepted wrong management interface 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("mgmt-interface:cp='{}' must match an existing connection-point"
 220                                    .format(test_vnfd["mgmt-interface"]["cp"])),
 221                               norm(str(e.exception)), "Wrong exception text")
 222             finally:
 223                 test_vnfd["mgmt-interface"]["cp"] = tmp
 224         with self.subTest(i=9, t='Check Input Validation: vdu[interface][external-connection-point-ref]'):
 225             tmp = test_vnfd["vdu"][0]["interface"][0]["external-connection-point-ref"]
 226             test_vnfd["vdu"][0]["interface"][0]["external-connection-point-ref"] = "wrong-cp"
 227             try:
 228                 with self.assertRaises(EngineException,
 229                                        msg="Accepted wrong VDU interface external connection point reference") as e:
 230                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 231                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 232                 self.assertIn(norm("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}'"
 233                                    " must match an existing connection-point"
 234                                    .format(test_vnfd["vdu"][0]["id"], test_vnfd["vdu"][0]["interface"][0]["name"],
 235                                            test_vnfd["vdu"][0]["interface"][0]["external-connection-point-ref"])),
 236                               norm(str(e.exception)), "Wrong exception text")
 237             finally:
 238                 test_vnfd["vdu"][0]["interface"][0]["external-connection-point-ref"] = tmp
 239         with self.subTest(i=10, t='Check Input Validation: vdu[interface][internal-connection-point-ref]'):
 240             tmp = test_vnfd["vdu"][1]["interface"][0]["internal-connection-point-ref"]
 241             test_vnfd["vdu"][1]["interface"][0]["internal-connection-point-ref"] = "wrong-cp"
 242             try:
 243                 with self.assertRaises(EngineException,
 244                                        msg="Accepted wrong VDU interface internal connection point reference") as e:
 245                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 246                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 247                 self.assertIn(norm("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}'"
 248                                    " must match an existing vdu:internal-connection-point"
 249                                    .format(test_vnfd["vdu"][1]["id"], test_vnfd["vdu"][1]["interface"][0]["name"],
 250                                            test_vnfd["vdu"][1]["interface"][0]["internal-connection-point-ref"])),
 251                               norm(str(e.exception)), "Wrong exception text")
 252             finally:
 253                 test_vnfd["vdu"][1]["interface"][0]["internal-connection-point-ref"] = tmp
 254         with self.subTest(i=11, t='Check Input Validation: vdu[vdu-configuration][juju]'):
 255             test_vnfd["vdu"][0]["vdu-configuration"] = {"juju": {"charm": "wrong-charm"}}
 256             try:
 257                 with self.assertRaises(EngineException, msg="Accepted non-existent charm in VDU configuration") as e:
 258                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 259                 self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 260                 self.assertIn(norm("{} defined in vnf[id={}]:vdu[id={}] but not present in package"
 261                                    .format("charm", test_vnfd["id"], test_vnfd["vdu"][0]["id"])),
 262                               norm(str(e.exception)), "Wrong exception text")
 263             finally:
 264                 del test_vnfd["vdu"][0]["vdu-configuration"]
 265         with self.subTest(i=12, t='Check Input Validation: Duplicated VLD name'):
 266             test_vnfd["internal-vld"].append(deepcopy(test_vnfd["internal-vld"][0]))
 267             test_vnfd["internal-vld"][1]["id"] = "wrong-internal-vld"
 268             try:
 269                 with self.assertRaises(EngineException, msg="Accepted duplicated VLD name") as e:
 270                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 271                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 272                 self.assertIn(norm("Duplicated VLD name '{}' in vnfd[id={}]:internal-vld[id={}]"
 273                                    .format(test_vnfd["internal-vld"][1]["name"], test_vnfd["id"],
 274                                            test_vnfd["internal-vld"][1]["id"])),
 275                               norm(str(e.exception)), "Wrong exception text")
 276             finally:
 277                 del test_vnfd["internal-vld"][1]
 278         with self.subTest(i=13, t='Check Input Validation: internal-vld[internal-connection-point][id-ref])'):
 279             tmp = test_vnfd["internal-vld"][0]["internal-connection-point"][0]["id-ref"]
 280             test_vnfd["internal-vld"][0]["internal-connection-point"][0]["id-ref"] = "wrong-icp-id-ref"
 281             try:
 282                 with self.assertRaises(EngineException, msg="Accepted non-existent internal VLD ICP id-ref") as e:
 283                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 284                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 285                 self.assertIn(norm("internal-vld[id='{}']:internal-connection-point='{}' must match an existing "
 286                                    "vdu:internal-connection-point"
 287                                    .format(test_vnfd["internal-vld"][0]["id"],
 288                                            test_vnfd["internal-vld"][0]["internal-connection-point"][0]["id-ref"])),
 289                               norm(str(e.exception)), "Wrong exception text")
 290             finally:
 291                 test_vnfd["internal-vld"][0]["internal-connection-point"][0]["id-ref"] = tmp
 292         with self.subTest(i=14, t='Check Input Validation: internal-vld[ip-profile-ref])'):
 293             test_vnfd["ip-profiles"] = [{"name": "fake-ip-profile-ref"}]
 294             test_vnfd["internal-vld"][0]["ip-profile-ref"] = "wrong-ip-profile-ref"
 295             try:
 296                 with self.assertRaises(EngineException, msg="Accepted non-existent IP Profile Ref") as e:
 297                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 298                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 299                 self.assertIn(norm("internal-vld[id='{}']:ip-profile-ref='{}' does not exist"
 300                                    .format(test_vnfd["internal-vld"][0]["id"],
 301                                            test_vnfd["internal-vld"][0]["ip-profile-ref"])),
 302                               norm(str(e.exception)), "Wrong exception text")
 303             finally:
 304                 del test_vnfd["ip-profiles"]
 305                 del test_vnfd["internal-vld"][0]["ip-profile-ref"]
 306         with self.subTest(i=15, t='Check Input Validation: vdu[monitoring-param])'):
 307             test_vnfd["monitoring-param"] = [{"id": "fake-mp-id", "vdu-monitoring-param": {
 308                 "vdu-monitoring-param-ref": "fake-vdu-mp-ref", "vdu-ref": "fake-vdu-ref"}}]
 309             test_vnfd["vdu"][0]["monitoring-param"] = [{"id": "wrong-vdu-mp-id"}]
 310             try:
 311                 with self.assertRaises(EngineException, msg="Accepted non-existent VDU Monitorimg Param") as e:
 312                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 313                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 314                 mp = test_vnfd["monitoring-param"][0]["vdu-monitoring-param"]
 315                 self.assertIn(norm("monitoring-param:vdu-monitoring-param:vdu-monitoring-param-ref='{}' not defined"
 316                                    " at vdu[id='{}'] or vdu does not exist"
 317                                    .format(mp["vdu-monitoring-param-ref"], mp["vdu-ref"])),
 318                               norm(str(e.exception)), "Wrong exception text")
 319             finally:
 320                 del test_vnfd["monitoring-param"]
 321                 del test_vnfd["vdu"][0]["monitoring-param"]
 322         with self.subTest(i=16, t='Check Input Validation: vdu[vdu-configuration][metrics]'):
 323             test_vnfd["monitoring-param"] = [{"id": "fake-mp-id", "vdu-metric": {
 324                 "vdu-metric-name-ref": "fake-vdu-mp-ref", "vdu-ref": "fake-vdu-ref"}}]
 325             test_vnfd["vdu"][0]["vdu-configuration"] = {"metrics": [{"name": "wrong-vdu-mp-id"}]}
 326             try:
 327                 with self.assertRaises(EngineException, msg="Accepted non-existent VDU Configuration Metric") as e:
 328                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 329                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 330                 mp = test_vnfd["monitoring-param"][0]["vdu-metric"]
 331                 self.assertIn(norm("monitoring-param:vdu-metric:vdu-metric-name-ref='{}' not defined"
 332                                    " at vdu[id='{}'] or vdu does not exist"
 333                                    .format(mp["vdu-metric-name-ref"], mp["vdu-ref"])),
 334                               norm(str(e.exception)), "Wrong exception text")
 335             finally:
 336                 del test_vnfd["monitoring-param"]
 337                 del test_vnfd["vdu"][0]["vdu-configuration"]
 338         with self.subTest(i=17, t='Check Input Validation: scaling-group-descriptor[scaling-policy][scaling-criteria]'):
 339             test_vnfd["monitoring-param"] = [{"id": "fake-mp-id"}]
 340             test_vnfd["scaling-group-descriptor"] = [{
 341                 "name": "fake-vnf-sg-name",
 342                 "vdu": [{"vdu-id-ref": "wrong-vdu-id-ref"}],
 343                 "scaling-policy": [{"name": "fake-vnf-sp-name", "scaling-criteria": [{
 344                     "name": "fake-vnf-sc-name", "vnf-monitoring-param-ref": "wrong-vnf-mp-id"}]}]}]
 345             with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group Policy Criteria") as e:
 346                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 347             self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 348             sg = test_vnfd["scaling-group-descriptor"][0]
 349             sc = sg["scaling-policy"][0]["scaling-criteria"][0]
 350             self.assertIn(norm("scaling-group-descriptor[name='{}']:scaling-criteria[name='{}']:"
 351                                "vnf-monitoring-param-ref='{}' not defined in any monitoring-param"
 352                                .format(sg["name"], sc["name"], sc["vnf-monitoring-param-ref"])),
 353                           norm(str(e.exception)), "Wrong exception text")
 354         with self.subTest(i=18, t='Check Input Validation: scaling-group-descriptor[vdu][vdu-id-ref]'):
 355             sc["vnf-monitoring-param-ref"] = "fake-mp-id"
 356             with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference") as e:
 357                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 358             self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 359             self.assertIn(norm("scaling-group-descriptor[name='{}']:vdu-id-ref={} does not match any vdu"
 360                                .format(sg["name"], sg["vdu"][0]["vdu-id-ref"])),
 361                           norm(str(e.exception)), "Wrong exception text")
 362         with self.subTest(i=19, t='Check Input Validation: scaling-group-descriptor[scaling-config-action]'):
 363             tmp = test_vnfd["vnf-configuration"]
 364             del test_vnfd["vnf-configuration"]
 365             sg["vdu"][0]["vdu-id-ref"] = test_vnfd["vdu"][0]["id"]
 366             sg["scaling-config-action"] = [{"trigger": "pre-scale-in"}]
 367             try:
 368                 with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference") \
 369                         as e:
 370                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 371                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 372                 self.assertIn(norm("'vnf-configuration' not defined in the descriptor but it is referenced"
 373                                    " by scaling-group-descriptor[name='{}']:scaling-config-action"
 374                                    .format(sg["name"])),
 375                               norm(str(e.exception)), "Wrong exception text")
 376             finally:
 377                 test_vnfd["vnf-configuration"] = tmp
 378         with self.subTest(i=20, t='Check Input Validation: scaling-group-descriptor[scaling-config-action]'
 379                                   '[vnf-config-primitive-name-ref]'):
 380             sg["scaling-config-action"][0]["vnf-config-primitive-name-ref"] = "wrong-sca-prim-name"
 381             with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference") as e:
 382                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 383             self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 384             self.assertIn(norm("scaling-group-descriptor[name='{}']:scaling-config-action:"
 385                                "vnf-config-primitive-name-ref='{}' does not match"
 386                                " any vnf-configuration:config-primitive:name"
 387                                .format(sg["name"], sg["scaling-config-action"][0]["vnf-config-primitive-name-ref"])),
 388                           norm(str(e.exception)), "Wrong exception text")
 389             # del test_vnfd["monitoring-param"]
 390             # del test_vnfd["scaling-group-descriptor"]
 391         with self.subTest(i=21, t='Check Input Validation: everything right'):
 392             sg["scaling-config-action"][0]["vnf-config-primitive-name-ref"] = "touch"
 393             test_vnfd["id"] = "fake-vnfd-id"
 394             self.db.get_one.side_effect = [{"_id": did, "_admin": db_vnfd_content["_admin"]}, None]
 395             rc = self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 396             self.assertTrue(rc, "Input Validation: Unexpected failure")
 397         return
 398
 399     def test_edit_vnfd(self):
 400         did = db_vnfd_content["_id"]
 401         self.fs.file_exists.return_value = True
 402         self.fs.dir_ls.return_value = True
 403         with self.subTest(i=1, t='Normal Edition'):
 404             now = time()
 405             self.db.get_one.side_effect = [db_vnfd_content, None]
 406             data = {"id": "new-vnfd-id", "name": "new-vnfd-name"}
 407             self.topic.edit(fake_session, did, data)
 408             db_args = self.db.replace.call_args[0]
 409             msg_args = self.msg.write.call_args[0]
 410             data["_id"] = did
 411             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 412             self.assertEqual(msg_args[1], "edited", "Wrong message action")
 413             self.assertEqual(msg_args[2], data, "Wrong message content")
 414             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 415             self.assertEqual(db_args[1], did, "Wrong DB ID")
 416             self.assertEqual(db_args[2]["_admin"]["created"], db_vnfd_content["_admin"]["created"],
 417                              "Wrong creation time")
 418             self.assertGreater(db_args[2]["_admin"]["modified"], now,
 419                                "Wrong modification time")
 420             self.assertEqual(db_args[2]["_admin"]["projects_read"], db_vnfd_content["_admin"]["projects_read"],
 421                              "Wrong read-only project list")
 422             self.assertEqual(db_args[2]["_admin"]["projects_write"], db_vnfd_content["_admin"]["projects_write"],
 423                              "Wrong read-write project list")
 424             self.assertEqual(db_args[2]["id"], data["id"], "Wrong VNFD ID")
 425             self.assertEqual(db_args[2]["name"], data["name"], "Wrong VNFD Name")
 426         with self.subTest(i=2, t='Conflict on Edit'):
 427             data = {"id": "fake-vnfd-id", "name": "new-vnfd-name"}
 428             self.db.get_one.side_effect = [db_vnfd_content, {"_id": str(uuid4()), "id": data["id"]}]
 429             with self.assertRaises(EngineException, msg="Accepted existing VNFD ID") as e:
 430                 self.topic.edit(fake_session, did, data)
 431             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 432             self.assertIn(norm("{} with id '{}' already exists for this project".format("vnfd", data["id"])),
 433                           norm(str(e.exception)), "Wrong exception text")
 434         with self.subTest(i=3, t='Check Envelope'):
 435             data = {"vnfd": {"id": "new-vnfd-id-1", "name": "new-vnfd-name"}}
 436             with self.assertRaises(EngineException, msg="Accepted VNFD with wrong envelope") as e:
 437                 self.topic.edit(fake_session, did, data)
 438             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 439             self.assertIn("'vnfd' must be a list of only one element", norm(str(e.exception)), "Wrong exception text")
 440         return
 441
 442     def test_delete_vnfd(self):
 443         did = db_vnfd_content["_id"]
 444         self.db.get_one.return_value = db_vnfd_content
 445         p_id = db_vnfd_content["_admin"]["projects_read"][0]
 446         with self.subTest(i=1, t='Normal Deletion'):
 447             self.db.get_list.return_value = []
 448             self.db.del_one.return_value = {"deleted": 1}
 449             self.topic.delete(fake_session, did)
 450             db_args = self.db.del_one.call_args[0]
 451             msg_args = self.msg.write.call_args[0]
 452             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 453             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 454             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 455             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 456             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
 457             self.assertEqual(db_args[1]["_admin.projects_write.cont"], [p_id, 'ANY'], "Wrong DB filter")
 458             db_g1_args = self.db.get_one.call_args[0]
 459             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 460             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 461             db_gl_calls = self.db.get_list.call_args_list
 462             self.assertEqual(db_gl_calls[0][0][0], "vnfrs", "Wrong DB topic")
 463             # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID")   # Filter changed after call
 464             self.assertEqual(db_gl_calls[1][0][0], "nsds", "Wrong DB topic")
 465             self.assertEqual(db_gl_calls[1][0][1]["constituent-vnfd.ANYINDEX.vnfd-id-ref"], db_vnfd_content["id"],
 466                              "Wrong DB NSD constituent-vnfd id-ref")
 467
 468             self.db.set_one.assert_not_called()
 469             fs_del_calls = self.fs.file_delete.call_args_list
 470             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
 471             self.assertEqual(fs_del_calls[1][0][0], did + '_', "Wrong FS folder id")
 472         with self.subTest(i=2, t='Conflict on Delete - VNFD in use by VNFR'):
 473             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-vnfr"}]
 474             with self.assertRaises(EngineException, msg="Accepted VNFD in use by VNFR") as e:
 475                 self.topic.delete(fake_session, did)
 476             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 477             self.assertIn("there is at least one vnf using this descriptor", norm(str(e.exception)),
 478                           "Wrong exception text")
 479         with self.subTest(i=3, t='Conflict on Delete - VNFD in use by NSD'):
 480             self.db.get_list.side_effect = [[], [{"_id": str(uuid4()), "name": "fake-nsd"}]]
 481             with self.assertRaises(EngineException, msg="Accepted VNFD in use by NSD") as e:
 482                 self.topic.delete(fake_session, did)
 483             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 484             self.assertIn("there is at least one nsd referencing this descriptor", norm(str(e.exception)),
 485                           "Wrong exception text")
 486         with self.subTest(i=4, t='Non-existent VNFD'):
 487             excp_msg = "Not found any {} with filter='{}'".format("VNFD", {"_id": did})
 488             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
 489             with self.assertRaises(DbException, msg="Accepted non-existent VNFD ID") as e:
 490                 self.topic.delete(fake_session, did)
 491             self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
 492             self.assertIn(norm(excp_msg), norm(str(e.exception)), "Wrong exception text")
 493         with self.subTest(i=5, t='No delete because referenced by other project'):
 494             db_vnfd_content["_admin"]["projects_read"].append("other_project")
 495             self.db.get_one = Mock(return_value=db_vnfd_content)
 496             self.db.get_list = Mock(return_value=[])
 497             self.msg.write.reset_mock()
 498             self.db.del_one.reset_mock()
 499             self.fs.file_delete.reset_mock()
 500
 501             self.topic.delete(fake_session, did)
 502             self.db.del_one.assert_not_called()
 503             self.msg.write.assert_not_called()
 504             db_g1_args = self.db.get_one.call_args[0]
 505             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 506             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 507             db_s1_args = self.db.set_one.call_args
 508             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
 509             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
 510             self.assertIn(p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter")
 511             self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary")
 512             self.assertEqual(db_s1_args[1]["pull_list"],
 513                              {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
 514                              "Wrong DB pull_list dictionary")
 515             self.fs.file_delete.assert_not_called()
 516         return
 517
 518     def test_validate_mgmt_interfaces_connection_points_on_valid_descriptor(self):
 519         indata = deepcopy(db_vnfd_content)
 520         self.topic.validate_mgmt_interfaces_connection_points(indata)
 521
 522     def test_validate_mgmt_interfaces_connection_points_when_missing_connection_point(self):
 523         indata = deepcopy(db_vnfd_content)
 524         indata['connection-point'] = []
 525         with self.assertRaises(EngineException) as e:
 526             self.topic.validate_mgmt_interfaces_connection_points(indata)
 527         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 528         self.assertIn(norm("mgmt-interface:cp='{}' must match an existing connection-point"
 529                            .format(indata["mgmt-interface"]["cp"])),
 530                       norm(str(e.exception)), "Wrong exception text")
 531
 532     def test_validate_mgmt_interfaces_connection_points_when_missing_mgmt_interface(self):
 533         indata = deepcopy(db_vnfd_content)
 534         indata.pop('mgmt-interface')
 535         with self.assertRaises(EngineException) as e:
 536             self.topic.validate_mgmt_interfaces_connection_points(indata)
 537         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 538         self.assertIn(norm("'mgmt-interface' is a mandatory field and it is not defined"),
 539                       norm(str(e.exception)), "Wrong exception text")
 540
 541     def test_validate_vdu_connection_point_refs_on_valid_descriptor(self):
 542         indata = db_vnfd_content
 543         vdu = indata['vdu'][0]
 544         self.topic.validate_vdu_connection_point_refs(vdu, indata)
 545
 546     def test_validate_vdu_connection_point_refs_when_missing_internal_connection_point(self):
 547         indata = deepcopy(db_vnfd_content)
 548         vdu = indata['vdu'][0]
 549         vdu.pop('internal-connection-point')
 550         with self.assertRaises(EngineException) as e:
 551             self.topic.validate_vdu_connection_point_refs(vdu, indata)
 552         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 553         self.assertIn(norm("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}' "
 554                            "must match an existing vdu:internal-connection-point"
 555                            .format(vdu["id"], vdu['interface'][1]["name"],
 556                                    vdu['interface'][1]["internal-connection-point-ref"])),
 557                       norm(str(e.exception)), "Wrong exception text")
 558
 559     def test_validate_vdu_connection_point_refs_when_missing_external_connection_point(self):
 560         indata = deepcopy(db_vnfd_content)
 561         vdu = indata['vdu'][0]
 562         indata.pop('connection-point')
 563         with self.assertRaises(EngineException) as e:
 564             self.topic.validate_vdu_connection_point_refs(vdu, indata)
 565         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 566         self.assertIn(norm("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}' "
 567                            "must match an existing connection-point"
 568                            .format(vdu["id"], vdu['interface'][0]["name"],
 569                                    vdu['interface'][0]["external-connection-point-ref"])),
 570                       norm(str(e.exception)), "Wrong exception text")
 571
 572     def test_validate_vdu_connection_point_refs_on_duplicated_internal_connection_point_ref(self):
 573         indata = deepcopy(db_vnfd_content)
 574         vdu = indata['vdu'][0]
 575         duplicated_interface = {'name': 'dup-mgmt-eth1', 'position': 3,
 576                                 'internal-connection-point-ref': 'mgmtVM-internal'}
 577         vdu['interface'].insert(0, duplicated_interface)
 578         with self.assertRaises(EngineException) as e:
 579             self.topic.validate_vdu_connection_point_refs(vdu, indata)
 580         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 581         self.assertIn(norm("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}' "
 582                            "is referenced by other interface"
 583                            .format(vdu["id"], vdu['interface'][2]["name"],
 584                                    vdu['interface'][2]["internal-connection-point-ref"])),
 585                       norm(str(e.exception)), "Wrong exception text")
 586
 587     def test_validate_vdu_connection_point_refs_on_duplicated_external_connection_point_ref(self):
 588         indata = deepcopy(db_vnfd_content)
 589         vdu = indata['vdu'][0]
 590         duplicated_interface = {'name': 'dup-mgmt-eth0', 'position': 3, 'external-connection-point-ref': 'vnf-mgmt'}
 591         vdu['interface'].insert(0, duplicated_interface)
 592         with self.assertRaises(EngineException) as e:
 593             self.topic.validate_vdu_connection_point_refs(vdu, indata)
 594         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 595         self.assertIn(norm("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}' "
 596                            "is referenced by other interface"
 597                            .format(vdu["id"], vdu['interface'][1]["name"],
 598                                    vdu['interface'][1]["external-connection-point-ref"])),
 599                       norm(str(e.exception)), "Wrong exception text")
 600
 601     def test_validate_internal_vlds_on_valid_descriptor(self):
 602         indata = db_vnfd_content
 603         self.topic.validate_internal_vlds(indata)
 604
 605     def test_validate_internal_vlds_on_duplicated_vld(self):
 606         indata = deepcopy(db_vnfd_content)
 607         duplicated_vld = {'id': 'internal', 'name': 'internal',
 608                           'internal-connection-point': [{'id-ref': 'mgmtVM-internal'}, {'id-ref': 'dataVM-internal'}]}
 609         indata['internal-vld'].insert(0, duplicated_vld)
 610         with self.assertRaises(EngineException) as e:
 611             self.topic.validate_internal_vlds(indata)
 612         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 613         self.assertIn(norm("Duplicated VLD name '{}' in vnfd[id={}]:internal-vld[id={}]"
 614                            .format(indata['internal-vld'][1]["name"], indata["id"], indata['internal-vld'][1]["id"])),
 615                       norm(str(e.exception)), "Wrong exception text")
 616
 617     def test_validate_internal_vlds_when_missing_internal_connection_point(self):
 618         indata = deepcopy(db_vnfd_content)
 619         ivld = indata['internal-vld'][0]
 620         icp = ivld['internal-connection-point'][0]
 621         indata['vdu'][0]['internal-connection-point'].pop()
 622         with self.assertRaises(EngineException) as e:
 623             self.topic.validate_internal_vlds(indata)
 624         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 625         self.assertIn(norm("internal-vld[id='{}']:internal-connection-point='{}' must match an existing "
 626                            "vdu:internal-connection-point".format(ivld["id"], icp["id-ref"])),
 627                       norm(str(e.exception)), "Wrong exception text")
 628
 629     def test_validate_internal_vlds_when_missing_ip_profile(self):
 630         indata = deepcopy(db_vnfd_content)
 631         ivld = indata['internal-vld'][0]
 632         ivld['ip-profile-ref'] = 'non-existing-ip-profile'
 633         with self.assertRaises(EngineException) as e:
 634             self.topic.validate_internal_vlds(indata)
 635         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 636         self.assertIn(norm("internal-vld[id='{}']:ip-profile-ref='{}' does not exist".format(
 637             ivld["id"], ivld["ip-profile-ref"])),
 638             norm(str(e.exception)), "Wrong exception text")
 639
 640     def test_validate_monitoring_params_on_valid_descriptor(self):
 641         indata = db_vnfd_content
 642         self.topic.validate_monitoring_params(indata)
 643
 644     def test_validate_monitoring_params_when_missing_vdu(self):
 645         indata = deepcopy(db_vnfd_content)
 646         monitoring_param = indata['monitoring-param'][0]
 647         indata['vdu'].pop()
 648         with self.assertRaises(EngineException) as e:
 649             self.topic.validate_monitoring_params(indata)
 650         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 651         self.assertIn(norm("monitoring-param:vdu-monitoring-param:vdu-monitoring-param-ref='{}' not "
 652                            "defined at vdu[id='{}'] or vdu does not exist"
 653                            .format(monitoring_param["vdu-monitoring-param"]["vdu-monitoring-param-ref"],
 654                                    monitoring_param["vdu-monitoring-param"]["vdu-ref"])),
 655                       norm(str(e.exception)), "Wrong exception text")
 656
 657     def test_validate_monitoring_params_when_missing_vdu_monitoring_param_ref(self):
 658         indata = deepcopy(db_vnfd_content)
 659         monitoring_param = indata['monitoring-param'][0]
 660         indata['vdu'][1]['monitoring-param'] = []
 661         with self.assertRaises(EngineException) as e:
 662             self.topic.validate_monitoring_params(indata)
 663         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 664         self.assertIn(norm("monitoring-param:vdu-monitoring-param:vdu-monitoring-param-ref='{}' not "
 665                            "defined at vdu[id='{}'] or vdu does not exist"
 666                            .format(monitoring_param["vdu-monitoring-param"]["vdu-monitoring-param-ref"],
 667                                    monitoring_param["vdu-monitoring-param"]["vdu-ref"])),
 668                       norm(str(e.exception)), "Wrong exception text")
 669
 670     def test_validate_scaling_group_descriptor_on_valid_descriptor(self):
 671         indata = db_vnfd_content
 672         self.topic.validate_scaling_group_descriptor(indata)
 673
 674     def test_validate_scaling_group_descriptor_when_missing_vnf_monitoring_param_ref(self):
 675         indata = deepcopy(db_vnfd_content)
 676         sgd = indata['scaling-group-descriptor'][0]
 677         sc = sgd['scaling-policy'][0]['scaling-criteria'][0]
 678         indata['monitoring-param'] = []
 679         with self.assertRaises(EngineException) as e:
 680             self.topic.validate_scaling_group_descriptor(indata)
 681         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 682         self.assertIn(norm("scaling-group-descriptor[name='{}']:scaling-criteria[name='{}']:"
 683                            "vnf-monitoring-param-ref='{}' not defined in any monitoring-param"
 684                            .format(sgd["name"], sc["name"], sc["vnf-monitoring-param-ref"])),
 685                       norm(str(e.exception)), "Wrong exception text")
 686
 687     def test_validate_scaling_group_descriptor_when_missing_vdu(self):
 688         indata = deepcopy(db_vnfd_content)
 689         sgd = indata['scaling-group-descriptor'][0]
 690         sgd_vdu = sgd['vdu'][0]
 691         indata['vdu'].pop()
 692         with self.assertRaises(EngineException) as e:
 693             self.topic.validate_scaling_group_descriptor(indata)
 694         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 695         self.assertIn(norm("scaling-group-descriptor[name='{}']:vdu-id-ref={} does not match any vdu"
 696                            .format(sgd["name"], sgd_vdu["vdu-id-ref"])),
 697                       norm(str(e.exception)), "Wrong exception text")
 698
 699     def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self):
 700         indata = deepcopy(db_vnfd_content)
 701         sgd = indata['scaling-group-descriptor'][0]
 702         indata.pop('vnf-configuration')
 703         with self.assertRaises(EngineException) as e:
 704             self.topic.validate_scaling_group_descriptor(indata)
 705         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 706         self.assertIn(norm("'vnf-configuration' not defined in the descriptor but it is referenced by "
 707                            "scaling-group-descriptor[name='{}']:scaling-config-action"
 708                            .format(sgd["name"])),
 709                       norm(str(e.exception)), "Wrong exception text")
 710
 711     def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(self):
 712         indata = deepcopy(db_vnfd_content)
 713         sgd = indata['scaling-group-descriptor'][0]
 714         sca = sgd['scaling-config-action'][0]
 715         indata['vnf-configuration']['config-primitive'] = []
 716         with self.assertRaises(EngineException) as e:
 717             self.topic.validate_scaling_group_descriptor(indata)
 718         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 719         self.assertIn(norm("scaling-group-descriptor[name='{}']:scaling-config-action:vnf-config-"
 720                            "primitive-name-ref='{}' does not match any "
 721                            "vnf-configuration:config-primitive:name"
 722                            .format(sgd["name"], sca["vnf-config-primitive-name-ref"])),
 723                       norm(str(e.exception)), "Wrong exception text")
 724
 725
 726 class Test_NsdTopic(TestCase):
 727
 728     @classmethod
 729     def setUpClass(cls):
 730         cls.test_name = "test-nsd-topic"
 731
 732     @classmethod
 733     def tearDownClass(cls):
 734         pass
 735
 736     def setUp(self):
 737         self.db = Mock(dbbase.DbBase())
 738         self.fs = Mock(fsbase.FsBase())
 739         self.msg = Mock(msgbase.MsgBase())
 740         self.auth = Mock(authconn.Authconn(None, None, None))
 741         self.topic = NsdTopic(self.db, self.fs, self.msg, self.auth)
 742         self.topic.check_quota = Mock(return_value=None)  # skip quota
 743
 744     def test_new_nsd(self):
 745         did = db_nsd_content["_id"]
 746         self.fs.get_params.return_value = {}
 747         self.fs.file_exists.return_value = False
 748         self.fs.file_open.side_effect = lambda path, mode: open("/tmp/" + str(uuid4()), "a+b")
 749         test_nsd = deepcopy(db_nsd_content)
 750         del test_nsd["_id"]
 751         del test_nsd["_admin"]
 752         with self.subTest(i=1, t='Normal Creation'):
 753             self.db.create.return_value = did
 754             rollback = []
 755             did2, oid = self.topic.new(rollback, fake_session, {})
 756             db_args = self.db.create.call_args[0]
 757             msg_args = self.msg.write.call_args[0]
 758             self.assertEqual(len(rollback), 1, "Wrong rollback length")
 759             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 760             self.assertEqual(msg_args[1], "created", "Wrong message action")
 761             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 762             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 763             self.assertEqual(did2, did, "Wrong DB NSD id")
 764             self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
 765             self.assertEqual(db_args[1]["_admin"]["modified"], db_args[1]["_admin"]["created"],
 766                              "Wrong modification time")
 767             self.assertEqual(db_args[1]["_admin"]["projects_read"], [test_pid], "Wrong read-only project list")
 768             self.assertEqual(db_args[1]["_admin"]["projects_write"], [test_pid], "Wrong read-write project list")
 769             try:
 770                 self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 771                 self.db.get_list.return_value = [db_vnfd_content]
 772                 self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 773                 msg_args = self.msg.write.call_args[0]
 774                 test_nsd["_id"] = did
 775                 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 776                 self.assertEqual(msg_args[1], "edited", "Wrong message action")
 777                 self.assertEqual(msg_args[2], test_nsd, "Wrong message content")
 778                 db_args = self.db.get_one.mock_calls[0][1]
 779                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 780                 self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id")
 781                 db_args = self.db.replace.call_args[0]
 782                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 783                 self.assertEqual(db_args[1], did, "Wrong DB NSD id")
 784                 admin = db_args[2]["_admin"]
 785                 db_admin = db_nsd_content["_admin"]
 786                 self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time")
 787                 self.assertGreater(admin["modified"], db_admin["created"], "Wrong modification time")
 788                 self.assertEqual(admin["projects_read"], db_admin["projects_read"], "Wrong read-only project list")
 789                 self.assertEqual(admin["projects_write"], db_admin["projects_write"], "Wrong read-write project list")
 790                 self.assertEqual(admin["onboardingState"], "ONBOARDED", "Wrong onboarding state")
 791                 self.assertEqual(admin["operationalState"], "ENABLED", "Wrong operational state")
 792                 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
 793                 storage = admin["storage"]
 794                 self.assertEqual(storage["folder"], did, "Wrong storage folder")
 795                 self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor")
 796                 compare_desc(self, test_nsd, db_args[2], "NSD")
 797             finally:
 798                 pass
 799         self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None: \
 800             {"_id": did, "_admin": db_nsd_content["_admin"]}
 801         with self.subTest(i=2, t='Check Pyangbind Validation: required properties'):
 802             tmp = test_nsd["id"]
 803             del test_nsd["id"]
 804             try:
 805                 with self.assertRaises(EngineException, msg="Accepted NSD with a missing required property") as e:
 806                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 807                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 808                 self.assertIn(norm("Error in pyangbind validation: '{}'".format("id")),
 809                               norm(str(e.exception)), "Wrong exception text")
 810             finally:
 811                 test_nsd["id"] = tmp
 812         with self.subTest(i=3, t='Check Pyangbind Validation: additional properties'):
 813             test_nsd["extra-property"] = 0
 814             try:
 815                 with self.assertRaises(EngineException, msg="Accepted NSD with an additional property") as e:
 816                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 817                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 818                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 819                                    .format("json object contained a key that did not exist", "extra-property")),
 820                               norm(str(e.exception)), "Wrong exception text")
 821             finally:
 822                 del test_nsd["extra-property"]
 823         with self.subTest(i=4, t='Check Pyangbind Validation: property types'):
 824             tmp = test_nsd["short-name"]
 825             test_nsd["short-name"] = {"key": 0}
 826             try:
 827                 with self.assertRaises(EngineException, msg="Accepted NSD with a wrongly typed property") as e:
 828                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 829                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 830                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 831                                    .format("json object contained a key that did not exist", "key")),
 832                               norm(str(e.exception)), "Wrong exception text")
 833             finally:
 834                 test_nsd["short-name"] = tmp
 835         with self.subTest(i=5, t='Check Input Validation: vld[mgmt-network+ip-profile]'):
 836             tmp = test_nsd["vld"][0]["vim-network-name"]
 837             del test_nsd["vld"][0]["vim-network-name"]
 838             test_nsd["vld"][0]["ip-profile-ref"] = "fake-ip-profile"
 839             try:
 840                 with self.assertRaises(EngineException, msg="Accepted VLD with mgmt-network+ip-profile") as e:
 841                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 842                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 843                 self.assertIn(norm("Error at vld[id='{}']:ip-profile-ref"
 844                                    " You cannot set an ip-profile when mgmt-network is True"
 845                                    .format(test_nsd["vld"][0]["id"])),
 846                               norm(str(e.exception)), "Wrong exception text")
 847             finally:
 848                 test_nsd["vld"][0]["vim-network-name"] = tmp
 849                 del test_nsd["vld"][0]["ip-profile-ref"]
 850         with self.subTest(i=6, t='Check Input Validation: vld[vnfd-connection-point-ref][vnfd-id-ref]'):
 851             tmp = test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"]
 852             test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"] = "wrong-vnfd-id-ref"
 853             try:
 854                 with self.assertRaises(EngineException, msg="Accepted VLD with wrong vnfd-connection-point-ref") as e:
 855                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 856                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 857                 self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[vnfd-id-ref='{}']"
 858                                    " does not match constituent-vnfd[member-vnf-index='{}']:vnfd-id-ref '{}'"
 859                                    .format(test_nsd["vld"][0]["id"],
 860                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"],
 861                                            test_nsd["constituent-vnfd"][0]["member-vnf-index"],
 862                                            test_nsd["constituent-vnfd"][0]["vnfd-id-ref"])),
 863                               norm(str(e.exception)), "Wrong exception text")
 864             finally:
 865                 test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"] = tmp
 866         with self.subTest(i=7, t='Check Input Validation: vld[vnfd-connection-point-ref][member-vnf-index-ref]'):
 867             tmp = test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"]
 868             test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"] = "wrong-member-vnf-index-ref"
 869             try:
 870                 with self.assertRaises(EngineException, msg="Accepted VLD with wrong vnfd-connection-point-ref") as e:
 871                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 872                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 873                 self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}']"
 874                                    " does not match any constituent-vnfd:member-vnf-index"
 875                                    .format(test_nsd["vld"][0]["id"],
 876                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"])),
 877                               norm(str(e.exception)), "Wrong exception text")
 878             finally:
 879                 test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"] = tmp
 880         with self.subTest(i=8, t='Check Input Validation: vnffgd[classifier][rsp-id-ref]'):
 881             test_nsd["vnffgd"] = [{"id": "fake-vnffgd-id",
 882                                    "rsp": [{"id": "fake-rsp-id"}],
 883                                    "classifier": [{"id": "fake-vnffgd-classifier-id", "rsp-id-ref": "wrong-rsp-id"}]}]
 884             try:
 885                 with self.assertRaises(EngineException, msg="Accepted VNF FGD with wrong classifier rsp-id-ref") as e:
 886                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 887                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 888                 self.assertIn(norm("Error at vnffgd[id='{}']:classifier[id='{}']:rsp-id-ref '{}'"
 889                                    " does not match any rsp:id"
 890                                    .format(test_nsd["vnffgd"][0]["id"],
 891                                            test_nsd["vnffgd"][0]["classifier"][0]["id"],
 892                                            test_nsd["vnffgd"][0]["classifier"][0]["rsp-id-ref"])),
 893                               norm(str(e.exception)), "Wrong exception text")
 894             finally:
 895                 test_nsd["vnffgd"][0]["classifier"][0]["rsp-id-ref"] = "fake-rsp-id"
 896         with self.subTest(i=9, t='Check Descriptor Dependencies: constituent-vnfd[vnfd-id-ref]'):
 897             self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 898             self.db.get_list.return_value = []
 899             try:
 900                 with self.assertRaises(EngineException, msg="Accepted wrong constituent VNFD ID reference") as e:
 901                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 902                 self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 903                 self.assertIn(norm("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}'"
 904                                    " references a non existing vnfd"
 905                                    .format(test_nsd["constituent-vnfd"][0]["vnfd-id-ref"])),
 906                               norm(str(e.exception)), "Wrong exception text")
 907             finally:
 908                 pass
 909         with self.subTest(i=10, t='Check Descriptor Dependencies: '
 910                                   'vld[vnfd-connection-point-ref][vnfd-connection-point-ref]'):
 911             tmp = test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-connection-point-ref"]
 912             test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-connection-point-ref"] = "wrong-vnfd-cp-ref"
 913             self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 914             self.db.get_list.return_value = [db_vnfd_content]
 915             try:
 916                 with self.assertRaises(EngineException, msg="Accepted wrong VLD CP reference") as e:
 917                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 918                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 919                 self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}']:"
 920                                    "vnfd-connection-point-ref='{}' references a non existing conection-point:name"
 921                                    " inside vnfd '{}'"
 922                                    .format(test_nsd["vld"][0]["id"],
 923                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"],
 924                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]
 925                                            ["vnfd-connection-point-ref"],
 926                                            db_vnfd_content["id"])),
 927                               norm(str(e.exception)), "Wrong exception text")
 928             finally:
 929                 test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-connection-point-ref"] = tmp
 930         return
 931
 932     def test_edit_nsd(self):
 933         did = db_nsd_content["_id"]
 934         self.fs.file_exists.return_value = True
 935         self.fs.dir_ls.return_value = True
 936         with self.subTest(i=1, t='Normal Edition'):
 937             now = time()
 938             self.db.get_one.side_effect = [db_nsd_content, None]
 939             self.db.get_list.return_value = [db_vnfd_content]
 940             data = {"id": "new-nsd-id", "name": "new-nsd-name"}
 941             self.topic.edit(fake_session, did, data)
 942             db_args = self.db.replace.call_args[0]
 943             msg_args = self.msg.write.call_args[0]
 944             data["_id"] = did
 945             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 946             self.assertEqual(msg_args[1], "edited", "Wrong message action")
 947             self.assertEqual(msg_args[2], data, "Wrong message content")
 948             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 949             self.assertEqual(db_args[1], did, "Wrong DB ID")
 950             self.assertEqual(db_args[2]["_admin"]["created"], db_nsd_content["_admin"]["created"],
 951                              "Wrong creation time")
 952             self.assertGreater(db_args[2]["_admin"]["modified"], now, "Wrong modification time")
 953             self.assertEqual(db_args[2]["_admin"]["projects_read"], db_nsd_content["_admin"]["projects_read"],
 954                              "Wrong read-only project list")
 955             self.assertEqual(db_args[2]["_admin"]["projects_write"], db_nsd_content["_admin"]["projects_write"],
 956                              "Wrong read-write project list")
 957             self.assertEqual(db_args[2]["id"], data["id"], "Wrong NSD ID")
 958             self.assertEqual(db_args[2]["name"], data["name"], "Wrong NSD Name")
 959         with self.subTest(i=2, t='Conflict on Edit'):
 960             data = {"id": "fake-nsd-id", "name": "new-nsd-name"}
 961             self.db.get_one.side_effect = [db_nsd_content, {"_id": str(uuid4()), "id": data["id"]}]
 962             with self.assertRaises(EngineException, msg="Accepted existing NSD ID") as e:
 963                 self.topic.edit(fake_session, did, data)
 964             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 965             self.assertIn(norm("{} with id '{}' already exists for this project".format("nsd", data["id"])),
 966                           norm(str(e.exception)), "Wrong exception text")
 967         with self.subTest(i=3, t='Check Envelope'):
 968             data = {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}
 969             with self.assertRaises(EngineException, msg="Accepted NSD with wrong envelope") as e:
 970                 self.topic.edit(fake_session, did, data)
 971             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 972             self.assertIn("'nsd' must be a list of only one element", norm(str(e.exception)), "Wrong exception text")
 973         return
 974
 975     def test_delete_nsd(self):
 976         did = db_nsd_content["_id"]
 977         self.db.get_one.return_value = db_nsd_content
 978         p_id = db_nsd_content["_admin"]["projects_read"][0]
 979         with self.subTest(i=1, t='Normal Deletion'):
 980             self.db.get_list.return_value = []
 981             self.db.del_one.return_value = {"deleted": 1}
 982             self.topic.delete(fake_session, did)
 983             db_args = self.db.del_one.call_args[0]
 984             msg_args = self.msg.write.call_args[0]
 985             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 986             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 987             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 988             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 989             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
 990             self.assertEqual(db_args[1]["_admin.projects_write.cont"], [p_id, 'ANY'], "Wrong DB filter")
 991             db_g1_args = self.db.get_one.call_args[0]
 992             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 993             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB NSD ID")
 994             db_gl_calls = self.db.get_list.call_args_list
 995             self.assertEqual(db_gl_calls[0][0][0], "nsrs", "Wrong DB topic")
 996             # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID")   # Filter changed after call
 997             self.assertEqual(db_gl_calls[1][0][0], "nsts", "Wrong DB topic")
 998             self.assertEqual(db_gl_calls[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"], db_nsd_content["id"],
 999                              "Wrong DB NSD netslice-subnet nsd-ref")
1000             self.db.set_one.assert_not_called()
1001             fs_del_calls = self.fs.file_delete.call_args_list
1002             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
1003             self.assertEqual(fs_del_calls[1][0][0], did + '_', "Wrong FS folder id")
1004         with self.subTest(i=2, t='Conflict on Delete - NSD in use by nsr'):
1005             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-nsr"}]
1006             with self.assertRaises(EngineException, msg="Accepted NSD in use by NSR") as e:
1007                 self.topic.delete(fake_session, did)
1008             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
1009             self.assertIn("there is at least one ns using this descriptor", norm(str(e.exception)),
1010                           "Wrong exception text")
1011         with self.subTest(i=3, t='Conflict on Delete - NSD in use by NST'):
1012             self.db.get_list.side_effect = [[], [{"_id": str(uuid4()), "name": "fake-nst"}]]
1013             with self.assertRaises(EngineException, msg="Accepted NSD in use by NST") as e:
1014                 self.topic.delete(fake_session, did)
1015             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
1016             self.assertIn("there is at least one netslice template referencing this descriptor", norm(str(e.exception)),
1017                           "Wrong exception text")
1018         with self.subTest(i=4, t='Non-existent NSD'):
1019             excp_msg = "Not found any {} with filter='{}'".format("NSD", {"_id": did})
1020             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
1021             with self.assertRaises(DbException, msg="Accepted non-existent NSD ID") as e:
1022                 self.topic.delete(fake_session, did)
1023             self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
1024             self.assertIn(norm(excp_msg), norm(str(e.exception)), "Wrong exception text")
1025         with self.subTest(i=5, t='No delete because referenced by other project'):
1026             db_nsd_content["_admin"]["projects_read"].append("other_project")
1027             self.db.get_one = Mock(return_value=db_nsd_content)
1028             self.db.get_list = Mock(return_value=[])
1029             self.msg.write.reset_mock()
1030             self.db.del_one.reset_mock()
1031             self.fs.file_delete.reset_mock()
1032
1033             self.topic.delete(fake_session, did)
1034             self.db.del_one.assert_not_called()
1035             self.msg.write.assert_not_called()
1036             db_g1_args = self.db.get_one.call_args[0]
1037             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
1038             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
1039             db_s1_args = self.db.set_one.call_args
1040             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
1041             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
1042             self.assertIn(p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter")
1043             self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary")
1044             self.assertEqual(db_s1_args[1]["pull_list"],
1045                              {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
1046                              "Wrong DB pull_list dictionary")
1047             self.fs.file_delete.assert_not_called()
1048         return
1049
1050     def test_validate_vld_mgmt_network_with_ip_profile_ref_on_valid_descriptor(self):
1051         indata = deepcopy(db_nsd_content)
1052         vld = indata['vld'][0]
1053         self.topic.validate_vld_mgmt_network_with_ip_profile_ref(vld)
1054
1055     def test_validate_vld_mgmt_network_with_ip_profile_ref_when_both_defined(self):
1056         indata = deepcopy(db_nsd_content)
1057         vld = indata['vld'][0]
1058         vld['ip-profile-ref'] = 'a-profile-ref'
1059         with self.assertRaises(EngineException) as e:
1060             self.topic.validate_vld_mgmt_network_with_ip_profile_ref(vld)
1061         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
1062         self.assertIn(norm("Error at vld[id='{}']:ip-profile-ref"
1063                            " You cannot set an ip-profile when mgmt-network is True"
1064                            .format(vld["id"])),
1065                       norm(str(e.exception)), "Wrong exception text")
1066
1067     def test_validate_vld_connection_point_refs_on_valid_descriptor(self):
1068         indata = deepcopy(db_nsd_content)
1069         vld = indata['vld'][0]
1070         self.topic.validate_vld_connection_point_refs(vld, indata)
1071
1072     def test_validate_vld_connection_point_refs_when_missing_constituent_vnfd(self):
1073         indata = deepcopy(db_nsd_content)
1074         vld = indata['vld'][0]
1075         vnfd_cp = vld['vnfd-connection-point-ref'][0]
1076         indata['constituent-vnfd'] = []
1077         with self.assertRaises(EngineException) as e:
1078             self.topic.validate_vld_connection_point_refs(vld, indata)
1079         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
1080         self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}'] "
1081                            "does not match any constituent-vnfd:member-vnf-index"
1082                            .format(vld["id"], vnfd_cp["member-vnf-index-ref"])),
1083                       norm(str(e.exception)), "Wrong exception text")
1084
1085     def test_validate_vld_connection_point_refs_on_unmatched_constituent_vnfd(self):
1086         indata = deepcopy(db_nsd_content)
1087         vld = indata['vld'][0]
1088         vnfd_cp = vld['vnfd-connection-point-ref'][0]
1089         constituent_vnfd = indata['constituent-vnfd'][0]
1090         constituent_vnfd['vnfd-id-ref'] = 'unmatched-vnfd-id-ref'
1091         with self.assertRaises(EngineException) as e:
1092             self.topic.validate_vld_connection_point_refs(vld, indata)
1093         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
1094         self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[vnfd-id-ref='{}'] "
1095                            "does not match constituent-vnfd[member-vnf-index='{}']:vnfd-id-ref"
1096                            " '{}'".format(vld["id"], vnfd_cp["vnfd-id-ref"],
1097                                           constituent_vnfd["member-vnf-index"],
1098                                           constituent_vnfd["vnfd-id-ref"])),
1099                       norm(str(e.exception)), "Wrong exception text")
1100
1101     def test_validate_vld_connection_point_refs_vnfd_connection_points_on_valid_descriptor(self):
1102         nsd_descriptor = deepcopy(db_nsd_content)
1103         vnfd_1_descriptor = deepcopy(db_vnfd_content)
1104         vnfd_2_descriptor = deepcopy(db_vnfd_content)
1105         vld = nsd_descriptor['vld'][0]
1106         member_vnfd_index = {'1': vnfd_1_descriptor, '2': vnfd_2_descriptor}
1107         self.topic.validate_vld_connection_point_refs_vnfd_connection_points(vld, member_vnfd_index)
1108
1109     def test_validate_vld_connection_point_refs_vnfd_connection_points_when_missing_connection_point(self):
1110         nsd_descriptor = deepcopy(db_nsd_content)
1111         vnfd_1_descriptor = deepcopy(db_vnfd_content)
1112         vnfd_2_descriptor = deepcopy(db_vnfd_content)
1113         vld = nsd_descriptor['vld'][0]
1114         referenced_vnfd_cp = vld['vnfd-connection-point-ref'][0]
1115         member_vnfd_index = {'1': vnfd_1_descriptor, '2': vnfd_2_descriptor}
1116         vnfd_1_descriptor['connection-point'] = []
1117         with self.assertRaises(EngineException) as e:
1118             self.topic.validate_vld_connection_point_refs_vnfd_connection_points(vld, member_vnfd_index)
1119         self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
1120         self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}']:vnfd-"
1121                            "connection-point-ref='{}' references a non existing conection-point:name inside vnfd '{}'"
1122                            .format(vld["id"], referenced_vnfd_cp["member-vnf-index-ref"],
1123                                    referenced_vnfd_cp["vnfd-connection-point-ref"], vnfd_1_descriptor["id"])),
1124                       norm(str(e.exception)), "Wrong exception text")
1125
1126     def test_check_conflict_on_edit_when_missing_vnfd_index(self):
1127         nsd_descriptor = deepcopy(db_nsd_content)
1128         invalid_vnfd_id_ref = 'invalid-vnfd-id-ref'
1129         nsd_descriptor['id'] = 'invalid-vnfd-id-ns'
1130         nsd_descriptor['constituent-vnfd'][0]['vnfd-id-ref'] = invalid_vnfd_id_ref
1131         nsd_descriptor['vld'][0]['vnfd-connection-point-ref'][0]['vnfd-id-ref'] = invalid_vnfd_id_ref
1132         nsd_descriptor['vld'][1]['vnfd-connection-point-ref'][0]['vnfd-id-ref'] = invalid_vnfd_id_ref
1133         with self.assertRaises(EngineException) as e:
1134             self.db.get_list.return_value = []
1135             self.topic.check_conflict_on_edit(fake_session, nsd_descriptor, [], 'id')
1136         self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
1137         self.assertIn(norm("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}' references a non "
1138                            "existing vnfd".format(invalid_vnfd_id_ref)),
1139                       norm(str(e.exception)), "Wrong exception text")
1140
1141
1142 if __name__ == '__main__':
1143     unittest.main()