27c0bd3d9a1ffb58deb824d404d6a4a1b282c8bb
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
8 # http://www.apache.org/licenses/LICENSE-2.0
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
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"
20 from contextlib
import contextmanager
22 from unittest
import TestCase
23 from unittest
.mock
import Mock
, patch
24 from uuid
import uuid4
25 from http
import HTTPStatus
26 from copy
import deepcopy
28 from osm_common
import dbbase
, fsbase
, msgbase
29 from osm_nbi
import authconn
30 from osm_nbi
.tests
.test_pkg_descriptors
import db_vnfds_text
, db_nsds_text
31 from osm_nbi
.descriptor_topics
import VnfdTopic
, NsdTopic
32 from osm_nbi
.engine
import EngineException
33 from osm_common
.dbbase
import DbException
36 test_name
= "test-user"
37 db_vnfd_content
= yaml
.load(db_vnfds_text
, Loader
=yaml
.Loader
)[0]
38 db_nsd_content
= yaml
.load(db_nsds_text
, Loader
=yaml
.Loader
)[0]
39 test_pid
= db_vnfd_content
["_admin"]["projects_read"][0]
41 "username": test_name
,
42 "project_id": (test_pid
,),
47 "allow_show_user_project_role": True,
52 """Normalize string for checking"""
53 return " ".join(str.strip().split()).lower()
56 def compare_desc(tc
, d1
, d2
, k
):
58 Compare two descriptors
59 We need this function because some methods are adding/removing items to/from the descriptors
60 before they are stored in the database, so the original and stored versions will differ
61 What we check is that COMMON LEAF ITEMS are equal
62 Lists of different length are not compared
63 :param tc: Test Case wich provides context (in particular the assert* methods)
64 :param d1,d2: Descriptors to be compared
65 :param k: key/item being compared
68 if isinstance(d1
, dict) and isinstance(d2
, dict):
71 compare_desc(tc
, d1
[key
], d2
[key
], k
+ "[{}]".format(key
))
72 elif isinstance(d1
, list) and isinstance(d2
, list) and len(d1
) == len(d2
):
73 for i
in range(len(d1
)):
74 compare_desc(tc
, d1
[i
], d2
[i
], k
+ "[{}]".format(i
))
76 tc
.assertEqual(d1
, d2
, "Wrong descriptor content: {}".format(k
))
79 class Test_VnfdTopic(TestCase
):
82 cls
.test_name
= "test-vnfd-topic"
85 def tearDownClass(cls
):
89 self
.db
= Mock(dbbase
.DbBase())
90 self
.fs
= Mock(fsbase
.FsBase())
91 self
.msg
= Mock(msgbase
.MsgBase())
92 self
.auth
= Mock(authconn
.Authconn(None, None, None))
93 self
.topic
= VnfdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
94 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
97 def assertNotRaises(self
, exception_type
):
100 except exception_type
:
101 raise self
.failureException("{} raised".format(exception_type
.__name
__))
103 def create_desc_temp(self
, template
):
104 old_desc
= deepcopy(template
)
105 new_desc
= deepcopy(template
)
106 return old_desc
, new_desc
108 def prepare_vnfd_creation(self
):
110 self
.fs
.get_params
.return_value
= {}
111 self
.fs
.file_exists
.return_value
= False
112 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
113 "/tmp/" + str(uuid4()), "a+b"
115 test_vnfd
= deepcopy(db_vnfd_content
)
116 did
= db_vnfd_content
["_id"]
117 self
.db
.create
.return_value
= did
118 self
.db
.get_one
.side_effect
= [
119 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
122 return did
, test_vnfd
124 def prepare_test_vnfd(self
, test_vnfd
):
126 del test_vnfd
["_admin"]
127 del test_vnfd
["vdu"][0]["cloud-init-file"]
128 del test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
130 ][0]["execution-environment-list"][0]["juju"]
133 @patch("osm_nbi.descriptor_topics.shutil")
134 @patch("osm_nbi.descriptor_topics.os.rename")
135 def test_new_vnfd_normal_creation(self
, mock_rename
, mock_shutil
):
136 did
, test_vnfd
= self
.prepare_vnfd_creation()
137 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
139 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
140 db_args
= self
.db
.create
.call_args
[0]
141 msg_args
= self
.msg
.write
.call_args
[0]
143 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
144 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
145 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
146 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
147 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
148 self
.assertEqual(did2
, did
, "Wrong DB VNFD id")
149 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
151 db_args
[1]["_admin"]["modified"],
152 db_args
[1]["_admin"]["created"],
153 "Wrong modification time",
156 db_args
[1]["_admin"]["projects_read"],
158 "Wrong read-only project list",
161 db_args
[1]["_admin"]["projects_write"],
163 "Wrong read-write project list",
166 self
.db
.get_one
.side_effect
= [
167 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
171 self
.topic
.upload_content(
172 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
174 msg_args
= self
.msg
.write
.call_args
[0]
175 test_vnfd
["_id"] = did
176 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
177 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
178 self
.assertEqual(msg_args
[2], test_vnfd
, "Wrong message content")
180 db_args
= self
.db
.get_one
.mock_calls
[0][1]
181 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
182 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB VNFD id")
184 db_args
= self
.db
.replace
.call_args
[0]
185 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
186 self
.assertEqual(db_args
[1], did
, "Wrong DB VNFD id")
188 admin
= db_args
[2]["_admin"]
189 db_admin
= deepcopy(db_vnfd_content
["_admin"])
190 self
.assertEqual(admin
["type"], "vnfd", "Wrong descriptor type")
191 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
193 admin
["modified"], db_admin
["created"], "Wrong modification time"
196 admin
["projects_read"],
197 db_admin
["projects_read"],
198 "Wrong read-only project list",
201 admin
["projects_write"],
202 db_admin
["projects_write"],
203 "Wrong read-write project list",
206 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
209 admin
["operationalState"], "ENABLED", "Wrong operational state"
211 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
213 storage
= admin
["storage"]
214 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
215 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
216 self
.assertEqual(admin
["revision"], 1, "Wrong revision number")
217 compare_desc(self
, test_vnfd
, db_args
[2], "VNFD")
219 @patch("osm_nbi.descriptor_topics.shutil")
220 @patch("osm_nbi.descriptor_topics.os.rename")
221 def test_new_vnfd_check_pyangbind_validation_additional_properties(
222 self
, mock_rename
, mock_shutil
224 did
, test_vnfd
= self
.prepare_vnfd_creation()
225 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
226 self
.topic
.upload_content(
227 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
229 test_vnfd
["_id"] = did
230 test_vnfd
["extra-property"] = 0
231 self
.db
.get_one
.side_effect
= lambda table
, filter, fail_on_empty
=None, fail_on_more
=None: {
233 "_admin": deepcopy(db_vnfd_content
["_admin"]),
236 with self
.assertRaises(
237 EngineException
, msg
="Accepted VNFD with an additional property"
239 self
.topic
.upload_content(
240 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
243 e
.exception
.http_code
,
244 HTTPStatus
.UNPROCESSABLE_ENTITY
,
245 "Wrong HTTP status code",
249 "Error in pyangbind validation: {} ({})".format(
250 "json object contained a key that did not exist", "extra-property"
253 norm(str(e
.exception
)),
254 "Wrong exception text",
256 db_args
= self
.db
.replace
.call_args
[0]
257 admin
= db_args
[2]["_admin"]
258 self
.assertEqual(admin
["revision"], 1, "Wrong revision number")
260 @patch("osm_nbi.descriptor_topics.shutil")
261 @patch("osm_nbi.descriptor_topics.os.rename")
262 def test_new_vnfd_check_pyangbind_validation_property_types(
263 self
, mock_rename
, mock_shutil
265 did
, test_vnfd
= self
.prepare_vnfd_creation()
266 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
267 test_vnfd
["_id"] = did
268 test_vnfd
["product-name"] = {"key": 0}
270 with self
.assertRaises(
271 EngineException
, msg
="Accepted VNFD with a wrongly typed property"
273 self
.topic
.upload_content(
274 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
277 e
.exception
.http_code
,
278 HTTPStatus
.UNPROCESSABLE_ENTITY
,
279 "Wrong HTTP status code",
283 "Error in pyangbind validation: {} ({})".format(
284 "json object contained a key that did not exist", "key"
287 norm(str(e
.exception
)),
288 "Wrong exception text",
291 @patch("osm_nbi.descriptor_topics.shutil")
292 @patch("osm_nbi.descriptor_topics.os.rename")
293 def test_new_vnfd_check_input_validation_cloud_init(self
, mock_rename
, mock_shutil
):
294 did
, test_vnfd
= self
.prepare_vnfd_creation()
295 del test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
297 ][0]["execution-environment-list"][0]["juju"]
299 with self
.assertRaises(
300 EngineException
, msg
="Accepted non-existent cloud_init file"
302 self
.topic
.upload_content(
303 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
306 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
310 "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format(
311 "cloud-init", test_vnfd
["id"], test_vnfd
["vdu"][0]["id"]
314 norm(str(e
.exception
)),
315 "Wrong exception text",
318 @patch("osm_nbi.descriptor_topics.shutil")
319 @patch("osm_nbi.descriptor_topics.os.rename")
320 def test_new_vnfd_check_input_validation_day12_configuration(
321 self
, mock_rename
, mock_shutil
323 did
, test_vnfd
= self
.prepare_vnfd_creation()
324 del test_vnfd
["vdu"][0]["cloud-init-file"]
326 with self
.assertRaises(
327 EngineException
, msg
="Accepted non-existent charm in VNF configuration"
329 self
.topic
.upload_content(
330 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
333 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
337 "{} defined in vnf[id={}] but not present in package".format(
338 "charm", test_vnfd
["id"]
341 norm(str(e
.exception
)),
342 "Wrong exception text",
345 @patch("osm_nbi.descriptor_topics.shutil")
346 @patch("osm_nbi.descriptor_topics.os.rename")
347 def test_new_vnfd_check_input_validation_mgmt_cp(self
, mock_rename
, mock_shutil
):
348 did
, test_vnfd
= self
.prepare_vnfd_creation()
349 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
350 del test_vnfd
["mgmt-cp"]
352 with self
.assertRaises(
353 EngineException
, msg
="Accepted VNFD without management interface"
355 self
.topic
.upload_content(
356 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
359 e
.exception
.http_code
,
360 HTTPStatus
.UNPROCESSABLE_ENTITY
,
361 "Wrong HTTP status code",
364 norm("'{}' is a mandatory field and it is not defined".format("mgmt-cp")),
365 norm(str(e
.exception
)),
366 "Wrong exception text",
369 @patch("osm_nbi.descriptor_topics.shutil")
370 @patch("osm_nbi.descriptor_topics.os.rename")
371 def test_new_vnfd_check_input_validation_mgmt_cp_connection_point(
372 self
, mock_rename
, mock_shutil
374 did
, test_vnfd
= self
.prepare_vnfd_creation()
375 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
376 test_vnfd
["mgmt-cp"] = "wrong-cp"
378 with self
.assertRaises(
379 EngineException
, msg
="Accepted wrong mgmt-cp connection point"
381 self
.topic
.upload_content(
382 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
385 e
.exception
.http_code
,
386 HTTPStatus
.UNPROCESSABLE_ENTITY
,
387 "Wrong HTTP status code",
391 "mgmt-cp='{}' must match an existing ext-cpd".format(
395 norm(str(e
.exception
)),
396 "Wrong exception text",
399 @patch("osm_nbi.descriptor_topics.shutil")
400 @patch("osm_nbi.descriptor_topics.os.rename")
401 def test_new_vnfd_check_input_validation_vdu_int_cpd(
402 self
, mock_rename
, mock_shutil
404 did
, test_vnfd
= self
.prepare_vnfd_creation()
405 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
406 ext_cpd
= test_vnfd
["ext-cpd"][1]
407 ext_cpd
["int-cpd"]["cpd"] = "wrong-cpd"
409 with self
.assertRaises(
410 EngineException
, msg
="Accepted wrong ext-cpd internal connection point"
412 self
.topic
.upload_content(
413 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
416 e
.exception
.http_code
,
417 HTTPStatus
.UNPROCESSABLE_ENTITY
,
418 "Wrong HTTP status code",
422 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
426 norm(str(e
.exception
)),
427 "Wrong exception text",
430 @patch("osm_nbi.descriptor_topics.shutil")
431 @patch("osm_nbi.descriptor_topics.os.rename")
432 def test_new_vnfd_check_input_validation_duplicated_vld(
433 self
, mock_rename
, mock_shutil
435 did
, test_vnfd
= self
.prepare_vnfd_creation()
436 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
437 test_vnfd
["int-virtual-link-desc"].insert(0, {"id": "internal"})
439 with self
.assertRaises(
440 EngineException
, msg
="Accepted duplicated VLD name"
442 self
.topic
.upload_content(
443 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
446 e
.exception
.http_code
,
447 HTTPStatus
.UNPROCESSABLE_ENTITY
,
448 "Wrong HTTP status code",
452 "identifier id '{}' is not unique".format(
453 test_vnfd
["int-virtual-link-desc"][0]["id"]
456 norm(str(e
.exception
)),
457 "Wrong exception text",
460 @patch("osm_nbi.descriptor_topics.shutil")
461 @patch("osm_nbi.descriptor_topics.os.rename")
462 def test_new_vnfd_check_input_validation_vdu_int_virtual_link_desc(
463 self
, mock_rename
, mock_shutil
465 did
, test_vnfd
= self
.prepare_vnfd_creation()
466 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
467 vdu
= test_vnfd
["vdu"][0]
468 int_cpd
= vdu
["int-cpd"][1]
469 int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
471 with self
.assertRaises(
472 EngineException
, msg
="Accepted int-virtual-link-desc"
474 self
.topic
.upload_content(
475 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
478 e
.exception
.http_code
,
479 HTTPStatus
.UNPROCESSABLE_ENTITY
,
480 "Wrong HTTP status code",
484 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
485 "int-virtual-link-desc".format(
486 vdu
["id"], int_cpd
["id"], int_cpd
["int-virtual-link-desc"]
489 norm(str(e
.exception
)),
490 "Wrong exception text",
493 @patch("osm_nbi.descriptor_topics.shutil")
494 @patch("osm_nbi.descriptor_topics.os.rename")
495 def test_new_vnfd_check_input_validation_virtual_link_profile(
496 self
, mock_rename
, mock_shutil
498 did
, test_vnfd
= self
.prepare_vnfd_creation()
499 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
500 fake_ivld_profile
= {"id": "fake-profile-ref", "flavour": "fake-flavour"}
501 df
= test_vnfd
["df"][0]
502 df
["virtual-link-profile"] = [fake_ivld_profile
]
504 with self
.assertRaises(
505 EngineException
, msg
="Accepted non-existent Profile Ref"
507 self
.topic
.upload_content(
508 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
511 e
.exception
.http_code
,
512 HTTPStatus
.UNPROCESSABLE_ENTITY
,
513 "Wrong HTTP status code",
517 "df[id='{}']:virtual-link-profile='{}' must match an existing "
518 "int-virtual-link-desc".format(df
["id"], fake_ivld_profile
["id"])
520 norm(str(e
.exception
)),
521 "Wrong exception text",
524 @patch("osm_nbi.descriptor_topics.shutil")
525 @patch("osm_nbi.descriptor_topics.os.rename")
526 def test_new_vnfd_check_input_validation_scaling_criteria_monitoring_param_ref(
527 self
, mock_rename
, mock_shutil
529 did
, test_vnfd
= self
.prepare_vnfd_creation()
530 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
531 vdu
= test_vnfd
["vdu"][1]
532 affected_df
= test_vnfd
["df"][0]
533 sa
= affected_df
["scaling-aspect"][0]
534 sp
= sa
["scaling-policy"][0]
535 sc
= sp
["scaling-criteria"][0]
536 vdu
.pop("monitoring-parameter")
538 with self
.assertRaises(
539 EngineException
, msg
="Accepted non-existent Scaling Group Policy Criteria"
541 self
.topic
.upload_content(
542 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
545 e
.exception
.http_code
,
546 HTTPStatus
.UNPROCESSABLE_ENTITY
,
547 "Wrong HTTP status code",
551 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
552 "[name='{}']:scaling-criteria[name='{}']: "
553 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
558 sc
["vnf-monitoring-param-ref"],
561 norm(str(e
.exception
)),
562 "Wrong exception text",
565 @patch("osm_nbi.descriptor_topics.shutil")
566 @patch("osm_nbi.descriptor_topics.os.rename")
567 def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration(
568 self
, mock_rename
, mock_shutil
570 did
, test_vnfd
= self
.prepare_vnfd_creation()
571 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
572 test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
575 df
= test_vnfd
["df"][0]
577 with self
.assertRaises(
578 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
580 self
.topic
.upload_content(
581 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
584 e
.exception
.http_code
,
585 HTTPStatus
.UNPROCESSABLE_ENTITY
,
586 "Wrong HTTP status code",
590 "'day1-2 configuration' not defined in the descriptor but it is referenced "
591 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
592 df
["id"], df
["scaling-aspect"][0]["id"]
595 norm(str(e
.exception
)),
596 "Wrong exception text",
599 @patch("osm_nbi.descriptor_topics.shutil")
600 @patch("osm_nbi.descriptor_topics.os.rename")
601 def test_new_vnfd_check_input_validation_scaling_config_action(
602 self
, mock_rename
, mock_shutil
604 did
, test_vnfd
= self
.prepare_vnfd_creation()
605 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
606 df
= test_vnfd
["df"][0]
607 affected_df
= test_vnfd
["df"][0]
608 sa
= affected_df
["scaling-aspect"][0]
609 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
610 "operate-vnf-op-config"
611 )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
613 with self
.assertRaises(
614 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
616 self
.topic
.upload_content(
617 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
620 e
.exception
.http_code
,
621 HTTPStatus
.UNPROCESSABLE_ENTITY
,
622 "Wrong HTTP status code",
626 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
627 "config-primitive-name-ref='{}' does not match any "
628 "day1-2 configuration:config-primitive:name".format(
630 df
["scaling-aspect"][0]["id"],
631 sa
["scaling-config-action"][0]["vnf-config-primitive-name-ref"],
634 norm(str(e
.exception
)),
635 "Wrong exception text",
638 @patch("osm_nbi.descriptor_topics.shutil")
639 @patch("osm_nbi.descriptor_topics.os.rename")
640 def test_new_vnfd_check_input_validation_everything_right(
641 self
, mock_rename
, mock_shutil
643 did
, test_vnfd
= self
.prepare_vnfd_creation()
644 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
645 test_vnfd
["id"] = "fake-vnfd-id"
646 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
647 "operate-vnf-op-config"
648 )["day1-2"][0]["id"] = "fake-vnfd-id"
649 self
.db
.get_one
.side_effect
= [
650 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
653 rc
= self
.topic
.upload_content(
654 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
656 self
.assertTrue(rc
, "Input Validation: Unexpected failure")
658 def test_edit_vnfd(self
):
659 vnfd_content
= deepcopy(db_vnfd_content
)
660 did
= vnfd_content
["_id"]
661 self
.fs
.file_exists
.return_value
= True
662 self
.fs
.dir_ls
.return_value
= True
663 with self
.subTest(i
=1, t
="Normal Edition"):
665 self
.db
.get_one
.side_effect
= [deepcopy(vnfd_content
), None]
666 data
= {"product-name": "new-vnfd-name"}
667 self
.topic
.edit(fake_session
, did
, data
)
668 db_args
= self
.db
.replace
.call_args
[0]
669 msg_args
= self
.msg
.write
.call_args
[0]
671 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
672 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
673 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
674 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
675 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
677 db_args
[2]["_admin"]["created"],
678 vnfd_content
["_admin"]["created"],
679 "Wrong creation time",
682 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
685 db_args
[2]["_admin"]["projects_read"],
686 vnfd_content
["_admin"]["projects_read"],
687 "Wrong read-only project list",
690 db_args
[2]["_admin"]["projects_write"],
691 vnfd_content
["_admin"]["projects_write"],
692 "Wrong read-write project list",
695 db_args
[2]["product-name"], data
["product-name"], "Wrong VNFD Name"
697 with self
.subTest(i
=2, t
="Conflict on Edit"):
698 data
= {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
699 self
.db
.get_one
.side_effect
= [
700 deepcopy(vnfd_content
),
701 {"_id": str(uuid4()), "id": data
["id"]},
703 with self
.assertRaises(
704 EngineException
, msg
="Accepted existing VNFD ID"
706 self
.topic
.edit(fake_session
, did
, data
)
708 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
712 "{} with id '{}' already exists for this project".format(
716 norm(str(e
.exception
)),
717 "Wrong exception text",
719 with self
.subTest(i
=3, t
="Check Envelope"):
720 data
= {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
721 with self
.assertRaises(
722 EngineException
, msg
="Accepted VNFD with wrong envelope"
724 self
.topic
.edit(fake_session
, did
, data
, content
=vnfd_content
)
726 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
729 "'vnfd' must be dict", norm(str(e
.exception
)), "Wrong exception text"
733 def test_delete_vnfd(self
):
734 did
= db_vnfd_content
["_id"]
735 self
.db
.get_one
.return_value
= db_vnfd_content
736 p_id
= db_vnfd_content
["_admin"]["projects_read"][0]
737 with self
.subTest(i
=1, t
="Normal Deletion"):
738 self
.db
.get_list
.return_value
= []
739 self
.db
.del_one
.return_value
= {"deleted": 1}
740 self
.topic
.delete(fake_session
, did
)
741 db_args
= self
.db
.del_one
.call_args
[0]
742 msg_args
= self
.msg
.write
.call_args
[0]
743 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
744 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
745 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
746 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
747 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
749 db_args
[1]["_admin.projects_write.cont"],
753 db_g1_args
= self
.db
.get_one
.call_args
[0]
754 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
755 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
756 db_gl_calls
= self
.db
.get_list
.call_args_list
757 self
.assertEqual(db_gl_calls
[0][0][0], "vnfrs", "Wrong DB topic")
758 # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID") # Filter changed after call
759 self
.assertEqual(db_gl_calls
[1][0][0], "nsds", "Wrong DB topic")
761 db_gl_calls
[1][0][1]["vnfd-id"],
762 db_vnfd_content
["id"],
763 "Wrong DB NSD vnfd-id",
766 self
.db
.del_list
.call_args
[0]
768 self
.db
.del_list
.call_args
[0][0],
769 self
.topic
.topic
+ "_revisions",
774 self
.db
.del_list
.call_args
[0][1]["_id"]["$regex"],
776 "Wrong ID for rexep delete",
779 self
.db
.set_one
.assert_not_called()
780 fs_del_calls
= self
.fs
.file_delete
.call_args_list
781 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
782 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
783 with self
.subTest(i
=2, t
="Conflict on Delete - VNFD in use by VNFR"):
784 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-vnfr"}]
785 with self
.assertRaises(
786 EngineException
, msg
="Accepted VNFD in use by VNFR"
788 self
.topic
.delete(fake_session
, did
)
790 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
793 "there is at least one vnf instance using this descriptor",
794 norm(str(e
.exception
)),
795 "Wrong exception text",
797 with self
.subTest(i
=3, t
="Conflict on Delete - VNFD in use by NSD"):
798 self
.db
.get_list
.side_effect
= [
800 [{"_id": str(uuid4()), "name": "fake-nsd"}],
802 with self
.assertRaises(
803 EngineException
, msg
="Accepted VNFD in use by NSD"
805 self
.topic
.delete(fake_session
, did
)
807 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
810 "there is at least one ns package referencing this descriptor",
811 norm(str(e
.exception
)),
812 "Wrong exception text",
814 with self
.subTest(i
=4, t
="Non-existent VNFD"):
815 excp_msg
= "Not found any {} with filter='{}'".format("VNFD", {"_id": did
})
816 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
817 with self
.assertRaises(
818 DbException
, msg
="Accepted non-existent VNFD ID"
820 self
.topic
.delete(fake_session
, did
)
822 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
825 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
827 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
828 db_vnfd_content
["_admin"]["projects_read"].append("other_project")
829 self
.db
.get_one
= Mock(return_value
=db_vnfd_content
)
830 self
.db
.get_list
= Mock(return_value
=[])
831 self
.msg
.write
.reset_mock()
832 self
.db
.del_one
.reset_mock()
833 self
.fs
.file_delete
.reset_mock()
835 self
.topic
.delete(fake_session
, did
)
836 self
.db
.del_one
.assert_not_called()
837 self
.msg
.write
.assert_not_called()
838 db_g1_args
= self
.db
.get_one
.call_args
[0]
839 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
840 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
841 db_s1_args
= self
.db
.set_one
.call_args
842 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
843 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
845 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
848 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
851 db_s1_args
[1]["pull_list"],
852 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
853 "Wrong DB pull_list dictionary",
855 self
.fs
.file_delete
.assert_not_called()
858 def prepare_vnfd_validation(self
):
859 descriptor_name
= "test_descriptor"
860 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
861 "/tmp/" + str(uuid4()), "a+b"
863 old_vnfd
, new_vnfd
= self
.create_desc_temp(db_vnfd_content
)
864 return descriptor_name
, old_vnfd
, new_vnfd
866 @patch("osm_nbi.descriptor_topics.yaml")
867 def test_validate_vnfd_changes_day12_config_primitive_changed(self
, mock_yaml
):
868 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
869 new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
871 ][0]["config-primitive"][0]["name"] = "new_action"
872 mock_yaml
.load
.side_effect
= [old_vnfd
, new_vnfd
]
873 with self
.assertNotRaises(EngineException
):
874 self
.topic
._validate
_descriptor
_changes
(descriptor_name
, "/tmp/", "/tmp:1/")
876 @patch("osm_nbi.descriptor_topics.yaml")
877 def test_validate_vnfd_changes_sw_version_changed(self
, mock_yaml
):
878 # old vnfd uses the default software version: 1.0
879 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
880 new_vnfd
["software-version"] = "1.3"
881 new_vnfd
["sw-image-desc"][0]["name"] = "new-image"
882 mock_yaml
.load
.side_effect
= [old_vnfd
, new_vnfd
]
883 with self
.assertNotRaises(EngineException
):
884 self
.topic
._validate
_descriptor
_changes
(descriptor_name
, "/tmp/", "/tmp:1/")
886 @patch("osm_nbi.descriptor_topics.yaml")
887 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed(
890 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
891 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
892 mock_yaml
.load
.side_effect
= [old_vnfd
, new_vnfd
]
893 with self
.assertRaises(EngineException
) as e
:
894 self
.topic
._validate
_descriptor
_changes
(descriptor_name
, "/tmp/", "/tmp:1/")
896 e
.exception
.http_code
,
897 HTTPStatus
.UNPROCESSABLE_ENTITY
,
898 "Wrong HTTP status code",
901 norm("there are disallowed changes in the vnf descriptor"),
902 norm(str(e
.exception
)),
903 "Wrong exception text",
906 def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self
):
907 indata
= deepcopy(db_vnfd_content
)
908 self
.topic
.validate_mgmt_interface_connection_point(indata
)
910 def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
913 indata
= deepcopy(db_vnfd_content
)
914 indata
["ext-cpd"] = []
915 with self
.assertRaises(EngineException
) as e
:
916 self
.topic
.validate_mgmt_interface_connection_point(indata
)
918 e
.exception
.http_code
,
919 HTTPStatus
.UNPROCESSABLE_ENTITY
,
920 "Wrong HTTP status code",
924 "mgmt-cp='{}' must match an existing ext-cpd".format(indata
["mgmt-cp"])
926 norm(str(e
.exception
)),
927 "Wrong exception text",
930 def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self
):
931 indata
= deepcopy(db_vnfd_content
)
932 indata
.pop("mgmt-cp")
933 with self
.assertRaises(EngineException
) as e
:
934 self
.topic
.validate_mgmt_interface_connection_point(indata
)
936 e
.exception
.http_code
,
937 HTTPStatus
.UNPROCESSABLE_ENTITY
,
938 "Wrong HTTP status code",
941 norm("'mgmt-cp' is a mandatory field and it is not defined"),
942 norm(str(e
.exception
)),
943 "Wrong exception text",
946 def test_validate_vdu_internal_connection_points_on_valid_descriptor(self
):
947 indata
= db_vnfd_content
948 vdu
= indata
["vdu"][0]
949 self
.topic
.validate_vdu_internal_connection_points(vdu
)
951 def test_validate_external_connection_points_on_valid_descriptor(self
):
952 indata
= db_vnfd_content
953 self
.topic
.validate_external_connection_points(indata
)
955 def test_validate_external_connection_points_when_missing_internal_connection_point(
958 indata
= deepcopy(db_vnfd_content
)
959 vdu
= indata
["vdu"][0]
961 affected_ext_cpd
= indata
["ext-cpd"][0]
962 with self
.assertRaises(EngineException
) as e
:
963 self
.topic
.validate_external_connection_points(indata
)
965 e
.exception
.http_code
,
966 HTTPStatus
.UNPROCESSABLE_ENTITY
,
967 "Wrong HTTP status code",
971 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
972 affected_ext_cpd
["id"]
975 norm(str(e
.exception
)),
976 "Wrong exception text",
979 def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
982 indata
= deepcopy(db_vnfd_content
)
983 vdu
= indata
["vdu"][0]
987 "virtual-network-interface-requirement": [{"name": "duplicated"}],
989 vdu
["int-cpd"].insert(0, duplicated_cpd
)
990 with self
.assertRaises(EngineException
) as e
:
991 self
.topic
.validate_vdu_internal_connection_points(vdu
)
993 e
.exception
.http_code
,
994 HTTPStatus
.UNPROCESSABLE_ENTITY
,
995 "Wrong HTTP status code",
999 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
1000 vdu
["id"], duplicated_cpd
["id"]
1003 norm(str(e
.exception
)),
1004 "Wrong exception text",
1007 def test_validate_external_connection_points_on_duplicated_external_connection_point(
1010 indata
= deepcopy(db_vnfd_content
)
1012 "id": "vnf-mgmt-ext",
1013 "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
1015 indata
["ext-cpd"].insert(0, duplicated_cpd
)
1016 with self
.assertRaises(EngineException
) as e
:
1017 self
.topic
.validate_external_connection_points(indata
)
1019 e
.exception
.http_code
,
1020 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1021 "Wrong HTTP status code",
1025 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
1026 duplicated_cpd
["id"]
1029 norm(str(e
.exception
)),
1030 "Wrong exception text",
1033 def test_validate_internal_virtual_links_on_valid_descriptor(self
):
1034 indata
= db_vnfd_content
1035 self
.topic
.validate_internal_virtual_links(indata
)
1037 def test_validate_internal_virtual_links_on_duplicated_ivld(self
):
1038 indata
= deepcopy(db_vnfd_content
)
1039 duplicated_vld
= {"id": "internal"}
1040 indata
["int-virtual-link-desc"].insert(0, duplicated_vld
)
1041 with self
.assertRaises(EngineException
) as e
:
1042 self
.topic
.validate_internal_virtual_links(indata
)
1044 e
.exception
.http_code
,
1045 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1046 "Wrong HTTP status code",
1050 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
1051 duplicated_vld
["id"]
1054 norm(str(e
.exception
)),
1055 "Wrong exception text",
1058 def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
1061 indata
= deepcopy(db_vnfd_content
)
1062 vdu
= indata
["vdu"][0]
1063 affected_int_cpd
= vdu
["int-cpd"][0]
1064 affected_int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
1065 with self
.assertRaises(EngineException
) as e
:
1066 self
.topic
.validate_internal_virtual_links(indata
)
1068 e
.exception
.http_code
,
1069 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1070 "Wrong HTTP status code",
1074 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
1075 "int-virtual-link-desc".format(
1077 affected_int_cpd
["id"],
1078 affected_int_cpd
["int-virtual-link-desc"],
1081 norm(str(e
.exception
)),
1082 "Wrong exception text",
1085 def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self
):
1086 indata
= deepcopy(db_vnfd_content
)
1087 affected_ivld_profile
= {"id": "non-existing-int-virtual-link-desc"}
1088 df
= indata
["df"][0]
1089 df
["virtual-link-profile"] = [affected_ivld_profile
]
1090 with self
.assertRaises(EngineException
) as e
:
1091 self
.topic
.validate_internal_virtual_links(indata
)
1093 e
.exception
.http_code
,
1094 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1095 "Wrong HTTP status code",
1099 "df[id='{}']:virtual-link-profile='{}' must match an existing "
1100 "int-virtual-link-desc".format(df
["id"], affected_ivld_profile
["id"])
1102 norm(str(e
.exception
)),
1103 "Wrong exception text",
1106 def test_validate_monitoring_params_on_valid_descriptor(self
):
1107 indata
= db_vnfd_content
1108 self
.topic
.validate_monitoring_params(indata
)
1110 def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self
):
1111 indata
= deepcopy(db_vnfd_content
)
1112 duplicated_mp
= {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1113 affected_ivld
= indata
["int-virtual-link-desc"][0]
1114 affected_ivld
["monitoring-parameters"] = [duplicated_mp
, duplicated_mp
]
1115 with self
.assertRaises(EngineException
) as e
:
1116 self
.topic
.validate_monitoring_params(indata
)
1118 e
.exception
.http_code
,
1119 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1120 "Wrong HTTP status code",
1124 "Duplicated monitoring-parameter id in "
1125 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1126 affected_ivld
["id"], duplicated_mp
["id"]
1129 norm(str(e
.exception
)),
1130 "Wrong exception text",
1133 def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self
):
1134 indata
= deepcopy(db_vnfd_content
)
1136 "id": "dataVM_cpu_util",
1137 "name": "dataVM_cpu_util",
1138 "performance_metric": "cpu",
1140 affected_vdu
= indata
["vdu"][1]
1141 affected_vdu
["monitoring-parameter"].insert(0, duplicated_mp
)
1142 with self
.assertRaises(EngineException
) as e
:
1143 self
.topic
.validate_monitoring_params(indata
)
1145 e
.exception
.http_code
,
1146 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1147 "Wrong HTTP status code",
1151 "Duplicated monitoring-parameter id in "
1152 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1153 affected_vdu
["id"], duplicated_mp
["id"]
1156 norm(str(e
.exception
)),
1157 "Wrong exception text",
1160 def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self
):
1161 indata
= deepcopy(db_vnfd_content
)
1165 "performance_metric": "memory",
1167 affected_df
= indata
["df"][0]
1168 affected_df
["monitoring-parameter"] = [duplicated_mp
, duplicated_mp
]
1169 with self
.assertRaises(EngineException
) as e
:
1170 self
.topic
.validate_monitoring_params(indata
)
1172 e
.exception
.http_code
,
1173 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1174 "Wrong HTTP status code",
1178 "Duplicated monitoring-parameter id in "
1179 "df[id='{}']:monitoring-parameter[id='{}']".format(
1180 affected_df
["id"], duplicated_mp
["id"]
1183 norm(str(e
.exception
)),
1184 "Wrong exception text",
1187 def test_validate_scaling_group_descriptor_on_valid_descriptor(self
):
1188 indata
= db_vnfd_content
1189 self
.topic
.validate_scaling_group_descriptor(indata
)
1191 def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self
):
1192 indata
= deepcopy(db_vnfd_content
)
1193 vdu
= indata
["vdu"][1]
1194 affected_df
= indata
["df"][0]
1195 affected_sa
= affected_df
["scaling-aspect"][0]
1196 affected_sp
= affected_sa
["scaling-policy"][0]
1197 affected_sc
= affected_sp
["scaling-criteria"][0]
1198 vdu
.pop("monitoring-parameter")
1199 with self
.assertRaises(EngineException
) as e
:
1200 self
.topic
.validate_scaling_group_descriptor(indata
)
1202 e
.exception
.http_code
,
1203 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1204 "Wrong HTTP status code",
1208 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1209 "[name='{}']:scaling-criteria[name='{}']: "
1210 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1213 affected_sp
["name"],
1214 affected_sc
["name"],
1215 affected_sc
["vnf-monitoring-param-ref"],
1218 norm(str(e
.exception
)),
1219 "Wrong exception text",
1222 def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self
):
1223 indata
= deepcopy(db_vnfd_content
)
1224 df
= indata
["df"][0]
1225 affected_sa
= df
["scaling-aspect"][0]
1226 indata
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1229 with self
.assertRaises(EngineException
) as e
:
1230 self
.topic
.validate_scaling_group_descriptor(indata
)
1232 e
.exception
.http_code
,
1233 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1234 "Wrong HTTP status code",
1238 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1239 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1240 df
["id"], affected_sa
["id"]
1243 norm(str(e
.exception
)),
1244 "Wrong exception text",
1247 def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1250 indata
= deepcopy(db_vnfd_content
)
1251 df
= indata
["df"][0]
1252 affected_sa
= df
["scaling-aspect"][0]
1253 affected_sca_primitive
= affected_sa
["scaling-config-action"][0][
1254 "vnf-config-primitive-name-ref"
1256 df
["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1259 with self
.assertRaises(EngineException
) as e
:
1260 self
.topic
.validate_scaling_group_descriptor(indata
)
1262 e
.exception
.http_code
,
1263 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1264 "Wrong HTTP status code",
1268 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1269 "config-primitive-name-ref='{}' does not match any "
1270 "day1-2 configuration:config-primitive:name".format(
1271 df
["id"], affected_sa
["id"], affected_sca_primitive
1274 norm(str(e
.exception
)),
1275 "Wrong exception text",
1278 def test_new_vnfd_revision(self
):
1279 did
= db_vnfd_content
["_id"]
1280 self
.fs
.get_params
.return_value
= {}
1281 self
.fs
.file_exists
.return_value
= False
1282 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1283 "/tmp/" + str(uuid4()), "a+b"
1285 test_vnfd
= deepcopy(db_vnfd_content
)
1286 del test_vnfd
["_id"]
1287 del test_vnfd
["_admin"]
1288 self
.db
.create
.return_value
= did
1290 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1291 db_args
= self
.db
.create
.call_args
[0]
1293 db_args
[1]["_admin"]["revision"], 0, "New package should be at revision 0"
1296 @patch("osm_nbi.descriptor_topics.shutil")
1297 @patch("osm_nbi.descriptor_topics.os.rename")
1298 def test_update_vnfd(self
, mock_rename
, mock_shutil
):
1300 did
= db_vnfd_content
["_id"]
1302 self
.fs
.get_params
.return_value
= {}
1303 self
.fs
.file_exists
.return_value
= False
1304 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1305 "/tmp/" + str(uuid4()), "a+b"
1307 new_vnfd
= deepcopy(db_vnfd_content
)
1309 self
.db
.create
.return_value
= did
1311 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1312 del new_vnfd
["vdu"][0]["cloud-init-file"]
1313 del new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1315 ][0]["execution-environment-list"][0]["juju"]
1317 old_vnfd
= {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])}
1318 old_vnfd
["_admin"]["revision"] = old_revision
1319 self
.db
.get_one
.side_effect
= [old_vnfd
, None]
1320 self
.topic
.upload_content(fake_session
, did
, new_vnfd
, {}, {"Content-Type": []})
1322 db_args
= self
.db
.replace
.call_args
[0]
1324 db_args
[2]["_admin"]["revision"],
1326 "Revision should increment",
1330 class Test_NsdTopic(TestCase
):
1332 def setUpClass(cls
):
1333 cls
.test_name
= "test-nsd-topic"
1336 def tearDownClass(cls
):
1340 self
.db
= Mock(dbbase
.DbBase())
1341 self
.fs
= Mock(fsbase
.FsBase())
1342 self
.msg
= Mock(msgbase
.MsgBase())
1343 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1344 self
.topic
= NsdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1345 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1348 def assertNotRaises(self
, exception_type
):
1351 except exception_type
:
1352 raise self
.failureException("{} raised".format(exception_type
.__name
__))
1354 def create_desc_temp(self
, template
):
1355 old_desc
= deepcopy(template
)
1356 new_desc
= deepcopy(template
)
1357 return old_desc
, new_desc
1359 def prepare_nsd_creation(self
):
1361 did
= db_nsd_content
["_id"]
1362 self
.fs
.get_params
.return_value
= {}
1363 self
.fs
.file_exists
.return_value
= False
1364 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1365 "/tmp/" + str(uuid4()), "a+b"
1367 self
.db
.get_one
.side_effect
= [
1368 {"_id": did
, "_admin": deepcopy(db_nsd_content
["_admin"])},
1371 test_nsd
= deepcopy(db_nsd_content
)
1373 del test_nsd
["_admin"]
1374 return did
, test_nsd
1376 @patch("osm_nbi.descriptor_topics.shutil")
1377 @patch("osm_nbi.descriptor_topics.os.rename")
1378 def test_new_nsd_normal_creation(self
, mock_rename
, mock_shutil
):
1379 did
, test_nsd
= self
.prepare_nsd_creation()
1380 self
.db
.create
.return_value
= did
1383 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1384 db_args
= self
.db
.create
.call_args
[0]
1385 msg_args
= self
.msg
.write
.call_args
[0]
1386 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1387 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1388 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
1389 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1390 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1391 self
.assertEqual(did2
, did
, "Wrong DB NSD id")
1392 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
1394 db_args
[1]["_admin"]["modified"],
1395 db_args
[1]["_admin"]["created"],
1396 "Wrong modification time",
1399 db_args
[1]["_admin"]["projects_read"],
1401 "Wrong read-only project list",
1404 db_args
[1]["_admin"]["projects_write"],
1406 "Wrong read-write project list",
1409 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1411 self
.topic
.upload_content(fake_session
, did
, test_nsd
, {}, {"Content-Type": []})
1412 msg_args
= self
.msg
.write
.call_args
[0]
1413 test_nsd
["_id"] = did
1414 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1415 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1416 self
.assertEqual(msg_args
[2], test_nsd
, "Wrong message content")
1418 db_args
= self
.db
.get_one
.mock_calls
[0][1]
1419 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1420 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB NSD id")
1422 db_args
= self
.db
.replace
.call_args
[0]
1423 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1424 self
.assertEqual(db_args
[1], did
, "Wrong DB NSD id")
1426 admin
= db_args
[2]["_admin"]
1427 db_admin
= db_nsd_content
["_admin"]
1428 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
1430 admin
["modified"], db_admin
["created"], "Wrong modification time"
1433 admin
["projects_read"],
1434 db_admin
["projects_read"],
1435 "Wrong read-only project list",
1438 admin
["projects_write"],
1439 db_admin
["projects_write"],
1440 "Wrong read-write project list",
1443 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1446 admin
["operationalState"], "ENABLED", "Wrong operational state"
1448 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
1450 storage
= admin
["storage"]
1451 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
1452 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
1454 compare_desc(self
, test_nsd
, db_args
[2], "NSD")
1455 revision_args
= self
.db
.create
.call_args
[0]
1457 revision_args
[0], self
.topic
.topic
+ "_revisions", "Wrong topic"
1459 self
.assertEqual(revision_args
[1]["id"], db_args
[2]["id"], "Wrong revision id")
1461 revision_args
[1]["_id"], db_args
[2]["_id"] + ":1", "Wrong revision _id"
1464 @patch("osm_nbi.descriptor_topics.shutil")
1465 @patch("osm_nbi.descriptor_topics.os.rename")
1466 def test_new_nsd_check_pyangbind_validation_required_properties(
1467 self
, mock_rename
, mock_shutil
1469 did
, test_nsd
= self
.prepare_nsd_creation()
1472 with self
.assertRaises(
1473 EngineException
, msg
="Accepted NSD with a missing required property"
1475 self
.topic
.upload_content(
1476 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1479 e
.exception
.http_code
,
1480 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1481 "Wrong HTTP status code",
1484 norm("Error in pyangbind validation: '{}'".format("id")),
1485 norm(str(e
.exception
)),
1486 "Wrong exception text",
1489 @patch("osm_nbi.descriptor_topics.shutil")
1490 @patch("osm_nbi.descriptor_topics.os.rename")
1491 def test_new_nsd_check_pyangbind_validation_additional_properties(
1492 self
, mock_rename
, mock_shutil
1494 did
, test_nsd
= self
.prepare_nsd_creation()
1495 test_nsd
["extra-property"] = 0
1497 with self
.assertRaises(
1498 EngineException
, msg
="Accepted NSD with an additional property"
1500 self
.topic
.upload_content(
1501 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1504 e
.exception
.http_code
,
1505 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1506 "Wrong HTTP status code",
1510 "Error in pyangbind validation: {} ({})".format(
1511 "json object contained a key that did not exist", "extra-property"
1514 norm(str(e
.exception
)),
1515 "Wrong exception text",
1518 @patch("osm_nbi.descriptor_topics.shutil")
1519 @patch("osm_nbi.descriptor_topics.os.rename")
1520 def test_new_nsd_check_pyangbind_validation_property_types(
1521 self
, mock_rename
, mock_shutil
1523 did
, test_nsd
= self
.prepare_nsd_creation()
1524 test_nsd
["designer"] = {"key": 0}
1526 with self
.assertRaises(
1527 EngineException
, msg
="Accepted NSD with a wrongly typed property"
1529 self
.topic
.upload_content(
1530 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1533 e
.exception
.http_code
,
1534 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1535 "Wrong HTTP status code",
1539 "Error in pyangbind validation: {} ({})".format(
1540 "json object contained a key that did not exist", "key"
1543 norm(str(e
.exception
)),
1544 "Wrong exception text",
1547 @patch("osm_nbi.descriptor_topics.shutil")
1548 @patch("osm_nbi.descriptor_topics.os.rename")
1549 def test_new_nsd_check_input_validation_mgmt_network_virtual_link_protocol_data(
1550 self
, mock_rename
, mock_shutil
1552 did
, test_nsd
= self
.prepare_nsd_creation()
1553 df
= test_nsd
["df"][0]
1556 "virtual-link-desc-id": "mgmt",
1557 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1559 df
["virtual-link-profile"] = [mgmt_profile
]
1561 with self
.assertRaises(
1562 EngineException
, msg
="Accepted VLD with mgmt-network+ip-profile"
1564 self
.topic
.upload_content(
1565 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1568 e
.exception
.http_code
,
1569 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1570 "Wrong HTTP status code",
1574 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1575 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1576 df
["id"], mgmt_profile
["id"]
1579 norm(str(e
.exception
)),
1580 "Wrong exception text",
1583 @patch("osm_nbi.descriptor_topics.shutil")
1584 @patch("osm_nbi.descriptor_topics.os.rename")
1585 def test_new_nsd_check_descriptor_dependencies_vnfd_id(
1586 self
, mock_rename
, mock_shutil
1588 did
, test_nsd
= self
.prepare_nsd_creation()
1589 self
.db
.get_list
.return_value
= []
1591 with self
.assertRaises(
1592 EngineException
, msg
="Accepted wrong VNFD ID reference"
1594 self
.topic
.upload_content(
1595 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1598 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1602 "'vnfd-id'='{}' references a non existing vnfd".format(
1603 test_nsd
["vnfd-id"][0]
1606 norm(str(e
.exception
)),
1607 "Wrong exception text",
1610 @patch("osm_nbi.descriptor_topics.shutil")
1611 @patch("osm_nbi.descriptor_topics.os.rename")
1612 def test_new_nsd_check_descriptor_dependencies_vld_vnfd_connection_point_ref(
1613 self
, mock_rename
, mock_shutil
1615 # Check Descriptor Dependencies: "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]
1616 did
, test_nsd
= self
.prepare_nsd_creation()
1617 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1618 df
= test_nsd
["df"][0]
1619 affected_vnf_profile
= df
["vnf-profile"][0]
1620 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
1621 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
1622 self
.db
.get_list
.return_value
= [vnfd_descriptor
]
1624 with self
.assertRaises(
1625 EngineException
, msg
="Accepted wrong VLD CP reference"
1627 self
.topic
.upload_content(
1628 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1631 e
.exception
.http_code
,
1632 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1633 "Wrong HTTP status code",
1637 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1638 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1639 "non existing ext-cpd:id inside vnfd '{}'".format(
1641 affected_vnf_profile
["id"],
1642 affected_virtual_link
["virtual-link-profile-id"],
1644 vnfd_descriptor
["id"],
1647 norm(str(e
.exception
)),
1648 "Wrong exception text",
1651 def test_edit_nsd(self
):
1652 nsd_content
= deepcopy(db_nsd_content
)
1653 did
= nsd_content
["_id"]
1654 self
.fs
.file_exists
.return_value
= True
1655 self
.fs
.dir_ls
.return_value
= True
1656 with self
.subTest(i
=1, t
="Normal Edition"):
1658 self
.db
.get_one
.side_effect
= [deepcopy(nsd_content
), None]
1659 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1660 data
= {"id": "new-nsd-id", "name": "new-nsd-name"}
1661 self
.topic
.edit(fake_session
, did
, data
)
1662 db_args
= self
.db
.replace
.call_args
[0]
1663 msg_args
= self
.msg
.write
.call_args
[0]
1665 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1666 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1667 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
1668 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1669 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
1671 db_args
[2]["_admin"]["created"],
1672 nsd_content
["_admin"]["created"],
1673 "Wrong creation time",
1676 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
1679 db_args
[2]["_admin"]["projects_read"],
1680 nsd_content
["_admin"]["projects_read"],
1681 "Wrong read-only project list",
1684 db_args
[2]["_admin"]["projects_write"],
1685 nsd_content
["_admin"]["projects_write"],
1686 "Wrong read-write project list",
1688 self
.assertEqual(db_args
[2]["id"], data
["id"], "Wrong NSD ID")
1689 self
.assertEqual(db_args
[2]["name"], data
["name"], "Wrong NSD Name")
1690 with self
.subTest(i
=2, t
="Conflict on Edit"):
1691 data
= {"id": "fake-nsd-id", "name": "new-nsd-name"}
1692 self
.db
.get_one
.side_effect
= [
1694 {"_id": str(uuid4()), "id": data
["id"]},
1696 with self
.assertRaises(
1697 EngineException
, msg
="Accepted existing NSD ID"
1699 self
.topic
.edit(fake_session
, did
, data
)
1701 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1705 "{} with id '{}' already exists for this project".format(
1709 norm(str(e
.exception
)),
1710 "Wrong exception text",
1712 with self
.subTest(i
=3, t
="Check Envelope"):
1713 data
= {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
1714 self
.db
.get_one
.side_effect
= [nsd_content
, None]
1715 with self
.assertRaises(
1716 EngineException
, msg
="Accepted NSD with wrong envelope"
1718 self
.topic
.edit(fake_session
, did
, data
, content
=nsd_content
)
1720 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
1723 "'nsd' must be a list of only one element",
1724 norm(str(e
.exception
)),
1725 "Wrong exception text",
1727 self
.db
.reset_mock()
1730 def test_delete_nsd(self
):
1731 did
= db_nsd_content
["_id"]
1732 self
.db
.get_one
.return_value
= db_nsd_content
1733 p_id
= db_nsd_content
["_admin"]["projects_read"][0]
1734 with self
.subTest(i
=1, t
="Normal Deletion"):
1735 self
.db
.get_list
.return_value
= []
1736 self
.db
.del_one
.return_value
= {"deleted": 1}
1737 self
.topic
.delete(fake_session
, did
)
1738 db_args
= self
.db
.del_one
.call_args
[0]
1739 msg_args
= self
.msg
.write
.call_args
[0]
1740 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1741 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
1742 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1743 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1744 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
1746 db_args
[1]["_admin.projects_write.cont"],
1750 db_g1_args
= self
.db
.get_one
.call_args
[0]
1751 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1752 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB NSD ID")
1753 db_gl_calls
= self
.db
.get_list
.call_args_list
1754 self
.assertEqual(db_gl_calls
[0][0][0], "nsrs", "Wrong DB topic")
1755 # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID") # Filter changed after call
1756 self
.assertEqual(db_gl_calls
[1][0][0], "nsts", "Wrong DB topic")
1758 db_gl_calls
[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
1759 db_nsd_content
["id"],
1760 "Wrong DB NSD netslice-subnet nsd-ref",
1762 self
.db
.set_one
.assert_not_called()
1763 fs_del_calls
= self
.fs
.file_delete
.call_args_list
1764 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
1765 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
1766 with self
.subTest(i
=2, t
="Conflict on Delete - NSD in use by nsr"):
1767 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-nsr"}]
1768 with self
.assertRaises(
1769 EngineException
, msg
="Accepted NSD in use by NSR"
1771 self
.topic
.delete(fake_session
, did
)
1773 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1776 "there is at least one ns instance using this descriptor",
1777 norm(str(e
.exception
)),
1778 "Wrong exception text",
1780 with self
.subTest(i
=3, t
="Conflict on Delete - NSD in use by NST"):
1781 self
.db
.get_list
.side_effect
= [
1783 [{"_id": str(uuid4()), "name": "fake-nst"}],
1785 with self
.assertRaises(
1786 EngineException
, msg
="Accepted NSD in use by NST"
1788 self
.topic
.delete(fake_session
, did
)
1790 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1793 "there is at least one netslice template referencing this descriptor",
1794 norm(str(e
.exception
)),
1795 "Wrong exception text",
1797 with self
.subTest(i
=4, t
="Non-existent NSD"):
1798 excp_msg
= "Not found any {} with filter='{}'".format("NSD", {"_id": did
})
1799 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
1800 with self
.assertRaises(
1801 DbException
, msg
="Accepted non-existent NSD ID"
1803 self
.topic
.delete(fake_session
, did
)
1805 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1808 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
1810 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
1811 db_nsd_content
["_admin"]["projects_read"].append("other_project")
1812 self
.db
.get_one
= Mock(return_value
=db_nsd_content
)
1813 self
.db
.get_list
= Mock(return_value
=[])
1814 self
.msg
.write
.reset_mock()
1815 self
.db
.del_one
.reset_mock()
1816 self
.fs
.file_delete
.reset_mock()
1818 self
.topic
.delete(fake_session
, did
)
1819 self
.db
.del_one
.assert_not_called()
1820 self
.msg
.write
.assert_not_called()
1821 db_g1_args
= self
.db
.get_one
.call_args
[0]
1822 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1823 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
1824 db_s1_args
= self
.db
.set_one
.call_args
1825 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
1826 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
1828 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1831 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
1834 db_s1_args
[1]["pull_list"],
1835 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
1836 "Wrong DB pull_list dictionary",
1838 self
.fs
.file_delete
.assert_not_called()
1839 self
.db
.reset_mock()
1842 def prepare_nsd_validation(self
):
1843 descriptor_name
= "test_ns_descriptor"
1844 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1845 "/tmp/" + str(uuid4()), "a+b"
1847 old_nsd
, new_nsd
= self
.create_desc_temp(db_nsd_content
)
1848 return descriptor_name
, old_nsd
, new_nsd
1850 @patch("osm_nbi.descriptor_topics.yaml")
1851 def test_validate_descriptor_ns_configuration_changed(self
, mock_yaml
):
1852 # NSD has changes in ns-configuration:config-primitive
1853 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
1855 {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
1858 {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
1860 mock_yaml
.load
.side_effect
= [old_nsd
, new_nsd
]
1862 with self
.assertNotRaises(EngineException
):
1863 self
.topic
._validate
_descriptor
_changes
(descriptor_name
, "/tmp", "/tmp:1")
1865 @patch("osm_nbi.descriptor_topics.yaml")
1866 def test_validate_descriptor_nsd_name_changed(self
, mock_yaml
):
1867 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
1868 new_nsd
["name"] = "nscharm-ns2"
1869 mock_yaml
.load
.side_effect
= [old_nsd
, new_nsd
]
1871 with self
.assertRaises(EngineException
) as e
:
1872 self
.topic
._validate
_descriptor
_changes
(descriptor_name
, "/tmp", "/tmp:1")
1875 e
.exception
.http_code
,
1876 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1877 "Wrong HTTP status code",
1880 norm("there are disallowed changes in the ns descriptor"),
1881 norm(str(e
.exception
)),
1882 "Wrong exception text",
1885 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
1888 indata
= deepcopy(db_nsd_content
)
1889 vld
= indata
["virtual-link-desc"][0]
1890 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
1894 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
1897 indata
= deepcopy(db_nsd_content
)
1898 vld
= indata
["virtual-link-desc"][0]
1899 df
= indata
["df"][0]
1902 "virtual-link-desc-id": "mgmt",
1903 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1905 df
["virtual-link-profile"] = [affected_vlp
]
1906 with self
.assertRaises(EngineException
) as e
:
1907 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
1911 e
.exception
.http_code
,
1912 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1913 "Wrong HTTP status code",
1917 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1918 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1919 df
["id"], affected_vlp
["id"]
1922 norm(str(e
.exception
)),
1923 "Wrong exception text",
1926 def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self
):
1927 indata
= deepcopy(db_nsd_content
)
1928 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
1930 def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self
):
1931 indata
= deepcopy(db_nsd_content
)
1932 df
= indata
["df"][0]
1933 affected_vnf_profile
= df
["vnf-profile"][0]
1934 indata
["vnfd-id"] = ["non-existing-vnfd"]
1935 with self
.assertRaises(EngineException
) as e
:
1936 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
1938 e
.exception
.http_code
,
1939 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1940 "Wrong HTTP status code",
1944 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
1945 "does not match any vnfd-id".format(
1947 affected_vnf_profile
["id"],
1948 affected_vnf_profile
["vnfd-id"],
1951 norm(str(e
.exception
)),
1952 "Wrong exception text",
1955 def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
1958 nsd_descriptor
= deepcopy(db_nsd_content
)
1959 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1960 df
= nsd_descriptor
["df"][0]
1961 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
1962 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
1966 def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
1969 nsd_descriptor
= deepcopy(db_nsd_content
)
1970 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1971 df
= nsd_descriptor
["df"][0]
1972 affected_vnf_profile
= df
["vnf-profile"][0]
1973 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
1974 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
1975 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
1976 with self
.assertRaises(EngineException
) as e
:
1977 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
1981 e
.exception
.http_code
,
1982 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1983 "Wrong HTTP status code",
1987 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1988 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1989 "non existing ext-cpd:id inside vnfd '{}'".format(
1991 affected_vnf_profile
["id"],
1992 affected_virtual_link
["virtual-link-profile-id"],
1994 vnfd_descriptor
["id"],
1997 norm(str(e
.exception
)),
1998 "Wrong exception text",
2001 def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self
):
2002 nsd_descriptor
= deepcopy(db_nsd_content
)
2003 invalid_vnfd_id
= "invalid-vnfd-id"
2004 nsd_descriptor
["id"] = "invalid-vnfd-id-ns"
2005 nsd_descriptor
["vnfd-id"][0] = invalid_vnfd_id
2006 nsd_descriptor
["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
2007 nsd_descriptor
["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
2008 with self
.assertRaises(EngineException
) as e
:
2009 self
.db
.get_list
.return_value
= []
2010 nsd_descriptor
= self
.topic
.check_conflict_on_edit(
2011 fake_session
, nsd_descriptor
, [], "id"
2014 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2018 "Descriptor error at 'vnfd-id'='{}' references a non "
2019 "existing vnfd".format(invalid_vnfd_id
)
2021 norm(str(e
.exception
)),
2022 "Wrong exception text",
2026 if __name__
== "__main__":