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