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