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