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