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
.safe_load(db_vnfds_text
)[0]
38 db_nsd_content
= yaml
.safe_load(db_nsds_text
)[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 """Testing input validation during new vnfd creation
405 for vdu internal connection point"""
406 did
, test_vnfd
= self
.prepare_vnfd_creation()
407 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
408 ext_cpd
= test_vnfd
["ext-cpd"][1]
409 ext_cpd
["int-cpd"]["cpd"] = "wrong-cpd"
411 with self
.assertRaises(
412 EngineException
, msg
="Accepted wrong ext-cpd internal connection point"
414 self
.topic
.upload_content(
415 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
418 e
.exception
.http_code
,
419 HTTPStatus
.UNPROCESSABLE_ENTITY
,
420 "Wrong HTTP status code",
424 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
428 norm(str(e
.exception
)),
429 "Wrong exception text",
432 @patch("osm_nbi.descriptor_topics.shutil")
433 @patch("osm_nbi.descriptor_topics.os.rename")
434 def test_new_vnfd_check_input_validation_duplicated_vld(
435 self
, mock_rename
, mock_shutil
437 """Testing input validation during new vnfd creation
438 for dublicated virtual link description"""
439 did
, test_vnfd
= self
.prepare_vnfd_creation()
440 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
441 test_vnfd
["int-virtual-link-desc"].insert(0, {"id": "internal"})
443 with self
.assertRaises(
444 EngineException
, msg
="Accepted duplicated VLD name"
446 self
.topic
.upload_content(
447 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
450 e
.exception
.http_code
,
451 HTTPStatus
.UNPROCESSABLE_ENTITY
,
452 "Wrong HTTP status code",
456 "identifier id '{}' is not unique".format(
457 test_vnfd
["int-virtual-link-desc"][0]["id"]
460 norm(str(e
.exception
)),
461 "Wrong exception text",
464 @patch("osm_nbi.descriptor_topics.shutil")
465 @patch("osm_nbi.descriptor_topics.os.rename")
466 def test_new_vnfd_check_input_validation_vdu_int_virtual_link_desc(
467 self
, mock_rename
, mock_shutil
469 """Testing input validation during new vnfd creation
470 for vdu internal virtual link description"""
471 did
, test_vnfd
= self
.prepare_vnfd_creation()
472 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
473 vdu
= test_vnfd
["vdu"][0]
474 int_cpd
= vdu
["int-cpd"][1]
475 int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
477 with self
.assertRaises(
478 EngineException
, msg
="Accepted int-virtual-link-desc"
480 self
.topic
.upload_content(
481 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
484 e
.exception
.http_code
,
485 HTTPStatus
.UNPROCESSABLE_ENTITY
,
486 "Wrong HTTP status code",
490 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
491 "int-virtual-link-desc".format(
492 vdu
["id"], int_cpd
["id"], int_cpd
["int-virtual-link-desc"]
495 norm(str(e
.exception
)),
496 "Wrong exception text",
499 @patch("osm_nbi.descriptor_topics.shutil")
500 @patch("osm_nbi.descriptor_topics.os.rename")
501 def test_new_vnfd_check_input_validation_virtual_link_profile(
502 self
, mock_rename
, mock_shutil
504 """Testing input validation during new vnfd creation
505 for virtual link profile"""
506 did
, test_vnfd
= self
.prepare_vnfd_creation()
507 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
508 fake_ivld_profile
= {"id": "fake-profile-ref", "flavour": "fake-flavour"}
509 df
= test_vnfd
["df"][0]
510 df
["virtual-link-profile"] = [fake_ivld_profile
]
512 with self
.assertRaises(
513 EngineException
, msg
="Accepted non-existent Profile Ref"
515 self
.topic
.upload_content(
516 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
519 e
.exception
.http_code
,
520 HTTPStatus
.UNPROCESSABLE_ENTITY
,
521 "Wrong HTTP status code",
525 "df[id='{}']:virtual-link-profile='{}' must match an existing "
526 "int-virtual-link-desc".format(df
["id"], fake_ivld_profile
["id"])
528 norm(str(e
.exception
)),
529 "Wrong exception text",
532 @patch("osm_nbi.descriptor_topics.shutil")
533 @patch("osm_nbi.descriptor_topics.os.rename")
534 def test_new_vnfd_check_input_validation_scaling_criteria_monitoring_param_ref(
535 self
, mock_rename
, mock_shutil
537 """Testing input validation during new vnfd creation
538 for scaling criteria without monitoring parameter"""
539 did
, test_vnfd
= self
.prepare_vnfd_creation()
540 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
541 vdu
= test_vnfd
["vdu"][1]
542 affected_df
= test_vnfd
["df"][0]
543 sa
= affected_df
["scaling-aspect"][0]
544 sp
= sa
["scaling-policy"][0]
545 sc
= sp
["scaling-criteria"][0]
546 vdu
.pop("monitoring-parameter")
548 with self
.assertRaises(
549 EngineException
, msg
="Accepted non-existent Scaling Group Policy Criteria"
551 self
.topic
.upload_content(
552 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
555 e
.exception
.http_code
,
556 HTTPStatus
.UNPROCESSABLE_ENTITY
,
557 "Wrong HTTP status code",
561 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
562 "[name='{}']:scaling-criteria[name='{}']: "
563 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
568 sc
["vnf-monitoring-param-ref"],
571 norm(str(e
.exception
)),
572 "Wrong exception text",
575 @patch("osm_nbi.descriptor_topics.shutil")
576 @patch("osm_nbi.descriptor_topics.os.rename")
577 def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration(
578 self
, mock_rename
, mock_shutil
580 """Testing input validation during new vnfd creation
581 for scaling criteria without day12 configuration"""
582 did
, test_vnfd
= self
.prepare_vnfd_creation()
583 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
584 test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
587 df
= test_vnfd
["df"][0]
589 with self
.assertRaises(
590 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
592 self
.topic
.upload_content(
593 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
596 e
.exception
.http_code
,
597 HTTPStatus
.UNPROCESSABLE_ENTITY
,
598 "Wrong HTTP status code",
602 "'day1-2 configuration' not defined in the descriptor but it is referenced "
603 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
604 df
["id"], df
["scaling-aspect"][0]["id"]
607 norm(str(e
.exception
)),
608 "Wrong exception text",
611 @patch("osm_nbi.descriptor_topics.shutil")
612 @patch("osm_nbi.descriptor_topics.os.rename")
613 def test_new_vnfd_check_input_validation_scaling_config_action(
614 self
, mock_rename
, mock_shutil
616 """Testing input validation during new vnfd creation
617 for scaling criteria wrong config primitive"""
618 did
, test_vnfd
= self
.prepare_vnfd_creation()
619 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
620 df
= test_vnfd
["df"][0]
621 affected_df
= test_vnfd
["df"][0]
622 sa
= affected_df
["scaling-aspect"][0]
623 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
624 "operate-vnf-op-config"
625 )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
627 with self
.assertRaises(
628 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
630 self
.topic
.upload_content(
631 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
634 e
.exception
.http_code
,
635 HTTPStatus
.UNPROCESSABLE_ENTITY
,
636 "Wrong HTTP status code",
640 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
641 "config-primitive-name-ref='{}' does not match any "
642 "day1-2 configuration:config-primitive:name".format(
644 df
["scaling-aspect"][0]["id"],
645 sa
["scaling-config-action"][0]["vnf-config-primitive-name-ref"],
648 norm(str(e
.exception
)),
649 "Wrong exception text",
652 @patch("osm_nbi.descriptor_topics.shutil")
653 @patch("osm_nbi.descriptor_topics.os.rename")
654 def test_new_vnfd_check_input_validation_everything_right(
655 self
, mock_rename
, mock_shutil
657 """Testing input validation during new vnfd creation
658 everything correct"""
659 did
, test_vnfd
= self
.prepare_vnfd_creation()
660 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
661 test_vnfd
["id"] = "fake-vnfd-id"
662 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
663 "operate-vnf-op-config"
664 )["day1-2"][0]["id"] = "fake-vnfd-id"
665 self
.db
.get_one
.side_effect
= [
666 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
669 rc
= self
.topic
.upload_content(
670 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
672 self
.assertTrue(rc
, "Input Validation: Unexpected failure")
674 def test_edit_vnfd(self
):
675 vnfd_content
= deepcopy(db_vnfd_content
)
676 did
= vnfd_content
["_id"]
677 self
.fs
.file_exists
.return_value
= True
678 self
.fs
.dir_ls
.return_value
= True
679 with self
.subTest(i
=1, t
="Normal Edition"):
681 self
.db
.get_one
.side_effect
= [deepcopy(vnfd_content
), None]
682 data
= {"product-name": "new-vnfd-name"}
683 self
.topic
.edit(fake_session
, did
, data
)
684 db_args
= self
.db
.replace
.call_args
[0]
685 msg_args
= self
.msg
.write
.call_args
[0]
687 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
688 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
689 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
690 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
691 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
693 db_args
[2]["_admin"]["created"],
694 vnfd_content
["_admin"]["created"],
695 "Wrong creation time",
698 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
701 db_args
[2]["_admin"]["projects_read"],
702 vnfd_content
["_admin"]["projects_read"],
703 "Wrong read-only project list",
706 db_args
[2]["_admin"]["projects_write"],
707 vnfd_content
["_admin"]["projects_write"],
708 "Wrong read-write project list",
711 db_args
[2]["product-name"], data
["product-name"], "Wrong VNFD Name"
713 with self
.subTest(i
=2, t
="Conflict on Edit"):
714 data
= {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
715 self
.db
.get_one
.side_effect
= [
716 deepcopy(vnfd_content
),
717 {"_id": str(uuid4()), "id": data
["id"]},
719 with self
.assertRaises(
720 EngineException
, msg
="Accepted existing VNFD ID"
722 self
.topic
.edit(fake_session
, did
, data
)
724 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
728 "{} with id '{}' already exists for this project".format(
732 norm(str(e
.exception
)),
733 "Wrong exception text",
735 with self
.subTest(i
=3, t
="Check Envelope"):
736 data
= {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
737 with self
.assertRaises(
738 EngineException
, msg
="Accepted VNFD with wrong envelope"
740 self
.topic
.edit(fake_session
, did
, data
, content
=vnfd_content
)
742 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
745 "'vnfd' must be dict", norm(str(e
.exception
)), "Wrong exception text"
749 def test_delete_vnfd(self
):
750 did
= db_vnfd_content
["_id"]
751 self
.db
.get_one
.return_value
= db_vnfd_content
752 p_id
= db_vnfd_content
["_admin"]["projects_read"][0]
753 with self
.subTest(i
=1, t
="Normal Deletion"):
754 self
.db
.get_list
.return_value
= []
755 self
.db
.del_one
.return_value
= {"deleted": 1}
756 self
.topic
.delete(fake_session
, did
)
757 db_args
= self
.db
.del_one
.call_args
[0]
758 msg_args
= self
.msg
.write
.call_args
[0]
759 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
760 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
761 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
762 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
763 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
765 db_args
[1]["_admin.projects_write.cont"],
769 db_g1_args
= self
.db
.get_one
.call_args
[0]
770 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
771 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
772 db_gl_calls
= self
.db
.get_list
.call_args_list
773 self
.assertEqual(db_gl_calls
[0][0][0], "vnfrs", "Wrong DB topic")
774 # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID") # Filter changed after call
775 self
.assertEqual(db_gl_calls
[1][0][0], "nsds", "Wrong DB topic")
777 db_gl_calls
[1][0][1]["vnfd-id"],
778 db_vnfd_content
["id"],
779 "Wrong DB NSD vnfd-id",
783 self
.db
.del_list
.call_args
[0][0],
784 self
.topic
.topic
+ "_revisions",
789 self
.db
.del_list
.call_args
[0][1]["_id"]["$regex"],
791 "Wrong ID for rexep delete",
794 self
.db
.set_one
.assert_not_called()
795 fs_del_calls
= self
.fs
.file_delete
.call_args_list
796 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
797 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
798 with self
.subTest(i
=2, t
="Conflict on Delete - VNFD in use by VNFR"):
799 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-vnfr"}]
800 with self
.assertRaises(
801 EngineException
, msg
="Accepted VNFD in use by VNFR"
803 self
.topic
.delete(fake_session
, did
)
805 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
808 "there is at least one vnf instance using this descriptor",
809 norm(str(e
.exception
)),
810 "Wrong exception text",
812 with self
.subTest(i
=3, t
="Conflict on Delete - VNFD in use by NSD"):
813 self
.db
.get_list
.side_effect
= [
815 [{"_id": str(uuid4()), "name": "fake-nsd"}],
817 with self
.assertRaises(
818 EngineException
, msg
="Accepted VNFD in use by NSD"
820 self
.topic
.delete(fake_session
, did
)
822 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
825 "there is at least one ns package referencing this descriptor",
826 norm(str(e
.exception
)),
827 "Wrong exception text",
829 with self
.subTest(i
=4, t
="Non-existent VNFD"):
830 excp_msg
= "Not found any {} with filter='{}'".format("VNFD", {"_id": did
})
831 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
832 with self
.assertRaises(
833 DbException
, msg
="Accepted non-existent VNFD ID"
835 self
.topic
.delete(fake_session
, did
)
837 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
840 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
842 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
843 db_vnfd_content
["_admin"]["projects_read"].append("other_project")
844 self
.db
.get_one
= Mock(return_value
=db_vnfd_content
)
845 self
.db
.get_list
= Mock(return_value
=[])
846 self
.msg
.write
.reset_mock()
847 self
.db
.del_one
.reset_mock()
848 self
.fs
.file_delete
.reset_mock()
850 self
.topic
.delete(fake_session
, did
)
851 self
.db
.del_one
.assert_not_called()
852 self
.msg
.write
.assert_not_called()
853 db_g1_args
= self
.db
.get_one
.call_args
[0]
854 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
855 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
856 db_s1_args
= self
.db
.set_one
.call_args
857 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
858 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
860 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
863 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
866 db_s1_args
[1]["pull_list"],
867 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
868 "Wrong DB pull_list dictionary",
870 self
.fs
.file_delete
.assert_not_called()
873 def prepare_vnfd_validation(self
):
874 descriptor_name
= "test_descriptor"
875 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
876 "/tmp/" + str(uuid4()), "a+b"
878 old_vnfd
, new_vnfd
= self
.create_desc_temp(db_vnfd_content
)
879 return descriptor_name
, old_vnfd
, new_vnfd
881 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
882 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
883 def test_validate_vnfd_changes_day12_config_primitive_changed(
884 self
, mock_safe_load
, mock_detect_usage
886 """Validating VNFD for VNFD updates, day1-2 config primitive has changed"""
887 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
888 did
= old_vnfd
["_id"]
889 new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
891 ][0]["config-primitive"][0]["name"] = "new_action"
892 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
893 mock_detect_usage
.return_value
= True
894 self
.db
.get_one
.return_value
= old_vnfd
896 with self
.assertNotRaises(EngineException
):
897 self
.topic
._validate
_descriptor
_changes
(
898 did
, descriptor_name
, "/tmp/", "/tmp:1/"
900 self
.db
.get_one
.assert_called_once()
901 mock_detect_usage
.assert_called_once()
902 self
.assertEqual(mock_safe_load
.call_count
, 2)
904 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
905 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
906 def test_validate_vnfd_changes_sw_version_changed(
907 self
, mock_safe_load
, mock_detect_usage
909 """Validating VNFD for updates, software version has changed"""
910 # old vnfd uses the default software version: 1.0
911 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
912 did
= old_vnfd
["_id"]
913 new_vnfd
["software-version"] = "1.3"
914 new_vnfd
["sw-image-desc"][0]["name"] = "new-image"
915 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
916 mock_detect_usage
.return_value
= True
917 self
.db
.get_one
.return_value
= old_vnfd
919 with self
.assertNotRaises(EngineException
):
920 self
.topic
._validate
_descriptor
_changes
(
921 did
, descriptor_name
, "/tmp/", "/tmp:1/"
923 self
.db
.get_one
.assert_called_once()
924 mock_detect_usage
.assert_called_once()
925 self
.assertEqual(mock_safe_load
.call_count
, 2)
927 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
928 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
929 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed(
930 self
, mock_safe_load
, mock_detect_usage
932 """Validating VNFD for updates, software version has not
933 changed, mgmt-cp has changed."""
934 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
935 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
936 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
937 did
= old_vnfd
["_id"]
938 mock_detect_usage
.return_value
= True
939 self
.db
.get_one
.return_value
= old_vnfd
941 with self
.assertRaises(
942 EngineException
, msg
="there are disallowed changes in the vnf descriptor"
944 self
.topic
._validate
_descriptor
_changes
(
945 did
, descriptor_name
, "/tmp/", "/tmp:1/"
949 e
.exception
.http_code
,
950 HTTPStatus
.UNPROCESSABLE_ENTITY
,
951 "Wrong HTTP status code",
954 norm("there are disallowed changes in the vnf descriptor"),
955 norm(str(e
.exception
)),
956 "Wrong exception text",
958 self
.db
.get_one
.assert_called_once()
959 mock_detect_usage
.assert_called_once()
960 self
.assertEqual(mock_safe_load
.call_count
, 2)
962 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
963 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
964 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed_vnfd_not_in_use(
965 self
, mock_safe_load
, mock_detect_usage
967 """Validating VNFD for updates, software version has not
968 changed, mgmt-cp has changed, vnfd is not in use."""
969 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
970 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
971 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
972 did
= old_vnfd
["_id"]
973 mock_detect_usage
.return_value
= None
974 self
.db
.get_one
.return_value
= old_vnfd
976 with self
.assertNotRaises(EngineException
):
977 self
.topic
._validate
_descriptor
_changes
(
978 did
, descriptor_name
, "/tmp/", "/tmp:1/"
981 self
.db
.get_one
.assert_called_once()
982 mock_detect_usage
.assert_called_once()
983 mock_safe_load
.assert_not_called()
985 def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self
):
986 indata
= deepcopy(db_vnfd_content
)
987 self
.topic
.validate_mgmt_interface_connection_point(indata
)
989 def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
992 indata
= deepcopy(db_vnfd_content
)
993 indata
["ext-cpd"] = []
994 with self
.assertRaises(EngineException
) as e
:
995 self
.topic
.validate_mgmt_interface_connection_point(indata
)
997 e
.exception
.http_code
,
998 HTTPStatus
.UNPROCESSABLE_ENTITY
,
999 "Wrong HTTP status code",
1003 "mgmt-cp='{}' must match an existing ext-cpd".format(indata
["mgmt-cp"])
1005 norm(str(e
.exception
)),
1006 "Wrong exception text",
1009 def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self
):
1010 indata
= deepcopy(db_vnfd_content
)
1011 indata
.pop("mgmt-cp")
1012 with self
.assertRaises(EngineException
) as e
:
1013 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1015 e
.exception
.http_code
,
1016 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1017 "Wrong HTTP status code",
1020 norm("'mgmt-cp' is a mandatory field and it is not defined"),
1021 norm(str(e
.exception
)),
1022 "Wrong exception text",
1025 def test_validate_vdu_internal_connection_points_on_valid_descriptor(self
):
1026 indata
= db_vnfd_content
1027 vdu
= indata
["vdu"][0]
1028 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1030 def test_validate_external_connection_points_on_valid_descriptor(self
):
1031 indata
= db_vnfd_content
1032 self
.topic
.validate_external_connection_points(indata
)
1034 def test_validate_external_connection_points_when_missing_internal_connection_point(
1037 indata
= deepcopy(db_vnfd_content
)
1038 vdu
= indata
["vdu"][0]
1040 affected_ext_cpd
= indata
["ext-cpd"][0]
1041 with self
.assertRaises(EngineException
) as e
:
1042 self
.topic
.validate_external_connection_points(indata
)
1044 e
.exception
.http_code
,
1045 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1046 "Wrong HTTP status code",
1050 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
1051 affected_ext_cpd
["id"]
1054 norm(str(e
.exception
)),
1055 "Wrong exception text",
1058 def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
1061 indata
= deepcopy(db_vnfd_content
)
1062 vdu
= indata
["vdu"][0]
1066 "virtual-network-interface-requirement": [{"name": "duplicated"}],
1068 vdu
["int-cpd"].insert(0, duplicated_cpd
)
1069 with self
.assertRaises(EngineException
) as e
:
1070 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1072 e
.exception
.http_code
,
1073 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1074 "Wrong HTTP status code",
1078 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
1079 vdu
["id"], duplicated_cpd
["id"]
1082 norm(str(e
.exception
)),
1083 "Wrong exception text",
1086 def test_validate_external_connection_points_on_duplicated_external_connection_point(
1089 indata
= deepcopy(db_vnfd_content
)
1091 "id": "vnf-mgmt-ext",
1092 "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
1094 indata
["ext-cpd"].insert(0, duplicated_cpd
)
1095 with self
.assertRaises(EngineException
) as e
:
1096 self
.topic
.validate_external_connection_points(indata
)
1098 e
.exception
.http_code
,
1099 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1100 "Wrong HTTP status code",
1104 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
1105 duplicated_cpd
["id"]
1108 norm(str(e
.exception
)),
1109 "Wrong exception text",
1112 def test_validate_internal_virtual_links_on_valid_descriptor(self
):
1113 indata
= db_vnfd_content
1114 self
.topic
.validate_internal_virtual_links(indata
)
1116 def test_validate_internal_virtual_links_on_duplicated_ivld(self
):
1117 indata
= deepcopy(db_vnfd_content
)
1118 duplicated_vld
= {"id": "internal"}
1119 indata
["int-virtual-link-desc"].insert(0, duplicated_vld
)
1120 with self
.assertRaises(EngineException
) as e
:
1121 self
.topic
.validate_internal_virtual_links(indata
)
1123 e
.exception
.http_code
,
1124 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1125 "Wrong HTTP status code",
1129 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
1130 duplicated_vld
["id"]
1133 norm(str(e
.exception
)),
1134 "Wrong exception text",
1137 def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
1140 indata
= deepcopy(db_vnfd_content
)
1141 vdu
= indata
["vdu"][0]
1142 affected_int_cpd
= vdu
["int-cpd"][0]
1143 affected_int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
1144 with self
.assertRaises(EngineException
) as e
:
1145 self
.topic
.validate_internal_virtual_links(indata
)
1147 e
.exception
.http_code
,
1148 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1149 "Wrong HTTP status code",
1153 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
1154 "int-virtual-link-desc".format(
1156 affected_int_cpd
["id"],
1157 affected_int_cpd
["int-virtual-link-desc"],
1160 norm(str(e
.exception
)),
1161 "Wrong exception text",
1164 def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self
):
1165 indata
= deepcopy(db_vnfd_content
)
1166 affected_ivld_profile
= {"id": "non-existing-int-virtual-link-desc"}
1167 df
= indata
["df"][0]
1168 df
["virtual-link-profile"] = [affected_ivld_profile
]
1169 with self
.assertRaises(EngineException
) as e
:
1170 self
.topic
.validate_internal_virtual_links(indata
)
1172 e
.exception
.http_code
,
1173 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1174 "Wrong HTTP status code",
1178 "df[id='{}']:virtual-link-profile='{}' must match an existing "
1179 "int-virtual-link-desc".format(df
["id"], affected_ivld_profile
["id"])
1181 norm(str(e
.exception
)),
1182 "Wrong exception text",
1185 def test_validate_monitoring_params_on_valid_descriptor(self
):
1186 indata
= db_vnfd_content
1187 self
.topic
.validate_monitoring_params(indata
)
1189 def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self
):
1190 indata
= deepcopy(db_vnfd_content
)
1191 duplicated_mp
= {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1192 affected_ivld
= indata
["int-virtual-link-desc"][0]
1193 affected_ivld
["monitoring-parameters"] = [duplicated_mp
, duplicated_mp
]
1194 with self
.assertRaises(EngineException
) as e
:
1195 self
.topic
.validate_monitoring_params(indata
)
1197 e
.exception
.http_code
,
1198 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1199 "Wrong HTTP status code",
1203 "Duplicated monitoring-parameter id in "
1204 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1205 affected_ivld
["id"], duplicated_mp
["id"]
1208 norm(str(e
.exception
)),
1209 "Wrong exception text",
1212 def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self
):
1213 indata
= deepcopy(db_vnfd_content
)
1215 "id": "dataVM_cpu_util",
1216 "name": "dataVM_cpu_util",
1217 "performance_metric": "cpu",
1219 affected_vdu
= indata
["vdu"][1]
1220 affected_vdu
["monitoring-parameter"].insert(0, duplicated_mp
)
1221 with self
.assertRaises(EngineException
) as e
:
1222 self
.topic
.validate_monitoring_params(indata
)
1224 e
.exception
.http_code
,
1225 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1226 "Wrong HTTP status code",
1230 "Duplicated monitoring-parameter id in "
1231 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1232 affected_vdu
["id"], duplicated_mp
["id"]
1235 norm(str(e
.exception
)),
1236 "Wrong exception text",
1239 def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self
):
1240 indata
= deepcopy(db_vnfd_content
)
1244 "performance_metric": "memory",
1246 affected_df
= indata
["df"][0]
1247 affected_df
["monitoring-parameter"] = [duplicated_mp
, duplicated_mp
]
1248 with self
.assertRaises(EngineException
) as e
:
1249 self
.topic
.validate_monitoring_params(indata
)
1251 e
.exception
.http_code
,
1252 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1253 "Wrong HTTP status code",
1257 "Duplicated monitoring-parameter id in "
1258 "df[id='{}']:monitoring-parameter[id='{}']".format(
1259 affected_df
["id"], duplicated_mp
["id"]
1262 norm(str(e
.exception
)),
1263 "Wrong exception text",
1266 def test_validate_scaling_group_descriptor_on_valid_descriptor(self
):
1267 indata
= db_vnfd_content
1268 self
.topic
.validate_scaling_group_descriptor(indata
)
1270 def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self
):
1271 indata
= deepcopy(db_vnfd_content
)
1272 vdu
= indata
["vdu"][1]
1273 affected_df
= indata
["df"][0]
1274 affected_sa
= affected_df
["scaling-aspect"][0]
1275 affected_sp
= affected_sa
["scaling-policy"][0]
1276 affected_sc
= affected_sp
["scaling-criteria"][0]
1277 vdu
.pop("monitoring-parameter")
1278 with self
.assertRaises(EngineException
) as e
:
1279 self
.topic
.validate_scaling_group_descriptor(indata
)
1281 e
.exception
.http_code
,
1282 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1283 "Wrong HTTP status code",
1287 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1288 "[name='{}']:scaling-criteria[name='{}']: "
1289 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1292 affected_sp
["name"],
1293 affected_sc
["name"],
1294 affected_sc
["vnf-monitoring-param-ref"],
1297 norm(str(e
.exception
)),
1298 "Wrong exception text",
1301 def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self
):
1302 indata
= deepcopy(db_vnfd_content
)
1303 df
= indata
["df"][0]
1304 affected_sa
= df
["scaling-aspect"][0]
1305 indata
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1308 with self
.assertRaises(EngineException
) as e
:
1309 self
.topic
.validate_scaling_group_descriptor(indata
)
1311 e
.exception
.http_code
,
1312 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1313 "Wrong HTTP status code",
1317 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1318 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1319 df
["id"], affected_sa
["id"]
1322 norm(str(e
.exception
)),
1323 "Wrong exception text",
1326 def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1329 indata
= deepcopy(db_vnfd_content
)
1330 df
= indata
["df"][0]
1331 affected_sa
= df
["scaling-aspect"][0]
1332 affected_sca_primitive
= affected_sa
["scaling-config-action"][0][
1333 "vnf-config-primitive-name-ref"
1335 df
["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1338 with self
.assertRaises(EngineException
) as e
:
1339 self
.topic
.validate_scaling_group_descriptor(indata
)
1341 e
.exception
.http_code
,
1342 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1343 "Wrong HTTP status code",
1347 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1348 "config-primitive-name-ref='{}' does not match any "
1349 "day1-2 configuration:config-primitive:name".format(
1350 df
["id"], affected_sa
["id"], affected_sca_primitive
1353 norm(str(e
.exception
)),
1354 "Wrong exception text",
1357 def test_new_vnfd_revision(self
):
1358 did
= db_vnfd_content
["_id"]
1359 self
.fs
.get_params
.return_value
= {}
1360 self
.fs
.file_exists
.return_value
= False
1361 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1362 "/tmp/" + str(uuid4()), "a+b"
1364 test_vnfd
= deepcopy(db_vnfd_content
)
1365 del test_vnfd
["_id"]
1366 del test_vnfd
["_admin"]
1367 self
.db
.create
.return_value
= did
1369 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1370 db_args
= self
.db
.create
.call_args
[0]
1372 db_args
[1]["_admin"]["revision"], 0, "New package should be at revision 0"
1375 @patch("osm_nbi.descriptor_topics.shutil")
1376 @patch("osm_nbi.descriptor_topics.os.rename")
1377 def test_update_vnfd(self
, mock_rename
, mock_shutil
):
1379 did
= db_vnfd_content
["_id"]
1381 self
.fs
.get_params
.return_value
= {}
1382 self
.fs
.file_exists
.return_value
= False
1383 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1384 "/tmp/" + str(uuid4()), "a+b"
1386 new_vnfd
= deepcopy(db_vnfd_content
)
1388 self
.db
.create
.return_value
= did
1390 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1391 del new_vnfd
["vdu"][0]["cloud-init-file"]
1392 del new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1394 ][0]["execution-environment-list"][0]["juju"]
1396 old_vnfd
= {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])}
1397 old_vnfd
["_admin"]["revision"] = old_revision
1399 self
.db
.get_one
.side_effect
= [old_vnfd
, old_vnfd
, None]
1400 self
.topic
.upload_content(fake_session
, did
, new_vnfd
, {}, {"Content-Type": []})
1402 db_args
= self
.db
.replace
.call_args
[0]
1404 db_args
[2]["_admin"]["revision"],
1406 "Revision should increment",
1410 class Test_NsdTopic(TestCase
):
1412 def setUpClass(cls
):
1413 cls
.test_name
= "test-nsd-topic"
1416 def tearDownClass(cls
):
1420 self
.db
= Mock(dbbase
.DbBase())
1421 self
.fs
= Mock(fsbase
.FsBase())
1422 self
.msg
= Mock(msgbase
.MsgBase())
1423 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1424 self
.topic
= NsdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1425 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1428 def assertNotRaises(self
, exception_type
):
1431 except exception_type
:
1432 raise self
.failureException("{} raised".format(exception_type
.__name
__))
1434 def create_desc_temp(self
, template
):
1435 old_desc
= deepcopy(template
)
1436 new_desc
= deepcopy(template
)
1437 return old_desc
, new_desc
1439 def prepare_nsd_creation(self
):
1441 did
= db_nsd_content
["_id"]
1442 self
.fs
.get_params
.return_value
= {}
1443 self
.fs
.file_exists
.return_value
= False
1444 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1445 "/tmp/" + str(uuid4()), "a+b"
1447 self
.db
.get_one
.side_effect
= [
1448 {"_id": did
, "_admin": deepcopy(db_nsd_content
["_admin"])},
1451 test_nsd
= deepcopy(db_nsd_content
)
1453 del test_nsd
["_admin"]
1454 return did
, test_nsd
1456 @patch("osm_nbi.descriptor_topics.shutil")
1457 @patch("osm_nbi.descriptor_topics.os.rename")
1458 def test_new_nsd_normal_creation(self
, mock_rename
, mock_shutil
):
1459 did
, test_nsd
= self
.prepare_nsd_creation()
1460 self
.db
.create
.return_value
= did
1463 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1464 db_args
= self
.db
.create
.call_args
[0]
1465 msg_args
= self
.msg
.write
.call_args
[0]
1466 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1467 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1468 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
1469 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1470 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1471 self
.assertEqual(did2
, did
, "Wrong DB NSD id")
1472 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
1474 db_args
[1]["_admin"]["modified"],
1475 db_args
[1]["_admin"]["created"],
1476 "Wrong modification time",
1479 db_args
[1]["_admin"]["projects_read"],
1481 "Wrong read-only project list",
1484 db_args
[1]["_admin"]["projects_write"],
1486 "Wrong read-write project list",
1489 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1491 self
.topic
.upload_content(fake_session
, did
, test_nsd
, {}, {"Content-Type": []})
1492 msg_args
= self
.msg
.write
.call_args
[0]
1493 test_nsd
["_id"] = did
1494 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1495 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1496 self
.assertEqual(msg_args
[2], test_nsd
, "Wrong message content")
1498 db_args
= self
.db
.get_one
.mock_calls
[0][1]
1499 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1500 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB NSD id")
1502 db_args
= self
.db
.replace
.call_args
[0]
1503 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1504 self
.assertEqual(db_args
[1], did
, "Wrong DB NSD id")
1506 admin
= db_args
[2]["_admin"]
1507 db_admin
= db_nsd_content
["_admin"]
1508 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
1510 admin
["modified"], db_admin
["created"], "Wrong modification time"
1513 admin
["projects_read"],
1514 db_admin
["projects_read"],
1515 "Wrong read-only project list",
1518 admin
["projects_write"],
1519 db_admin
["projects_write"],
1520 "Wrong read-write project list",
1523 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1526 admin
["operationalState"], "ENABLED", "Wrong operational state"
1528 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
1530 storage
= admin
["storage"]
1531 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
1532 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
1534 compare_desc(self
, test_nsd
, db_args
[2], "NSD")
1535 revision_args
= self
.db
.create
.call_args
[0]
1537 revision_args
[0], self
.topic
.topic
+ "_revisions", "Wrong topic"
1539 self
.assertEqual(revision_args
[1]["id"], db_args
[2]["id"], "Wrong revision id")
1541 revision_args
[1]["_id"], db_args
[2]["_id"] + ":1", "Wrong revision _id"
1544 @patch("osm_nbi.descriptor_topics.shutil")
1545 @patch("osm_nbi.descriptor_topics.os.rename")
1546 def test_new_nsd_check_pyangbind_validation_required_properties(
1547 self
, mock_rename
, mock_shutil
1549 did
, test_nsd
= self
.prepare_nsd_creation()
1552 with self
.assertRaises(
1553 EngineException
, msg
="Accepted NSD with a missing required property"
1555 self
.topic
.upload_content(
1556 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1559 e
.exception
.http_code
,
1560 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1561 "Wrong HTTP status code",
1564 norm("Error in pyangbind validation: '{}'".format("id")),
1565 norm(str(e
.exception
)),
1566 "Wrong exception text",
1569 @patch("osm_nbi.descriptor_topics.shutil")
1570 @patch("osm_nbi.descriptor_topics.os.rename")
1571 def test_new_nsd_check_pyangbind_validation_additional_properties(
1572 self
, mock_rename
, mock_shutil
1574 did
, test_nsd
= self
.prepare_nsd_creation()
1575 test_nsd
["extra-property"] = 0
1577 with self
.assertRaises(
1578 EngineException
, msg
="Accepted NSD with an additional property"
1580 self
.topic
.upload_content(
1581 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1584 e
.exception
.http_code
,
1585 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1586 "Wrong HTTP status code",
1590 "Error in pyangbind validation: {} ({})".format(
1591 "json object contained a key that did not exist", "extra-property"
1594 norm(str(e
.exception
)),
1595 "Wrong exception text",
1598 @patch("osm_nbi.descriptor_topics.shutil")
1599 @patch("osm_nbi.descriptor_topics.os.rename")
1600 def test_new_nsd_check_pyangbind_validation_property_types(
1601 self
, mock_rename
, mock_shutil
1603 did
, test_nsd
= self
.prepare_nsd_creation()
1604 test_nsd
["designer"] = {"key": 0}
1606 with self
.assertRaises(
1607 EngineException
, msg
="Accepted NSD with a wrongly typed property"
1609 self
.topic
.upload_content(
1610 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1613 e
.exception
.http_code
,
1614 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1615 "Wrong HTTP status code",
1619 "Error in pyangbind validation: {} ({})".format(
1620 "json object contained a key that did not exist", "key"
1623 norm(str(e
.exception
)),
1624 "Wrong exception text",
1627 @patch("osm_nbi.descriptor_topics.shutil")
1628 @patch("osm_nbi.descriptor_topics.os.rename")
1629 def test_new_nsd_check_input_validation_mgmt_network_virtual_link_protocol_data(
1630 self
, mock_rename
, mock_shutil
1632 did
, test_nsd
= self
.prepare_nsd_creation()
1633 df
= test_nsd
["df"][0]
1636 "virtual-link-desc-id": "mgmt",
1637 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1639 df
["virtual-link-profile"] = [mgmt_profile
]
1641 with self
.assertRaises(
1642 EngineException
, msg
="Accepted VLD with mgmt-network+ip-profile"
1644 self
.topic
.upload_content(
1645 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1648 e
.exception
.http_code
,
1649 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1650 "Wrong HTTP status code",
1654 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1655 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1656 df
["id"], mgmt_profile
["id"]
1659 norm(str(e
.exception
)),
1660 "Wrong exception text",
1663 @patch("osm_nbi.descriptor_topics.shutil")
1664 @patch("osm_nbi.descriptor_topics.os.rename")
1665 def test_new_nsd_check_descriptor_dependencies_vnfd_id(
1666 self
, mock_rename
, mock_shutil
1668 did
, test_nsd
= self
.prepare_nsd_creation()
1669 self
.db
.get_list
.return_value
= []
1671 with self
.assertRaises(
1672 EngineException
, msg
="Accepted wrong VNFD ID reference"
1674 self
.topic
.upload_content(
1675 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1678 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1682 "'vnfd-id'='{}' references a non existing vnfd".format(
1683 test_nsd
["vnfd-id"][0]
1686 norm(str(e
.exception
)),
1687 "Wrong exception text",
1690 @patch("osm_nbi.descriptor_topics.shutil")
1691 @patch("osm_nbi.descriptor_topics.os.rename")
1692 def test_new_nsd_check_descriptor_dependencies_vld_vnfd_connection_point_ref(
1693 self
, mock_rename
, mock_shutil
1695 # Check Descriptor Dependencies: "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]
1696 did
, test_nsd
= self
.prepare_nsd_creation()
1697 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1698 df
= test_nsd
["df"][0]
1699 affected_vnf_profile
= df
["vnf-profile"][0]
1700 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
1701 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
1702 self
.db
.get_list
.return_value
= [vnfd_descriptor
]
1704 with self
.assertRaises(
1705 EngineException
, msg
="Accepted wrong VLD CP reference"
1707 self
.topic
.upload_content(
1708 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1711 e
.exception
.http_code
,
1712 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1713 "Wrong HTTP status code",
1717 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1718 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1719 "non existing ext-cpd:id inside vnfd '{}'".format(
1721 affected_vnf_profile
["id"],
1722 affected_virtual_link
["virtual-link-profile-id"],
1724 vnfd_descriptor
["id"],
1727 norm(str(e
.exception
)),
1728 "Wrong exception text",
1731 def test_edit_nsd(self
):
1732 nsd_content
= deepcopy(db_nsd_content
)
1733 did
= nsd_content
["_id"]
1734 self
.fs
.file_exists
.return_value
= True
1735 self
.fs
.dir_ls
.return_value
= True
1736 with self
.subTest(i
=1, t
="Normal Edition"):
1738 self
.db
.get_one
.side_effect
= [deepcopy(nsd_content
), None]
1739 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1740 data
= {"id": "new-nsd-id", "name": "new-nsd-name"}
1741 self
.topic
.edit(fake_session
, did
, data
)
1742 db_args
= self
.db
.replace
.call_args
[0]
1743 msg_args
= self
.msg
.write
.call_args
[0]
1745 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1746 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1747 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
1748 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1749 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
1751 db_args
[2]["_admin"]["created"],
1752 nsd_content
["_admin"]["created"],
1753 "Wrong creation time",
1756 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
1759 db_args
[2]["_admin"]["projects_read"],
1760 nsd_content
["_admin"]["projects_read"],
1761 "Wrong read-only project list",
1764 db_args
[2]["_admin"]["projects_write"],
1765 nsd_content
["_admin"]["projects_write"],
1766 "Wrong read-write project list",
1768 self
.assertEqual(db_args
[2]["id"], data
["id"], "Wrong NSD ID")
1769 self
.assertEqual(db_args
[2]["name"], data
["name"], "Wrong NSD Name")
1770 with self
.subTest(i
=2, t
="Conflict on Edit"):
1771 data
= {"id": "fake-nsd-id", "name": "new-nsd-name"}
1772 self
.db
.get_one
.side_effect
= [
1774 {"_id": str(uuid4()), "id": data
["id"]},
1776 with self
.assertRaises(
1777 EngineException
, msg
="Accepted existing NSD ID"
1779 self
.topic
.edit(fake_session
, did
, data
)
1781 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1785 "{} with id '{}' already exists for this project".format(
1789 norm(str(e
.exception
)),
1790 "Wrong exception text",
1792 with self
.subTest(i
=3, t
="Check Envelope"):
1793 data
= {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
1794 self
.db
.get_one
.side_effect
= [nsd_content
, None]
1795 with self
.assertRaises(
1796 EngineException
, msg
="Accepted NSD with wrong envelope"
1798 self
.topic
.edit(fake_session
, did
, data
, content
=nsd_content
)
1800 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
1803 "'nsd' must be a list of only one element",
1804 norm(str(e
.exception
)),
1805 "Wrong exception text",
1807 self
.db
.reset_mock()
1810 def test_delete_nsd(self
):
1811 did
= db_nsd_content
["_id"]
1812 self
.db
.get_one
.return_value
= db_nsd_content
1813 p_id
= db_nsd_content
["_admin"]["projects_read"][0]
1814 with self
.subTest(i
=1, t
="Normal Deletion"):
1815 self
.db
.get_list
.return_value
= []
1816 self
.db
.del_one
.return_value
= {"deleted": 1}
1817 self
.topic
.delete(fake_session
, did
)
1818 db_args
= self
.db
.del_one
.call_args
[0]
1819 msg_args
= self
.msg
.write
.call_args
[0]
1820 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1821 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
1822 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1823 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1824 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
1826 db_args
[1]["_admin.projects_write.cont"],
1830 db_g1_args
= self
.db
.get_one
.call_args
[0]
1831 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1832 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB NSD ID")
1833 db_gl_calls
= self
.db
.get_list
.call_args_list
1834 self
.assertEqual(db_gl_calls
[0][0][0], "nsrs", "Wrong DB topic")
1835 # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID") # Filter changed after call
1836 self
.assertEqual(db_gl_calls
[1][0][0], "nsts", "Wrong DB topic")
1838 db_gl_calls
[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
1839 db_nsd_content
["id"],
1840 "Wrong DB NSD netslice-subnet nsd-ref",
1842 self
.db
.set_one
.assert_not_called()
1843 fs_del_calls
= self
.fs
.file_delete
.call_args_list
1844 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
1845 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
1846 with self
.subTest(i
=2, t
="Conflict on Delete - NSD in use by nsr"):
1847 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-nsr"}]
1848 with self
.assertRaises(
1849 EngineException
, msg
="Accepted NSD in use by NSR"
1851 self
.topic
.delete(fake_session
, did
)
1853 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1856 "there is at least one ns instance using this descriptor",
1857 norm(str(e
.exception
)),
1858 "Wrong exception text",
1860 with self
.subTest(i
=3, t
="Conflict on Delete - NSD in use by NST"):
1861 self
.db
.get_list
.side_effect
= [
1863 [{"_id": str(uuid4()), "name": "fake-nst"}],
1865 with self
.assertRaises(
1866 EngineException
, msg
="Accepted NSD in use by NST"
1868 self
.topic
.delete(fake_session
, did
)
1870 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1873 "there is at least one netslice template referencing this descriptor",
1874 norm(str(e
.exception
)),
1875 "Wrong exception text",
1877 with self
.subTest(i
=4, t
="Non-existent NSD"):
1878 excp_msg
= "Not found any {} with filter='{}'".format("NSD", {"_id": did
})
1879 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
1880 with self
.assertRaises(
1881 DbException
, msg
="Accepted non-existent NSD ID"
1883 self
.topic
.delete(fake_session
, did
)
1885 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1888 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
1890 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
1891 db_nsd_content
["_admin"]["projects_read"].append("other_project")
1892 self
.db
.get_one
= Mock(return_value
=db_nsd_content
)
1893 self
.db
.get_list
= Mock(return_value
=[])
1894 self
.msg
.write
.reset_mock()
1895 self
.db
.del_one
.reset_mock()
1896 self
.fs
.file_delete
.reset_mock()
1898 self
.topic
.delete(fake_session
, did
)
1899 self
.db
.del_one
.assert_not_called()
1900 self
.msg
.write
.assert_not_called()
1901 db_g1_args
= self
.db
.get_one
.call_args
[0]
1902 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1903 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
1904 db_s1_args
= self
.db
.set_one
.call_args
1905 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
1906 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
1908 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1911 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
1914 db_s1_args
[1]["pull_list"],
1915 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
1916 "Wrong DB pull_list dictionary",
1918 self
.fs
.file_delete
.assert_not_called()
1919 self
.db
.reset_mock()
1922 def prepare_nsd_validation(self
):
1923 descriptor_name
= "test_ns_descriptor"
1924 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1925 "/tmp/" + str(uuid4()), "a+b"
1927 old_nsd
, new_nsd
= self
.create_desc_temp(db_nsd_content
)
1928 return descriptor_name
, old_nsd
, new_nsd
1930 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1931 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1932 def test_validate_descriptor_ns_configuration_changed(
1933 self
, mock_safe_load
, mock_detect_usage
1935 """Validating NSD and NSD has changes in ns-configuration:config-primitive"""
1936 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
1937 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
1938 mock_detect_usage
.return_value
= True
1939 self
.db
.get_one
.return_value
= old_nsd
1941 {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
1944 {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
1948 with self
.assertNotRaises(EngineException
):
1949 self
.topic
._validate
_descriptor
_changes
(
1950 old_nsd
["_id"], descriptor_name
, "/tmp", "/tmp:1"
1952 self
.db
.get_one
.assert_called_once()
1953 mock_detect_usage
.assert_called_once()
1954 self
.assertEqual(mock_safe_load
.call_count
, 2)
1956 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1957 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1958 def test_validate_descriptor_nsd_name_changed(
1959 self
, mock_safe_load
, mock_detect_usage
1961 """Validating NSD, NSD name has changed."""
1962 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
1963 did
= old_nsd
["_id"]
1964 new_nsd
["name"] = "nscharm-ns2"
1965 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
1966 mock_detect_usage
.return_value
= True
1967 self
.db
.get_one
.return_value
= old_nsd
1969 with self
.assertRaises(
1970 EngineException
, msg
="there are disallowed changes in the ns descriptor"
1972 self
.topic
._validate
_descriptor
_changes
(
1973 did
, descriptor_name
, "/tmp", "/tmp:1"
1976 e
.exception
.http_code
,
1977 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1978 "Wrong HTTP status code",
1981 norm("there are disallowed changes in the ns descriptor"),
1982 norm(str(e
.exception
)),
1983 "Wrong exception text",
1986 self
.db
.get_one
.assert_called_once()
1987 mock_detect_usage
.assert_called_once()
1988 self
.assertEqual(mock_safe_load
.call_count
, 2)
1990 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1991 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1992 def test_validate_descriptor_nsd_name_changed_nsd_not_in_use(
1993 self
, mock_safe_load
, mock_detect_usage
1995 """Validating NSD, NSD name has changed, NSD is not in use."""
1996 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
1997 did
= old_nsd
["_id"]
1998 new_nsd
["name"] = "nscharm-ns2"
1999 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
2000 mock_detect_usage
.return_value
= None
2001 self
.db
.get_one
.return_value
= old_nsd
2003 with self
.assertNotRaises(Exception):
2004 self
.topic
._validate
_descriptor
_changes
(
2005 did
, descriptor_name
, "/tmp", "/tmp:1"
2008 self
.db
.get_one
.assert_called_once()
2009 mock_detect_usage
.assert_called_once()
2010 mock_safe_load
.assert_not_called()
2012 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
2015 indata
= deepcopy(db_nsd_content
)
2016 vld
= indata
["virtual-link-desc"][0]
2017 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2021 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
2024 indata
= deepcopy(db_nsd_content
)
2025 vld
= indata
["virtual-link-desc"][0]
2026 df
= indata
["df"][0]
2029 "virtual-link-desc-id": "mgmt",
2030 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
2032 df
["virtual-link-profile"] = [affected_vlp
]
2033 with self
.assertRaises(EngineException
) as e
:
2034 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2038 e
.exception
.http_code
,
2039 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2040 "Wrong HTTP status code",
2044 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
2045 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
2046 df
["id"], affected_vlp
["id"]
2049 norm(str(e
.exception
)),
2050 "Wrong exception text",
2053 def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self
):
2054 indata
= deepcopy(db_nsd_content
)
2055 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2057 def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self
):
2058 indata
= deepcopy(db_nsd_content
)
2059 df
= indata
["df"][0]
2060 affected_vnf_profile
= df
["vnf-profile"][0]
2061 indata
["vnfd-id"] = ["non-existing-vnfd"]
2062 with self
.assertRaises(EngineException
) as e
:
2063 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2065 e
.exception
.http_code
,
2066 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2067 "Wrong HTTP status code",
2071 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
2072 "does not match any vnfd-id".format(
2074 affected_vnf_profile
["id"],
2075 affected_vnf_profile
["vnfd-id"],
2078 norm(str(e
.exception
)),
2079 "Wrong exception text",
2082 def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
2085 nsd_descriptor
= deepcopy(db_nsd_content
)
2086 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2087 df
= nsd_descriptor
["df"][0]
2088 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2089 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2093 def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
2096 nsd_descriptor
= deepcopy(db_nsd_content
)
2097 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2098 df
= nsd_descriptor
["df"][0]
2099 affected_vnf_profile
= df
["vnf-profile"][0]
2100 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
2101 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2102 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
2103 with self
.assertRaises(EngineException
) as e
:
2104 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2108 e
.exception
.http_code
,
2109 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2110 "Wrong HTTP status code",
2114 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
2115 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
2116 "non existing ext-cpd:id inside vnfd '{}'".format(
2118 affected_vnf_profile
["id"],
2119 affected_virtual_link
["virtual-link-profile-id"],
2121 vnfd_descriptor
["id"],
2124 norm(str(e
.exception
)),
2125 "Wrong exception text",
2128 def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self
):
2129 nsd_descriptor
= deepcopy(db_nsd_content
)
2130 invalid_vnfd_id
= "invalid-vnfd-id"
2131 nsd_descriptor
["id"] = "invalid-vnfd-id-ns"
2132 nsd_descriptor
["vnfd-id"][0] = invalid_vnfd_id
2133 nsd_descriptor
["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
2134 nsd_descriptor
["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
2135 with self
.assertRaises(EngineException
) as e
:
2136 self
.db
.get_list
.return_value
= []
2137 nsd_descriptor
= self
.topic
.check_conflict_on_edit(
2138 fake_session
, nsd_descriptor
, [], "id"
2141 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2145 "Descriptor error at 'vnfd-id'='{}' references a non "
2146 "existing vnfd".format(invalid_vnfd_id
)
2148 norm(str(e
.exception
)),
2149 "Wrong exception text",
2153 if __name__
== "__main__":