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
  36 test_name = "test-user"
  37 db_vnfd_content = yaml.load(db_vnfds_text, Loader=yaml.Loader)[0]
  38 db_nsd_content = yaml.load(db_nsds_text, Loader=yaml.Loader)[0]
  39 test_pid = db_vnfd_content["_admin"]["projects_read"][0]
  40 fake_session = {"username": test_name, "project_id": (test_pid,), "method": None,
  41                 "admin": True, "force": False, "public": False, "allow_show_user_project_role": True}
  42
  43
  44 def norm(str):
  45     """Normalize string for checking"""
  46     return ' '.join(str.strip().split()).lower()
  47
  48
  49 def compare_desc(tc, d1, d2, k):
  50     """
  51     Compare two descriptors
  52     We need this function because some methods are adding/removing items to/from the descriptors
  53     before they are stored in the database, so the original and stored versions will differ
  54     What we check is that COMMON LEAF ITEMS are equal
  55     Lists of different length are not compared
  56     :param tc: Test Case wich provides context (in particular the assert* methods)
  57     :param d1,d2: Descriptors to be compared
  58     :param key/item being compared
  59     :return: Nothing
  60     """
  61     if isinstance(d1, dict) and isinstance(d2, dict):
  62         for key in d1.keys():
  63             if key in d2:
  64                 compare_desc(tc, d1[key], d2[key], k+"[{}]".format(key))
  65     elif isinstance(d1, list) and isinstance(d2, list) and len(d1) == len(d2):
  66         for i in range(len(d1)):
  67             compare_desc(tc, d1[i], d2[i], k+"[{}]".format(i))
  68     else:
  69         tc.assertEqual(d1, d2, "Wrong descriptor content: {}".format(k))
  70
  71
  72 class Test_VnfdTopic(TestCase):
  73
  74     @classmethod
  75     def setUpClass(cls):
  76         cls.test_name = "test-vnfd-topic"
  77
  78     @classmethod
  79     def tearDownClass(cls):
  80         pass
  81
  82     def setUp(self):
  83         self.db = Mock(dbbase.DbBase())
  84         self.fs = Mock(fsbase.FsBase())
  85         self.msg = Mock(msgbase.MsgBase())
  86         self.auth = Mock(authconn.Authconn(None, None, None))
  87         self.topic = VnfdTopic(self.db, self.fs, self.msg, self.auth)
  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.assertIn("_admin.projects_read." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary")
 513             self.assertIn("_admin.projects_write." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary")
 514             self.fs.file_delete.assert_not_called()
 515         return
 516
 517
 518 class Test_NsdTopic(TestCase):
 519
 520     @classmethod
 521     def setUpClass(cls):
 522         cls.test_name = "test-nsd-topic"
 523
 524     @classmethod
 525     def tearDownClass(cls):
 526         pass
 527
 528     def setUp(self):
 529         self.db = Mock(dbbase.DbBase())
 530         self.fs = Mock(fsbase.FsBase())
 531         self.msg = Mock(msgbase.MsgBase())
 532         self.auth = Mock(authconn.Authconn(None, None, None))
 533         self.topic = NsdTopic(self.db, self.fs, self.msg, self.auth)
 534
 535     def test_new_nsd(self):
 536         did = db_nsd_content["_id"]
 537         self.fs.get_params.return_value = {}
 538         self.fs.file_exists.return_value = False
 539         self.fs.file_open.side_effect = lambda path, mode: open("/tmp/" + str(uuid4()), "a+b")
 540         test_nsd = deepcopy(db_nsd_content)
 541         del test_nsd["_id"]
 542         del test_nsd["_admin"]
 543         with self.subTest(i=1, t='Normal Creation'):
 544             self.db.create.return_value = did
 545             rollback = []
 546             did2, oid = self.topic.new(rollback, fake_session, {})
 547             db_args = self.db.create.call_args[0]
 548             msg_args = self.msg.write.call_args[0]
 549             self.assertEqual(len(rollback), 1, "Wrong rollback length")
 550             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 551             self.assertEqual(msg_args[1], "created", "Wrong message action")
 552             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 553             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 554             self.assertEqual(did2, did, "Wrong DB NSD id")
 555             self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
 556             self.assertEqual(db_args[1]["_admin"]["modified"], db_args[1]["_admin"]["created"],
 557                              "Wrong modification time")
 558             self.assertEqual(db_args[1]["_admin"]["projects_read"], [test_pid], "Wrong read-only project list")
 559             self.assertEqual(db_args[1]["_admin"]["projects_write"], [test_pid], "Wrong read-write project list")
 560             try:
 561                 self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 562                 self.db.get_list.return_value = [db_vnfd_content]
 563                 self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 564                 msg_args = self.msg.write.call_args[0]
 565                 test_nsd["_id"] = did
 566                 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 567                 self.assertEqual(msg_args[1], "edited", "Wrong message action")
 568                 self.assertEqual(msg_args[2], test_nsd, "Wrong message content")
 569                 db_args = self.db.get_one.mock_calls[0][1]
 570                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 571                 self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id")
 572                 db_args = self.db.replace.call_args[0]
 573                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 574                 self.assertEqual(db_args[1], did, "Wrong DB NSD id")
 575                 admin = db_args[2]["_admin"]
 576                 db_admin = db_nsd_content["_admin"]
 577                 self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time")
 578                 self.assertGreater(admin["modified"], db_admin["created"], "Wrong modification time")
 579                 self.assertEqual(admin["projects_read"], db_admin["projects_read"], "Wrong read-only project list")
 580                 self.assertEqual(admin["projects_write"], db_admin["projects_write"], "Wrong read-write project list")
 581                 self.assertEqual(admin["onboardingState"], "ONBOARDED", "Wrong onboarding state")
 582                 self.assertEqual(admin["operationalState"], "ENABLED", "Wrong operational state")
 583                 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
 584                 storage = admin["storage"]
 585                 self.assertEqual(storage["folder"], did, "Wrong storage folder")
 586                 self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor")
 587                 compare_desc(self, test_nsd, db_args[2], "NSD")
 588             finally:
 589                 pass
 590         self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None:\
 591             {"_id": did, "_admin": db_nsd_content["_admin"]}
 592         with self.subTest(i=2, t='Check Pyangbind Validation: required properties'):
 593             tmp = test_nsd["id"]
 594             del test_nsd["id"]
 595             try:
 596                 with self.assertRaises(EngineException, msg="Accepted NSD with a missing required property") as e:
 597                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 598                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 599                 self.assertIn(norm("Error in pyangbind validation: '{}'".format("id")),
 600                               norm(str(e.exception)), "Wrong exception text")
 601             finally:
 602                 test_nsd["id"] = tmp
 603         with self.subTest(i=3, t='Check Pyangbind Validation: additional properties'):
 604             test_nsd["extra-property"] = 0
 605             try:
 606                 with self.assertRaises(EngineException, msg="Accepted NSD with an additional property") as e:
 607                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 608                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 609                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 610                                    .format("json object contained a key that did not exist", "extra-property")),
 611                               norm(str(e.exception)), "Wrong exception text")
 612             finally:
 613                 del test_nsd["extra-property"]
 614         with self.subTest(i=4, t='Check Pyangbind Validation: property types'):
 615             tmp = test_nsd["short-name"]
 616             test_nsd["short-name"] = {"key": 0}
 617             try:
 618                 with self.assertRaises(EngineException, msg="Accepted NSD with a wrongly typed property") as e:
 619                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 620                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 621                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 622                                    .format("json object contained a key that did not exist", "key")),
 623                               norm(str(e.exception)), "Wrong exception text")
 624             finally:
 625                 test_nsd["short-name"] = tmp
 626         with self.subTest(i=5, t='Check Input Validation: vld[mgmt-network+ip-profile]'):
 627             tmp = test_nsd["vld"][0]["vim-network-name"]
 628             del test_nsd["vld"][0]["vim-network-name"]
 629             test_nsd["vld"][0]["ip-profile-ref"] = "fake-ip-profile"
 630             try:
 631                 with self.assertRaises(EngineException, msg="Accepted VLD with mgmt-network+ip-profile") as e:
 632                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 633                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 634                 self.assertIn(norm("Error at vld[id='{}']:ip-profile-ref"
 635                                    " You cannot set an ip-profile when mgmt-network is True"
 636                                    .format(test_nsd["vld"][0]["id"])),
 637                               norm(str(e.exception)), "Wrong exception text")
 638             finally:
 639                 test_nsd["vld"][0]["vim-network-name"] = tmp
 640                 del test_nsd["vld"][0]["ip-profile-ref"]
 641         with self.subTest(i=6, t='Check Input Validation: vld[vnfd-connection-point-ref][vnfd-id-ref]'):
 642             tmp = test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"]
 643             test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"] = "wrong-vnfd-id-ref"
 644             try:
 645                 with self.assertRaises(EngineException, msg="Accepted VLD with wrong vnfd-connection-point-ref") as e:
 646                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 647                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 648                 self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[vnfd-id-ref='{}']"
 649                                    " does not match constituent-vnfd[member-vnf-index='{}']:vnfd-id-ref '{}'"
 650                                    .format(test_nsd["vld"][0]["id"],
 651                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"],
 652                                            test_nsd["constituent-vnfd"][0]["member-vnf-index"],
 653                                            test_nsd["constituent-vnfd"][0]["vnfd-id-ref"])),
 654                               norm(str(e.exception)), "Wrong exception text")
 655             finally:
 656                 test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"] = tmp
 657         with self.subTest(i=7, t='Check Input Validation: vld[vnfd-connection-point-ref][member-vnf-index-ref]'):
 658             tmp = test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"]
 659             test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"] = "wrong-member-vnf-index-ref"
 660             try:
 661                 with self.assertRaises(EngineException, msg="Accepted VLD with wrong vnfd-connection-point-ref") as e:
 662                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 663                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 664                 self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}']"
 665                                    " does not match any constituent-vnfd:member-vnf-index"
 666                                    .format(test_nsd["vld"][0]["id"],
 667                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"])),
 668                               norm(str(e.exception)), "Wrong exception text")
 669             finally:
 670                 test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"] = tmp
 671         with self.subTest(i=8, t='Check Input Validation: vnffgd[classifier][rsp-id-ref]'):
 672             test_nsd["vnffgd"] = [{"id": "fake-vnffgd-id",
 673                                    "rsp": [{"id": "fake-rsp-id"}],
 674                                    "classifier": [{"id": "fake-vnffgd-classifier-id", "rsp-id-ref": "wrong-rsp-id"}]}]
 675             try:
 676                 with self.assertRaises(EngineException, msg="Accepted VNF FGD with wrong classifier rsp-id-ref") as e:
 677                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 678                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 679                 self.assertIn(norm("Error at vnffgd[id='{}']:classifier[id='{}']:rsp-id-ref '{}'"
 680                                    " does not match any rsp:id"
 681                                    .format(test_nsd["vnffgd"][0]["id"],
 682                                            test_nsd["vnffgd"][0]["classifier"][0]["id"],
 683                                            test_nsd["vnffgd"][0]["classifier"][0]["rsp-id-ref"])),
 684                               norm(str(e.exception)), "Wrong exception text")
 685             finally:
 686                 test_nsd["vnffgd"][0]["classifier"][0]["rsp-id-ref"] = "fake-rsp-id"
 687         with self.subTest(i=9, t='Check Descriptor Dependencies: constituent-vnfd[vnfd-id-ref]'):
 688             self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 689             self.db.get_list.return_value = []
 690             try:
 691                 with self.assertRaises(EngineException, msg="Accepted wrong constituent VNFD ID reference") as e:
 692                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 693                 self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 694                 self.assertIn(norm("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}'"
 695                                    " references a non existing vnfd"
 696                                    .format(test_nsd["constituent-vnfd"][0]["vnfd-id-ref"])),
 697                               norm(str(e.exception)), "Wrong exception text")
 698             finally:
 699                 pass
 700         with self.subTest(i=10, t='Check Descriptor Dependencies: '
 701                                   'vld[vnfd-connection-point-ref][vnfd-connection-point-ref]'):
 702             tmp = test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-connection-point-ref"]
 703             test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-connection-point-ref"] = "wrong-vnfd-cp-ref"
 704             self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 705             self.db.get_list.return_value = [db_vnfd_content]
 706             try:
 707                 with self.assertRaises(EngineException, msg="Accepted wrong VLD CP reference") as e:
 708                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 709                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 710                 self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}']:"
 711                                    "vnfd-connection-point-ref='{}' references a non existing conection-point:name"
 712                                    " inside vnfd '{}'"
 713                                    .format(test_nsd["vld"][0]["id"],
 714                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"],
 715                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]
 716                                                    ["vnfd-connection-point-ref"],
 717                                            db_vnfd_content["id"])),
 718                               norm(str(e.exception)), "Wrong exception text")
 719             finally:
 720                 test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-connection-point-ref"] = tmp
 721         return
 722         with self.subTest(i=11, t='Check Input Validation: everything right'):
 723             test_nsd["id"] = "fake-nsd-id"
 724             self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 725             self.db.get_list.return_value = [db_vnfd_content]
 726             rc = self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 727             self.assertTrue(rc, "Input Validation: Unexpected failure")
 728         return
 729
 730     def test_edit_nsd(self):
 731         did = db_nsd_content["_id"]
 732         self.fs.file_exists.return_value = True
 733         self.fs.dir_ls.return_value = True
 734         with self.subTest(i=1, t='Normal Edition'):
 735             now = time()
 736             self.db.get_one.side_effect = [db_nsd_content, None]
 737             self.db.get_list.return_value = [db_vnfd_content]
 738             data = {"id": "new-nsd-id", "name": "new-nsd-name"}
 739             self.topic.edit(fake_session, did, data)
 740             db_args = self.db.replace.call_args[0]
 741             msg_args = self.msg.write.call_args[0]
 742             data["_id"] = did
 743             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 744             self.assertEqual(msg_args[1], "edited", "Wrong message action")
 745             self.assertEqual(msg_args[2], data, "Wrong message content")
 746             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 747             self.assertEqual(db_args[1], did, "Wrong DB ID")
 748             self.assertEqual(db_args[2]["_admin"]["created"], db_nsd_content["_admin"]["created"],
 749                              "Wrong creation time")
 750             self.assertGreater(db_args[2]["_admin"]["modified"], now, "Wrong modification time")
 751             self.assertEqual(db_args[2]["_admin"]["projects_read"], db_nsd_content["_admin"]["projects_read"],
 752                              "Wrong read-only project list")
 753             self.assertEqual(db_args[2]["_admin"]["projects_write"], db_nsd_content["_admin"]["projects_write"],
 754                              "Wrong read-write project list")
 755             self.assertEqual(db_args[2]["id"], data["id"], "Wrong NSD ID")
 756             self.assertEqual(db_args[2]["name"], data["name"], "Wrong NSD Name")
 757         with self.subTest(i=2, t='Conflict on Edit'):
 758             data = {"id": "fake-nsd-id", "name": "new-nsd-name"}
 759             self.db.get_one.side_effect = [db_nsd_content, {"_id": str(uuid4()), "id": data["id"]}]
 760             with self.assertRaises(EngineException, msg="Accepted existing NSD ID") as e:
 761                 self.topic.edit(fake_session, did, data)
 762             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 763             self.assertIn(norm("{} with id '{}' already exists for this project".format("nsd", data["id"])),
 764                           norm(str(e.exception)), "Wrong exception text")
 765         with self.subTest(i=3, t='Check Envelope'):
 766             data = {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}
 767             with self.assertRaises(EngineException, msg="Accepted NSD with wrong envelope") as e:
 768                 self.topic.edit(fake_session, did, data)
 769             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 770             self.assertIn("'nsd' must be a list of only one element", norm(str(e.exception)), "Wrong exception text")
 771         return
 772
 773     def test_delete_nsd(self):
 774         did = db_nsd_content["_id"]
 775         self.db.get_one.return_value = db_nsd_content
 776         p_id = db_nsd_content["_admin"]["projects_read"][0]
 777         with self.subTest(i=1, t='Normal Deletion'):
 778             self.db.get_list.return_value = []
 779             self.db.del_one.return_value = {"deleted": 1}
 780             self.topic.delete(fake_session, did)
 781             db_args = self.db.del_one.call_args[0]
 782             msg_args = self.msg.write.call_args[0]
 783             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 784             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 785             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 786             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 787             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
 788             self.assertEqual(db_args[1]["_admin.projects_write.cont"], [p_id, 'ANY'], "Wrong DB filter")
 789             db_g1_args = self.db.get_one.call_args[0]
 790             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 791             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB NSD ID")
 792             db_gl_calls = self.db.get_list.call_args_list
 793             self.assertEqual(db_gl_calls[0][0][0], "nsrs", "Wrong DB topic")
 794             # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID")   # Filter changed after call
 795             self.assertEqual(db_gl_calls[1][0][0], "nsts", "Wrong DB topic")
 796             self.assertEqual(db_gl_calls[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"], db_nsd_content["id"],
 797                              "Wrong DB NSD netslice-subnet nsd-ref")
 798             self.db.set_one.assert_not_called()
 799             fs_del_calls = self.fs.file_delete.call_args_list
 800             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
 801             self.assertEqual(fs_del_calls[1][0][0], did+'_', "Wrong FS folder id")
 802         with self.subTest(i=2, t='Conflict on Delete - NSD in use by nsr'):
 803             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-nsr"}]
 804             with self.assertRaises(EngineException, msg="Accepted NSD in use by NSR") as e:
 805                 self.topic.delete(fake_session, did)
 806             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 807             self.assertIn("there is at least one ns using this descriptor", norm(str(e.exception)),
 808                           "Wrong exception text")
 809         with self.subTest(i=3, t='Conflict on Delete - NSD in use by NST'):
 810             self.db.get_list.side_effect = [[], [{"_id": str(uuid4()), "name": "fake-nst"}]]
 811             with self.assertRaises(EngineException, msg="Accepted NSD in use by NST") as e:
 812                 self.topic.delete(fake_session, did)
 813             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 814             self.assertIn("there is at least one netslice template referencing this descriptor", norm(str(e.exception)),
 815                           "Wrong exception text")
 816         with self.subTest(i=4, t='Non-existent NSD'):
 817             excp_msg = "Not found any {} with filter='{}'".format("NSD", {"_id": did})
 818             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
 819             with self.assertRaises(DbException, msg="Accepted non-existent NSD ID") as e:
 820                 self.topic.delete(fake_session, did)
 821             self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
 822             self.assertIn(norm(excp_msg), norm(str(e.exception)), "Wrong exception text")
 823         with self.subTest(i=5, t='No delete because referenced by other project'):
 824             db_nsd_content["_admin"]["projects_read"].append("other_project")
 825             self.db.get_one = Mock(return_value=db_nsd_content)
 826             self.db.get_list = Mock(return_value=[])
 827             self.msg.write.reset_mock()
 828             self.db.del_one.reset_mock()
 829             self.fs.file_delete.reset_mock()
 830
 831             self.topic.delete(fake_session, did)
 832             self.db.del_one.assert_not_called()
 833             self.msg.write.assert_not_called()
 834             db_g1_args = self.db.get_one.call_args[0]
 835             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 836             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 837             db_s1_args = self.db.set_one.call_args
 838             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
 839             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
 840             self.assertIn(p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter")
 841             self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary")
 842             self.assertIn("_admin.projects_read." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary")
 843             self.assertIn("_admin.projects_write." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary")
 844             self.fs.file_delete.assert_not_called()
 845         return
 846
 847
 848 if __name__ == '__main__':
 849     unittest.main()