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         self.topic.check_quota = Mock(return_value=None)  # skip quota
  89
  90     def test_new_vnfd(self):
  91         did = db_vnfd_content["_id"]
  92         self.fs.get_params.return_value = {}
  93         self.fs.file_exists.return_value = False
  94         self.fs.file_open.side_effect = lambda path, mode: open("/tmp/" + str(uuid4()), "a+b")
  95         test_vnfd = deepcopy(db_vnfd_content)
  96         del test_vnfd["_id"]
  97         del test_vnfd["_admin"]
  98         with self.subTest(i=1, t='Normal Creation'):
  99             self.db.create.return_value = did
 100             rollback = []
 101             did2, oid = self.topic.new(rollback, fake_session, {})
 102             db_args = self.db.create.call_args[0]
 103             msg_args = self.msg.write.call_args[0]
 104             self.assertEqual(len(rollback), 1, "Wrong rollback length")
 105             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 106             self.assertEqual(msg_args[1], "created", "Wrong message action")
 107             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 108             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 109             self.assertEqual(did2, did, "Wrong DB VNFD id")
 110             self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
 111             self.assertEqual(db_args[1]["_admin"]["modified"], db_args[1]["_admin"]["created"],
 112                              "Wrong modification time")
 113             self.assertEqual(db_args[1]["_admin"]["projects_read"], [test_pid], "Wrong read-only project list")
 114             self.assertEqual(db_args[1]["_admin"]["projects_write"], [test_pid], "Wrong read-write project list")
 115             tmp1 = test_vnfd["vdu"][0]["cloud-init-file"]
 116             tmp2 = test_vnfd["vnf-configuration"]["juju"]
 117             del test_vnfd["vdu"][0]["cloud-init-file"]
 118             del test_vnfd["vnf-configuration"]["juju"]
 119             try:
 120                 self.db.get_one.side_effect = [{"_id": did, "_admin": db_vnfd_content["_admin"]}, None]
 121                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 122                 msg_args = self.msg.write.call_args[0]
 123                 test_vnfd["_id"] = did
 124                 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 125                 self.assertEqual(msg_args[1], "edited", "Wrong message action")
 126                 self.assertEqual(msg_args[2], test_vnfd, "Wrong message content")
 127                 db_args = self.db.get_one.mock_calls[0][1]
 128                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 129                 self.assertEqual(db_args[1]["_id"], did, "Wrong DB VNFD id")
 130                 db_args = self.db.replace.call_args[0]
 131                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 132                 self.assertEqual(db_args[1], did, "Wrong DB VNFD id")
 133                 admin = db_args[2]["_admin"]
 134                 db_admin = db_vnfd_content["_admin"]
 135                 self.assertEqual(admin["type"], "vnfd", "Wrong descriptor type")
 136                 self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time")
 137                 self.assertGreater(admin["modified"], db_admin["created"], "Wrong modification time")
 138                 self.assertEqual(admin["projects_read"], db_admin["projects_read"], "Wrong read-only project list")
 139                 self.assertEqual(admin["projects_write"], db_admin["projects_write"], "Wrong read-write project list")
 140                 self.assertEqual(admin["onboardingState"], "ONBOARDED", "Wrong onboarding state")
 141                 self.assertEqual(admin["operationalState"], "ENABLED", "Wrong operational state")
 142                 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
 143                 storage = admin["storage"]
 144                 self.assertEqual(storage["folder"], did, "Wrong storage folder")
 145                 self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor")
 146                 compare_desc(self, test_vnfd, db_args[2], "VNFD")
 147             finally:
 148                 test_vnfd["vdu"][0]["cloud-init-file"] = tmp1
 149                 test_vnfd["vnf-configuration"]["juju"] = tmp2
 150         self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None:\
 151             {"_id": did, "_admin": db_vnfd_content["_admin"]}
 152         with self.subTest(i=2, t='Check Pyangbind Validation: required properties'):
 153             tmp = test_vnfd["id"]
 154             del test_vnfd["id"]
 155             try:
 156                 with self.assertRaises(EngineException, msg="Accepted VNFD with a missing required property") as e:
 157                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 158                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 159                 self.assertIn(norm("Error in pyangbind validation: '{}'".format("id")),
 160                               norm(str(e.exception)), "Wrong exception text")
 161             finally:
 162                 test_vnfd["id"] = tmp
 163         with self.subTest(i=3, t='Check Pyangbind Validation: additional properties'):
 164             test_vnfd["extra-property"] = 0
 165             try:
 166                 with self.assertRaises(EngineException, msg="Accepted VNFD with an additional property") as e:
 167                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 168                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 169                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 170                                    .format("json object contained a key that did not exist", "extra-property")),
 171                               norm(str(e.exception)), "Wrong exception text")
 172             finally:
 173                 del test_vnfd["extra-property"]
 174         with self.subTest(i=4, t='Check Pyangbind Validation: property types'):
 175             tmp = test_vnfd["short-name"]
 176             test_vnfd["short-name"] = {"key": 0}
 177             try:
 178                 with self.assertRaises(EngineException, msg="Accepted VNFD with a wrongly typed property") as e:
 179                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 180                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 181                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 182                                    .format("json object contained a key that did not exist", "key")),
 183                               norm(str(e.exception)), "Wrong exception text")
 184             finally:
 185                 test_vnfd["short-name"] = tmp
 186         with self.subTest(i=5, t='Check Input Validation: cloud-init'):
 187             with self.assertRaises(EngineException, msg="Accepted non-existent cloud_init file") as e:
 188                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 189             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 190             self.assertIn(norm("{} defined in vnf[id={}]:vdu[id={}] but not present in package"
 191                                .format("cloud-init", test_vnfd["id"], test_vnfd["vdu"][0]["id"])),
 192                           norm(str(e.exception)), "Wrong exception text")
 193         with self.subTest(i=6, t='Check Input Validation: vnf-configuration[juju]'):
 194             del test_vnfd["vdu"][0]["cloud-init-file"]
 195             with self.assertRaises(EngineException, msg="Accepted non-existent charm in VNF configuration") as e:
 196                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 197             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 198             self.assertIn(norm("{} defined in vnf[id={}] but not present in package".format("charm", test_vnfd["id"])),
 199                           norm(str(e.exception)), "Wrong exception text")
 200         with self.subTest(i=7, t='Check Input Validation: mgmt-interface'):
 201             del test_vnfd["vnf-configuration"]["juju"]
 202             tmp = test_vnfd["mgmt-interface"]
 203             del test_vnfd["mgmt-interface"]
 204             try:
 205                 with self.assertRaises(EngineException, msg="Accepted VNFD without management interface") as e:
 206                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 207                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 208                 self.assertIn(norm("'{}' is a mandatory field and it is not defined".format("mgmt-interface")),
 209                               norm(str(e.exception)), "Wrong exception text")
 210             finally:
 211                 test_vnfd["mgmt-interface"] = tmp
 212         with self.subTest(i=8, t='Check Input Validation: mgmt-interface[cp]'):
 213             tmp = test_vnfd["mgmt-interface"]["cp"]
 214             test_vnfd["mgmt-interface"]["cp"] = "wrong-cp"
 215             try:
 216                 with self.assertRaises(EngineException,
 217                                        msg="Accepted wrong management interface connection point") as e:
 218                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 219                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 220                 self.assertIn(norm("mgmt-interface:cp='{}' must match an existing connection-point"
 221                                    .format(test_vnfd["mgmt-interface"]["cp"])),
 222                               norm(str(e.exception)), "Wrong exception text")
 223             finally:
 224                 test_vnfd["mgmt-interface"]["cp"] = tmp
 225         with self.subTest(i=9, t='Check Input Validation: vdu[interface][external-connection-point-ref]'):
 226             tmp = test_vnfd["vdu"][0]["interface"][0]["external-connection-point-ref"]
 227             test_vnfd["vdu"][0]["interface"][0]["external-connection-point-ref"] = "wrong-cp"
 228             try:
 229                 with self.assertRaises(EngineException,
 230                                        msg="Accepted wrong VDU interface external connection point reference") as e:
 231                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 232                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 233                 self.assertIn(norm("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}'"
 234                                    " must match an existing connection-point"
 235                                    .format(test_vnfd["vdu"][0]["id"], test_vnfd["vdu"][0]["interface"][0]["name"],
 236                                            test_vnfd["vdu"][0]["interface"][0]["external-connection-point-ref"])),
 237                               norm(str(e.exception)), "Wrong exception text")
 238             finally:
 239                 test_vnfd["vdu"][0]["interface"][0]["external-connection-point-ref"] = tmp
 240         with self.subTest(i=10, t='Check Input Validation: vdu[interface][internal-connection-point-ref]'):
 241             tmp = test_vnfd["vdu"][1]["interface"][0]["internal-connection-point-ref"]
 242             test_vnfd["vdu"][1]["interface"][0]["internal-connection-point-ref"] = "wrong-cp"
 243             try:
 244                 with self.assertRaises(EngineException,
 245                                        msg="Accepted wrong VDU interface internal connection point reference") as e:
 246                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 247                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 248                 self.assertIn(norm("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}'"
 249                                    " must match an existing vdu:internal-connection-point"
 250                                    .format(test_vnfd["vdu"][1]["id"], test_vnfd["vdu"][1]["interface"][0]["name"],
 251                                            test_vnfd["vdu"][1]["interface"][0]["internal-connection-point-ref"])),
 252                               norm(str(e.exception)), "Wrong exception text")
 253             finally:
 254                 test_vnfd["vdu"][1]["interface"][0]["internal-connection-point-ref"] = tmp
 255         with self.subTest(i=11, t='Check Input Validation: vdu[vdu-configuration][juju]'):
 256             test_vnfd["vdu"][0]["vdu-configuration"] = {"juju": {"charm": "wrong-charm"}}
 257             try:
 258                 with self.assertRaises(EngineException, msg="Accepted non-existent charm in VDU configuration") as e:
 259                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 260                 self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 261                 self.assertIn(norm("{} defined in vnf[id={}]:vdu[id={}] but not present in package"
 262                                    .format("charm", test_vnfd["id"], test_vnfd["vdu"][0]["id"])),
 263                               norm(str(e.exception)), "Wrong exception text")
 264             finally:
 265                 del test_vnfd["vdu"][0]["vdu-configuration"]
 266         with self.subTest(i=12, t='Check Input Validation: Duplicated VLD name'):
 267             test_vnfd["internal-vld"].append(deepcopy(test_vnfd["internal-vld"][0]))
 268             test_vnfd["internal-vld"][1]["id"] = "wrong-internal-vld"
 269             try:
 270                 with self.assertRaises(EngineException, msg="Accepted duplicated VLD name") as e:
 271                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 272                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 273                 self.assertIn(norm("Duplicated VLD name '{}' in vnfd[id={}]:internal-vld[id={}]"
 274                                    .format(test_vnfd["internal-vld"][1]["name"], test_vnfd["id"],
 275                                            test_vnfd["internal-vld"][1]["id"])),
 276                               norm(str(e.exception)), "Wrong exception text")
 277             finally:
 278                 del test_vnfd["internal-vld"][1]
 279         with self.subTest(i=13, t='Check Input Validation: internal-vld[internal-connection-point][id-ref])'):
 280             tmp = test_vnfd["internal-vld"][0]["internal-connection-point"][0]["id-ref"]
 281             test_vnfd["internal-vld"][0]["internal-connection-point"][0]["id-ref"] = "wrong-icp-id-ref"
 282             try:
 283                 with self.assertRaises(EngineException, msg="Accepted non-existent internal VLD ICP id-ref") as e:
 284                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 285                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 286                 self.assertIn(norm("internal-vld[id='{}']:internal-connection-point='{}' must match an existing "
 287                                    "vdu:internal-connection-point"
 288                                    .format(test_vnfd["internal-vld"][0]["id"],
 289                                            test_vnfd["internal-vld"][0]["internal-connection-point"][0]["id-ref"])),
 290                               norm(str(e.exception)), "Wrong exception text")
 291             finally:
 292                 test_vnfd["internal-vld"][0]["internal-connection-point"][0]["id-ref"] = tmp
 293         with self.subTest(i=14, t='Check Input Validation: internal-vld[ip-profile-ref])'):
 294             test_vnfd["ip-profiles"] = [{"name": "fake-ip-profile-ref"}]
 295             test_vnfd["internal-vld"][0]["ip-profile-ref"] = "wrong-ip-profile-ref"
 296             try:
 297                 with self.assertRaises(EngineException, msg="Accepted non-existent IP Profile Ref") as e:
 298                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 299                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 300                 self.assertIn(norm("internal-vld[id='{}']:ip-profile-ref='{}' does not exist"
 301                                    .format(test_vnfd["internal-vld"][0]["id"],
 302                                            test_vnfd["internal-vld"][0]["ip-profile-ref"])),
 303                               norm(str(e.exception)), "Wrong exception text")
 304             finally:
 305                 del test_vnfd["ip-profiles"]
 306                 del test_vnfd["internal-vld"][0]["ip-profile-ref"]
 307         with self.subTest(i=15, t='Check Input Validation: vdu[monitoring-param])'):
 308             test_vnfd["monitoring-param"] = [{"id": "fake-mp-id", "vdu-monitoring-param": {
 309                 "vdu-monitoring-param-ref": "fake-vdu-mp-ref", "vdu-ref": "fake-vdu-ref"}}]
 310             test_vnfd["vdu"][0]["monitoring-param"] = [{"id": "wrong-vdu-mp-id"}]
 311             try:
 312                 with self.assertRaises(EngineException, msg="Accepted non-existent VDU Monitorimg Param") as e:
 313                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 314                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 315                 mp = test_vnfd["monitoring-param"][0]["vdu-monitoring-param"]
 316                 self.assertIn(norm("monitoring-param:vdu-monitoring-param:vdu-monitoring-param-ref='{}' not defined"
 317                                    " at vdu[id='{}'] or vdu does not exist"
 318                                    .format(mp["vdu-monitoring-param-ref"], mp["vdu-ref"])),
 319                               norm(str(e.exception)), "Wrong exception text")
 320             finally:
 321                 del test_vnfd["monitoring-param"]
 322                 del test_vnfd["vdu"][0]["monitoring-param"]
 323         with self.subTest(i=16, t='Check Input Validation: vdu[vdu-configuration][metrics]'):
 324             test_vnfd["monitoring-param"] = [{"id": "fake-mp-id", "vdu-metric": {
 325                 "vdu-metric-name-ref": "fake-vdu-mp-ref", "vdu-ref": "fake-vdu-ref"}}]
 326             test_vnfd["vdu"][0]["vdu-configuration"] = {"metrics": [{"name": "wrong-vdu-mp-id"}]}
 327             try:
 328                 with self.assertRaises(EngineException, msg="Accepted non-existent VDU Configuration Metric") as e:
 329                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 330                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 331                 mp = test_vnfd["monitoring-param"][0]["vdu-metric"]
 332                 self.assertIn(norm("monitoring-param:vdu-metric:vdu-metric-name-ref='{}' not defined"
 333                                    " at vdu[id='{}'] or vdu does not exist"
 334                                    .format(mp["vdu-metric-name-ref"], mp["vdu-ref"])),
 335                               norm(str(e.exception)), "Wrong exception text")
 336             finally:
 337                 del test_vnfd["monitoring-param"]
 338                 del test_vnfd["vdu"][0]["vdu-configuration"]
 339         with self.subTest(i=17, t='Check Input Validation: scaling-group-descriptor[scaling-policy][scaling-criteria]'):
 340             test_vnfd["monitoring-param"] = [{"id": "fake-mp-id"}]
 341             test_vnfd["scaling-group-descriptor"] = [{
 342                 "name": "fake-vnf-sg-name",
 343                 "vdu": [{"vdu-id-ref": "wrong-vdu-id-ref"}],
 344                 "scaling-policy": [{"name": "fake-vnf-sp-name", "scaling-criteria": [{
 345                     "name": "fake-vnf-sc-name", "vnf-monitoring-param-ref": "wrong-vnf-mp-id"}]}]}]
 346             with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group Policy Criteria") as e:
 347                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 348             self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 349             sg = test_vnfd["scaling-group-descriptor"][0]
 350             sc = sg["scaling-policy"][0]["scaling-criteria"][0]
 351             self.assertIn(norm("scaling-group-descriptor[name='{}']:scaling-criteria[name='{}']:"
 352                                "vnf-monitoring-param-ref='{}' not defined in any monitoring-param"
 353                                .format(sg["name"], sc["name"], sc["vnf-monitoring-param-ref"])),
 354                           norm(str(e.exception)), "Wrong exception text")
 355         with self.subTest(i=18, t='Check Input Validation: scaling-group-descriptor[vdu][vdu-id-ref]'):
 356             sc["vnf-monitoring-param-ref"] = "fake-mp-id"
 357             with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference") as e:
 358                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 359             self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 360             self.assertIn(norm("scaling-group-descriptor[name='{}']:vdu-id-ref={} does not match any vdu"
 361                                .format(sg["name"], sg["vdu"][0]["vdu-id-ref"])),
 362                           norm(str(e.exception)), "Wrong exception text")
 363         with self.subTest(i=19, t='Check Input Validation: scaling-group-descriptor[scaling-config-action]'):
 364             tmp = test_vnfd["vnf-configuration"]
 365             del test_vnfd["vnf-configuration"]
 366             sg["vdu"][0]["vdu-id-ref"] = test_vnfd["vdu"][0]["id"]
 367             sg["scaling-config-action"] = [{"trigger": "pre-scale-in"}]
 368             try:
 369                 with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference")\
 370                         as e:
 371                     self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 372                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 373                 self.assertIn(norm("'vnf-configuration' not defined in the descriptor but it is referenced"
 374                                    " by scaling-group-descriptor[name='{}']:scaling-config-action"
 375                                    .format(sg["name"])),
 376                               norm(str(e.exception)), "Wrong exception text")
 377             finally:
 378                 test_vnfd["vnf-configuration"] = tmp
 379         with self.subTest(i=20, t='Check Input Validation: scaling-group-descriptor[scaling-config-action]'
 380                                   '[vnf-config-primitive-name-ref]'):
 381             sg["scaling-config-action"][0]["vnf-config-primitive-name-ref"] = "wrong-sca-prim-name"
 382             with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference") as e:
 383                 self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 384             self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 385             self.assertIn(norm("scaling-group-descriptor[name='{}']:scaling-config-action:"
 386                                "vnf-config-primitive-name-ref='{}' does not match"
 387                                " any vnf-configuration:config-primitive:name"
 388                                .format(sg["name"], sg["scaling-config-action"][0]["vnf-config-primitive-name-ref"])),
 389                           norm(str(e.exception)), "Wrong exception text")
 390             # del test_vnfd["monitoring-param"]
 391             # del test_vnfd["scaling-group-descriptor"]
 392         with self.subTest(i=21, t='Check Input Validation: everything right'):
 393             sg["scaling-config-action"][0]["vnf-config-primitive-name-ref"] = "touch"
 394             test_vnfd["id"] = "fake-vnfd-id"
 395             self.db.get_one.side_effect = [{"_id": did, "_admin": db_vnfd_content["_admin"]}, None]
 396             rc = self.topic.upload_content(fake_session, did, test_vnfd, {}, {"Content-Type": []})
 397             self.assertTrue(rc, "Input Validation: Unexpected failure")
 398         return
 399
 400     def test_edit_vnfd(self):
 401         did = db_vnfd_content["_id"]
 402         self.fs.file_exists.return_value = True
 403         self.fs.dir_ls.return_value = True
 404         with self.subTest(i=1, t='Normal Edition'):
 405             now = time()
 406             self.db.get_one.side_effect = [db_vnfd_content, None]
 407             data = {"id": "new-vnfd-id", "name": "new-vnfd-name"}
 408             self.topic.edit(fake_session, did, data)
 409             db_args = self.db.replace.call_args[0]
 410             msg_args = self.msg.write.call_args[0]
 411             data["_id"] = did
 412             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 413             self.assertEqual(msg_args[1], "edited", "Wrong message action")
 414             self.assertEqual(msg_args[2], data, "Wrong message content")
 415             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 416             self.assertEqual(db_args[1], did, "Wrong DB ID")
 417             self.assertEqual(db_args[2]["_admin"]["created"], db_vnfd_content["_admin"]["created"],
 418                              "Wrong creation time")
 419             self.assertGreater(db_args[2]["_admin"]["modified"], now,
 420                                "Wrong modification time")
 421             self.assertEqual(db_args[2]["_admin"]["projects_read"], db_vnfd_content["_admin"]["projects_read"],
 422                              "Wrong read-only project list")
 423             self.assertEqual(db_args[2]["_admin"]["projects_write"], db_vnfd_content["_admin"]["projects_write"],
 424                              "Wrong read-write project list")
 425             self.assertEqual(db_args[2]["id"], data["id"], "Wrong VNFD ID")
 426             self.assertEqual(db_args[2]["name"], data["name"], "Wrong VNFD Name")
 427         with self.subTest(i=2, t='Conflict on Edit'):
 428             data = {"id": "fake-vnfd-id", "name": "new-vnfd-name"}
 429             self.db.get_one.side_effect = [db_vnfd_content, {"_id": str(uuid4()), "id": data["id"]}]
 430             with self.assertRaises(EngineException, msg="Accepted existing VNFD ID") as e:
 431                 self.topic.edit(fake_session, did, data)
 432             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 433             self.assertIn(norm("{} with id '{}' already exists for this project".format("vnfd", data["id"])),
 434                           norm(str(e.exception)), "Wrong exception text")
 435         with self.subTest(i=3, t='Check Envelope'):
 436             data = {"vnfd": {"id": "new-vnfd-id-1", "name": "new-vnfd-name"}}
 437             with self.assertRaises(EngineException, msg="Accepted VNFD with wrong envelope") as e:
 438                 self.topic.edit(fake_session, did, data)
 439             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 440             self.assertIn("'vnfd' must be a list of only one element", norm(str(e.exception)), "Wrong exception text")
 441         return
 442
 443     def test_delete_vnfd(self):
 444         did = db_vnfd_content["_id"]
 445         self.db.get_one.return_value = db_vnfd_content
 446         p_id = db_vnfd_content["_admin"]["projects_read"][0]
 447         with self.subTest(i=1, t='Normal Deletion'):
 448             self.db.get_list.return_value = []
 449             self.db.del_one.return_value = {"deleted": 1}
 450             self.topic.delete(fake_session, did)
 451             db_args = self.db.del_one.call_args[0]
 452             msg_args = self.msg.write.call_args[0]
 453             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 454             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 455             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 456             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 457             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
 458             self.assertEqual(db_args[1]["_admin.projects_write.cont"], [p_id, 'ANY'], "Wrong DB filter")
 459             db_g1_args = self.db.get_one.call_args[0]
 460             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 461             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 462             db_gl_calls = self.db.get_list.call_args_list
 463             self.assertEqual(db_gl_calls[0][0][0], "vnfrs", "Wrong DB topic")
 464             # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID")   # Filter changed after call
 465             self.assertEqual(db_gl_calls[1][0][0], "nsds", "Wrong DB topic")
 466             self.assertEqual(db_gl_calls[1][0][1]["constituent-vnfd.ANYINDEX.vnfd-id-ref"], db_vnfd_content["id"],
 467                              "Wrong DB NSD constituent-vnfd id-ref")
 468
 469             self.db.set_one.assert_not_called()
 470             fs_del_calls = self.fs.file_delete.call_args_list
 471             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
 472             self.assertEqual(fs_del_calls[1][0][0], did+'_', "Wrong FS folder id")
 473         with self.subTest(i=2, t='Conflict on Delete - VNFD in use by VNFR'):
 474             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-vnfr"}]
 475             with self.assertRaises(EngineException, msg="Accepted VNFD in use by VNFR") as e:
 476                 self.topic.delete(fake_session, did)
 477             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 478             self.assertIn("there is at least one vnf using this descriptor", norm(str(e.exception)),
 479                           "Wrong exception text")
 480         with self.subTest(i=3, t='Conflict on Delete - VNFD in use by NSD'):
 481             self.db.get_list.side_effect = [[], [{"_id": str(uuid4()), "name": "fake-nsd"}]]
 482             with self.assertRaises(EngineException, msg="Accepted VNFD in use by NSD") as e:
 483                 self.topic.delete(fake_session, did)
 484             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 485             self.assertIn("there is at least one nsd referencing this descriptor", norm(str(e.exception)),
 486                           "Wrong exception text")
 487         with self.subTest(i=4, t='Non-existent VNFD'):
 488             excp_msg = "Not found any {} with filter='{}'".format("VNFD", {"_id": did})
 489             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
 490             with self.assertRaises(DbException, msg="Accepted non-existent VNFD ID") as e:
 491                 self.topic.delete(fake_session, did)
 492             self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
 493             self.assertIn(norm(excp_msg), norm(str(e.exception)), "Wrong exception text")
 494         with self.subTest(i=5, t='No delete because referenced by other project'):
 495             db_vnfd_content["_admin"]["projects_read"].append("other_project")
 496             self.db.get_one = Mock(return_value=db_vnfd_content)
 497             self.db.get_list = Mock(return_value=[])
 498             self.msg.write.reset_mock()
 499             self.db.del_one.reset_mock()
 500             self.fs.file_delete.reset_mock()
 501
 502             self.topic.delete(fake_session, did)
 503             self.db.del_one.assert_not_called()
 504             self.msg.write.assert_not_called()
 505             db_g1_args = self.db.get_one.call_args[0]
 506             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 507             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 508             db_s1_args = self.db.set_one.call_args
 509             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
 510             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
 511             self.assertIn(p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter")
 512             self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary")
 513             self.assertIn("_admin.projects_read." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary")
 514             self.assertIn("_admin.projects_write." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary")
 515             self.fs.file_delete.assert_not_called()
 516         return
 517
 518
 519 class Test_NsdTopic(TestCase):
 520
 521     @classmethod
 522     def setUpClass(cls):
 523         cls.test_name = "test-nsd-topic"
 524
 525     @classmethod
 526     def tearDownClass(cls):
 527         pass
 528
 529     def setUp(self):
 530         self.db = Mock(dbbase.DbBase())
 531         self.fs = Mock(fsbase.FsBase())
 532         self.msg = Mock(msgbase.MsgBase())
 533         self.auth = Mock(authconn.Authconn(None, None, None))
 534         self.topic = NsdTopic(self.db, self.fs, self.msg, self.auth)
 535         self.topic.check_quota = Mock(return_value=None)  # skip quota
 536
 537     def test_new_nsd(self):
 538         did = db_nsd_content["_id"]
 539         self.fs.get_params.return_value = {}
 540         self.fs.file_exists.return_value = False
 541         self.fs.file_open.side_effect = lambda path, mode: open("/tmp/" + str(uuid4()), "a+b")
 542         test_nsd = deepcopy(db_nsd_content)
 543         del test_nsd["_id"]
 544         del test_nsd["_admin"]
 545         with self.subTest(i=1, t='Normal Creation'):
 546             self.db.create.return_value = did
 547             rollback = []
 548             did2, oid = self.topic.new(rollback, fake_session, {})
 549             db_args = self.db.create.call_args[0]
 550             msg_args = self.msg.write.call_args[0]
 551             self.assertEqual(len(rollback), 1, "Wrong rollback length")
 552             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 553             self.assertEqual(msg_args[1], "created", "Wrong message action")
 554             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 555             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 556             self.assertEqual(did2, did, "Wrong DB NSD id")
 557             self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
 558             self.assertEqual(db_args[1]["_admin"]["modified"], db_args[1]["_admin"]["created"],
 559                              "Wrong modification time")
 560             self.assertEqual(db_args[1]["_admin"]["projects_read"], [test_pid], "Wrong read-only project list")
 561             self.assertEqual(db_args[1]["_admin"]["projects_write"], [test_pid], "Wrong read-write project list")
 562             try:
 563                 self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 564                 self.db.get_list.return_value = [db_vnfd_content]
 565                 self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 566                 msg_args = self.msg.write.call_args[0]
 567                 test_nsd["_id"] = did
 568                 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 569                 self.assertEqual(msg_args[1], "edited", "Wrong message action")
 570                 self.assertEqual(msg_args[2], test_nsd, "Wrong message content")
 571                 db_args = self.db.get_one.mock_calls[0][1]
 572                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 573                 self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id")
 574                 db_args = self.db.replace.call_args[0]
 575                 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 576                 self.assertEqual(db_args[1], did, "Wrong DB NSD id")
 577                 admin = db_args[2]["_admin"]
 578                 db_admin = db_nsd_content["_admin"]
 579                 self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time")
 580                 self.assertGreater(admin["modified"], db_admin["created"], "Wrong modification time")
 581                 self.assertEqual(admin["projects_read"], db_admin["projects_read"], "Wrong read-only project list")
 582                 self.assertEqual(admin["projects_write"], db_admin["projects_write"], "Wrong read-write project list")
 583                 self.assertEqual(admin["onboardingState"], "ONBOARDED", "Wrong onboarding state")
 584                 self.assertEqual(admin["operationalState"], "ENABLED", "Wrong operational state")
 585                 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
 586                 storage = admin["storage"]
 587                 self.assertEqual(storage["folder"], did, "Wrong storage folder")
 588                 self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor")
 589                 compare_desc(self, test_nsd, db_args[2], "NSD")
 590             finally:
 591                 pass
 592         self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None:\
 593             {"_id": did, "_admin": db_nsd_content["_admin"]}
 594         with self.subTest(i=2, t='Check Pyangbind Validation: required properties'):
 595             tmp = test_nsd["id"]
 596             del test_nsd["id"]
 597             try:
 598                 with self.assertRaises(EngineException, msg="Accepted NSD with a missing required property") as e:
 599                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 600                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 601                 self.assertIn(norm("Error in pyangbind validation: '{}'".format("id")),
 602                               norm(str(e.exception)), "Wrong exception text")
 603             finally:
 604                 test_nsd["id"] = tmp
 605         with self.subTest(i=3, t='Check Pyangbind Validation: additional properties'):
 606             test_nsd["extra-property"] = 0
 607             try:
 608                 with self.assertRaises(EngineException, msg="Accepted NSD with an additional property") as e:
 609                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 610                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 611                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 612                                    .format("json object contained a key that did not exist", "extra-property")),
 613                               norm(str(e.exception)), "Wrong exception text")
 614             finally:
 615                 del test_nsd["extra-property"]
 616         with self.subTest(i=4, t='Check Pyangbind Validation: property types'):
 617             tmp = test_nsd["short-name"]
 618             test_nsd["short-name"] = {"key": 0}
 619             try:
 620                 with self.assertRaises(EngineException, msg="Accepted NSD with a wrongly typed property") as e:
 621                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 622                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 623                 self.assertIn(norm("Error in pyangbind validation: {} ({})"
 624                                    .format("json object contained a key that did not exist", "key")),
 625                               norm(str(e.exception)), "Wrong exception text")
 626             finally:
 627                 test_nsd["short-name"] = tmp
 628         with self.subTest(i=5, t='Check Input Validation: vld[mgmt-network+ip-profile]'):
 629             tmp = test_nsd["vld"][0]["vim-network-name"]
 630             del test_nsd["vld"][0]["vim-network-name"]
 631             test_nsd["vld"][0]["ip-profile-ref"] = "fake-ip-profile"
 632             try:
 633                 with self.assertRaises(EngineException, msg="Accepted VLD with mgmt-network+ip-profile") as e:
 634                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 635                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 636                 self.assertIn(norm("Error at vld[id='{}']:ip-profile-ref"
 637                                    " You cannot set an ip-profile when mgmt-network is True"
 638                                    .format(test_nsd["vld"][0]["id"])),
 639                               norm(str(e.exception)), "Wrong exception text")
 640             finally:
 641                 test_nsd["vld"][0]["vim-network-name"] = tmp
 642                 del test_nsd["vld"][0]["ip-profile-ref"]
 643         with self.subTest(i=6, t='Check Input Validation: vld[vnfd-connection-point-ref][vnfd-id-ref]'):
 644             tmp = test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"]
 645             test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"] = "wrong-vnfd-id-ref"
 646             try:
 647                 with self.assertRaises(EngineException, msg="Accepted VLD with wrong vnfd-connection-point-ref") as e:
 648                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 649                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 650                 self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[vnfd-id-ref='{}']"
 651                                    " does not match constituent-vnfd[member-vnf-index='{}']:vnfd-id-ref '{}'"
 652                                    .format(test_nsd["vld"][0]["id"],
 653                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"],
 654                                            test_nsd["constituent-vnfd"][0]["member-vnf-index"],
 655                                            test_nsd["constituent-vnfd"][0]["vnfd-id-ref"])),
 656                               norm(str(e.exception)), "Wrong exception text")
 657             finally:
 658                 test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-id-ref"] = tmp
 659         with self.subTest(i=7, t='Check Input Validation: vld[vnfd-connection-point-ref][member-vnf-index-ref]'):
 660             tmp = test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"]
 661             test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"] = "wrong-member-vnf-index-ref"
 662             try:
 663                 with self.assertRaises(EngineException, msg="Accepted VLD with wrong vnfd-connection-point-ref") as e:
 664                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 665                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 666                 self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}']"
 667                                    " does not match any constituent-vnfd:member-vnf-index"
 668                                    .format(test_nsd["vld"][0]["id"],
 669                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"])),
 670                               norm(str(e.exception)), "Wrong exception text")
 671             finally:
 672                 test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"] = tmp
 673         with self.subTest(i=8, t='Check Input Validation: vnffgd[classifier][rsp-id-ref]'):
 674             test_nsd["vnffgd"] = [{"id": "fake-vnffgd-id",
 675                                    "rsp": [{"id": "fake-rsp-id"}],
 676                                    "classifier": [{"id": "fake-vnffgd-classifier-id", "rsp-id-ref": "wrong-rsp-id"}]}]
 677             try:
 678                 with self.assertRaises(EngineException, msg="Accepted VNF FGD with wrong classifier rsp-id-ref") as e:
 679                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 680                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 681                 self.assertIn(norm("Error at vnffgd[id='{}']:classifier[id='{}']:rsp-id-ref '{}'"
 682                                    " does not match any rsp:id"
 683                                    .format(test_nsd["vnffgd"][0]["id"],
 684                                            test_nsd["vnffgd"][0]["classifier"][0]["id"],
 685                                            test_nsd["vnffgd"][0]["classifier"][0]["rsp-id-ref"])),
 686                               norm(str(e.exception)), "Wrong exception text")
 687             finally:
 688                 test_nsd["vnffgd"][0]["classifier"][0]["rsp-id-ref"] = "fake-rsp-id"
 689         with self.subTest(i=9, t='Check Descriptor Dependencies: constituent-vnfd[vnfd-id-ref]'):
 690             self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 691             self.db.get_list.return_value = []
 692             try:
 693                 with self.assertRaises(EngineException, msg="Accepted wrong constituent VNFD ID reference") as e:
 694                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 695                 self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 696                 self.assertIn(norm("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}'"
 697                                    " references a non existing vnfd"
 698                                    .format(test_nsd["constituent-vnfd"][0]["vnfd-id-ref"])),
 699                               norm(str(e.exception)), "Wrong exception text")
 700             finally:
 701                 pass
 702         with self.subTest(i=10, t='Check Descriptor Dependencies: '
 703                                   'vld[vnfd-connection-point-ref][vnfd-connection-point-ref]'):
 704             tmp = test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-connection-point-ref"]
 705             test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-connection-point-ref"] = "wrong-vnfd-cp-ref"
 706             self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 707             self.db.get_list.return_value = [db_vnfd_content]
 708             try:
 709                 with self.assertRaises(EngineException, msg="Accepted wrong VLD CP reference") as e:
 710                     self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 711                 self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
 712                 self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}']:"
 713                                    "vnfd-connection-point-ref='{}' references a non existing conection-point:name"
 714                                    " inside vnfd '{}'"
 715                                    .format(test_nsd["vld"][0]["id"],
 716                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"],
 717                                            test_nsd["vld"][0]["vnfd-connection-point-ref"][0]
 718                                                    ["vnfd-connection-point-ref"],
 719                                            db_vnfd_content["id"])),
 720                               norm(str(e.exception)), "Wrong exception text")
 721             finally:
 722                 test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-connection-point-ref"] = tmp
 723         return
 724         with self.subTest(i=11, t='Check Input Validation: everything right'):
 725             test_nsd["id"] = "fake-nsd-id"
 726             self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
 727             self.db.get_list.return_value = [db_vnfd_content]
 728             rc = self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
 729             self.assertTrue(rc, "Input Validation: Unexpected failure")
 730         return
 731
 732     def test_edit_nsd(self):
 733         did = db_nsd_content["_id"]
 734         self.fs.file_exists.return_value = True
 735         self.fs.dir_ls.return_value = True
 736         with self.subTest(i=1, t='Normal Edition'):
 737             now = time()
 738             self.db.get_one.side_effect = [db_nsd_content, None]
 739             self.db.get_list.return_value = [db_vnfd_content]
 740             data = {"id": "new-nsd-id", "name": "new-nsd-name"}
 741             self.topic.edit(fake_session, did, data)
 742             db_args = self.db.replace.call_args[0]
 743             msg_args = self.msg.write.call_args[0]
 744             data["_id"] = did
 745             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 746             self.assertEqual(msg_args[1], "edited", "Wrong message action")
 747             self.assertEqual(msg_args[2], data, "Wrong message content")
 748             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 749             self.assertEqual(db_args[1], did, "Wrong DB ID")
 750             self.assertEqual(db_args[2]["_admin"]["created"], db_nsd_content["_admin"]["created"],
 751                              "Wrong creation time")
 752             self.assertGreater(db_args[2]["_admin"]["modified"], now, "Wrong modification time")
 753             self.assertEqual(db_args[2]["_admin"]["projects_read"], db_nsd_content["_admin"]["projects_read"],
 754                              "Wrong read-only project list")
 755             self.assertEqual(db_args[2]["_admin"]["projects_write"], db_nsd_content["_admin"]["projects_write"],
 756                              "Wrong read-write project list")
 757             self.assertEqual(db_args[2]["id"], data["id"], "Wrong NSD ID")
 758             self.assertEqual(db_args[2]["name"], data["name"], "Wrong NSD Name")
 759         with self.subTest(i=2, t='Conflict on Edit'):
 760             data = {"id": "fake-nsd-id", "name": "new-nsd-name"}
 761             self.db.get_one.side_effect = [db_nsd_content, {"_id": str(uuid4()), "id": data["id"]}]
 762             with self.assertRaises(EngineException, msg="Accepted existing NSD ID") as e:
 763                 self.topic.edit(fake_session, did, data)
 764             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 765             self.assertIn(norm("{} with id '{}' already exists for this project".format("nsd", data["id"])),
 766                           norm(str(e.exception)), "Wrong exception text")
 767         with self.subTest(i=3, t='Check Envelope'):
 768             data = {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}
 769             with self.assertRaises(EngineException, msg="Accepted NSD with wrong envelope") as e:
 770                 self.topic.edit(fake_session, did, data)
 771             self.assertEqual(e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code")
 772             self.assertIn("'nsd' must be a list of only one element", norm(str(e.exception)), "Wrong exception text")
 773         return
 774
 775     def test_delete_nsd(self):
 776         did = db_nsd_content["_id"]
 777         self.db.get_one.return_value = db_nsd_content
 778         p_id = db_nsd_content["_admin"]["projects_read"][0]
 779         with self.subTest(i=1, t='Normal Deletion'):
 780             self.db.get_list.return_value = []
 781             self.db.del_one.return_value = {"deleted": 1}
 782             self.topic.delete(fake_session, did)
 783             db_args = self.db.del_one.call_args[0]
 784             msg_args = self.msg.write.call_args[0]
 785             self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
 786             self.assertEqual(msg_args[1], "deleted", "Wrong message action")
 787             self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
 788             self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
 789             self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
 790             self.assertEqual(db_args[1]["_admin.projects_write.cont"], [p_id, 'ANY'], "Wrong DB filter")
 791             db_g1_args = self.db.get_one.call_args[0]
 792             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 793             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB NSD ID")
 794             db_gl_calls = self.db.get_list.call_args_list
 795             self.assertEqual(db_gl_calls[0][0][0], "nsrs", "Wrong DB topic")
 796             # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID")   # Filter changed after call
 797             self.assertEqual(db_gl_calls[1][0][0], "nsts", "Wrong DB topic")
 798             self.assertEqual(db_gl_calls[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"], db_nsd_content["id"],
 799                              "Wrong DB NSD netslice-subnet nsd-ref")
 800             self.db.set_one.assert_not_called()
 801             fs_del_calls = self.fs.file_delete.call_args_list
 802             self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
 803             self.assertEqual(fs_del_calls[1][0][0], did+'_', "Wrong FS folder id")
 804         with self.subTest(i=2, t='Conflict on Delete - NSD in use by nsr'):
 805             self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-nsr"}]
 806             with self.assertRaises(EngineException, msg="Accepted NSD in use by NSR") as e:
 807                 self.topic.delete(fake_session, did)
 808             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 809             self.assertIn("there is at least one ns using this descriptor", norm(str(e.exception)),
 810                           "Wrong exception text")
 811         with self.subTest(i=3, t='Conflict on Delete - NSD in use by NST'):
 812             self.db.get_list.side_effect = [[], [{"_id": str(uuid4()), "name": "fake-nst"}]]
 813             with self.assertRaises(EngineException, msg="Accepted NSD in use by NST") as e:
 814                 self.topic.delete(fake_session, did)
 815             self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
 816             self.assertIn("there is at least one netslice template referencing this descriptor", norm(str(e.exception)),
 817                           "Wrong exception text")
 818         with self.subTest(i=4, t='Non-existent NSD'):
 819             excp_msg = "Not found any {} with filter='{}'".format("NSD", {"_id": did})
 820             self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
 821             with self.assertRaises(DbException, msg="Accepted non-existent NSD ID") as e:
 822                 self.topic.delete(fake_session, did)
 823             self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code")
 824             self.assertIn(norm(excp_msg), norm(str(e.exception)), "Wrong exception text")
 825         with self.subTest(i=5, t='No delete because referenced by other project'):
 826             db_nsd_content["_admin"]["projects_read"].append("other_project")
 827             self.db.get_one = Mock(return_value=db_nsd_content)
 828             self.db.get_list = Mock(return_value=[])
 829             self.msg.write.reset_mock()
 830             self.db.del_one.reset_mock()
 831             self.fs.file_delete.reset_mock()
 832
 833             self.topic.delete(fake_session, did)
 834             self.db.del_one.assert_not_called()
 835             self.msg.write.assert_not_called()
 836             db_g1_args = self.db.get_one.call_args[0]
 837             self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
 838             self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
 839             db_s1_args = self.db.set_one.call_args
 840             self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
 841             self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
 842             self.assertIn(p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter")
 843             self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary")
 844             self.assertIn("_admin.projects_read." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary")
 845             self.assertIn("_admin.projects_write." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary")
 846             self.fs.file_delete.assert_not_called()
 847         return
 848
 849
 850 if __name__ == '__main__':
 851     unittest.main()