Coverage for osm_nbi/tests/test_descriptor_topics.py: 99%
1104 statements
« prev ^ index » next coverage.py v7.3.1, created at 2024-06-25 09:03 +0000
1#! /usr/bin/python3
2# -*- coding: utf-8 -*-
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.
17__author__ = "Pedro de la Cruz Ramos, pedro.delacruzramos@altran.com"
18__date__ = "2019-11-20"
20from contextlib import contextmanager
21import unittest
22from unittest import TestCase
23from unittest.mock import Mock, patch
24from uuid import uuid4
25from http import HTTPStatus
26from copy import deepcopy
27from time import time
28from osm_common import dbbase, fsbase, msgbase
29from osm_nbi import authconn
30from osm_nbi.tests.test_pkg_descriptors import (
31 db_vnfds_text,
32 db_nsds_text,
33 vnfd_exploit_text,
34 vnfd_exploit_fixed_text,
35 db_sfc_nsds_text,
36)
37from osm_nbi.descriptor_topics import VnfdTopic, NsdTopic
38from osm_nbi.engine import EngineException
39from osm_common.dbbase import DbException
40import yaml
41import tempfile
42import collections
43import collections.abc
45collections.MutableSequence = collections.abc.MutableSequence
47test_name = "test-user"
48db_vnfd_content = yaml.safe_load(db_vnfds_text)[0]
49db_nsd_content = yaml.safe_load(db_nsds_text)[0]
50test_pid = db_vnfd_content["_admin"]["projects_read"][0]
51fake_session = {
52 "username": test_name,
53 "project_id": (test_pid,),
54 "method": None,
55 "admin": True,
56 "force": False,
57 "public": False,
58 "allow_show_user_project_role": True,
59}
60UUID = "00000000-0000-0000-0000-000000000000"
63def admin_value():
64 return {"projects_read": []}
67def setup_mock_fs(fs):
68 fs.path = ""
69 fs.get_params.return_value = {}
70 fs.file_exists.return_value = False
71 fs.file_open.side_effect = lambda path, mode: tempfile.TemporaryFile(mode="a+b")
74def norm(s: str):
75 """Normalize string for checking"""
76 return " ".join(s.strip().split()).lower()
79def compare_desc(tc, d1, d2, k):
80 """
81 Compare two descriptors
82 We need this function because some methods are adding/removing items to/from the descriptors
83 before they are stored in the database, so the original and stored versions will differ
84 What we check is that COMMON LEAF ITEMS are equal
85 Lists of different length are not compared
86 :param tc: Test Case wich provides context (in particular the assert* methods)
87 :param d1,d2: Descriptors to be compared
88 :param k: key/item being compared
89 :return: Nothing
90 """
91 if isinstance(d1, dict) and isinstance(d2, dict):
92 for key in d1.keys():
93 if key in d2:
94 compare_desc(tc, d1[key], d2[key], k + "[{}]".format(key))
95 elif isinstance(d1, list) and isinstance(d2, list) and len(d1) == len(d2):
96 for i in range(len(d1)):
97 compare_desc(tc, d1[i], d2[i], k + "[{}]".format(i))
98 else:
99 tc.assertEqual(d1, d2, "Wrong descriptor content: {}".format(k))
102class Test_VnfdTopic(TestCase):
103 @classmethod
104 def setUpClass(cls):
105 cls.test_name = "test-vnfd-topic"
107 @classmethod
108 def tearDownClass(cls):
109 pass
111 def setUp(self):
112 self.db = Mock(dbbase.DbBase())
113 self.fs = Mock(fsbase.FsBase())
114 self.msg = Mock(msgbase.MsgBase())
115 self.auth = Mock(authconn.Authconn(None, None, None))
116 self.topic = VnfdTopic(self.db, self.fs, self.msg, self.auth)
117 self.topic.check_quota = Mock(return_value=None) # skip quota
119 @contextmanager
120 def assertNotRaises(self, exception_type=Exception):
121 try:
122 yield None
123 except exception_type:
124 raise self.failureException("{} raised".format(exception_type.__name__))
126 def create_desc_temp(self, template):
127 old_desc = deepcopy(template)
128 new_desc = deepcopy(template)
129 return old_desc, new_desc
131 def prepare_vnfd_creation(self):
132 setup_mock_fs(self.fs)
133 test_vnfd = deepcopy(db_vnfd_content)
134 did = db_vnfd_content["_id"]
135 self.db.create.return_value = did
136 self.db.get_one.side_effect = [
137 {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])},
138 None,
139 ]
140 return did, test_vnfd
142 def prepare_vnfd(self, vnfd_text):
143 setup_mock_fs(self.fs)
144 test_vnfd = yaml.safe_load(vnfd_text)
145 self.db.create.return_value = UUID
146 self.db.get_one.side_effect = [
147 {"_id": UUID, "_admin": admin_value()},
148 None,
149 ]
150 return UUID, test_vnfd
152 def prepare_test_vnfd(self, test_vnfd):
153 del test_vnfd["_id"]
154 del test_vnfd["_admin"]
155 del test_vnfd["vdu"][0]["cloud-init-file"]
156 del test_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
157 "day1-2"
158 ][0]["execution-environment-list"][0]["juju"]
159 return test_vnfd
161 @patch("osm_nbi.descriptor_topics.shutil")
162 @patch("osm_nbi.descriptor_topics.os.rename")
163 def test_new_vnfd_normal_creation(self, mock_rename, mock_shutil):
164 did, test_vnfd = self.prepare_vnfd_creation()
165 test_vnfd = self.prepare_test_vnfd(test_vnfd)
166 rollback = []
167 did2, oid = self.topic.new(rollback, fake_session, {})
168 db_args = self.db.create.call_args[0]
169 msg_args = self.msg.write.call_args[0]
171 self.assertEqual(len(rollback), 1, "Wrong rollback length")
172 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
173 self.assertEqual(msg_args[1], "created", "Wrong message action")
174 self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
175 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
176 self.assertEqual(did2, did, "Wrong DB VNFD id")
177 self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
178 self.assertEqual(
179 db_args[1]["_admin"]["modified"],
180 db_args[1]["_admin"]["created"],
181 "Wrong modification time",
182 )
183 self.assertEqual(
184 db_args[1]["_admin"]["projects_read"],
185 [test_pid],
186 "Wrong read-only project list",
187 )
188 self.assertEqual(
189 db_args[1]["_admin"]["projects_write"],
190 [test_pid],
191 "Wrong read-write project list",
192 )
194 self.db.get_one.side_effect = [
195 {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])},
196 None,
197 ]
199 self.topic.upload_content(
200 fake_session, did, test_vnfd, {}, {"Content-Type": []}
201 )
202 msg_args = self.msg.write.call_args[0]
203 test_vnfd["_id"] = did
204 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
205 self.assertEqual(msg_args[1], "edited", "Wrong message action")
206 self.assertEqual(msg_args[2], test_vnfd, "Wrong message content")
208 db_args = self.db.get_one.mock_calls[0][1]
209 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
210 self.assertEqual(db_args[1]["_id"], did, "Wrong DB VNFD id")
212 db_args = self.db.replace.call_args[0]
213 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
214 self.assertEqual(db_args[1], did, "Wrong DB VNFD id")
216 admin = db_args[2]["_admin"]
217 db_admin = deepcopy(db_vnfd_content["_admin"])
218 self.assertEqual(admin["type"], "vnfd", "Wrong descriptor type")
219 self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time")
220 self.assertGreater(
221 admin["modified"], db_admin["created"], "Wrong modification time"
222 )
223 self.assertEqual(
224 admin["projects_read"],
225 db_admin["projects_read"],
226 "Wrong read-only project list",
227 )
228 self.assertEqual(
229 admin["projects_write"],
230 db_admin["projects_write"],
231 "Wrong read-write project list",
232 )
233 self.assertEqual(
234 admin["onboardingState"], "ONBOARDED", "Wrong onboarding state"
235 )
236 self.assertEqual(
237 admin["operationalState"], "ENABLED", "Wrong operational state"
238 )
239 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
241 storage = admin["storage"]
242 self.assertEqual(storage["folder"], did + ":1", "Wrong storage folder")
243 self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor")
244 self.assertEqual(admin["revision"], 1, "Wrong revision number")
245 compare_desc(self, test_vnfd, db_args[2], "VNFD")
247 @patch("osm_nbi.descriptor_topics.shutil")
248 @patch("osm_nbi.descriptor_topics.os.rename")
249 def test_new_vnfd_exploit(self, mock_rename, mock_shutil):
250 id, test_vnfd = self.prepare_vnfd(vnfd_exploit_text)
252 with self.assertRaises(EngineException):
253 self.topic.upload_content(
254 fake_session, id, test_vnfd, {}, {"Content-Type": []}
255 )
257 @patch("osm_nbi.descriptor_topics.shutil")
258 @patch("osm_nbi.descriptor_topics.os.rename")
259 def test_new_vnfd_valid_helm_chart(self, mock_rename, mock_shutil):
260 id, test_vnfd = self.prepare_vnfd(vnfd_exploit_fixed_text)
262 with self.assertNotRaises():
263 self.topic.upload_content(
264 fake_session, id, test_vnfd, {}, {"Content-Type": []}
265 )
267 @patch("osm_nbi.descriptor_topics.shutil")
268 @patch("osm_nbi.descriptor_topics.os.rename")
269 def test_new_vnfd_check_pyangbind_validation_additional_properties(
270 self, mock_rename, mock_shutil
271 ):
272 did, test_vnfd = self.prepare_vnfd_creation()
273 test_vnfd = self.prepare_test_vnfd(test_vnfd)
274 self.topic.upload_content(
275 fake_session, did, test_vnfd, {}, {"Content-Type": []}
276 )
277 test_vnfd["_id"] = did
278 test_vnfd["extra-property"] = 0
279 self.db.get_one.side_effect = (
280 lambda table, filter, fail_on_empty=None, fail_on_more=None: {
281 "_id": did,
282 "_admin": deepcopy(db_vnfd_content["_admin"]),
283 }
284 )
286 with self.assertRaises(
287 EngineException, msg="Accepted VNFD with an additional property"
288 ) as e:
289 self.topic.upload_content(
290 fake_session, did, test_vnfd, {}, {"Content-Type": []}
291 )
292 self.assertEqual(
293 e.exception.http_code,
294 HTTPStatus.UNPROCESSABLE_ENTITY,
295 "Wrong HTTP status code",
296 )
297 self.assertIn(
298 norm(
299 "Error in pyangbind validation: {} ({})".format(
300 "json object contained a key that did not exist", "extra-property"
301 )
302 ),
303 norm(str(e.exception)),
304 "Wrong exception text",
305 )
306 db_args = self.db.replace.call_args[0]
307 admin = db_args[2]["_admin"]
308 self.assertEqual(admin["revision"], 1, "Wrong revision number")
310 @patch("osm_nbi.descriptor_topics.shutil")
311 @patch("osm_nbi.descriptor_topics.os.rename")
312 def test_new_vnfd_check_pyangbind_validation_property_types(
313 self, mock_rename, mock_shutil
314 ):
315 did, test_vnfd = self.prepare_vnfd_creation()
316 test_vnfd = self.prepare_test_vnfd(test_vnfd)
317 test_vnfd["_id"] = did
318 test_vnfd["product-name"] = {"key": 0}
320 with self.assertRaises(
321 EngineException, msg="Accepted VNFD with a wrongly typed property"
322 ) as e:
323 self.topic.upload_content(
324 fake_session, did, test_vnfd, {}, {"Content-Type": []}
325 )
326 self.assertEqual(
327 e.exception.http_code,
328 HTTPStatus.UNPROCESSABLE_ENTITY,
329 "Wrong HTTP status code",
330 )
331 self.assertIn(
332 norm(
333 "Error in pyangbind validation: {} ({})".format(
334 "json object contained a key that did not exist", "key"
335 )
336 ),
337 norm(str(e.exception)),
338 "Wrong exception text",
339 )
341 @patch("osm_nbi.descriptor_topics.shutil")
342 @patch("osm_nbi.descriptor_topics.os.rename")
343 def test_new_vnfd_check_input_validation_cloud_init(self, mock_rename, mock_shutil):
344 did, test_vnfd = self.prepare_vnfd_creation()
345 del test_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
346 "day1-2"
347 ][0]["execution-environment-list"][0]["juju"]
349 with self.assertRaises(
350 EngineException, msg="Accepted non-existent cloud_init file"
351 ) as e:
352 self.topic.upload_content(
353 fake_session, did, test_vnfd, {}, {"Content-Type": []}
354 )
355 self.assertEqual(
356 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
357 )
358 self.assertIn(
359 norm(
360 "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format(
361 "cloud-init", test_vnfd["id"], test_vnfd["vdu"][0]["id"]
362 )
363 ),
364 norm(str(e.exception)),
365 "Wrong exception text",
366 )
368 @patch("osm_nbi.descriptor_topics.shutil")
369 @patch("osm_nbi.descriptor_topics.os.rename")
370 def test_new_vnfd_check_input_validation_day12_configuration(
371 self, mock_rename, mock_shutil
372 ):
373 did, test_vnfd = self.prepare_vnfd_creation()
374 del test_vnfd["vdu"][0]["cloud-init-file"]
376 with self.assertRaises(
377 EngineException, msg="Accepted non-existent charm in VNF configuration"
378 ) as e:
379 self.topic.upload_content(
380 fake_session, did, test_vnfd, {}, {"Content-Type": []}
381 )
382 self.assertEqual(
383 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
384 )
385 self.assertIn(
386 norm(
387 "{} defined in vnf[id={}] but not present in package".format(
388 "charm", test_vnfd["id"]
389 )
390 ),
391 norm(str(e.exception)),
392 "Wrong exception text",
393 )
395 @patch("osm_nbi.descriptor_topics.shutil")
396 @patch("osm_nbi.descriptor_topics.os.rename")
397 def test_new_vnfd_check_input_validation_mgmt_cp(self, mock_rename, mock_shutil):
398 did, test_vnfd = self.prepare_vnfd_creation()
399 test_vnfd = self.prepare_test_vnfd(test_vnfd)
400 del test_vnfd["mgmt-cp"]
402 with self.assertRaises(
403 EngineException, msg="Accepted VNFD without management interface"
404 ) as e:
405 self.topic.upload_content(
406 fake_session, did, test_vnfd, {}, {"Content-Type": []}
407 )
408 self.assertEqual(
409 e.exception.http_code,
410 HTTPStatus.UNPROCESSABLE_ENTITY,
411 "Wrong HTTP status code",
412 )
413 self.assertIn(
414 norm("'{}' is a mandatory field and it is not defined".format("mgmt-cp")),
415 norm(str(e.exception)),
416 "Wrong exception text",
417 )
419 @patch("osm_nbi.descriptor_topics.shutil")
420 @patch("osm_nbi.descriptor_topics.os.rename")
421 def test_new_vnfd_check_input_validation_mgmt_cp_connection_point(
422 self, mock_rename, mock_shutil
423 ):
424 did, test_vnfd = self.prepare_vnfd_creation()
425 test_vnfd = self.prepare_test_vnfd(test_vnfd)
426 test_vnfd["mgmt-cp"] = "wrong-cp"
428 with self.assertRaises(
429 EngineException, msg="Accepted wrong mgmt-cp connection point"
430 ) as e:
431 self.topic.upload_content(
432 fake_session, did, test_vnfd, {}, {"Content-Type": []}
433 )
434 self.assertEqual(
435 e.exception.http_code,
436 HTTPStatus.UNPROCESSABLE_ENTITY,
437 "Wrong HTTP status code",
438 )
439 self.assertIn(
440 norm(
441 "mgmt-cp='{}' must match an existing ext-cpd".format(
442 test_vnfd["mgmt-cp"]
443 )
444 ),
445 norm(str(e.exception)),
446 "Wrong exception text",
447 )
449 @patch("osm_nbi.descriptor_topics.shutil")
450 @patch("osm_nbi.descriptor_topics.os.rename")
451 def test_new_vnfd_check_input_validation_vdu_int_cpd(
452 self, mock_rename, mock_shutil
453 ):
454 """Testing input validation during new vnfd creation
455 for vdu internal connection point"""
456 did, test_vnfd = self.prepare_vnfd_creation()
457 test_vnfd = self.prepare_test_vnfd(test_vnfd)
458 ext_cpd = test_vnfd["ext-cpd"][1]
459 ext_cpd["int-cpd"]["cpd"] = "wrong-cpd"
461 with self.assertRaises(
462 EngineException, msg="Accepted wrong ext-cpd internal connection point"
463 ) as e:
464 self.topic.upload_content(
465 fake_session, did, test_vnfd, {}, {"Content-Type": []}
466 )
467 self.assertEqual(
468 e.exception.http_code,
469 HTTPStatus.UNPROCESSABLE_ENTITY,
470 "Wrong HTTP status code",
471 )
472 self.assertIn(
473 norm(
474 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
475 ext_cpd["id"]
476 )
477 ),
478 norm(str(e.exception)),
479 "Wrong exception text",
480 )
482 @patch("osm_nbi.descriptor_topics.shutil")
483 @patch("osm_nbi.descriptor_topics.os.rename")
484 def test_new_vnfd_check_input_validation_duplicated_vld(
485 self, mock_rename, mock_shutil
486 ):
487 """Testing input validation during new vnfd creation
488 for dublicated virtual link description"""
489 did, test_vnfd = self.prepare_vnfd_creation()
490 test_vnfd = self.prepare_test_vnfd(test_vnfd)
491 test_vnfd["int-virtual-link-desc"].insert(0, {"id": "internal"})
493 with self.assertRaises(
494 EngineException, msg="Accepted duplicated VLD name"
495 ) as e:
496 self.topic.upload_content(
497 fake_session, did, test_vnfd, {}, {"Content-Type": []}
498 )
499 self.assertEqual(
500 e.exception.http_code,
501 HTTPStatus.UNPROCESSABLE_ENTITY,
502 "Wrong HTTP status code",
503 )
504 self.assertIn(
505 norm(
506 "identifier id '{}' is not unique".format(
507 test_vnfd["int-virtual-link-desc"][0]["id"]
508 )
509 ),
510 norm(str(e.exception)),
511 "Wrong exception text",
512 )
514 @patch("osm_nbi.descriptor_topics.shutil")
515 @patch("osm_nbi.descriptor_topics.os.rename")
516 def test_new_vnfd_check_input_validation_vdu_int_virtual_link_desc(
517 self, mock_rename, mock_shutil
518 ):
519 """Testing input validation during new vnfd creation
520 for vdu internal virtual link description"""
521 did, test_vnfd = self.prepare_vnfd_creation()
522 test_vnfd = self.prepare_test_vnfd(test_vnfd)
523 vdu = test_vnfd["vdu"][0]
524 int_cpd = vdu["int-cpd"][1]
525 int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
527 with self.assertRaises(
528 EngineException, msg="Accepted int-virtual-link-desc"
529 ) as e:
530 self.topic.upload_content(
531 fake_session, did, test_vnfd, {}, {"Content-Type": []}
532 )
533 self.assertEqual(
534 e.exception.http_code,
535 HTTPStatus.UNPROCESSABLE_ENTITY,
536 "Wrong HTTP status code",
537 )
538 self.assertIn(
539 norm(
540 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
541 "int-virtual-link-desc".format(
542 vdu["id"], int_cpd["id"], int_cpd["int-virtual-link-desc"]
543 )
544 ),
545 norm(str(e.exception)),
546 "Wrong exception text",
547 )
549 @patch("osm_nbi.descriptor_topics.shutil")
550 @patch("osm_nbi.descriptor_topics.os.rename")
551 def test_new_vnfd_check_input_validation_virtual_link_profile(
552 self, mock_rename, mock_shutil
553 ):
554 """Testing input validation during new vnfd creation
555 for virtual link profile"""
556 did, test_vnfd = self.prepare_vnfd_creation()
557 test_vnfd = self.prepare_test_vnfd(test_vnfd)
558 fake_ivld_profile = {"id": "fake-profile-ref", "flavour": "fake-flavour"}
559 df = test_vnfd["df"][0]
560 df["virtual-link-profile"] = [fake_ivld_profile]
562 with self.assertRaises(
563 EngineException, msg="Accepted non-existent Profile Ref"
564 ) as e:
565 self.topic.upload_content(
566 fake_session, did, test_vnfd, {}, {"Content-Type": []}
567 )
568 self.assertEqual(
569 e.exception.http_code,
570 HTTPStatus.UNPROCESSABLE_ENTITY,
571 "Wrong HTTP status code",
572 )
573 self.assertIn(
574 norm(
575 "df[id='{}']:virtual-link-profile='{}' must match an existing "
576 "int-virtual-link-desc".format(df["id"], fake_ivld_profile["id"])
577 ),
578 norm(str(e.exception)),
579 "Wrong exception text",
580 )
582 @patch("osm_nbi.descriptor_topics.shutil")
583 @patch("osm_nbi.descriptor_topics.os.rename")
584 def test_new_vnfd_check_input_validation_scaling_criteria_vdu_id(
585 self, mock_rename, mock_shutil
586 ):
587 """Testing input validation during new vnfd creation
588 for scaling criteria with invalid vdu-id"""
589 did, test_vnfd = self.prepare_vnfd_creation()
590 test_vnfd = self.prepare_test_vnfd(test_vnfd)
591 test_vnfd["df"][0]["scaling-aspect"][0]["aspect-delta-details"]["deltas"][0][
592 "vdu-delta"
593 ][0]["id"] = "vdudelta1"
594 affected_df = test_vnfd["df"][0]
595 sa = affected_df["scaling-aspect"][0]
596 delta = sa["aspect-delta-details"]["deltas"][0]
597 vdu_delta = delta["vdu-delta"][0]
599 with self.assertRaises(
600 EngineException, msg="Accepted invalid Scaling Group Policy Criteria"
601 ) as e:
602 self.topic.upload_content(
603 fake_session, did, test_vnfd, {}, {"Content-Type": []}
604 )
605 self.assertEqual(
606 e.exception.http_code,
607 HTTPStatus.UNPROCESSABLE_ENTITY,
608 "Wrong HTTP status code",
609 )
610 self.assertIn(
611 norm(
612 "df[id='{}']:scaling-aspect[id='{}']:aspect-delta-details"
613 "[delta='{}']: "
614 "vdu-id='{}' not defined in vdu".format(
615 affected_df["id"],
616 sa["id"],
617 delta["id"],
618 vdu_delta["id"],
619 )
620 ),
621 norm(str(e.exception)),
622 "Wrong exception text",
623 )
625 @patch("osm_nbi.descriptor_topics.shutil")
626 @patch("osm_nbi.descriptor_topics.os.rename")
627 def test_new_vnfd_check_input_validation_scaling_criteria_monitoring_param_ref(
628 self, mock_rename, mock_shutil
629 ):
630 """Testing input validation during new vnfd creation
631 for scaling criteria without monitoring parameter"""
632 did, test_vnfd = self.prepare_vnfd_creation()
633 test_vnfd = self.prepare_test_vnfd(test_vnfd)
634 vdu = test_vnfd["vdu"][1]
635 affected_df = test_vnfd["df"][0]
636 sa = affected_df["scaling-aspect"][0]
637 sp = sa["scaling-policy"][0]
638 sc = sp["scaling-criteria"][0]
639 vdu.pop("monitoring-parameter")
641 with self.assertRaises(
642 EngineException, msg="Accepted non-existent Scaling Group Policy Criteria"
643 ) as e:
644 self.topic.upload_content(
645 fake_session, did, test_vnfd, {}, {"Content-Type": []}
646 )
647 self.assertEqual(
648 e.exception.http_code,
649 HTTPStatus.UNPROCESSABLE_ENTITY,
650 "Wrong HTTP status code",
651 )
652 self.assertIn(
653 norm(
654 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
655 "[name='{}']:scaling-criteria[name='{}']: "
656 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
657 affected_df["id"],
658 sa["id"],
659 sp["name"],
660 sc["name"],
661 sc["vnf-monitoring-param-ref"],
662 )
663 ),
664 norm(str(e.exception)),
665 "Wrong exception text",
666 )
668 @patch("osm_nbi.descriptor_topics.shutil")
669 @patch("osm_nbi.descriptor_topics.os.rename")
670 def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration(
671 self, mock_rename, mock_shutil
672 ):
673 """Testing input validation during new vnfd creation
674 for scaling criteria without day12 configuration"""
675 did, test_vnfd = self.prepare_vnfd_creation()
676 test_vnfd = self.prepare_test_vnfd(test_vnfd)
677 test_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
678 "day1-2"
679 ].pop()
680 df = test_vnfd["df"][0]
682 with self.assertRaises(
683 EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference"
684 ) as e:
685 self.topic.upload_content(
686 fake_session, did, test_vnfd, {}, {"Content-Type": []}
687 )
688 self.assertEqual(
689 e.exception.http_code,
690 HTTPStatus.UNPROCESSABLE_ENTITY,
691 "Wrong HTTP status code",
692 )
693 self.assertIn(
694 norm(
695 "'day1-2 configuration' not defined in the descriptor but it is referenced "
696 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
697 df["id"], df["scaling-aspect"][0]["id"]
698 )
699 ),
700 norm(str(e.exception)),
701 "Wrong exception text",
702 )
704 @patch("osm_nbi.descriptor_topics.shutil")
705 @patch("osm_nbi.descriptor_topics.os.rename")
706 def test_new_vnfd_check_input_validation_scaling_config_action(
707 self, mock_rename, mock_shutil
708 ):
709 """Testing input validation during new vnfd creation
710 for scaling criteria wrong config primitive"""
711 did, test_vnfd = self.prepare_vnfd_creation()
712 test_vnfd = self.prepare_test_vnfd(test_vnfd)
713 df = test_vnfd["df"][0]
714 affected_df = test_vnfd["df"][0]
715 sa = affected_df["scaling-aspect"][0]
716 test_vnfd["df"][0].get("lcm-operations-configuration").get(
717 "operate-vnf-op-config"
718 )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
720 with self.assertRaises(
721 EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference"
722 ) as e:
723 self.topic.upload_content(
724 fake_session, did, test_vnfd, {}, {"Content-Type": []}
725 )
726 self.assertEqual(
727 e.exception.http_code,
728 HTTPStatus.UNPROCESSABLE_ENTITY,
729 "Wrong HTTP status code",
730 )
731 self.assertIn(
732 norm(
733 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
734 "config-primitive-name-ref='{}' does not match any "
735 "day1-2 configuration:config-primitive:name".format(
736 df["id"],
737 df["scaling-aspect"][0]["id"],
738 sa["scaling-config-action"][0]["vnf-config-primitive-name-ref"],
739 )
740 ),
741 norm(str(e.exception)),
742 "Wrong exception text",
743 )
745 @patch("osm_nbi.descriptor_topics.shutil")
746 @patch("osm_nbi.descriptor_topics.os.rename")
747 def test_new_vnfd_check_input_validation_healing_criteria_vdu_id(
748 self, mock_rename, mock_shutil
749 ):
750 """Testing input validation during new vnfd creation
751 for healing criteria with invalid vdu-id"""
752 did, test_vnfd = self.prepare_vnfd_creation()
753 test_vnfd = self.prepare_test_vnfd(test_vnfd)
754 test_vnfd["df"][0]["healing-aspect"][0]["healing-policy"][0][
755 "vdu-id"
756 ] = "vduid1"
757 affected_df = test_vnfd["df"][0]
758 ha = affected_df["healing-aspect"][0]
759 hp = ha["healing-policy"][0]
760 hp_vdu_id = hp["vdu-id"]
762 with self.assertRaises(
763 EngineException, msg="Accepted invalid Healing Group Policy Criteria"
764 ) as e:
765 self.topic.upload_content(
766 fake_session, did, test_vnfd, {}, {"Content-Type": []}
767 )
768 self.assertEqual(
769 e.exception.http_code,
770 HTTPStatus.UNPROCESSABLE_ENTITY,
771 "Wrong HTTP status code",
772 )
773 self.assertIn(
774 norm(
775 "df[id='{}']:healing-aspect[id='{}']:healing-policy"
776 "[name='{}']: "
777 "vdu-id='{}' not defined in vdu".format(
778 affected_df["id"],
779 ha["id"],
780 hp["event-name"],
781 hp_vdu_id,
782 )
783 ),
784 norm(str(e.exception)),
785 "Wrong exception text",
786 )
788 @patch("osm_nbi.descriptor_topics.shutil")
789 @patch("osm_nbi.descriptor_topics.os.rename")
790 def test_new_vnfd_check_input_validation_alarm_criteria_monitoring_param_ref(
791 self, mock_rename, mock_shutil
792 ):
793 """Testing input validation during new vnfd creation
794 for alarm with invalid monitoring parameter reference"""
795 did, test_vnfd = self.prepare_vnfd_creation()
796 test_vnfd = self.prepare_test_vnfd(test_vnfd)
797 test_vnfd["vdu"][1]["alarm"][0]["vnf-monitoring-param-ref"] = "unit_test_alarm"
798 vdu = test_vnfd["vdu"][1]
799 alarm = vdu["alarm"][0]
800 alarm_monitoring_param = alarm["vnf-monitoring-param-ref"]
802 with self.assertRaises(
803 EngineException, msg="Accepted invalid Alarm Criteria"
804 ) as e:
805 self.topic.upload_content(
806 fake_session, did, test_vnfd, {}, {"Content-Type": []}
807 )
808 self.assertEqual(
809 e.exception.http_code,
810 HTTPStatus.UNPROCESSABLE_ENTITY,
811 "Wrong HTTP status code",
812 )
813 self.assertIn(
814 norm(
815 "vdu[id='{}']:alarm[id='{}']:"
816 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
817 vdu["id"],
818 alarm["alarm-id"],
819 alarm_monitoring_param,
820 )
821 ),
822 norm(str(e.exception)),
823 "Wrong exception text",
824 )
826 @patch("osm_nbi.descriptor_topics.shutil")
827 @patch("osm_nbi.descriptor_topics.os.rename")
828 def test_new_vnfd_check_input_validation_storage_reference_criteria(
829 self, mock_rename, mock_shutil
830 ):
831 """Testing input validation during new vnfd creation
832 for invalid virtual-storge-desc reference"""
833 did, test_vnfd = self.prepare_vnfd_creation()
834 test_vnfd = self.prepare_test_vnfd(test_vnfd)
835 test_vnfd["vdu"][1]["virtual-storage-desc"] = "unit_test_storage"
836 vdu = test_vnfd["vdu"][1]
837 vsd_ref = vdu["virtual-storage-desc"]
839 with self.assertRaises(
840 EngineException, msg="Accepted invalid virtual-storage-desc"
841 ) as e:
842 self.topic.upload_content(
843 fake_session, did, test_vnfd, {}, {"Content-Type": []}
844 )
845 self.assertEqual(
846 e.exception.http_code,
847 HTTPStatus.UNPROCESSABLE_ENTITY,
848 "Wrong HTTP status code",
849 )
850 self.assertIn(
851 norm(
852 "vdu[virtual-storage-desc='{}']"
853 "not defined in vnfd".format(
854 vsd_ref,
855 )
856 ),
857 norm(str(e.exception)),
858 "Wrong exception text",
859 )
861 @patch("osm_nbi.descriptor_topics.shutil")
862 @patch("osm_nbi.descriptor_topics.os.rename")
863 def test_new_vnfd_check_input_validation_compute_reference_criteria(
864 self, mock_rename, mock_shutil
865 ):
866 """Testing input validation during new vnfd creation
867 for invalid virtual-compute-desc reference"""
868 did, test_vnfd = self.prepare_vnfd_creation()
869 test_vnfd = self.prepare_test_vnfd(test_vnfd)
870 test_vnfd["vdu"][1]["virtual-compute-desc"] = "unit_test_compute"
871 vdu = test_vnfd["vdu"][1]
872 vcd_ref = vdu["virtual-compute-desc"]
874 with self.assertRaises(
875 EngineException, msg="Accepted invalid virtual-compute-desc"
876 ) as e:
877 self.topic.upload_content(
878 fake_session, did, test_vnfd, {}, {"Content-Type": []}
879 )
880 self.assertEqual(
881 e.exception.http_code,
882 HTTPStatus.UNPROCESSABLE_ENTITY,
883 "Wrong HTTP status code",
884 )
885 self.assertIn(
886 norm(
887 "vdu[virtual-compute-desc='{}']"
888 "not defined in vnfd".format(
889 vcd_ref,
890 )
891 ),
892 norm(str(e.exception)),
893 "Wrong exception text",
894 )
896 @patch("osm_nbi.descriptor_topics.shutil")
897 @patch("osm_nbi.descriptor_topics.os.rename")
898 def test_new_vnfd_check_input_validation_everything_right(
899 self, mock_rename, mock_shutil
900 ):
901 """Testing input validation during new vnfd creation
902 everything correct"""
903 did, test_vnfd = self.prepare_vnfd_creation()
904 test_vnfd = self.prepare_test_vnfd(test_vnfd)
905 test_vnfd["id"] = "fake-vnfd-id"
906 test_vnfd["df"][0].get("lcm-operations-configuration").get(
907 "operate-vnf-op-config"
908 )["day1-2"][0]["id"] = "fake-vnfd-id"
909 self.db.get_one.side_effect = [
910 {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])},
911 None,
912 ]
913 rc = self.topic.upload_content(
914 fake_session, did, test_vnfd, {}, {"Content-Type": []}
915 )
916 self.assertTrue(rc, "Input Validation: Unexpected failure")
918 def test_edit_vnfd(self):
919 vnfd_content = deepcopy(db_vnfd_content)
920 did = vnfd_content["_id"]
921 self.fs.file_exists.return_value = True
922 self.fs.dir_ls.return_value = True
923 with self.subTest(i=1, t="Normal Edition"):
924 now = time()
925 self.db.get_one.side_effect = [deepcopy(vnfd_content), None]
926 data = {"product-name": "new-vnfd-name"}
927 self.topic.edit(fake_session, did, data)
928 db_args = self.db.replace.call_args[0]
929 msg_args = self.msg.write.call_args[0]
930 data["_id"] = did
931 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
932 self.assertEqual(msg_args[1], "edited", "Wrong message action")
933 self.assertEqual(msg_args[2], data, "Wrong message content")
934 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
935 self.assertEqual(db_args[1], did, "Wrong DB ID")
936 self.assertEqual(
937 db_args[2]["_admin"]["created"],
938 vnfd_content["_admin"]["created"],
939 "Wrong creation time",
940 )
941 self.assertGreater(
942 db_args[2]["_admin"]["modified"], now, "Wrong modification time"
943 )
944 self.assertEqual(
945 db_args[2]["_admin"]["projects_read"],
946 vnfd_content["_admin"]["projects_read"],
947 "Wrong read-only project list",
948 )
949 self.assertEqual(
950 db_args[2]["_admin"]["projects_write"],
951 vnfd_content["_admin"]["projects_write"],
952 "Wrong read-write project list",
953 )
954 self.assertEqual(
955 db_args[2]["product-name"], data["product-name"], "Wrong VNFD Name"
956 )
957 with self.subTest(i=2, t="Conflict on Edit"):
958 data = {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
959 self.db.get_one.side_effect = [
960 deepcopy(vnfd_content),
961 {"_id": str(uuid4()), "id": data["id"]},
962 ]
963 with self.assertRaises(
964 EngineException, msg="Accepted existing VNFD ID"
965 ) as e:
966 self.topic.edit(fake_session, did, data)
967 self.assertEqual(
968 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
969 )
970 self.assertIn(
971 norm(
972 "{} with id '{}' already exists for this project".format(
973 "vnfd", data["id"]
974 )
975 ),
976 norm(str(e.exception)),
977 "Wrong exception text",
978 )
979 with self.subTest(i=3, t="Check Envelope"):
980 data = {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
981 with self.assertRaises(
982 EngineException, msg="Accepted VNFD with wrong envelope"
983 ) as e:
984 self.topic.edit(fake_session, did, data, content=vnfd_content)
985 self.assertEqual(
986 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
987 )
988 self.assertIn(
989 "'vnfd' must be dict", norm(str(e.exception)), "Wrong exception text"
990 )
991 return
993 def test_delete_vnfd(self):
994 did = db_vnfd_content["_id"]
995 self.db.get_one.return_value = db_vnfd_content
996 p_id = db_vnfd_content["_admin"]["projects_read"][0]
997 with self.subTest(i=1, t="Normal Deletion"):
998 self.db.get_list.return_value = []
999 self.db.del_one.return_value = {"deleted": 1}
1000 self.topic.delete(fake_session, did)
1001 db_args = self.db.del_one.call_args[0]
1002 msg_args = self.msg.write.call_args[0]
1003 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1004 self.assertEqual(msg_args[1], "deleted", "Wrong message action")
1005 self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
1006 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1007 self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
1008 self.assertEqual(
1009 db_args[1]["_admin.projects_write.cont"],
1010 [p_id, "ANY"],
1011 "Wrong DB filter",
1012 )
1013 db_g1_args = self.db.get_one.call_args[0]
1014 self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
1015 self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
1016 db_gl_calls = self.db.get_list.call_args_list
1017 self.assertEqual(db_gl_calls[0][0][0], "vnfrs", "Wrong DB topic")
1018 # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID") # Filter changed after call
1019 self.assertEqual(db_gl_calls[1][0][0], "nsds", "Wrong DB topic")
1020 self.assertEqual(
1021 db_gl_calls[1][0][1]["vnfd-id"],
1022 db_vnfd_content["id"],
1023 "Wrong DB NSD vnfd-id",
1024 )
1026 self.assertEqual(
1027 self.db.del_list.call_args[0][0],
1028 self.topic.topic + "_revisions",
1029 "Wrong DB topic",
1030 )
1032 self.assertEqual(
1033 self.db.del_list.call_args[0][1]["_id"]["$regex"],
1034 did,
1035 "Wrong ID for rexep delete",
1036 )
1038 self.db.set_one.assert_not_called()
1039 fs_del_calls = self.fs.file_delete.call_args_list
1040 self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
1041 self.assertEqual(fs_del_calls[1][0][0], did + "_", "Wrong FS folder id")
1042 with self.subTest(i=2, t="Conflict on Delete - VNFD in use by VNFR"):
1043 self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-vnfr"}]
1044 with self.assertRaises(
1045 EngineException, msg="Accepted VNFD in use by VNFR"
1046 ) as e:
1047 self.topic.delete(fake_session, did)
1048 self.assertEqual(
1049 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1050 )
1051 self.assertIn(
1052 "there is at least one vnf instance using this descriptor",
1053 norm(str(e.exception)),
1054 "Wrong exception text",
1055 )
1056 with self.subTest(i=3, t="Conflict on Delete - VNFD in use by NSD"):
1057 self.db.get_list.side_effect = [
1058 [],
1059 [{"_id": str(uuid4()), "name": "fake-nsd"}],
1060 ]
1061 with self.assertRaises(
1062 EngineException, msg="Accepted VNFD in use by NSD"
1063 ) as e:
1064 self.topic.delete(fake_session, did)
1065 self.assertEqual(
1066 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1067 )
1068 self.assertIn(
1069 "there is at least one ns package referencing this descriptor",
1070 norm(str(e.exception)),
1071 "Wrong exception text",
1072 )
1073 with self.subTest(i=4, t="Non-existent VNFD"):
1074 excp_msg = "Not found any {} with filter='{}'".format("VNFD", {"_id": did})
1075 self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
1076 with self.assertRaises(
1077 DbException, msg="Accepted non-existent VNFD ID"
1078 ) as e:
1079 self.topic.delete(fake_session, did)
1080 self.assertEqual(
1081 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
1082 )
1083 self.assertIn(
1084 norm(excp_msg), norm(str(e.exception)), "Wrong exception text"
1085 )
1086 with self.subTest(i=5, t="No delete because referenced by other project"):
1087 db_vnfd_content["_admin"]["projects_read"].append("other_project")
1088 self.db.get_one = Mock(return_value=db_vnfd_content)
1089 self.db.get_list = Mock(return_value=[])
1090 self.msg.write.reset_mock()
1091 self.db.del_one.reset_mock()
1092 self.fs.file_delete.reset_mock()
1094 self.topic.delete(fake_session, did)
1095 self.db.del_one.assert_not_called()
1096 self.msg.write.assert_not_called()
1097 db_g1_args = self.db.get_one.call_args[0]
1098 self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
1099 self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
1100 db_s1_args = self.db.set_one.call_args
1101 self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
1102 self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
1103 self.assertIn(
1104 p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1105 )
1106 self.assertIsNone(
1107 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
1108 )
1109 self.assertEqual(
1110 db_s1_args[1]["pull_list"],
1111 {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
1112 "Wrong DB pull_list dictionary",
1113 )
1114 self.fs.file_delete.assert_not_called()
1115 return
1117 def prepare_vnfd_validation(self):
1118 descriptor_name = "test_descriptor"
1119 self.fs.file_open.side_effect = lambda path, mode: open(
1120 "/tmp/" + str(uuid4()), "a+b"
1121 )
1122 old_vnfd, new_vnfd = self.create_desc_temp(db_vnfd_content)
1123 return descriptor_name, old_vnfd, new_vnfd
1125 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1126 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1127 def test_validate_vnfd_changes_day12_config_primitive_changed(
1128 self, mock_safe_load, mock_detect_usage
1129 ):
1130 """Validating VNFD for VNFD updates, day1-2 config primitive has changed"""
1131 descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
1132 did = old_vnfd["_id"]
1133 new_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1134 "day1-2"
1135 ][0]["config-primitive"][0]["name"] = "new_action"
1136 mock_safe_load.side_effect = [old_vnfd, new_vnfd]
1137 mock_detect_usage.return_value = True
1138 self.db.get_one.return_value = old_vnfd
1140 with self.assertNotRaises(EngineException):
1141 self.topic._validate_descriptor_changes(
1142 did, descriptor_name, "/tmp/", "/tmp:1/"
1143 )
1144 self.db.get_one.assert_called_once()
1145 mock_detect_usage.assert_called_once()
1146 self.assertEqual(mock_safe_load.call_count, 2)
1148 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1149 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1150 def test_validate_vnfd_changes_sw_version_changed(
1151 self, mock_safe_load, mock_detect_usage
1152 ):
1153 """Validating VNFD for updates, software version has changed"""
1154 # old vnfd uses the default software version: 1.0
1155 descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
1156 did = old_vnfd["_id"]
1157 new_vnfd["software-version"] = "1.3"
1158 new_vnfd["sw-image-desc"][0]["name"] = "new-image"
1159 mock_safe_load.side_effect = [old_vnfd, new_vnfd]
1160 mock_detect_usage.return_value = True
1161 self.db.get_one.return_value = old_vnfd
1163 with self.assertNotRaises(EngineException):
1164 self.topic._validate_descriptor_changes(
1165 did, descriptor_name, "/tmp/", "/tmp:1/"
1166 )
1167 self.db.get_one.assert_called_once()
1168 mock_detect_usage.assert_called_once()
1169 self.assertEqual(mock_safe_load.call_count, 2)
1171 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1172 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1173 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed(
1174 self, mock_safe_load, mock_detect_usage
1175 ):
1176 """Validating VNFD for updates, software version has not
1177 changed, mgmt-cp has changed."""
1178 descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
1179 new_vnfd["mgmt-cp"] = "new-mgmt-cp"
1180 mock_safe_load.side_effect = [old_vnfd, new_vnfd]
1181 did = old_vnfd["_id"]
1182 mock_detect_usage.return_value = True
1183 self.db.get_one.return_value = old_vnfd
1185 with self.assertRaises(
1186 EngineException, msg="there are disallowed changes in the vnf descriptor"
1187 ) as e:
1188 self.topic._validate_descriptor_changes(
1189 did, descriptor_name, "/tmp/", "/tmp:1/"
1190 )
1192 self.assertEqual(
1193 e.exception.http_code,
1194 HTTPStatus.UNPROCESSABLE_ENTITY,
1195 "Wrong HTTP status code",
1196 )
1197 self.assertIn(
1198 norm("there are disallowed changes in the vnf descriptor"),
1199 norm(str(e.exception)),
1200 "Wrong exception text",
1201 )
1202 self.db.get_one.assert_called_once()
1203 mock_detect_usage.assert_called_once()
1204 self.assertEqual(mock_safe_load.call_count, 2)
1206 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1207 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1208 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed_vnfd_not_in_use(
1209 self, mock_safe_load, mock_detect_usage
1210 ):
1211 """Validating VNFD for updates, software version has not
1212 changed, mgmt-cp has changed, vnfd is not in use."""
1213 descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
1214 new_vnfd["mgmt-cp"] = "new-mgmt-cp"
1215 mock_safe_load.side_effect = [old_vnfd, new_vnfd]
1216 did = old_vnfd["_id"]
1217 mock_detect_usage.return_value = None
1218 self.db.get_one.return_value = old_vnfd
1220 with self.assertNotRaises(EngineException):
1221 self.topic._validate_descriptor_changes(
1222 did, descriptor_name, "/tmp/", "/tmp:1/"
1223 )
1225 self.db.get_one.assert_called_once()
1226 mock_detect_usage.assert_called_once()
1227 mock_safe_load.assert_not_called()
1229 def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self):
1230 indata = deepcopy(db_vnfd_content)
1231 self.topic.validate_mgmt_interface_connection_point(indata)
1233 def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
1234 self,
1235 ):
1236 indata = deepcopy(db_vnfd_content)
1237 indata["ext-cpd"] = []
1238 with self.assertRaises(EngineException) as e:
1239 self.topic.validate_mgmt_interface_connection_point(indata)
1240 self.assertEqual(
1241 e.exception.http_code,
1242 HTTPStatus.UNPROCESSABLE_ENTITY,
1243 "Wrong HTTP status code",
1244 )
1245 self.assertIn(
1246 norm(
1247 "mgmt-cp='{}' must match an existing ext-cpd".format(indata["mgmt-cp"])
1248 ),
1249 norm(str(e.exception)),
1250 "Wrong exception text",
1251 )
1253 def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self):
1254 indata = deepcopy(db_vnfd_content)
1255 indata.pop("mgmt-cp")
1256 with self.assertRaises(EngineException) as e:
1257 self.topic.validate_mgmt_interface_connection_point(indata)
1258 self.assertEqual(
1259 e.exception.http_code,
1260 HTTPStatus.UNPROCESSABLE_ENTITY,
1261 "Wrong HTTP status code",
1262 )
1263 self.assertIn(
1264 norm("'mgmt-cp' is a mandatory field and it is not defined"),
1265 norm(str(e.exception)),
1266 "Wrong exception text",
1267 )
1269 def test_validate_vdu_internal_connection_points_on_valid_descriptor(self):
1270 indata = db_vnfd_content
1271 vdu = indata["vdu"][0]
1272 self.topic.validate_vdu_internal_connection_points(vdu)
1274 def test_validate_external_connection_points_on_valid_descriptor(self):
1275 indata = db_vnfd_content
1276 self.topic.validate_external_connection_points(indata)
1278 def test_validate_external_connection_points_when_missing_internal_connection_point(
1279 self,
1280 ):
1281 indata = deepcopy(db_vnfd_content)
1282 vdu = indata["vdu"][0]
1283 vdu.pop("int-cpd")
1284 affected_ext_cpd = indata["ext-cpd"][0]
1285 with self.assertRaises(EngineException) as e:
1286 self.topic.validate_external_connection_points(indata)
1287 self.assertEqual(
1288 e.exception.http_code,
1289 HTTPStatus.UNPROCESSABLE_ENTITY,
1290 "Wrong HTTP status code",
1291 )
1292 self.assertIn(
1293 norm(
1294 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
1295 affected_ext_cpd["id"]
1296 )
1297 ),
1298 norm(str(e.exception)),
1299 "Wrong exception text",
1300 )
1302 def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
1303 self,
1304 ):
1305 indata = deepcopy(db_vnfd_content)
1306 vdu = indata["vdu"][0]
1307 duplicated_cpd = {
1308 "id": "vnf-mgmt",
1309 "order": 3,
1310 "virtual-network-interface-requirement": [{"name": "duplicated"}],
1311 }
1312 vdu["int-cpd"].insert(0, duplicated_cpd)
1313 with self.assertRaises(EngineException) as e:
1314 self.topic.validate_vdu_internal_connection_points(vdu)
1315 self.assertEqual(
1316 e.exception.http_code,
1317 HTTPStatus.UNPROCESSABLE_ENTITY,
1318 "Wrong HTTP status code",
1319 )
1320 self.assertIn(
1321 norm(
1322 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
1323 vdu["id"], duplicated_cpd["id"]
1324 )
1325 ),
1326 norm(str(e.exception)),
1327 "Wrong exception text",
1328 )
1330 def test_validate_external_connection_points_on_duplicated_external_connection_point(
1331 self,
1332 ):
1333 indata = deepcopy(db_vnfd_content)
1334 duplicated_cpd = {
1335 "id": "vnf-mgmt-ext",
1336 "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
1337 }
1338 indata["ext-cpd"].insert(0, duplicated_cpd)
1339 with self.assertRaises(EngineException) as e:
1340 self.topic.validate_external_connection_points(indata)
1341 self.assertEqual(
1342 e.exception.http_code,
1343 HTTPStatus.UNPROCESSABLE_ENTITY,
1344 "Wrong HTTP status code",
1345 )
1346 self.assertIn(
1347 norm(
1348 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
1349 duplicated_cpd["id"]
1350 )
1351 ),
1352 norm(str(e.exception)),
1353 "Wrong exception text",
1354 )
1356 def test_validate_internal_virtual_links_on_valid_descriptor(self):
1357 indata = db_vnfd_content
1358 self.topic.validate_internal_virtual_links(indata)
1360 def test_validate_internal_virtual_links_on_duplicated_ivld(self):
1361 indata = deepcopy(db_vnfd_content)
1362 duplicated_vld = {"id": "internal"}
1363 indata["int-virtual-link-desc"].insert(0, duplicated_vld)
1364 with self.assertRaises(EngineException) as e:
1365 self.topic.validate_internal_virtual_links(indata)
1366 self.assertEqual(
1367 e.exception.http_code,
1368 HTTPStatus.UNPROCESSABLE_ENTITY,
1369 "Wrong HTTP status code",
1370 )
1371 self.assertIn(
1372 norm(
1373 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
1374 duplicated_vld["id"]
1375 )
1376 ),
1377 norm(str(e.exception)),
1378 "Wrong exception text",
1379 )
1381 def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
1382 self,
1383 ):
1384 indata = deepcopy(db_vnfd_content)
1385 vdu = indata["vdu"][0]
1386 affected_int_cpd = vdu["int-cpd"][0]
1387 affected_int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
1388 with self.assertRaises(EngineException) as e:
1389 self.topic.validate_internal_virtual_links(indata)
1390 self.assertEqual(
1391 e.exception.http_code,
1392 HTTPStatus.UNPROCESSABLE_ENTITY,
1393 "Wrong HTTP status code",
1394 )
1395 self.assertIn(
1396 norm(
1397 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
1398 "int-virtual-link-desc".format(
1399 vdu["id"],
1400 affected_int_cpd["id"],
1401 affected_int_cpd["int-virtual-link-desc"],
1402 )
1403 ),
1404 norm(str(e.exception)),
1405 "Wrong exception text",
1406 )
1408 def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self):
1409 indata = deepcopy(db_vnfd_content)
1410 affected_ivld_profile = {"id": "non-existing-int-virtual-link-desc"}
1411 df = indata["df"][0]
1412 df["virtual-link-profile"] = [affected_ivld_profile]
1413 with self.assertRaises(EngineException) as e:
1414 self.topic.validate_internal_virtual_links(indata)
1415 self.assertEqual(
1416 e.exception.http_code,
1417 HTTPStatus.UNPROCESSABLE_ENTITY,
1418 "Wrong HTTP status code",
1419 )
1420 self.assertIn(
1421 norm(
1422 "df[id='{}']:virtual-link-profile='{}' must match an existing "
1423 "int-virtual-link-desc".format(df["id"], affected_ivld_profile["id"])
1424 ),
1425 norm(str(e.exception)),
1426 "Wrong exception text",
1427 )
1429 def test_validate_monitoring_params_on_valid_descriptor(self):
1430 indata = db_vnfd_content
1431 self.topic.validate_monitoring_params(indata)
1433 def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self):
1434 indata = deepcopy(db_vnfd_content)
1435 duplicated_mp = {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1436 affected_ivld = indata["int-virtual-link-desc"][0]
1437 affected_ivld["monitoring-parameters"] = [duplicated_mp, duplicated_mp]
1438 with self.assertRaises(EngineException) as e:
1439 self.topic.validate_monitoring_params(indata)
1440 self.assertEqual(
1441 e.exception.http_code,
1442 HTTPStatus.UNPROCESSABLE_ENTITY,
1443 "Wrong HTTP status code",
1444 )
1445 self.assertIn(
1446 norm(
1447 "Duplicated monitoring-parameter id in "
1448 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1449 affected_ivld["id"], duplicated_mp["id"]
1450 )
1451 ),
1452 norm(str(e.exception)),
1453 "Wrong exception text",
1454 )
1456 def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self):
1457 indata = deepcopy(db_vnfd_content)
1458 duplicated_mp = {
1459 "id": "dataVM_cpu_util",
1460 "name": "dataVM_cpu_util",
1461 "performance_metric": "cpu",
1462 }
1463 affected_vdu = indata["vdu"][1]
1464 affected_vdu["monitoring-parameter"].insert(0, duplicated_mp)
1465 with self.assertRaises(EngineException) as e:
1466 self.topic.validate_monitoring_params(indata)
1467 self.assertEqual(
1468 e.exception.http_code,
1469 HTTPStatus.UNPROCESSABLE_ENTITY,
1470 "Wrong HTTP status code",
1471 )
1472 self.assertIn(
1473 norm(
1474 "Duplicated monitoring-parameter id in "
1475 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1476 affected_vdu["id"], duplicated_mp["id"]
1477 )
1478 ),
1479 norm(str(e.exception)),
1480 "Wrong exception text",
1481 )
1483 def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self):
1484 indata = deepcopy(db_vnfd_content)
1485 duplicated_mp = {
1486 "id": "memory",
1487 "name": "memory",
1488 "performance_metric": "memory",
1489 }
1490 affected_df = indata["df"][0]
1491 affected_df["monitoring-parameter"] = [duplicated_mp, duplicated_mp]
1492 with self.assertRaises(EngineException) as e:
1493 self.topic.validate_monitoring_params(indata)
1494 self.assertEqual(
1495 e.exception.http_code,
1496 HTTPStatus.UNPROCESSABLE_ENTITY,
1497 "Wrong HTTP status code",
1498 )
1499 self.assertIn(
1500 norm(
1501 "Duplicated monitoring-parameter id in "
1502 "df[id='{}']:monitoring-parameter[id='{}']".format(
1503 affected_df["id"], duplicated_mp["id"]
1504 )
1505 ),
1506 norm(str(e.exception)),
1507 "Wrong exception text",
1508 )
1510 def test_validate_scaling_group_descriptor_on_valid_descriptor(self):
1511 indata = db_vnfd_content
1512 self.topic.validate_scaling_group_descriptor(indata)
1514 def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self):
1515 indata = deepcopy(db_vnfd_content)
1516 vdu = indata["vdu"][1]
1517 affected_df = indata["df"][0]
1518 affected_sa = affected_df["scaling-aspect"][0]
1519 affected_sp = affected_sa["scaling-policy"][0]
1520 affected_sc = affected_sp["scaling-criteria"][0]
1521 vdu.pop("monitoring-parameter")
1522 with self.assertRaises(EngineException) as e:
1523 self.topic.validate_scaling_group_descriptor(indata)
1524 self.assertEqual(
1525 e.exception.http_code,
1526 HTTPStatus.UNPROCESSABLE_ENTITY,
1527 "Wrong HTTP status code",
1528 )
1529 self.assertIn(
1530 norm(
1531 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1532 "[name='{}']:scaling-criteria[name='{}']: "
1533 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1534 affected_df["id"],
1535 affected_sa["id"],
1536 affected_sp["name"],
1537 affected_sc["name"],
1538 affected_sc["vnf-monitoring-param-ref"],
1539 )
1540 ),
1541 norm(str(e.exception)),
1542 "Wrong exception text",
1543 )
1545 def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self):
1546 indata = deepcopy(db_vnfd_content)
1547 df = indata["df"][0]
1548 affected_sa = df["scaling-aspect"][0]
1549 indata["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1550 "day1-2"
1551 ].pop()
1552 with self.assertRaises(EngineException) as e:
1553 self.topic.validate_scaling_group_descriptor(indata)
1554 self.assertEqual(
1555 e.exception.http_code,
1556 HTTPStatus.UNPROCESSABLE_ENTITY,
1557 "Wrong HTTP status code",
1558 )
1559 self.assertIn(
1560 norm(
1561 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1562 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1563 df["id"], affected_sa["id"]
1564 )
1565 ),
1566 norm(str(e.exception)),
1567 "Wrong exception text",
1568 )
1570 def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1571 self,
1572 ):
1573 indata = deepcopy(db_vnfd_content)
1574 df = indata["df"][0]
1575 affected_sa = df["scaling-aspect"][0]
1576 affected_sca_primitive = affected_sa["scaling-config-action"][0][
1577 "vnf-config-primitive-name-ref"
1578 ]
1579 df["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1580 "config-primitive"
1581 ] = []
1582 with self.assertRaises(EngineException) as e:
1583 self.topic.validate_scaling_group_descriptor(indata)
1584 self.assertEqual(
1585 e.exception.http_code,
1586 HTTPStatus.UNPROCESSABLE_ENTITY,
1587 "Wrong HTTP status code",
1588 )
1589 self.assertIn(
1590 norm(
1591 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1592 "config-primitive-name-ref='{}' does not match any "
1593 "day1-2 configuration:config-primitive:name".format(
1594 df["id"], affected_sa["id"], affected_sca_primitive
1595 )
1596 ),
1597 norm(str(e.exception)),
1598 "Wrong exception text",
1599 )
1601 def test_new_vnfd_revision(self):
1602 did = db_vnfd_content["_id"]
1603 self.fs.get_params.return_value = {}
1604 self.fs.file_exists.return_value = False
1605 self.fs.file_open.side_effect = lambda path, mode: open(
1606 "/tmp/" + str(uuid4()), "a+b"
1607 )
1608 test_vnfd = deepcopy(db_vnfd_content)
1609 del test_vnfd["_id"]
1610 del test_vnfd["_admin"]
1611 self.db.create.return_value = did
1612 rollback = []
1613 did2, oid = self.topic.new(rollback, fake_session, {})
1614 db_args = self.db.create.call_args[0]
1615 self.assertEqual(
1616 db_args[1]["_admin"]["revision"], 0, "New package should be at revision 0"
1617 )
1619 @patch("osm_nbi.descriptor_topics.shutil")
1620 @patch("osm_nbi.descriptor_topics.os.rename")
1621 def test_update_vnfd(self, mock_rename, mock_shutil):
1622 old_revision = 5
1623 did = db_vnfd_content["_id"]
1624 self.fs.path = ""
1625 self.fs.get_params.return_value = {}
1626 self.fs.file_exists.return_value = False
1627 self.fs.file_open.side_effect = lambda path, mode: open(
1628 "/tmp/" + str(uuid4()), "a+b"
1629 )
1630 new_vnfd = deepcopy(db_vnfd_content)
1631 del new_vnfd["_id"]
1632 self.db.create.return_value = did
1633 rollback = []
1634 did2, oid = self.topic.new(rollback, fake_session, {})
1635 del new_vnfd["vdu"][0]["cloud-init-file"]
1636 del new_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1637 "day1-2"
1638 ][0]["execution-environment-list"][0]["juju"]
1640 old_vnfd = {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}
1641 old_vnfd["_admin"]["revision"] = old_revision
1643 self.db.get_one.side_effect = [old_vnfd, old_vnfd, None]
1644 self.topic.upload_content(fake_session, did, new_vnfd, {}, {"Content-Type": []})
1646 db_args = self.db.replace.call_args[0]
1647 self.assertEqual(
1648 db_args[2]["_admin"]["revision"],
1649 old_revision + 1,
1650 "Revision should increment",
1651 )
1654class Test_NsdTopic(TestCase):
1655 @classmethod
1656 def setUpClass(cls):
1657 cls.test_name = "test-nsd-topic"
1659 @classmethod
1660 def tearDownClass(cls):
1661 pass
1663 def setUp(self):
1664 self.db = Mock(dbbase.DbBase())
1665 self.fs = Mock(fsbase.FsBase())
1666 self.msg = Mock(msgbase.MsgBase())
1667 self.auth = Mock(authconn.Authconn(None, None, None))
1668 self.topic = NsdTopic(self.db, self.fs, self.msg, self.auth)
1669 self.topic.check_quota = Mock(return_value=None) # skip quota
1671 @contextmanager
1672 def assertNotRaises(self, exception_type):
1673 try:
1674 yield None
1675 except exception_type:
1676 raise self.failureException("{} raised".format(exception_type.__name__))
1678 def create_desc_temp(self, template):
1679 old_desc = deepcopy(template)
1680 new_desc = deepcopy(template)
1681 return old_desc, new_desc
1683 def prepare_nsd_creation(self):
1684 self.fs.path = ""
1685 did = db_nsd_content["_id"]
1686 self.fs.get_params.return_value = {}
1687 self.fs.file_exists.return_value = False
1688 self.fs.file_open.side_effect = lambda path, mode: tempfile.TemporaryFile(
1689 mode="a+b"
1690 )
1691 self.db.get_one.side_effect = [
1692 {"_id": did, "_admin": deepcopy(db_nsd_content["_admin"])},
1693 None,
1694 ]
1695 test_nsd = deepcopy(db_nsd_content)
1696 del test_nsd["_id"]
1697 del test_nsd["_admin"]
1698 return did, test_nsd
1700 @patch("osm_nbi.descriptor_topics.shutil")
1701 @patch("osm_nbi.descriptor_topics.os.rename")
1702 def test_new_nsd_normal_creation(self, mock_rename, mock_shutil):
1703 did, test_nsd = self.prepare_nsd_creation()
1704 self.db.create.return_value = did
1705 rollback = []
1707 did2, oid = self.topic.new(rollback, fake_session, {})
1708 db_args = self.db.create.call_args[0]
1709 msg_args = self.msg.write.call_args[0]
1710 self.assertEqual(len(rollback), 1, "Wrong rollback length")
1711 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1712 self.assertEqual(msg_args[1], "created", "Wrong message action")
1713 self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
1714 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1715 self.assertEqual(did2, did, "Wrong DB NSD id")
1716 self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
1717 self.assertEqual(
1718 db_args[1]["_admin"]["modified"],
1719 db_args[1]["_admin"]["created"],
1720 "Wrong modification time",
1721 )
1722 self.assertEqual(
1723 db_args[1]["_admin"]["projects_read"],
1724 [test_pid],
1725 "Wrong read-only project list",
1726 )
1727 self.assertEqual(
1728 db_args[1]["_admin"]["projects_write"],
1729 [test_pid],
1730 "Wrong read-write project list",
1731 )
1733 self.db.get_list.return_value = [db_vnfd_content]
1735 self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
1736 msg_args = self.msg.write.call_args[0]
1737 test_nsd["_id"] = did
1738 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1739 self.assertEqual(msg_args[1], "edited", "Wrong message action")
1740 self.assertEqual(msg_args[2], test_nsd, "Wrong message content")
1742 db_args = self.db.get_one.mock_calls[0][1]
1743 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1744 self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id")
1746 db_args = self.db.replace.call_args[0]
1747 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1748 self.assertEqual(db_args[1], did, "Wrong DB NSD id")
1750 admin = db_args[2]["_admin"]
1751 db_admin = db_nsd_content["_admin"]
1752 self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time")
1753 self.assertGreater(
1754 admin["modified"], db_admin["created"], "Wrong modification time"
1755 )
1756 self.assertEqual(
1757 admin["projects_read"],
1758 db_admin["projects_read"],
1759 "Wrong read-only project list",
1760 )
1761 self.assertEqual(
1762 admin["projects_write"],
1763 db_admin["projects_write"],
1764 "Wrong read-write project list",
1765 )
1766 self.assertEqual(
1767 admin["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1768 )
1769 self.assertEqual(
1770 admin["operationalState"], "ENABLED", "Wrong operational state"
1771 )
1772 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
1774 storage = admin["storage"]
1775 self.assertEqual(storage["folder"], did + ":1", "Wrong storage folder")
1776 self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor")
1778 compare_desc(self, test_nsd, db_args[2], "NSD")
1779 revision_args = self.db.create.call_args[0]
1780 self.assertEqual(
1781 revision_args[0], self.topic.topic + "_revisions", "Wrong topic"
1782 )
1783 self.assertEqual(revision_args[1]["id"], db_args[2]["id"], "Wrong revision id")
1784 self.assertEqual(
1785 revision_args[1]["_id"], db_args[2]["_id"] + ":1", "Wrong revision _id"
1786 )
1788 @patch("osm_nbi.descriptor_topics.shutil")
1789 @patch("osm_nbi.descriptor_topics.os.rename")
1790 def test_new_nsd_check_pyangbind_validation_required_properties(
1791 self, mock_rename, mock_shutil
1792 ):
1793 did, test_nsd = self.prepare_nsd_creation()
1794 del test_nsd["id"]
1796 with self.assertRaises(
1797 EngineException, msg="Accepted NSD with a missing required property"
1798 ) as e:
1799 self.topic.upload_content(
1800 fake_session, did, test_nsd, {}, {"Content-Type": []}
1801 )
1802 self.assertEqual(
1803 e.exception.http_code,
1804 HTTPStatus.UNPROCESSABLE_ENTITY,
1805 "Wrong HTTP status code",
1806 )
1807 self.assertIn(
1808 norm("Error in pyangbind validation: '{}'".format("id")),
1809 norm(str(e.exception)),
1810 "Wrong exception text",
1811 )
1813 @patch("osm_nbi.descriptor_topics.shutil")
1814 @patch("osm_nbi.descriptor_topics.os.rename")
1815 def test_new_nsd_check_pyangbind_validation_additional_properties(
1816 self, mock_rename, mock_shutil
1817 ):
1818 did, test_nsd = self.prepare_nsd_creation()
1819 test_nsd["extra-property"] = 0
1821 with self.assertRaises(
1822 EngineException, msg="Accepted NSD with an additional property"
1823 ) as e:
1824 self.topic.upload_content(
1825 fake_session, did, test_nsd, {}, {"Content-Type": []}
1826 )
1827 self.assertEqual(
1828 e.exception.http_code,
1829 HTTPStatus.UNPROCESSABLE_ENTITY,
1830 "Wrong HTTP status code",
1831 )
1832 self.assertIn(
1833 norm(
1834 "Error in pyangbind validation: {} ({})".format(
1835 "json object contained a key that did not exist", "extra-property"
1836 )
1837 ),
1838 norm(str(e.exception)),
1839 "Wrong exception text",
1840 )
1842 @patch("osm_nbi.descriptor_topics.shutil")
1843 @patch("osm_nbi.descriptor_topics.os.rename")
1844 def test_new_nsd_check_pyangbind_validation_property_types(
1845 self, mock_rename, mock_shutil
1846 ):
1847 did, test_nsd = self.prepare_nsd_creation()
1848 test_nsd["designer"] = {"key": 0}
1850 with self.assertRaises(
1851 EngineException, msg="Accepted NSD with a wrongly typed property"
1852 ) as e:
1853 self.topic.upload_content(
1854 fake_session, did, test_nsd, {}, {"Content-Type": []}
1855 )
1856 self.assertEqual(
1857 e.exception.http_code,
1858 HTTPStatus.UNPROCESSABLE_ENTITY,
1859 "Wrong HTTP status code",
1860 )
1861 self.assertIn(
1862 norm(
1863 "Error in pyangbind validation: {} ({})".format(
1864 "json object contained a key that did not exist", "key"
1865 )
1866 ),
1867 norm(str(e.exception)),
1868 "Wrong exception text",
1869 )
1871 @patch("osm_nbi.descriptor_topics.shutil")
1872 @patch("osm_nbi.descriptor_topics.os.rename")
1873 def test_new_nsd_check_input_validation_mgmt_network_virtual_link_protocol_data(
1874 self, mock_rename, mock_shutil
1875 ):
1876 did, test_nsd = self.prepare_nsd_creation()
1877 df = test_nsd["df"][0]
1878 mgmt_profile = {
1879 "id": "id",
1880 "virtual-link-desc-id": "mgmt",
1881 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1882 }
1883 df["virtual-link-profile"] = [mgmt_profile]
1885 with self.assertRaises(
1886 EngineException, msg="Accepted VLD with mgmt-network+ip-profile"
1887 ) as e:
1888 self.topic.upload_content(
1889 fake_session, did, test_nsd, {}, {"Content-Type": []}
1890 )
1891 self.assertEqual(
1892 e.exception.http_code,
1893 HTTPStatus.UNPROCESSABLE_ENTITY,
1894 "Wrong HTTP status code",
1895 )
1896 self.assertIn(
1897 norm(
1898 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1899 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1900 df["id"], mgmt_profile["id"]
1901 )
1902 ),
1903 norm(str(e.exception)),
1904 "Wrong exception text",
1905 )
1907 @patch("osm_nbi.descriptor_topics.shutil")
1908 @patch("osm_nbi.descriptor_topics.os.rename")
1909 def test_new_nsd_check_descriptor_dependencies_vnfd_id(
1910 self, mock_rename, mock_shutil
1911 ):
1912 did, test_nsd = self.prepare_nsd_creation()
1913 self.db.get_list.return_value = []
1915 with self.assertRaises(
1916 EngineException, msg="Accepted wrong VNFD ID reference"
1917 ) as e:
1918 self.topic.upload_content(
1919 fake_session, did, test_nsd, {}, {"Content-Type": []}
1920 )
1921 self.assertEqual(
1922 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1923 )
1924 self.assertIn(
1925 norm(
1926 "'vnfd-id'='{}' references a non existing vnfd".format(
1927 test_nsd["vnfd-id"][0]
1928 )
1929 ),
1930 norm(str(e.exception)),
1931 "Wrong exception text",
1932 )
1934 @patch("osm_nbi.descriptor_topics.shutil")
1935 @patch("osm_nbi.descriptor_topics.os.rename")
1936 def test_new_nsd_check_descriptor_dependencies_vld_vnfd_connection_point_ref(
1937 self, mock_rename, mock_shutil
1938 ):
1939 # Check Descriptor Dependencies: "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]
1940 did, test_nsd = self.prepare_nsd_creation()
1941 vnfd_descriptor = deepcopy(db_vnfd_content)
1942 df = test_nsd["df"][0]
1943 affected_vnf_profile = df["vnf-profile"][0]
1944 affected_virtual_link = affected_vnf_profile["virtual-link-connectivity"][1]
1945 affected_cpd = vnfd_descriptor["ext-cpd"].pop()
1946 self.db.get_list.return_value = [vnfd_descriptor]
1948 with self.assertRaises(
1949 EngineException, msg="Accepted wrong VLD CP reference"
1950 ) as e:
1951 self.topic.upload_content(
1952 fake_session, did, test_nsd, {}, {"Content-Type": []}
1953 )
1954 self.assertEqual(
1955 e.exception.http_code,
1956 HTTPStatus.UNPROCESSABLE_ENTITY,
1957 "Wrong HTTP status code",
1958 )
1959 self.assertIn(
1960 norm(
1961 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1962 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1963 "non existing ext-cpd:id inside vnfd '{}'".format(
1964 df["id"],
1965 affected_vnf_profile["id"],
1966 affected_virtual_link["virtual-link-profile-id"],
1967 affected_cpd["id"],
1968 vnfd_descriptor["id"],
1969 )
1970 ),
1971 norm(str(e.exception)),
1972 "Wrong exception text",
1973 )
1975 def test_edit_nsd(self):
1976 nsd_content = deepcopy(db_nsd_content)
1977 did = nsd_content["_id"]
1978 self.fs.file_exists.return_value = True
1979 self.fs.dir_ls.return_value = True
1980 with self.subTest(i=1, t="Normal Edition"):
1981 now = time()
1982 self.db.get_one.side_effect = [deepcopy(nsd_content), None]
1983 self.db.get_list.return_value = [db_vnfd_content]
1984 data = {"id": "new-nsd-id", "name": "new-nsd-name"}
1985 self.topic.edit(fake_session, did, data)
1986 db_args = self.db.replace.call_args[0]
1987 msg_args = self.msg.write.call_args[0]
1988 data["_id"] = did
1989 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1990 self.assertEqual(msg_args[1], "edited", "Wrong message action")
1991 self.assertEqual(msg_args[2], data, "Wrong message content")
1992 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1993 self.assertEqual(db_args[1], did, "Wrong DB ID")
1994 self.assertEqual(
1995 db_args[2]["_admin"]["created"],
1996 nsd_content["_admin"]["created"],
1997 "Wrong creation time",
1998 )
1999 self.assertGreater(
2000 db_args[2]["_admin"]["modified"], now, "Wrong modification time"
2001 )
2002 self.assertEqual(
2003 db_args[2]["_admin"]["projects_read"],
2004 nsd_content["_admin"]["projects_read"],
2005 "Wrong read-only project list",
2006 )
2007 self.assertEqual(
2008 db_args[2]["_admin"]["projects_write"],
2009 nsd_content["_admin"]["projects_write"],
2010 "Wrong read-write project list",
2011 )
2012 self.assertEqual(db_args[2]["id"], data["id"], "Wrong NSD ID")
2013 self.assertEqual(db_args[2]["name"], data["name"], "Wrong NSD Name")
2014 with self.subTest(i=2, t="Conflict on Edit"):
2015 data = {"id": "fake-nsd-id", "name": "new-nsd-name"}
2016 self.db.get_one.side_effect = [
2017 nsd_content,
2018 {"_id": str(uuid4()), "id": data["id"]},
2019 ]
2020 with self.assertRaises(
2021 EngineException, msg="Accepted existing NSD ID"
2022 ) as e:
2023 self.topic.edit(fake_session, did, data)
2024 self.assertEqual(
2025 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
2026 )
2027 self.assertIn(
2028 norm(
2029 "{} with id '{}' already exists for this project".format(
2030 "nsd", data["id"]
2031 )
2032 ),
2033 norm(str(e.exception)),
2034 "Wrong exception text",
2035 )
2036 with self.subTest(i=3, t="Check Envelope"):
2037 data = {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
2038 self.db.get_one.side_effect = [nsd_content, None]
2039 with self.assertRaises(
2040 EngineException, msg="Accepted NSD with wrong envelope"
2041 ) as e:
2042 self.topic.edit(fake_session, did, data, content=nsd_content)
2043 self.assertEqual(
2044 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
2045 )
2046 self.assertIn(
2047 "'nsd' must be a list of only one element",
2048 norm(str(e.exception)),
2049 "Wrong exception text",
2050 )
2051 self.db.reset_mock()
2052 return
2054 def test_delete_nsd(self):
2055 did = db_nsd_content["_id"]
2056 self.db.get_one.return_value = db_nsd_content
2057 p_id = db_nsd_content["_admin"]["projects_read"][0]
2058 with self.subTest(i=1, t="Normal Deletion"):
2059 self.db.get_list.return_value = []
2060 self.db.del_one.return_value = {"deleted": 1}
2061 self.topic.delete(fake_session, did)
2062 db_args = self.db.del_one.call_args[0]
2063 msg_args = self.msg.write.call_args[0]
2064 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
2065 self.assertEqual(msg_args[1], "deleted", "Wrong message action")
2066 self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
2067 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
2068 self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
2069 self.assertEqual(
2070 db_args[1]["_admin.projects_write.cont"],
2071 [p_id, "ANY"],
2072 "Wrong DB filter",
2073 )
2074 db_g1_args = self.db.get_one.call_args[0]
2075 self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
2076 self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB NSD ID")
2077 db_gl_calls = self.db.get_list.call_args_list
2078 self.assertEqual(db_gl_calls[0][0][0], "nsrs", "Wrong DB topic")
2079 # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID") # Filter changed after call
2080 self.assertEqual(db_gl_calls[1][0][0], "nsts", "Wrong DB topic")
2081 self.assertEqual(
2082 db_gl_calls[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
2083 db_nsd_content["id"],
2084 "Wrong DB NSD netslice-subnet nsd-ref",
2085 )
2086 self.db.set_one.assert_not_called()
2087 fs_del_calls = self.fs.file_delete.call_args_list
2088 self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
2089 self.assertEqual(fs_del_calls[1][0][0], did + "_", "Wrong FS folder id")
2090 with self.subTest(i=2, t="Conflict on Delete - NSD in use by nsr"):
2091 self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-nsr"}]
2092 with self.assertRaises(
2093 EngineException, msg="Accepted NSD in use by NSR"
2094 ) as e:
2095 self.topic.delete(fake_session, did)
2096 self.assertEqual(
2097 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
2098 )
2099 self.assertIn(
2100 "there is at least one ns instance using this descriptor",
2101 norm(str(e.exception)),
2102 "Wrong exception text",
2103 )
2104 with self.subTest(i=3, t="Conflict on Delete - NSD in use by NST"):
2105 self.db.get_list.side_effect = [
2106 [],
2107 [{"_id": str(uuid4()), "name": "fake-nst"}],
2108 ]
2109 with self.assertRaises(
2110 EngineException, msg="Accepted NSD in use by NST"
2111 ) as e:
2112 self.topic.delete(fake_session, did)
2113 self.assertEqual(
2114 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
2115 )
2116 self.assertIn(
2117 "there is at least one netslice template referencing this descriptor",
2118 norm(str(e.exception)),
2119 "Wrong exception text",
2120 )
2121 with self.subTest(i=4, t="Non-existent NSD"):
2122 excp_msg = "Not found any {} with filter='{}'".format("NSD", {"_id": did})
2123 self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
2124 with self.assertRaises(
2125 DbException, msg="Accepted non-existent NSD ID"
2126 ) as e:
2127 self.topic.delete(fake_session, did)
2128 self.assertEqual(
2129 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
2130 )
2131 self.assertIn(
2132 norm(excp_msg), norm(str(e.exception)), "Wrong exception text"
2133 )
2134 with self.subTest(i=5, t="No delete because referenced by other project"):
2135 db_nsd_content["_admin"]["projects_read"].append("other_project")
2136 self.db.get_one = Mock(return_value=db_nsd_content)
2137 self.db.get_list = Mock(return_value=[])
2138 self.msg.write.reset_mock()
2139 self.db.del_one.reset_mock()
2140 self.fs.file_delete.reset_mock()
2142 self.topic.delete(fake_session, did)
2143 self.db.del_one.assert_not_called()
2144 self.msg.write.assert_not_called()
2145 db_g1_args = self.db.get_one.call_args[0]
2146 self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
2147 self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
2148 db_s1_args = self.db.set_one.call_args
2149 self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
2150 self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
2151 self.assertIn(
2152 p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
2153 )
2154 self.assertIsNone(
2155 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
2156 )
2157 self.assertEqual(
2158 db_s1_args[1]["pull_list"],
2159 {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
2160 "Wrong DB pull_list dictionary",
2161 )
2162 self.fs.file_delete.assert_not_called()
2163 self.db.reset_mock()
2164 return
2166 def prepare_nsd_validation(self):
2167 descriptor_name = "test_ns_descriptor"
2168 self.fs.file_open.side_effect = lambda path, mode: open(
2169 "/tmp/" + str(uuid4()), "a+b"
2170 )
2171 old_nsd, new_nsd = self.create_desc_temp(db_nsd_content)
2172 return descriptor_name, old_nsd, new_nsd
2174 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2175 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2176 def test_validate_descriptor_ns_configuration_changed(
2177 self, mock_safe_load, mock_detect_usage
2178 ):
2179 """Validating NSD and NSD has changes in ns-configuration:config-primitive"""
2180 descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
2181 mock_safe_load.side_effect = [old_nsd, new_nsd]
2182 mock_detect_usage.return_value = True
2183 self.db.get_one.return_value = old_nsd
2184 old_nsd.update(
2185 {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
2186 )
2187 new_nsd.update(
2188 {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
2189 )
2191 with self.assertNotRaises(EngineException):
2192 self.topic._validate_descriptor_changes(
2193 old_nsd["_id"], descriptor_name, "/tmp", "/tmp:1"
2194 )
2195 self.db.get_one.assert_called_once()
2196 mock_detect_usage.assert_called_once()
2197 self.assertEqual(mock_safe_load.call_count, 2)
2199 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2200 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2201 def test_validate_descriptor_nsd_name_changed(
2202 self, mock_safe_load, mock_detect_usage
2203 ):
2204 """Validating NSD, NSD name has changed."""
2205 descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
2206 did = old_nsd["_id"]
2207 new_nsd["name"] = "nscharm-ns2"
2208 mock_safe_load.side_effect = [old_nsd, new_nsd]
2209 mock_detect_usage.return_value = True
2210 self.db.get_one.return_value = old_nsd
2212 with self.assertRaises(
2213 EngineException, msg="there are disallowed changes in the ns descriptor"
2214 ) as e:
2215 self.topic._validate_descriptor_changes(
2216 did, descriptor_name, "/tmp", "/tmp:1"
2217 )
2218 self.assertEqual(
2219 e.exception.http_code,
2220 HTTPStatus.UNPROCESSABLE_ENTITY,
2221 "Wrong HTTP status code",
2222 )
2223 self.assertIn(
2224 norm("there are disallowed changes in the ns descriptor"),
2225 norm(str(e.exception)),
2226 "Wrong exception text",
2227 )
2229 self.db.get_one.assert_called_once()
2230 mock_detect_usage.assert_called_once()
2231 self.assertEqual(mock_safe_load.call_count, 2)
2233 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2234 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2235 def test_validate_descriptor_nsd_name_changed_nsd_not_in_use(
2236 self, mock_safe_load, mock_detect_usage
2237 ):
2238 """Validating NSD, NSD name has changed, NSD is not in use."""
2239 descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
2240 did = old_nsd["_id"]
2241 new_nsd["name"] = "nscharm-ns2"
2242 mock_safe_load.side_effect = [old_nsd, new_nsd]
2243 mock_detect_usage.return_value = None
2244 self.db.get_one.return_value = old_nsd
2246 with self.assertNotRaises(Exception):
2247 self.topic._validate_descriptor_changes(
2248 did, descriptor_name, "/tmp", "/tmp:1"
2249 )
2251 self.db.get_one.assert_called_once()
2252 mock_detect_usage.assert_called_once()
2253 mock_safe_load.assert_not_called()
2255 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
2256 self,
2257 ):
2258 indata = deepcopy(db_nsd_content)
2259 vld = indata["virtual-link-desc"][0]
2260 self.topic.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2261 vld, indata
2262 )
2264 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
2265 self,
2266 ):
2267 indata = deepcopy(db_nsd_content)
2268 vld = indata["virtual-link-desc"][0]
2269 df = indata["df"][0]
2270 affected_vlp = {
2271 "id": "id",
2272 "virtual-link-desc-id": "mgmt",
2273 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
2274 }
2275 df["virtual-link-profile"] = [affected_vlp]
2276 with self.assertRaises(EngineException) as e:
2277 self.topic.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2278 vld, indata
2279 )
2280 self.assertEqual(
2281 e.exception.http_code,
2282 HTTPStatus.UNPROCESSABLE_ENTITY,
2283 "Wrong HTTP status code",
2284 )
2285 self.assertIn(
2286 norm(
2287 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
2288 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
2289 df["id"], affected_vlp["id"]
2290 )
2291 ),
2292 norm(str(e.exception)),
2293 "Wrong exception text",
2294 )
2296 def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self):
2297 indata = deepcopy(db_nsd_content)
2298 self.topic.validate_vnf_profiles_vnfd_id(indata)
2300 def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self):
2301 indata = deepcopy(db_nsd_content)
2302 df = indata["df"][0]
2303 affected_vnf_profile = df["vnf-profile"][0]
2304 indata["vnfd-id"] = ["non-existing-vnfd"]
2305 with self.assertRaises(EngineException) as e:
2306 self.topic.validate_vnf_profiles_vnfd_id(indata)
2307 self.assertEqual(
2308 e.exception.http_code,
2309 HTTPStatus.UNPROCESSABLE_ENTITY,
2310 "Wrong HTTP status code",
2311 )
2312 self.assertIn(
2313 norm(
2314 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
2315 "does not match any vnfd-id".format(
2316 df["id"],
2317 affected_vnf_profile["id"],
2318 affected_vnf_profile["vnfd-id"],
2319 )
2320 ),
2321 norm(str(e.exception)),
2322 "Wrong exception text",
2323 )
2325 def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
2326 self,
2327 ):
2328 nsd_descriptor = deepcopy(db_nsd_content)
2329 vnfd_descriptor = deepcopy(db_vnfd_content)
2330 df = nsd_descriptor["df"][0]
2331 vnfds_index = {vnfd_descriptor["id"]: vnfd_descriptor}
2332 self.topic.validate_df_vnf_profiles_constituent_connection_points(
2333 df, vnfds_index
2334 )
2336 def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
2337 self,
2338 ):
2339 nsd_descriptor = deepcopy(db_nsd_content)
2340 vnfd_descriptor = deepcopy(db_vnfd_content)
2341 df = nsd_descriptor["df"][0]
2342 affected_vnf_profile = df["vnf-profile"][0]
2343 affected_virtual_link = affected_vnf_profile["virtual-link-connectivity"][1]
2344 vnfds_index = {vnfd_descriptor["id"]: vnfd_descriptor}
2345 affected_cpd = vnfd_descriptor["ext-cpd"].pop()
2346 with self.assertRaises(EngineException) as e:
2347 self.topic.validate_df_vnf_profiles_constituent_connection_points(
2348 df, vnfds_index
2349 )
2350 self.assertEqual(
2351 e.exception.http_code,
2352 HTTPStatus.UNPROCESSABLE_ENTITY,
2353 "Wrong HTTP status code",
2354 )
2355 self.assertIn(
2356 norm(
2357 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
2358 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
2359 "non existing ext-cpd:id inside vnfd '{}'".format(
2360 df["id"],
2361 affected_vnf_profile["id"],
2362 affected_virtual_link["virtual-link-profile-id"],
2363 affected_cpd["id"],
2364 vnfd_descriptor["id"],
2365 )
2366 ),
2367 norm(str(e.exception)),
2368 "Wrong exception text",
2369 )
2371 def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self):
2372 nsd_descriptor = deepcopy(db_nsd_content)
2373 invalid_vnfd_id = "invalid-vnfd-id"
2374 nsd_descriptor["id"] = "invalid-vnfd-id-ns"
2375 nsd_descriptor["vnfd-id"][0] = invalid_vnfd_id
2376 nsd_descriptor["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
2377 nsd_descriptor["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
2378 with self.assertRaises(EngineException) as e:
2379 self.db.get_list.return_value = []
2380 nsd_descriptor = self.topic.check_conflict_on_edit(
2381 fake_session, nsd_descriptor, [], "id"
2382 )
2383 self.assertEqual(
2384 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
2385 )
2386 self.assertIn(
2387 norm(
2388 "Descriptor error at 'vnfd-id'='{}' references a non "
2389 "existing vnfd".format(invalid_vnfd_id)
2390 ),
2391 norm(str(e.exception)),
2392 "Wrong exception text",
2393 )
2395 def test_validate_vnffgd_descriptor_on_valid_descriptor(self):
2396 indata = yaml.safe_load(db_sfc_nsds_text)[0]
2397 vnffgd = indata.get("vnffgd")
2398 fg = vnffgd[0]
2399 self.topic.validate_vnffgd_data(fg, indata)
2401 def test_validate_vnffgd_descriptor_not_matching_nfp_position_element(self):
2402 indata = yaml.safe_load(db_sfc_nsds_text)[0]
2403 vnffgd = indata.get("vnffgd")
2404 fg = vnffgd[0]
2405 nfpd = fg.get("nfpd")[0]
2406 with self.assertRaises(EngineException) as e:
2407 fg.update({"nfp-position-element": [{"id": "test1"}]})
2408 self.topic.validate_vnffgd_data(fg, indata)
2409 self.assertEqual(
2410 e.exception.http_code,
2411 HTTPStatus.UNPROCESSABLE_ENTITY,
2412 "Wrong HTTP status code",
2413 )
2414 self.assertIn(
2415 norm(
2416 "Error at vnffgd nfpd[id='{}']:nfp-position-element-id='{}' "
2417 "does not match any nfp-position-element".format(nfpd["id"], "test")
2418 ),
2419 norm(str(e.exception)),
2420 "Wrong exception text",
2421 )
2423 def test_validate_vnffgd_descriptor_not_matching_constituent_base_element_id(
2424 self,
2425 ):
2426 indata = yaml.safe_load(db_sfc_nsds_text)[0]
2427 vnffgd = indata.get("vnffgd")
2428 fg = vnffgd[0]
2429 fg["nfpd"][0]["position-desc-id"][0]["cp-profile-id"][0][
2430 "constituent-profile-elements"
2431 ][0]["constituent-base-element-id"] = "error_vnf"
2432 with self.assertRaises(EngineException) as e:
2433 self.topic.validate_vnffgd_data(fg, indata)
2434 self.assertEqual(
2435 e.exception.http_code,
2436 HTTPStatus.UNPROCESSABLE_ENTITY,
2437 "Wrong HTTP status code",
2438 )
2439 self.assertIn(
2440 norm(
2441 "Error at vnffgd constituent_profile[id='{}']:vnfd-id='{}' "
2442 "does not match any constituent-base-element-id".format(
2443 "vnf1", "error_vnf"
2444 )
2445 ),
2446 norm(str(e.exception)),
2447 "Wrong exception text",
2448 )
2451if __name__ == "__main__":
2452 unittest.main()