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