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
  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, 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 if __name__ == '__main__':
 502     unittest.main()