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 (
34 vnfd_exploit_fixed_text
,
37 from osm_nbi
.descriptor_topics
import VnfdTopic
, NsdTopic
38 from osm_nbi
.engine
import EngineException
39 from osm_common
.dbbase
import DbException
43 import collections
.abc
45 collections
.MutableSequence
= collections
.abc
.MutableSequence
47 test_name
= "test-user"
48 db_vnfd_content
= yaml
.safe_load(db_vnfds_text
)[0]
49 db_nsd_content
= yaml
.safe_load(db_nsds_text
)[0]
50 test_pid
= db_vnfd_content
["_admin"]["projects_read"][0]
52 "username": test_name
,
53 "project_id": (test_pid
,),
58 "allow_show_user_project_role": True,
60 UUID
= "00000000-0000-0000-0000-000000000000"
64 return {"projects_read": []}
67 def setup_mock_fs(fs
):
69 fs
.get_params
.return_value
= {}
70 fs
.file_exists
.return_value
= False
71 fs
.file_open
.side_effect
= lambda path
, mode
: tempfile
.TemporaryFile(mode
="a+b")
75 """Normalize string for checking"""
76 return " ".join(s
.strip().split()).lower()
79 def compare_desc(tc
, d1
, d2
, k
):
81 Compare two descriptors
82 We need this function because some methods are adding/removing items to/from the descriptors
83 before they are stored in the database, so the original and stored versions will differ
84 What we check is that COMMON LEAF ITEMS are equal
85 Lists of different length are not compared
86 :param tc: Test Case wich provides context (in particular the assert* methods)
87 :param d1,d2: Descriptors to be compared
88 :param k: key/item being compared
91 if isinstance(d1
, dict) and isinstance(d2
, dict):
94 compare_desc(tc
, d1
[key
], d2
[key
], k
+ "[{}]".format(key
))
95 elif isinstance(d1
, list) and isinstance(d2
, list) and len(d1
) == len(d2
):
96 for i
in range(len(d1
)):
97 compare_desc(tc
, d1
[i
], d2
[i
], k
+ "[{}]".format(i
))
99 tc
.assertEqual(d1
, d2
, "Wrong descriptor content: {}".format(k
))
102 class Test_VnfdTopic(TestCase
):
105 cls
.test_name
= "test-vnfd-topic"
108 def tearDownClass(cls
):
112 self
.db
= Mock(dbbase
.DbBase())
113 self
.fs
= Mock(fsbase
.FsBase())
114 self
.msg
= Mock(msgbase
.MsgBase())
115 self
.auth
= Mock(authconn
.Authconn(None, None, None))
116 self
.topic
= VnfdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
117 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
120 def assertNotRaises(self
, exception_type
=Exception):
123 except exception_type
:
124 raise self
.failureException("{} raised".format(exception_type
.__name
__))
126 def create_desc_temp(self
, template
):
127 old_desc
= deepcopy(template
)
128 new_desc
= deepcopy(template
)
129 return old_desc
, new_desc
131 def prepare_vnfd_creation(self
):
132 setup_mock_fs(self
.fs
)
133 test_vnfd
= deepcopy(db_vnfd_content
)
134 did
= db_vnfd_content
["_id"]
135 self
.db
.create
.return_value
= did
136 self
.db
.get_one
.side_effect
= [
137 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
140 return did
, test_vnfd
142 def prepare_vnfd(self
, vnfd_text
):
143 setup_mock_fs(self
.fs
)
144 test_vnfd
= yaml
.safe_load(vnfd_text
)
145 self
.db
.create
.return_value
= UUID
146 self
.db
.get_one
.side_effect
= [
147 {"_id": UUID
, "_admin": admin_value()},
150 return UUID
, test_vnfd
152 def prepare_test_vnfd(self
, test_vnfd
):
154 del test_vnfd
["_admin"]
155 del test_vnfd
["vdu"][0]["cloud-init-file"]
156 del test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
158 ][0]["execution-environment-list"][0]["juju"]
161 @patch("osm_nbi.descriptor_topics.shutil")
162 @patch("osm_nbi.descriptor_topics.os.rename")
163 def test_new_vnfd_normal_creation(self
, mock_rename
, mock_shutil
):
164 did
, test_vnfd
= self
.prepare_vnfd_creation()
165 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
167 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
168 db_args
= self
.db
.create
.call_args
[0]
169 msg_args
= self
.msg
.write
.call_args
[0]
171 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
172 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
173 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
174 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
175 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
176 self
.assertEqual(did2
, did
, "Wrong DB VNFD id")
177 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
179 db_args
[1]["_admin"]["modified"],
180 db_args
[1]["_admin"]["created"],
181 "Wrong modification time",
184 db_args
[1]["_admin"]["projects_read"],
186 "Wrong read-only project list",
189 db_args
[1]["_admin"]["projects_write"],
191 "Wrong read-write project list",
194 self
.db
.get_one
.side_effect
= [
195 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
199 self
.topic
.upload_content(
200 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
202 msg_args
= self
.msg
.write
.call_args
[0]
203 test_vnfd
["_id"] = did
204 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
205 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
206 self
.assertEqual(msg_args
[2], test_vnfd
, "Wrong message content")
208 db_args
= self
.db
.get_one
.mock_calls
[0][1]
209 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
210 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB VNFD id")
212 db_args
= self
.db
.replace
.call_args
[0]
213 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
214 self
.assertEqual(db_args
[1], did
, "Wrong DB VNFD id")
216 admin
= db_args
[2]["_admin"]
217 db_admin
= deepcopy(db_vnfd_content
["_admin"])
218 self
.assertEqual(admin
["type"], "vnfd", "Wrong descriptor type")
219 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
221 admin
["modified"], db_admin
["created"], "Wrong modification time"
224 admin
["projects_read"],
225 db_admin
["projects_read"],
226 "Wrong read-only project list",
229 admin
["projects_write"],
230 db_admin
["projects_write"],
231 "Wrong read-write project list",
234 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
237 admin
["operationalState"], "ENABLED", "Wrong operational state"
239 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
241 storage
= admin
["storage"]
242 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
243 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
244 self
.assertEqual(admin
["revision"], 1, "Wrong revision number")
245 compare_desc(self
, test_vnfd
, db_args
[2], "VNFD")
247 @patch("osm_nbi.descriptor_topics.shutil")
248 @patch("osm_nbi.descriptor_topics.os.rename")
249 def test_new_vnfd_exploit(self
, mock_rename
, mock_shutil
):
250 id, test_vnfd
= self
.prepare_vnfd(vnfd_exploit_text
)
252 with self
.assertRaises(EngineException
):
253 self
.topic
.upload_content(
254 fake_session
, id, test_vnfd
, {}, {"Content-Type": []}
257 @patch("osm_nbi.descriptor_topics.shutil")
258 @patch("osm_nbi.descriptor_topics.os.rename")
259 def test_new_vnfd_valid_helm_chart(self
, mock_rename
, mock_shutil
):
260 id, test_vnfd
= self
.prepare_vnfd(vnfd_exploit_fixed_text
)
262 with self
.assertNotRaises():
263 self
.topic
.upload_content(
264 fake_session
, id, test_vnfd
, {}, {"Content-Type": []}
267 @patch("osm_nbi.descriptor_topics.shutil")
268 @patch("osm_nbi.descriptor_topics.os.rename")
269 def test_new_vnfd_check_pyangbind_validation_additional_properties(
270 self
, mock_rename
, mock_shutil
272 did
, test_vnfd
= self
.prepare_vnfd_creation()
273 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
274 self
.topic
.upload_content(
275 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
277 test_vnfd
["_id"] = did
278 test_vnfd
["extra-property"] = 0
279 self
.db
.get_one
.side_effect
= (
280 lambda table
, filter, fail_on_empty
=None, fail_on_more
=None: {
282 "_admin": deepcopy(db_vnfd_content
["_admin"]),
286 with self
.assertRaises(
287 EngineException
, msg
="Accepted VNFD with an additional property"
289 self
.topic
.upload_content(
290 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
293 e
.exception
.http_code
,
294 HTTPStatus
.UNPROCESSABLE_ENTITY
,
295 "Wrong HTTP status code",
299 "Error in pyangbind validation: {} ({})".format(
300 "json object contained a key that did not exist", "extra-property"
303 norm(str(e
.exception
)),
304 "Wrong exception text",
306 db_args
= self
.db
.replace
.call_args
[0]
307 admin
= db_args
[2]["_admin"]
308 self
.assertEqual(admin
["revision"], 1, "Wrong revision number")
310 @patch("osm_nbi.descriptor_topics.shutil")
311 @patch("osm_nbi.descriptor_topics.os.rename")
312 def test_new_vnfd_check_pyangbind_validation_property_types(
313 self
, mock_rename
, mock_shutil
315 did
, test_vnfd
= self
.prepare_vnfd_creation()
316 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
317 test_vnfd
["_id"] = did
318 test_vnfd
["product-name"] = {"key": 0}
320 with self
.assertRaises(
321 EngineException
, msg
="Accepted VNFD with a wrongly typed property"
323 self
.topic
.upload_content(
324 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
327 e
.exception
.http_code
,
328 HTTPStatus
.UNPROCESSABLE_ENTITY
,
329 "Wrong HTTP status code",
333 "Error in pyangbind validation: {} ({})".format(
334 "json object contained a key that did not exist", "key"
337 norm(str(e
.exception
)),
338 "Wrong exception text",
341 @patch("osm_nbi.descriptor_topics.shutil")
342 @patch("osm_nbi.descriptor_topics.os.rename")
343 def test_new_vnfd_check_input_validation_cloud_init(self
, mock_rename
, mock_shutil
):
344 did
, test_vnfd
= self
.prepare_vnfd_creation()
345 del test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
347 ][0]["execution-environment-list"][0]["juju"]
349 with self
.assertRaises(
350 EngineException
, msg
="Accepted non-existent cloud_init file"
352 self
.topic
.upload_content(
353 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
356 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
360 "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format(
361 "cloud-init", test_vnfd
["id"], test_vnfd
["vdu"][0]["id"]
364 norm(str(e
.exception
)),
365 "Wrong exception text",
368 @patch("osm_nbi.descriptor_topics.shutil")
369 @patch("osm_nbi.descriptor_topics.os.rename")
370 def test_new_vnfd_check_input_validation_day12_configuration(
371 self
, mock_rename
, mock_shutil
373 did
, test_vnfd
= self
.prepare_vnfd_creation()
374 del test_vnfd
["vdu"][0]["cloud-init-file"]
376 with self
.assertRaises(
377 EngineException
, msg
="Accepted non-existent charm in VNF configuration"
379 self
.topic
.upload_content(
380 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
383 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
387 "{} defined in vnf[id={}] but not present in package".format(
388 "charm", test_vnfd
["id"]
391 norm(str(e
.exception
)),
392 "Wrong exception text",
395 @patch("osm_nbi.descriptor_topics.shutil")
396 @patch("osm_nbi.descriptor_topics.os.rename")
397 def test_new_vnfd_check_input_validation_mgmt_cp(self
, mock_rename
, mock_shutil
):
398 did
, test_vnfd
= self
.prepare_vnfd_creation()
399 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
400 del test_vnfd
["mgmt-cp"]
402 with self
.assertRaises(
403 EngineException
, msg
="Accepted VNFD without management interface"
405 self
.topic
.upload_content(
406 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
409 e
.exception
.http_code
,
410 HTTPStatus
.UNPROCESSABLE_ENTITY
,
411 "Wrong HTTP status code",
414 norm("'{}' is a mandatory field and it is not defined".format("mgmt-cp")),
415 norm(str(e
.exception
)),
416 "Wrong exception text",
419 @patch("osm_nbi.descriptor_topics.shutil")
420 @patch("osm_nbi.descriptor_topics.os.rename")
421 def test_new_vnfd_check_input_validation_mgmt_cp_connection_point(
422 self
, mock_rename
, mock_shutil
424 did
, test_vnfd
= self
.prepare_vnfd_creation()
425 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
426 test_vnfd
["mgmt-cp"] = "wrong-cp"
428 with self
.assertRaises(
429 EngineException
, msg
="Accepted wrong mgmt-cp connection point"
431 self
.topic
.upload_content(
432 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
435 e
.exception
.http_code
,
436 HTTPStatus
.UNPROCESSABLE_ENTITY
,
437 "Wrong HTTP status code",
441 "mgmt-cp='{}' must match an existing ext-cpd".format(
445 norm(str(e
.exception
)),
446 "Wrong exception text",
449 @patch("osm_nbi.descriptor_topics.shutil")
450 @patch("osm_nbi.descriptor_topics.os.rename")
451 def test_new_vnfd_check_input_validation_vdu_int_cpd(
452 self
, mock_rename
, mock_shutil
454 """Testing input validation during new vnfd creation
455 for vdu internal connection point"""
456 did
, test_vnfd
= self
.prepare_vnfd_creation()
457 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
458 ext_cpd
= test_vnfd
["ext-cpd"][1]
459 ext_cpd
["int-cpd"]["cpd"] = "wrong-cpd"
461 with self
.assertRaises(
462 EngineException
, msg
="Accepted wrong ext-cpd internal connection point"
464 self
.topic
.upload_content(
465 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
468 e
.exception
.http_code
,
469 HTTPStatus
.UNPROCESSABLE_ENTITY
,
470 "Wrong HTTP status code",
474 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
478 norm(str(e
.exception
)),
479 "Wrong exception text",
482 @patch("osm_nbi.descriptor_topics.shutil")
483 @patch("osm_nbi.descriptor_topics.os.rename")
484 def test_new_vnfd_check_input_validation_duplicated_vld(
485 self
, mock_rename
, mock_shutil
487 """Testing input validation during new vnfd creation
488 for dublicated virtual link description"""
489 did
, test_vnfd
= self
.prepare_vnfd_creation()
490 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
491 test_vnfd
["int-virtual-link-desc"].insert(0, {"id": "internal"})
493 with self
.assertRaises(
494 EngineException
, msg
="Accepted duplicated VLD name"
496 self
.topic
.upload_content(
497 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
500 e
.exception
.http_code
,
501 HTTPStatus
.UNPROCESSABLE_ENTITY
,
502 "Wrong HTTP status code",
506 "identifier id '{}' is not unique".format(
507 test_vnfd
["int-virtual-link-desc"][0]["id"]
510 norm(str(e
.exception
)),
511 "Wrong exception text",
514 @patch("osm_nbi.descriptor_topics.shutil")
515 @patch("osm_nbi.descriptor_topics.os.rename")
516 def test_new_vnfd_check_input_validation_vdu_int_virtual_link_desc(
517 self
, mock_rename
, mock_shutil
519 """Testing input validation during new vnfd creation
520 for vdu internal virtual link description"""
521 did
, test_vnfd
= self
.prepare_vnfd_creation()
522 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
523 vdu
= test_vnfd
["vdu"][0]
524 int_cpd
= vdu
["int-cpd"][1]
525 int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
527 with self
.assertRaises(
528 EngineException
, msg
="Accepted int-virtual-link-desc"
530 self
.topic
.upload_content(
531 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
534 e
.exception
.http_code
,
535 HTTPStatus
.UNPROCESSABLE_ENTITY
,
536 "Wrong HTTP status code",
540 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
541 "int-virtual-link-desc".format(
542 vdu
["id"], int_cpd
["id"], int_cpd
["int-virtual-link-desc"]
545 norm(str(e
.exception
)),
546 "Wrong exception text",
549 @patch("osm_nbi.descriptor_topics.shutil")
550 @patch("osm_nbi.descriptor_topics.os.rename")
551 def test_new_vnfd_check_input_validation_virtual_link_profile(
552 self
, mock_rename
, mock_shutil
554 """Testing input validation during new vnfd creation
555 for virtual link profile"""
556 did
, test_vnfd
= self
.prepare_vnfd_creation()
557 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
558 fake_ivld_profile
= {"id": "fake-profile-ref", "flavour": "fake-flavour"}
559 df
= test_vnfd
["df"][0]
560 df
["virtual-link-profile"] = [fake_ivld_profile
]
562 with self
.assertRaises(
563 EngineException
, msg
="Accepted non-existent Profile Ref"
565 self
.topic
.upload_content(
566 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
569 e
.exception
.http_code
,
570 HTTPStatus
.UNPROCESSABLE_ENTITY
,
571 "Wrong HTTP status code",
575 "df[id='{}']:virtual-link-profile='{}' must match an existing "
576 "int-virtual-link-desc".format(df
["id"], fake_ivld_profile
["id"])
578 norm(str(e
.exception
)),
579 "Wrong exception text",
582 @patch("osm_nbi.descriptor_topics.shutil")
583 @patch("osm_nbi.descriptor_topics.os.rename")
584 def test_new_vnfd_check_input_validation_scaling_criteria_monitoring_param_ref(
585 self
, mock_rename
, mock_shutil
587 """Testing input validation during new vnfd creation
588 for scaling criteria without monitoring parameter"""
589 did
, test_vnfd
= self
.prepare_vnfd_creation()
590 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
591 vdu
= test_vnfd
["vdu"][1]
592 affected_df
= test_vnfd
["df"][0]
593 sa
= affected_df
["scaling-aspect"][0]
594 sp
= sa
["scaling-policy"][0]
595 sc
= sp
["scaling-criteria"][0]
596 vdu
.pop("monitoring-parameter")
598 with self
.assertRaises(
599 EngineException
, msg
="Accepted non-existent Scaling Group Policy Criteria"
601 self
.topic
.upload_content(
602 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
605 e
.exception
.http_code
,
606 HTTPStatus
.UNPROCESSABLE_ENTITY
,
607 "Wrong HTTP status code",
611 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
612 "[name='{}']:scaling-criteria[name='{}']: "
613 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
618 sc
["vnf-monitoring-param-ref"],
621 norm(str(e
.exception
)),
622 "Wrong exception text",
625 @patch("osm_nbi.descriptor_topics.shutil")
626 @patch("osm_nbi.descriptor_topics.os.rename")
627 def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration(
628 self
, mock_rename
, mock_shutil
630 """Testing input validation during new vnfd creation
631 for scaling criteria without day12 configuration"""
632 did
, test_vnfd
= self
.prepare_vnfd_creation()
633 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
634 test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
637 df
= test_vnfd
["df"][0]
639 with self
.assertRaises(
640 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
642 self
.topic
.upload_content(
643 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
646 e
.exception
.http_code
,
647 HTTPStatus
.UNPROCESSABLE_ENTITY
,
648 "Wrong HTTP status code",
652 "'day1-2 configuration' not defined in the descriptor but it is referenced "
653 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
654 df
["id"], df
["scaling-aspect"][0]["id"]
657 norm(str(e
.exception
)),
658 "Wrong exception text",
661 @patch("osm_nbi.descriptor_topics.shutil")
662 @patch("osm_nbi.descriptor_topics.os.rename")
663 def test_new_vnfd_check_input_validation_scaling_config_action(
664 self
, mock_rename
, mock_shutil
666 """Testing input validation during new vnfd creation
667 for scaling criteria wrong config primitive"""
668 did
, test_vnfd
= self
.prepare_vnfd_creation()
669 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
670 df
= test_vnfd
["df"][0]
671 affected_df
= test_vnfd
["df"][0]
672 sa
= affected_df
["scaling-aspect"][0]
673 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
674 "operate-vnf-op-config"
675 )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
677 with self
.assertRaises(
678 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
680 self
.topic
.upload_content(
681 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
684 e
.exception
.http_code
,
685 HTTPStatus
.UNPROCESSABLE_ENTITY
,
686 "Wrong HTTP status code",
690 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
691 "config-primitive-name-ref='{}' does not match any "
692 "day1-2 configuration:config-primitive:name".format(
694 df
["scaling-aspect"][0]["id"],
695 sa
["scaling-config-action"][0]["vnf-config-primitive-name-ref"],
698 norm(str(e
.exception
)),
699 "Wrong exception text",
702 @patch("osm_nbi.descriptor_topics.shutil")
703 @patch("osm_nbi.descriptor_topics.os.rename")
704 def test_new_vnfd_check_input_validation_everything_right(
705 self
, mock_rename
, mock_shutil
707 """Testing input validation during new vnfd creation
708 everything correct"""
709 did
, test_vnfd
= self
.prepare_vnfd_creation()
710 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
711 test_vnfd
["id"] = "fake-vnfd-id"
712 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
713 "operate-vnf-op-config"
714 )["day1-2"][0]["id"] = "fake-vnfd-id"
715 self
.db
.get_one
.side_effect
= [
716 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
719 rc
= self
.topic
.upload_content(
720 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
722 self
.assertTrue(rc
, "Input Validation: Unexpected failure")
724 def test_edit_vnfd(self
):
725 vnfd_content
= deepcopy(db_vnfd_content
)
726 did
= vnfd_content
["_id"]
727 self
.fs
.file_exists
.return_value
= True
728 self
.fs
.dir_ls
.return_value
= True
729 with self
.subTest(i
=1, t
="Normal Edition"):
731 self
.db
.get_one
.side_effect
= [deepcopy(vnfd_content
), None]
732 data
= {"product-name": "new-vnfd-name"}
733 self
.topic
.edit(fake_session
, did
, data
)
734 db_args
= self
.db
.replace
.call_args
[0]
735 msg_args
= self
.msg
.write
.call_args
[0]
737 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
738 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
739 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
740 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
741 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
743 db_args
[2]["_admin"]["created"],
744 vnfd_content
["_admin"]["created"],
745 "Wrong creation time",
748 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
751 db_args
[2]["_admin"]["projects_read"],
752 vnfd_content
["_admin"]["projects_read"],
753 "Wrong read-only project list",
756 db_args
[2]["_admin"]["projects_write"],
757 vnfd_content
["_admin"]["projects_write"],
758 "Wrong read-write project list",
761 db_args
[2]["product-name"], data
["product-name"], "Wrong VNFD Name"
763 with self
.subTest(i
=2, t
="Conflict on Edit"):
764 data
= {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
765 self
.db
.get_one
.side_effect
= [
766 deepcopy(vnfd_content
),
767 {"_id": str(uuid4()), "id": data
["id"]},
769 with self
.assertRaises(
770 EngineException
, msg
="Accepted existing VNFD ID"
772 self
.topic
.edit(fake_session
, did
, data
)
774 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
778 "{} with id '{}' already exists for this project".format(
782 norm(str(e
.exception
)),
783 "Wrong exception text",
785 with self
.subTest(i
=3, t
="Check Envelope"):
786 data
= {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
787 with self
.assertRaises(
788 EngineException
, msg
="Accepted VNFD with wrong envelope"
790 self
.topic
.edit(fake_session
, did
, data
, content
=vnfd_content
)
792 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
795 "'vnfd' must be dict", norm(str(e
.exception
)), "Wrong exception text"
799 def test_delete_vnfd(self
):
800 did
= db_vnfd_content
["_id"]
801 self
.db
.get_one
.return_value
= db_vnfd_content
802 p_id
= db_vnfd_content
["_admin"]["projects_read"][0]
803 with self
.subTest(i
=1, t
="Normal Deletion"):
804 self
.db
.get_list
.return_value
= []
805 self
.db
.del_one
.return_value
= {"deleted": 1}
806 self
.topic
.delete(fake_session
, did
)
807 db_args
= self
.db
.del_one
.call_args
[0]
808 msg_args
= self
.msg
.write
.call_args
[0]
809 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
810 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
811 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
812 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
813 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
815 db_args
[1]["_admin.projects_write.cont"],
819 db_g1_args
= self
.db
.get_one
.call_args
[0]
820 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
821 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
822 db_gl_calls
= self
.db
.get_list
.call_args_list
823 self
.assertEqual(db_gl_calls
[0][0][0], "vnfrs", "Wrong DB topic")
824 # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID") # Filter changed after call
825 self
.assertEqual(db_gl_calls
[1][0][0], "nsds", "Wrong DB topic")
827 db_gl_calls
[1][0][1]["vnfd-id"],
828 db_vnfd_content
["id"],
829 "Wrong DB NSD vnfd-id",
833 self
.db
.del_list
.call_args
[0][0],
834 self
.topic
.topic
+ "_revisions",
839 self
.db
.del_list
.call_args
[0][1]["_id"]["$regex"],
841 "Wrong ID for rexep delete",
844 self
.db
.set_one
.assert_not_called()
845 fs_del_calls
= self
.fs
.file_delete
.call_args_list
846 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
847 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
848 with self
.subTest(i
=2, t
="Conflict on Delete - VNFD in use by VNFR"):
849 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-vnfr"}]
850 with self
.assertRaises(
851 EngineException
, msg
="Accepted VNFD in use by VNFR"
853 self
.topic
.delete(fake_session
, did
)
855 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
858 "there is at least one vnf instance using this descriptor",
859 norm(str(e
.exception
)),
860 "Wrong exception text",
862 with self
.subTest(i
=3, t
="Conflict on Delete - VNFD in use by NSD"):
863 self
.db
.get_list
.side_effect
= [
865 [{"_id": str(uuid4()), "name": "fake-nsd"}],
867 with self
.assertRaises(
868 EngineException
, msg
="Accepted VNFD in use by NSD"
870 self
.topic
.delete(fake_session
, did
)
872 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
875 "there is at least one ns package referencing this descriptor",
876 norm(str(e
.exception
)),
877 "Wrong exception text",
879 with self
.subTest(i
=4, t
="Non-existent VNFD"):
880 excp_msg
= "Not found any {} with filter='{}'".format("VNFD", {"_id": did
})
881 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
882 with self
.assertRaises(
883 DbException
, msg
="Accepted non-existent VNFD ID"
885 self
.topic
.delete(fake_session
, did
)
887 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
890 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
892 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
893 db_vnfd_content
["_admin"]["projects_read"].append("other_project")
894 self
.db
.get_one
= Mock(return_value
=db_vnfd_content
)
895 self
.db
.get_list
= Mock(return_value
=[])
896 self
.msg
.write
.reset_mock()
897 self
.db
.del_one
.reset_mock()
898 self
.fs
.file_delete
.reset_mock()
900 self
.topic
.delete(fake_session
, did
)
901 self
.db
.del_one
.assert_not_called()
902 self
.msg
.write
.assert_not_called()
903 db_g1_args
= self
.db
.get_one
.call_args
[0]
904 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
905 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
906 db_s1_args
= self
.db
.set_one
.call_args
907 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
908 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
910 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
913 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
916 db_s1_args
[1]["pull_list"],
917 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
918 "Wrong DB pull_list dictionary",
920 self
.fs
.file_delete
.assert_not_called()
923 def prepare_vnfd_validation(self
):
924 descriptor_name
= "test_descriptor"
925 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
926 "/tmp/" + str(uuid4()), "a+b"
928 old_vnfd
, new_vnfd
= self
.create_desc_temp(db_vnfd_content
)
929 return descriptor_name
, old_vnfd
, new_vnfd
931 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
932 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
933 def test_validate_vnfd_changes_day12_config_primitive_changed(
934 self
, mock_safe_load
, mock_detect_usage
936 """Validating VNFD for VNFD updates, day1-2 config primitive has changed"""
937 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
938 did
= old_vnfd
["_id"]
939 new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
941 ][0]["config-primitive"][0]["name"] = "new_action"
942 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
943 mock_detect_usage
.return_value
= True
944 self
.db
.get_one
.return_value
= old_vnfd
946 with self
.assertNotRaises(EngineException
):
947 self
.topic
._validate
_descriptor
_changes
(
948 did
, descriptor_name
, "/tmp/", "/tmp:1/"
950 self
.db
.get_one
.assert_called_once()
951 mock_detect_usage
.assert_called_once()
952 self
.assertEqual(mock_safe_load
.call_count
, 2)
954 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
955 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
956 def test_validate_vnfd_changes_sw_version_changed(
957 self
, mock_safe_load
, mock_detect_usage
959 """Validating VNFD for updates, software version has changed"""
960 # old vnfd uses the default software version: 1.0
961 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
962 did
= old_vnfd
["_id"]
963 new_vnfd
["software-version"] = "1.3"
964 new_vnfd
["sw-image-desc"][0]["name"] = "new-image"
965 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
966 mock_detect_usage
.return_value
= True
967 self
.db
.get_one
.return_value
= old_vnfd
969 with self
.assertNotRaises(EngineException
):
970 self
.topic
._validate
_descriptor
_changes
(
971 did
, descriptor_name
, "/tmp/", "/tmp:1/"
973 self
.db
.get_one
.assert_called_once()
974 mock_detect_usage
.assert_called_once()
975 self
.assertEqual(mock_safe_load
.call_count
, 2)
977 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
978 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
979 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed(
980 self
, mock_safe_load
, mock_detect_usage
982 """Validating VNFD for updates, software version has not
983 changed, mgmt-cp has changed."""
984 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
985 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
986 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
987 did
= old_vnfd
["_id"]
988 mock_detect_usage
.return_value
= True
989 self
.db
.get_one
.return_value
= old_vnfd
991 with self
.assertRaises(
992 EngineException
, msg
="there are disallowed changes in the vnf descriptor"
994 self
.topic
._validate
_descriptor
_changes
(
995 did
, descriptor_name
, "/tmp/", "/tmp:1/"
999 e
.exception
.http_code
,
1000 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1001 "Wrong HTTP status code",
1004 norm("there are disallowed changes in the vnf descriptor"),
1005 norm(str(e
.exception
)),
1006 "Wrong exception text",
1008 self
.db
.get_one
.assert_called_once()
1009 mock_detect_usage
.assert_called_once()
1010 self
.assertEqual(mock_safe_load
.call_count
, 2)
1012 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1013 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1014 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed_vnfd_not_in_use(
1015 self
, mock_safe_load
, mock_detect_usage
1017 """Validating VNFD for updates, software version has not
1018 changed, mgmt-cp has changed, vnfd is not in use."""
1019 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
1020 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
1021 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
1022 did
= old_vnfd
["_id"]
1023 mock_detect_usage
.return_value
= None
1024 self
.db
.get_one
.return_value
= old_vnfd
1026 with self
.assertNotRaises(EngineException
):
1027 self
.topic
._validate
_descriptor
_changes
(
1028 did
, descriptor_name
, "/tmp/", "/tmp:1/"
1031 self
.db
.get_one
.assert_called_once()
1032 mock_detect_usage
.assert_called_once()
1033 mock_safe_load
.assert_not_called()
1035 def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self
):
1036 indata
= deepcopy(db_vnfd_content
)
1037 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1039 def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
1042 indata
= deepcopy(db_vnfd_content
)
1043 indata
["ext-cpd"] = []
1044 with self
.assertRaises(EngineException
) as e
:
1045 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1047 e
.exception
.http_code
,
1048 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1049 "Wrong HTTP status code",
1053 "mgmt-cp='{}' must match an existing ext-cpd".format(indata
["mgmt-cp"])
1055 norm(str(e
.exception
)),
1056 "Wrong exception text",
1059 def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self
):
1060 indata
= deepcopy(db_vnfd_content
)
1061 indata
.pop("mgmt-cp")
1062 with self
.assertRaises(EngineException
) as e
:
1063 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1065 e
.exception
.http_code
,
1066 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1067 "Wrong HTTP status code",
1070 norm("'mgmt-cp' is a mandatory field and it is not defined"),
1071 norm(str(e
.exception
)),
1072 "Wrong exception text",
1075 def test_validate_vdu_internal_connection_points_on_valid_descriptor(self
):
1076 indata
= db_vnfd_content
1077 vdu
= indata
["vdu"][0]
1078 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1080 def test_validate_external_connection_points_on_valid_descriptor(self
):
1081 indata
= db_vnfd_content
1082 self
.topic
.validate_external_connection_points(indata
)
1084 def test_validate_external_connection_points_when_missing_internal_connection_point(
1087 indata
= deepcopy(db_vnfd_content
)
1088 vdu
= indata
["vdu"][0]
1090 affected_ext_cpd
= indata
["ext-cpd"][0]
1091 with self
.assertRaises(EngineException
) as e
:
1092 self
.topic
.validate_external_connection_points(indata
)
1094 e
.exception
.http_code
,
1095 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1096 "Wrong HTTP status code",
1100 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
1101 affected_ext_cpd
["id"]
1104 norm(str(e
.exception
)),
1105 "Wrong exception text",
1108 def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
1111 indata
= deepcopy(db_vnfd_content
)
1112 vdu
= indata
["vdu"][0]
1116 "virtual-network-interface-requirement": [{"name": "duplicated"}],
1118 vdu
["int-cpd"].insert(0, duplicated_cpd
)
1119 with self
.assertRaises(EngineException
) as e
:
1120 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1122 e
.exception
.http_code
,
1123 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1124 "Wrong HTTP status code",
1128 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
1129 vdu
["id"], duplicated_cpd
["id"]
1132 norm(str(e
.exception
)),
1133 "Wrong exception text",
1136 def test_validate_external_connection_points_on_duplicated_external_connection_point(
1139 indata
= deepcopy(db_vnfd_content
)
1141 "id": "vnf-mgmt-ext",
1142 "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
1144 indata
["ext-cpd"].insert(0, duplicated_cpd
)
1145 with self
.assertRaises(EngineException
) as e
:
1146 self
.topic
.validate_external_connection_points(indata
)
1148 e
.exception
.http_code
,
1149 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1150 "Wrong HTTP status code",
1154 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
1155 duplicated_cpd
["id"]
1158 norm(str(e
.exception
)),
1159 "Wrong exception text",
1162 def test_validate_internal_virtual_links_on_valid_descriptor(self
):
1163 indata
= db_vnfd_content
1164 self
.topic
.validate_internal_virtual_links(indata
)
1166 def test_validate_internal_virtual_links_on_duplicated_ivld(self
):
1167 indata
= deepcopy(db_vnfd_content
)
1168 duplicated_vld
= {"id": "internal"}
1169 indata
["int-virtual-link-desc"].insert(0, duplicated_vld
)
1170 with self
.assertRaises(EngineException
) as e
:
1171 self
.topic
.validate_internal_virtual_links(indata
)
1173 e
.exception
.http_code
,
1174 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1175 "Wrong HTTP status code",
1179 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
1180 duplicated_vld
["id"]
1183 norm(str(e
.exception
)),
1184 "Wrong exception text",
1187 def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
1190 indata
= deepcopy(db_vnfd_content
)
1191 vdu
= indata
["vdu"][0]
1192 affected_int_cpd
= vdu
["int-cpd"][0]
1193 affected_int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
1194 with self
.assertRaises(EngineException
) as e
:
1195 self
.topic
.validate_internal_virtual_links(indata
)
1197 e
.exception
.http_code
,
1198 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1199 "Wrong HTTP status code",
1203 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
1204 "int-virtual-link-desc".format(
1206 affected_int_cpd
["id"],
1207 affected_int_cpd
["int-virtual-link-desc"],
1210 norm(str(e
.exception
)),
1211 "Wrong exception text",
1214 def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self
):
1215 indata
= deepcopy(db_vnfd_content
)
1216 affected_ivld_profile
= {"id": "non-existing-int-virtual-link-desc"}
1217 df
= indata
["df"][0]
1218 df
["virtual-link-profile"] = [affected_ivld_profile
]
1219 with self
.assertRaises(EngineException
) as e
:
1220 self
.topic
.validate_internal_virtual_links(indata
)
1222 e
.exception
.http_code
,
1223 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1224 "Wrong HTTP status code",
1228 "df[id='{}']:virtual-link-profile='{}' must match an existing "
1229 "int-virtual-link-desc".format(df
["id"], affected_ivld_profile
["id"])
1231 norm(str(e
.exception
)),
1232 "Wrong exception text",
1235 def test_validate_monitoring_params_on_valid_descriptor(self
):
1236 indata
= db_vnfd_content
1237 self
.topic
.validate_monitoring_params(indata
)
1239 def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self
):
1240 indata
= deepcopy(db_vnfd_content
)
1241 duplicated_mp
= {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1242 affected_ivld
= indata
["int-virtual-link-desc"][0]
1243 affected_ivld
["monitoring-parameters"] = [duplicated_mp
, duplicated_mp
]
1244 with self
.assertRaises(EngineException
) as e
:
1245 self
.topic
.validate_monitoring_params(indata
)
1247 e
.exception
.http_code
,
1248 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1249 "Wrong HTTP status code",
1253 "Duplicated monitoring-parameter id in "
1254 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1255 affected_ivld
["id"], duplicated_mp
["id"]
1258 norm(str(e
.exception
)),
1259 "Wrong exception text",
1262 def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self
):
1263 indata
= deepcopy(db_vnfd_content
)
1265 "id": "dataVM_cpu_util",
1266 "name": "dataVM_cpu_util",
1267 "performance_metric": "cpu",
1269 affected_vdu
= indata
["vdu"][1]
1270 affected_vdu
["monitoring-parameter"].insert(0, duplicated_mp
)
1271 with self
.assertRaises(EngineException
) as e
:
1272 self
.topic
.validate_monitoring_params(indata
)
1274 e
.exception
.http_code
,
1275 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1276 "Wrong HTTP status code",
1280 "Duplicated monitoring-parameter id in "
1281 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1282 affected_vdu
["id"], duplicated_mp
["id"]
1285 norm(str(e
.exception
)),
1286 "Wrong exception text",
1289 def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self
):
1290 indata
= deepcopy(db_vnfd_content
)
1294 "performance_metric": "memory",
1296 affected_df
= indata
["df"][0]
1297 affected_df
["monitoring-parameter"] = [duplicated_mp
, duplicated_mp
]
1298 with self
.assertRaises(EngineException
) as e
:
1299 self
.topic
.validate_monitoring_params(indata
)
1301 e
.exception
.http_code
,
1302 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1303 "Wrong HTTP status code",
1307 "Duplicated monitoring-parameter id in "
1308 "df[id='{}']:monitoring-parameter[id='{}']".format(
1309 affected_df
["id"], duplicated_mp
["id"]
1312 norm(str(e
.exception
)),
1313 "Wrong exception text",
1316 def test_validate_scaling_group_descriptor_on_valid_descriptor(self
):
1317 indata
= db_vnfd_content
1318 self
.topic
.validate_scaling_group_descriptor(indata
)
1320 def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self
):
1321 indata
= deepcopy(db_vnfd_content
)
1322 vdu
= indata
["vdu"][1]
1323 affected_df
= indata
["df"][0]
1324 affected_sa
= affected_df
["scaling-aspect"][0]
1325 affected_sp
= affected_sa
["scaling-policy"][0]
1326 affected_sc
= affected_sp
["scaling-criteria"][0]
1327 vdu
.pop("monitoring-parameter")
1328 with self
.assertRaises(EngineException
) as e
:
1329 self
.topic
.validate_scaling_group_descriptor(indata
)
1331 e
.exception
.http_code
,
1332 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1333 "Wrong HTTP status code",
1337 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1338 "[name='{}']:scaling-criteria[name='{}']: "
1339 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1342 affected_sp
["name"],
1343 affected_sc
["name"],
1344 affected_sc
["vnf-monitoring-param-ref"],
1347 norm(str(e
.exception
)),
1348 "Wrong exception text",
1351 def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self
):
1352 indata
= deepcopy(db_vnfd_content
)
1353 df
= indata
["df"][0]
1354 affected_sa
= df
["scaling-aspect"][0]
1355 indata
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1358 with self
.assertRaises(EngineException
) as e
:
1359 self
.topic
.validate_scaling_group_descriptor(indata
)
1361 e
.exception
.http_code
,
1362 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1363 "Wrong HTTP status code",
1367 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1368 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1369 df
["id"], affected_sa
["id"]
1372 norm(str(e
.exception
)),
1373 "Wrong exception text",
1376 def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1379 indata
= deepcopy(db_vnfd_content
)
1380 df
= indata
["df"][0]
1381 affected_sa
= df
["scaling-aspect"][0]
1382 affected_sca_primitive
= affected_sa
["scaling-config-action"][0][
1383 "vnf-config-primitive-name-ref"
1385 df
["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1388 with self
.assertRaises(EngineException
) as e
:
1389 self
.topic
.validate_scaling_group_descriptor(indata
)
1391 e
.exception
.http_code
,
1392 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1393 "Wrong HTTP status code",
1397 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1398 "config-primitive-name-ref='{}' does not match any "
1399 "day1-2 configuration:config-primitive:name".format(
1400 df
["id"], affected_sa
["id"], affected_sca_primitive
1403 norm(str(e
.exception
)),
1404 "Wrong exception text",
1407 def test_new_vnfd_revision(self
):
1408 did
= db_vnfd_content
["_id"]
1409 self
.fs
.get_params
.return_value
= {}
1410 self
.fs
.file_exists
.return_value
= False
1411 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1412 "/tmp/" + str(uuid4()), "a+b"
1414 test_vnfd
= deepcopy(db_vnfd_content
)
1415 del test_vnfd
["_id"]
1416 del test_vnfd
["_admin"]
1417 self
.db
.create
.return_value
= did
1419 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1420 db_args
= self
.db
.create
.call_args
[0]
1422 db_args
[1]["_admin"]["revision"], 0, "New package should be at revision 0"
1425 @patch("osm_nbi.descriptor_topics.shutil")
1426 @patch("osm_nbi.descriptor_topics.os.rename")
1427 def test_update_vnfd(self
, mock_rename
, mock_shutil
):
1429 did
= db_vnfd_content
["_id"]
1431 self
.fs
.get_params
.return_value
= {}
1432 self
.fs
.file_exists
.return_value
= False
1433 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1434 "/tmp/" + str(uuid4()), "a+b"
1436 new_vnfd
= deepcopy(db_vnfd_content
)
1438 self
.db
.create
.return_value
= did
1440 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1441 del new_vnfd
["vdu"][0]["cloud-init-file"]
1442 del new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1444 ][0]["execution-environment-list"][0]["juju"]
1446 old_vnfd
= {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])}
1447 old_vnfd
["_admin"]["revision"] = old_revision
1449 self
.db
.get_one
.side_effect
= [old_vnfd
, old_vnfd
, None]
1450 self
.topic
.upload_content(fake_session
, did
, new_vnfd
, {}, {"Content-Type": []})
1452 db_args
= self
.db
.replace
.call_args
[0]
1454 db_args
[2]["_admin"]["revision"],
1456 "Revision should increment",
1460 class Test_NsdTopic(TestCase
):
1462 def setUpClass(cls
):
1463 cls
.test_name
= "test-nsd-topic"
1466 def tearDownClass(cls
):
1470 self
.db
= Mock(dbbase
.DbBase())
1471 self
.fs
= Mock(fsbase
.FsBase())
1472 self
.msg
= Mock(msgbase
.MsgBase())
1473 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1474 self
.topic
= NsdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1475 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1478 def assertNotRaises(self
, exception_type
):
1481 except exception_type
:
1482 raise self
.failureException("{} raised".format(exception_type
.__name
__))
1484 def create_desc_temp(self
, template
):
1485 old_desc
= deepcopy(template
)
1486 new_desc
= deepcopy(template
)
1487 return old_desc
, new_desc
1489 def prepare_nsd_creation(self
):
1491 did
= db_nsd_content
["_id"]
1492 self
.fs
.get_params
.return_value
= {}
1493 self
.fs
.file_exists
.return_value
= False
1494 self
.fs
.file_open
.side_effect
= lambda path
, mode
: tempfile
.TemporaryFile(
1497 self
.db
.get_one
.side_effect
= [
1498 {"_id": did
, "_admin": deepcopy(db_nsd_content
["_admin"])},
1501 test_nsd
= deepcopy(db_nsd_content
)
1503 del test_nsd
["_admin"]
1504 return did
, test_nsd
1506 @patch("osm_nbi.descriptor_topics.shutil")
1507 @patch("osm_nbi.descriptor_topics.os.rename")
1508 def test_new_nsd_normal_creation(self
, mock_rename
, mock_shutil
):
1509 did
, test_nsd
= self
.prepare_nsd_creation()
1510 self
.db
.create
.return_value
= did
1513 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1514 db_args
= self
.db
.create
.call_args
[0]
1515 msg_args
= self
.msg
.write
.call_args
[0]
1516 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1517 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1518 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
1519 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1520 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1521 self
.assertEqual(did2
, did
, "Wrong DB NSD id")
1522 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
1524 db_args
[1]["_admin"]["modified"],
1525 db_args
[1]["_admin"]["created"],
1526 "Wrong modification time",
1529 db_args
[1]["_admin"]["projects_read"],
1531 "Wrong read-only project list",
1534 db_args
[1]["_admin"]["projects_write"],
1536 "Wrong read-write project list",
1539 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1541 self
.topic
.upload_content(fake_session
, did
, test_nsd
, {}, {"Content-Type": []})
1542 msg_args
= self
.msg
.write
.call_args
[0]
1543 test_nsd
["_id"] = did
1544 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1545 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1546 self
.assertEqual(msg_args
[2], test_nsd
, "Wrong message content")
1548 db_args
= self
.db
.get_one
.mock_calls
[0][1]
1549 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1550 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB NSD id")
1552 db_args
= self
.db
.replace
.call_args
[0]
1553 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1554 self
.assertEqual(db_args
[1], did
, "Wrong DB NSD id")
1556 admin
= db_args
[2]["_admin"]
1557 db_admin
= db_nsd_content
["_admin"]
1558 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
1560 admin
["modified"], db_admin
["created"], "Wrong modification time"
1563 admin
["projects_read"],
1564 db_admin
["projects_read"],
1565 "Wrong read-only project list",
1568 admin
["projects_write"],
1569 db_admin
["projects_write"],
1570 "Wrong read-write project list",
1573 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1576 admin
["operationalState"], "ENABLED", "Wrong operational state"
1578 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
1580 storage
= admin
["storage"]
1581 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
1582 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
1584 compare_desc(self
, test_nsd
, db_args
[2], "NSD")
1585 revision_args
= self
.db
.create
.call_args
[0]
1587 revision_args
[0], self
.topic
.topic
+ "_revisions", "Wrong topic"
1589 self
.assertEqual(revision_args
[1]["id"], db_args
[2]["id"], "Wrong revision id")
1591 revision_args
[1]["_id"], db_args
[2]["_id"] + ":1", "Wrong revision _id"
1594 @patch("osm_nbi.descriptor_topics.shutil")
1595 @patch("osm_nbi.descriptor_topics.os.rename")
1596 def test_new_nsd_check_pyangbind_validation_required_properties(
1597 self
, mock_rename
, mock_shutil
1599 did
, test_nsd
= self
.prepare_nsd_creation()
1602 with self
.assertRaises(
1603 EngineException
, msg
="Accepted NSD with a missing required property"
1605 self
.topic
.upload_content(
1606 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1609 e
.exception
.http_code
,
1610 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1611 "Wrong HTTP status code",
1614 norm("Error in pyangbind validation: '{}'".format("id")),
1615 norm(str(e
.exception
)),
1616 "Wrong exception text",
1619 @patch("osm_nbi.descriptor_topics.shutil")
1620 @patch("osm_nbi.descriptor_topics.os.rename")
1621 def test_new_nsd_check_pyangbind_validation_additional_properties(
1622 self
, mock_rename
, mock_shutil
1624 did
, test_nsd
= self
.prepare_nsd_creation()
1625 test_nsd
["extra-property"] = 0
1627 with self
.assertRaises(
1628 EngineException
, msg
="Accepted NSD with an additional property"
1630 self
.topic
.upload_content(
1631 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1634 e
.exception
.http_code
,
1635 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1636 "Wrong HTTP status code",
1640 "Error in pyangbind validation: {} ({})".format(
1641 "json object contained a key that did not exist", "extra-property"
1644 norm(str(e
.exception
)),
1645 "Wrong exception text",
1648 @patch("osm_nbi.descriptor_topics.shutil")
1649 @patch("osm_nbi.descriptor_topics.os.rename")
1650 def test_new_nsd_check_pyangbind_validation_property_types(
1651 self
, mock_rename
, mock_shutil
1653 did
, test_nsd
= self
.prepare_nsd_creation()
1654 test_nsd
["designer"] = {"key": 0}
1656 with self
.assertRaises(
1657 EngineException
, msg
="Accepted NSD with a wrongly typed property"
1659 self
.topic
.upload_content(
1660 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1663 e
.exception
.http_code
,
1664 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1665 "Wrong HTTP status code",
1669 "Error in pyangbind validation: {} ({})".format(
1670 "json object contained a key that did not exist", "key"
1673 norm(str(e
.exception
)),
1674 "Wrong exception text",
1677 @patch("osm_nbi.descriptor_topics.shutil")
1678 @patch("osm_nbi.descriptor_topics.os.rename")
1679 def test_new_nsd_check_input_validation_mgmt_network_virtual_link_protocol_data(
1680 self
, mock_rename
, mock_shutil
1682 did
, test_nsd
= self
.prepare_nsd_creation()
1683 df
= test_nsd
["df"][0]
1686 "virtual-link-desc-id": "mgmt",
1687 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1689 df
["virtual-link-profile"] = [mgmt_profile
]
1691 with self
.assertRaises(
1692 EngineException
, msg
="Accepted VLD with mgmt-network+ip-profile"
1694 self
.topic
.upload_content(
1695 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1698 e
.exception
.http_code
,
1699 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1700 "Wrong HTTP status code",
1704 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1705 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1706 df
["id"], mgmt_profile
["id"]
1709 norm(str(e
.exception
)),
1710 "Wrong exception text",
1713 @patch("osm_nbi.descriptor_topics.shutil")
1714 @patch("osm_nbi.descriptor_topics.os.rename")
1715 def test_new_nsd_check_descriptor_dependencies_vnfd_id(
1716 self
, mock_rename
, mock_shutil
1718 did
, test_nsd
= self
.prepare_nsd_creation()
1719 self
.db
.get_list
.return_value
= []
1721 with self
.assertRaises(
1722 EngineException
, msg
="Accepted wrong VNFD ID reference"
1724 self
.topic
.upload_content(
1725 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1728 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1732 "'vnfd-id'='{}' references a non existing vnfd".format(
1733 test_nsd
["vnfd-id"][0]
1736 norm(str(e
.exception
)),
1737 "Wrong exception text",
1740 @patch("osm_nbi.descriptor_topics.shutil")
1741 @patch("osm_nbi.descriptor_topics.os.rename")
1742 def test_new_nsd_check_descriptor_dependencies_vld_vnfd_connection_point_ref(
1743 self
, mock_rename
, mock_shutil
1745 # Check Descriptor Dependencies: "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]
1746 did
, test_nsd
= self
.prepare_nsd_creation()
1747 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1748 df
= test_nsd
["df"][0]
1749 affected_vnf_profile
= df
["vnf-profile"][0]
1750 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
1751 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
1752 self
.db
.get_list
.return_value
= [vnfd_descriptor
]
1754 with self
.assertRaises(
1755 EngineException
, msg
="Accepted wrong VLD CP reference"
1757 self
.topic
.upload_content(
1758 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1761 e
.exception
.http_code
,
1762 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1763 "Wrong HTTP status code",
1767 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1768 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1769 "non existing ext-cpd:id inside vnfd '{}'".format(
1771 affected_vnf_profile
["id"],
1772 affected_virtual_link
["virtual-link-profile-id"],
1774 vnfd_descriptor
["id"],
1777 norm(str(e
.exception
)),
1778 "Wrong exception text",
1781 def test_edit_nsd(self
):
1782 nsd_content
= deepcopy(db_nsd_content
)
1783 did
= nsd_content
["_id"]
1784 self
.fs
.file_exists
.return_value
= True
1785 self
.fs
.dir_ls
.return_value
= True
1786 with self
.subTest(i
=1, t
="Normal Edition"):
1788 self
.db
.get_one
.side_effect
= [deepcopy(nsd_content
), None]
1789 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1790 data
= {"id": "new-nsd-id", "name": "new-nsd-name"}
1791 self
.topic
.edit(fake_session
, did
, data
)
1792 db_args
= self
.db
.replace
.call_args
[0]
1793 msg_args
= self
.msg
.write
.call_args
[0]
1795 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1796 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1797 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
1798 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1799 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
1801 db_args
[2]["_admin"]["created"],
1802 nsd_content
["_admin"]["created"],
1803 "Wrong creation time",
1806 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
1809 db_args
[2]["_admin"]["projects_read"],
1810 nsd_content
["_admin"]["projects_read"],
1811 "Wrong read-only project list",
1814 db_args
[2]["_admin"]["projects_write"],
1815 nsd_content
["_admin"]["projects_write"],
1816 "Wrong read-write project list",
1818 self
.assertEqual(db_args
[2]["id"], data
["id"], "Wrong NSD ID")
1819 self
.assertEqual(db_args
[2]["name"], data
["name"], "Wrong NSD Name")
1820 with self
.subTest(i
=2, t
="Conflict on Edit"):
1821 data
= {"id": "fake-nsd-id", "name": "new-nsd-name"}
1822 self
.db
.get_one
.side_effect
= [
1824 {"_id": str(uuid4()), "id": data
["id"]},
1826 with self
.assertRaises(
1827 EngineException
, msg
="Accepted existing NSD ID"
1829 self
.topic
.edit(fake_session
, did
, data
)
1831 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1835 "{} with id '{}' already exists for this project".format(
1839 norm(str(e
.exception
)),
1840 "Wrong exception text",
1842 with self
.subTest(i
=3, t
="Check Envelope"):
1843 data
= {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
1844 self
.db
.get_one
.side_effect
= [nsd_content
, None]
1845 with self
.assertRaises(
1846 EngineException
, msg
="Accepted NSD with wrong envelope"
1848 self
.topic
.edit(fake_session
, did
, data
, content
=nsd_content
)
1850 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
1853 "'nsd' must be a list of only one element",
1854 norm(str(e
.exception
)),
1855 "Wrong exception text",
1857 self
.db
.reset_mock()
1860 def test_delete_nsd(self
):
1861 did
= db_nsd_content
["_id"]
1862 self
.db
.get_one
.return_value
= db_nsd_content
1863 p_id
= db_nsd_content
["_admin"]["projects_read"][0]
1864 with self
.subTest(i
=1, t
="Normal Deletion"):
1865 self
.db
.get_list
.return_value
= []
1866 self
.db
.del_one
.return_value
= {"deleted": 1}
1867 self
.topic
.delete(fake_session
, did
)
1868 db_args
= self
.db
.del_one
.call_args
[0]
1869 msg_args
= self
.msg
.write
.call_args
[0]
1870 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1871 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
1872 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1873 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1874 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
1876 db_args
[1]["_admin.projects_write.cont"],
1880 db_g1_args
= self
.db
.get_one
.call_args
[0]
1881 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1882 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB NSD ID")
1883 db_gl_calls
= self
.db
.get_list
.call_args_list
1884 self
.assertEqual(db_gl_calls
[0][0][0], "nsrs", "Wrong DB topic")
1885 # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID") # Filter changed after call
1886 self
.assertEqual(db_gl_calls
[1][0][0], "nsts", "Wrong DB topic")
1888 db_gl_calls
[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
1889 db_nsd_content
["id"],
1890 "Wrong DB NSD netslice-subnet nsd-ref",
1892 self
.db
.set_one
.assert_not_called()
1893 fs_del_calls
= self
.fs
.file_delete
.call_args_list
1894 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
1895 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
1896 with self
.subTest(i
=2, t
="Conflict on Delete - NSD in use by nsr"):
1897 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-nsr"}]
1898 with self
.assertRaises(
1899 EngineException
, msg
="Accepted NSD in use by NSR"
1901 self
.topic
.delete(fake_session
, did
)
1903 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1906 "there is at least one ns instance using this descriptor",
1907 norm(str(e
.exception
)),
1908 "Wrong exception text",
1910 with self
.subTest(i
=3, t
="Conflict on Delete - NSD in use by NST"):
1911 self
.db
.get_list
.side_effect
= [
1913 [{"_id": str(uuid4()), "name": "fake-nst"}],
1915 with self
.assertRaises(
1916 EngineException
, msg
="Accepted NSD in use by NST"
1918 self
.topic
.delete(fake_session
, did
)
1920 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1923 "there is at least one netslice template referencing this descriptor",
1924 norm(str(e
.exception
)),
1925 "Wrong exception text",
1927 with self
.subTest(i
=4, t
="Non-existent NSD"):
1928 excp_msg
= "Not found any {} with filter='{}'".format("NSD", {"_id": did
})
1929 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
1930 with self
.assertRaises(
1931 DbException
, msg
="Accepted non-existent NSD ID"
1933 self
.topic
.delete(fake_session
, did
)
1935 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1938 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
1940 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
1941 db_nsd_content
["_admin"]["projects_read"].append("other_project")
1942 self
.db
.get_one
= Mock(return_value
=db_nsd_content
)
1943 self
.db
.get_list
= Mock(return_value
=[])
1944 self
.msg
.write
.reset_mock()
1945 self
.db
.del_one
.reset_mock()
1946 self
.fs
.file_delete
.reset_mock()
1948 self
.topic
.delete(fake_session
, did
)
1949 self
.db
.del_one
.assert_not_called()
1950 self
.msg
.write
.assert_not_called()
1951 db_g1_args
= self
.db
.get_one
.call_args
[0]
1952 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1953 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
1954 db_s1_args
= self
.db
.set_one
.call_args
1955 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
1956 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
1958 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1961 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
1964 db_s1_args
[1]["pull_list"],
1965 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
1966 "Wrong DB pull_list dictionary",
1968 self
.fs
.file_delete
.assert_not_called()
1969 self
.db
.reset_mock()
1972 def prepare_nsd_validation(self
):
1973 descriptor_name
= "test_ns_descriptor"
1974 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1975 "/tmp/" + str(uuid4()), "a+b"
1977 old_nsd
, new_nsd
= self
.create_desc_temp(db_nsd_content
)
1978 return descriptor_name
, old_nsd
, new_nsd
1980 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1981 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1982 def test_validate_descriptor_ns_configuration_changed(
1983 self
, mock_safe_load
, mock_detect_usage
1985 """Validating NSD and NSD has changes in ns-configuration:config-primitive"""
1986 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
1987 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
1988 mock_detect_usage
.return_value
= True
1989 self
.db
.get_one
.return_value
= old_nsd
1991 {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
1994 {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
1997 with self
.assertNotRaises(EngineException
):
1998 self
.topic
._validate
_descriptor
_changes
(
1999 old_nsd
["_id"], descriptor_name
, "/tmp", "/tmp:1"
2001 self
.db
.get_one
.assert_called_once()
2002 mock_detect_usage
.assert_called_once()
2003 self
.assertEqual(mock_safe_load
.call_count
, 2)
2005 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2006 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2007 def test_validate_descriptor_nsd_name_changed(
2008 self
, mock_safe_load
, mock_detect_usage
2010 """Validating NSD, NSD name has changed."""
2011 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
2012 did
= old_nsd
["_id"]
2013 new_nsd
["name"] = "nscharm-ns2"
2014 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
2015 mock_detect_usage
.return_value
= True
2016 self
.db
.get_one
.return_value
= old_nsd
2018 with self
.assertRaises(
2019 EngineException
, msg
="there are disallowed changes in the ns descriptor"
2021 self
.topic
._validate
_descriptor
_changes
(
2022 did
, descriptor_name
, "/tmp", "/tmp:1"
2025 e
.exception
.http_code
,
2026 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2027 "Wrong HTTP status code",
2030 norm("there are disallowed changes in the ns descriptor"),
2031 norm(str(e
.exception
)),
2032 "Wrong exception text",
2035 self
.db
.get_one
.assert_called_once()
2036 mock_detect_usage
.assert_called_once()
2037 self
.assertEqual(mock_safe_load
.call_count
, 2)
2039 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2040 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2041 def test_validate_descriptor_nsd_name_changed_nsd_not_in_use(
2042 self
, mock_safe_load
, mock_detect_usage
2044 """Validating NSD, NSD name has changed, NSD is not in use."""
2045 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
2046 did
= old_nsd
["_id"]
2047 new_nsd
["name"] = "nscharm-ns2"
2048 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
2049 mock_detect_usage
.return_value
= None
2050 self
.db
.get_one
.return_value
= old_nsd
2052 with self
.assertNotRaises(Exception):
2053 self
.topic
._validate
_descriptor
_changes
(
2054 did
, descriptor_name
, "/tmp", "/tmp:1"
2057 self
.db
.get_one
.assert_called_once()
2058 mock_detect_usage
.assert_called_once()
2059 mock_safe_load
.assert_not_called()
2061 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
2064 indata
= deepcopy(db_nsd_content
)
2065 vld
= indata
["virtual-link-desc"][0]
2066 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2070 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
2073 indata
= deepcopy(db_nsd_content
)
2074 vld
= indata
["virtual-link-desc"][0]
2075 df
= indata
["df"][0]
2078 "virtual-link-desc-id": "mgmt",
2079 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
2081 df
["virtual-link-profile"] = [affected_vlp
]
2082 with self
.assertRaises(EngineException
) as e
:
2083 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2087 e
.exception
.http_code
,
2088 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2089 "Wrong HTTP status code",
2093 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
2094 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
2095 df
["id"], affected_vlp
["id"]
2098 norm(str(e
.exception
)),
2099 "Wrong exception text",
2102 def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self
):
2103 indata
= deepcopy(db_nsd_content
)
2104 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2106 def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self
):
2107 indata
= deepcopy(db_nsd_content
)
2108 df
= indata
["df"][0]
2109 affected_vnf_profile
= df
["vnf-profile"][0]
2110 indata
["vnfd-id"] = ["non-existing-vnfd"]
2111 with self
.assertRaises(EngineException
) as e
:
2112 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2114 e
.exception
.http_code
,
2115 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2116 "Wrong HTTP status code",
2120 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
2121 "does not match any vnfd-id".format(
2123 affected_vnf_profile
["id"],
2124 affected_vnf_profile
["vnfd-id"],
2127 norm(str(e
.exception
)),
2128 "Wrong exception text",
2131 def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
2134 nsd_descriptor
= deepcopy(db_nsd_content
)
2135 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2136 df
= nsd_descriptor
["df"][0]
2137 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2138 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2142 def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
2145 nsd_descriptor
= deepcopy(db_nsd_content
)
2146 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2147 df
= nsd_descriptor
["df"][0]
2148 affected_vnf_profile
= df
["vnf-profile"][0]
2149 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
2150 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2151 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
2152 with self
.assertRaises(EngineException
) as e
:
2153 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2157 e
.exception
.http_code
,
2158 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2159 "Wrong HTTP status code",
2163 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
2164 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
2165 "non existing ext-cpd:id inside vnfd '{}'".format(
2167 affected_vnf_profile
["id"],
2168 affected_virtual_link
["virtual-link-profile-id"],
2170 vnfd_descriptor
["id"],
2173 norm(str(e
.exception
)),
2174 "Wrong exception text",
2177 def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self
):
2178 nsd_descriptor
= deepcopy(db_nsd_content
)
2179 invalid_vnfd_id
= "invalid-vnfd-id"
2180 nsd_descriptor
["id"] = "invalid-vnfd-id-ns"
2181 nsd_descriptor
["vnfd-id"][0] = invalid_vnfd_id
2182 nsd_descriptor
["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
2183 nsd_descriptor
["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
2184 with self
.assertRaises(EngineException
) as e
:
2185 self
.db
.get_list
.return_value
= []
2186 nsd_descriptor
= self
.topic
.check_conflict_on_edit(
2187 fake_session
, nsd_descriptor
, [], "id"
2190 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2194 "Descriptor error at 'vnfd-id'='{}' references a non "
2195 "existing vnfd".format(invalid_vnfd_id
)
2197 norm(str(e
.exception
)),
2198 "Wrong exception text",
2201 def test_validate_vnffgd_descriptor_on_valid_descriptor(self
):
2202 indata
= yaml
.safe_load(db_sfc_nsds_text
)[0]
2203 vnffgd
= indata
.get("vnffgd")
2205 self
.topic
.validate_vnffgd_data(fg
, indata
)
2207 def test_validate_vnffgd_descriptor_not_matching_nfp_position_element(self
):
2208 indata
= yaml
.safe_load(db_sfc_nsds_text
)[0]
2209 vnffgd
= indata
.get("vnffgd")
2211 nfpd
= fg
.get("nfpd")[0]
2212 with self
.assertRaises(EngineException
) as e
:
2213 fg
.update({"nfp-position-element": [{"id": "test1"}]})
2214 self
.topic
.validate_vnffgd_data(fg
, indata
)
2216 e
.exception
.http_code
,
2217 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2218 "Wrong HTTP status code",
2222 "Error at vnffgd nfpd[id='{}']:nfp-position-element-id='{}' "
2223 "does not match any nfp-position-element".format(nfpd
["id"], "test")
2225 norm(str(e
.exception
)),
2226 "Wrong exception text",
2229 def test_validate_vnffgd_descriptor_not_matching_constituent_base_element_id(
2232 indata
= yaml
.safe_load(db_sfc_nsds_text
)[0]
2233 vnffgd
= indata
.get("vnffgd")
2235 fg
["nfpd"][0]["position-desc-id"][0]["cp-profile-id"][0][
2236 "constituent-profile-elements"
2237 ][0]["constituent-base-element-id"] = "error_vnf"
2238 with self
.assertRaises(EngineException
) as e
:
2239 self
.topic
.validate_vnffgd_data(fg
, indata
)
2241 e
.exception
.http_code
,
2242 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2243 "Wrong HTTP status code",
2247 "Error at vnffgd constituent_profile[id='{}']:vnfd-id='{}' "
2248 "does not match any constituent-base-element-id".format(
2252 norm(str(e
.exception
)),
2253 "Wrong exception text",
2257 if __name__
== "__main__":