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_vdu_id(
585 self
, mock_rename
, mock_shutil
587 """Testing input validation during new vnfd creation
588 for scaling criteria with invalid vdu-id"""
589 did
, test_vnfd
= self
.prepare_vnfd_creation()
590 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
591 test_vnfd
["df"][0]["scaling-aspect"][0]["aspect-delta-details"]["deltas"][0][
593 ][0]["id"] = "vdudelta1"
594 affected_df
= test_vnfd
["df"][0]
595 sa
= affected_df
["scaling-aspect"][0]
596 delta
= sa
["aspect-delta-details"]["deltas"][0]
597 vdu_delta
= delta
["vdu-delta"][0]
599 with self
.assertRaises(
600 EngineException
, msg
="Accepted invalid Scaling Group Policy Criteria"
602 self
.topic
.upload_content(
603 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
606 e
.exception
.http_code
,
607 HTTPStatus
.UNPROCESSABLE_ENTITY
,
608 "Wrong HTTP status code",
612 "df[id='{}']:scaling-aspect[id='{}']:aspect-delta-details"
614 "vdu-id='{}' not defined in vdu".format(
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_criteria_monitoring_param_ref(
628 self
, mock_rename
, mock_shutil
630 """Testing input validation during new vnfd creation
631 for scaling criteria without monitoring parameter"""
632 did
, test_vnfd
= self
.prepare_vnfd_creation()
633 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
634 vdu
= test_vnfd
["vdu"][1]
635 affected_df
= test_vnfd
["df"][0]
636 sa
= affected_df
["scaling-aspect"][0]
637 sp
= sa
["scaling-policy"][0]
638 sc
= sp
["scaling-criteria"][0]
639 vdu
.pop("monitoring-parameter")
641 with self
.assertRaises(
642 EngineException
, msg
="Accepted non-existent Scaling Group Policy Criteria"
644 self
.topic
.upload_content(
645 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
648 e
.exception
.http_code
,
649 HTTPStatus
.UNPROCESSABLE_ENTITY
,
650 "Wrong HTTP status code",
654 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
655 "[name='{}']:scaling-criteria[name='{}']: "
656 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
661 sc
["vnf-monitoring-param-ref"],
664 norm(str(e
.exception
)),
665 "Wrong exception text",
668 @patch("osm_nbi.descriptor_topics.shutil")
669 @patch("osm_nbi.descriptor_topics.os.rename")
670 def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration(
671 self
, mock_rename
, mock_shutil
673 """Testing input validation during new vnfd creation
674 for scaling criteria without day12 configuration"""
675 did
, test_vnfd
= self
.prepare_vnfd_creation()
676 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
677 test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
680 df
= test_vnfd
["df"][0]
682 with self
.assertRaises(
683 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
685 self
.topic
.upload_content(
686 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
689 e
.exception
.http_code
,
690 HTTPStatus
.UNPROCESSABLE_ENTITY
,
691 "Wrong HTTP status code",
695 "'day1-2 configuration' not defined in the descriptor but it is referenced "
696 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
697 df
["id"], df
["scaling-aspect"][0]["id"]
700 norm(str(e
.exception
)),
701 "Wrong exception text",
704 @patch("osm_nbi.descriptor_topics.shutil")
705 @patch("osm_nbi.descriptor_topics.os.rename")
706 def test_new_vnfd_check_input_validation_scaling_config_action(
707 self
, mock_rename
, mock_shutil
709 """Testing input validation during new vnfd creation
710 for scaling criteria wrong config primitive"""
711 did
, test_vnfd
= self
.prepare_vnfd_creation()
712 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
713 df
= test_vnfd
["df"][0]
714 affected_df
= test_vnfd
["df"][0]
715 sa
= affected_df
["scaling-aspect"][0]
716 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
717 "operate-vnf-op-config"
718 )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
720 with self
.assertRaises(
721 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
723 self
.topic
.upload_content(
724 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
727 e
.exception
.http_code
,
728 HTTPStatus
.UNPROCESSABLE_ENTITY
,
729 "Wrong HTTP status code",
733 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
734 "config-primitive-name-ref='{}' does not match any "
735 "day1-2 configuration:config-primitive:name".format(
737 df
["scaling-aspect"][0]["id"],
738 sa
["scaling-config-action"][0]["vnf-config-primitive-name-ref"],
741 norm(str(e
.exception
)),
742 "Wrong exception text",
745 @patch("osm_nbi.descriptor_topics.shutil")
746 @patch("osm_nbi.descriptor_topics.os.rename")
747 def test_new_vnfd_check_input_validation_healing_criteria_vdu_id(
748 self
, mock_rename
, mock_shutil
750 """Testing input validation during new vnfd creation
751 for healing criteria with invalid vdu-id"""
752 did
, test_vnfd
= self
.prepare_vnfd_creation()
753 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
754 test_vnfd
["df"][0]["healing-aspect"][0]["healing-policy"][0][
757 affected_df
= test_vnfd
["df"][0]
758 ha
= affected_df
["healing-aspect"][0]
759 hp
= ha
["healing-policy"][0]
760 hp_vdu_id
= hp
["vdu-id"]
762 with self
.assertRaises(
763 EngineException
, msg
="Accepted invalid Healing Group Policy Criteria"
765 self
.topic
.upload_content(
766 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
769 e
.exception
.http_code
,
770 HTTPStatus
.UNPROCESSABLE_ENTITY
,
771 "Wrong HTTP status code",
775 "df[id='{}']:healing-aspect[id='{}']:healing-policy"
777 "vdu-id='{}' not defined in vdu".format(
784 norm(str(e
.exception
)),
785 "Wrong exception text",
788 @patch("osm_nbi.descriptor_topics.shutil")
789 @patch("osm_nbi.descriptor_topics.os.rename")
790 def test_new_vnfd_check_input_validation_alarm_criteria_monitoring_param_ref(
791 self
, mock_rename
, mock_shutil
793 """Testing input validation during new vnfd creation
794 for alarm with invalid monitoring parameter reference"""
795 did
, test_vnfd
= self
.prepare_vnfd_creation()
796 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
797 test_vnfd
["vdu"][1]["alarm"][0]["vnf-monitoring-param-ref"] = "unit_test_alarm"
798 vdu
= test_vnfd
["vdu"][1]
799 alarm
= vdu
["alarm"][0]
800 alarm_monitoring_param
= alarm
["vnf-monitoring-param-ref"]
802 with self
.assertRaises(
803 EngineException
, msg
="Accepted invalid Alarm Criteria"
805 self
.topic
.upload_content(
806 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
809 e
.exception
.http_code
,
810 HTTPStatus
.UNPROCESSABLE_ENTITY
,
811 "Wrong HTTP status code",
815 "vdu[id='{}']:alarm[id='{}']:"
816 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
819 alarm_monitoring_param
,
822 norm(str(e
.exception
)),
823 "Wrong exception text",
826 @patch("osm_nbi.descriptor_topics.shutil")
827 @patch("osm_nbi.descriptor_topics.os.rename")
828 def test_new_vnfd_check_input_validation_storage_reference_criteria(
829 self
, mock_rename
, mock_shutil
831 """Testing input validation during new vnfd creation
832 for invalid virtual-storge-desc reference"""
833 did
, test_vnfd
= self
.prepare_vnfd_creation()
834 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
835 test_vnfd
["vdu"][1]["virtual-storage-desc"] = "unit_test_storage"
836 vdu
= test_vnfd
["vdu"][1]
837 vsd_ref
= vdu
["virtual-storage-desc"]
839 with self
.assertRaises(
840 EngineException
, msg
="Accepted invalid virtual-storage-desc"
842 self
.topic
.upload_content(
843 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
846 e
.exception
.http_code
,
847 HTTPStatus
.UNPROCESSABLE_ENTITY
,
848 "Wrong HTTP status code",
852 "vdu[virtual-storage-desc='{}']"
853 "not defined in vnfd".format(
857 norm(str(e
.exception
)),
858 "Wrong exception text",
861 @patch("osm_nbi.descriptor_topics.shutil")
862 @patch("osm_nbi.descriptor_topics.os.rename")
863 def test_new_vnfd_check_input_validation_compute_reference_criteria(
864 self
, mock_rename
, mock_shutil
866 """Testing input validation during new vnfd creation
867 for invalid virtual-compute-desc reference"""
868 did
, test_vnfd
= self
.prepare_vnfd_creation()
869 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
870 test_vnfd
["vdu"][1]["virtual-compute-desc"] = "unit_test_compute"
871 vdu
= test_vnfd
["vdu"][1]
872 vcd_ref
= vdu
["virtual-compute-desc"]
874 with self
.assertRaises(
875 EngineException
, msg
="Accepted invalid virtual-compute-desc"
877 self
.topic
.upload_content(
878 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
881 e
.exception
.http_code
,
882 HTTPStatus
.UNPROCESSABLE_ENTITY
,
883 "Wrong HTTP status code",
887 "vdu[virtual-compute-desc='{}']"
888 "not defined in vnfd".format(
892 norm(str(e
.exception
)),
893 "Wrong exception text",
896 @patch("osm_nbi.descriptor_topics.shutil")
897 @patch("osm_nbi.descriptor_topics.os.rename")
898 def test_new_vnfd_check_input_validation_everything_right(
899 self
, mock_rename
, mock_shutil
901 """Testing input validation during new vnfd creation
902 everything correct"""
903 did
, test_vnfd
= self
.prepare_vnfd_creation()
904 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
905 test_vnfd
["id"] = "fake-vnfd-id"
906 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
907 "operate-vnf-op-config"
908 )["day1-2"][0]["id"] = "fake-vnfd-id"
909 self
.db
.get_one
.side_effect
= [
910 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
913 rc
= self
.topic
.upload_content(
914 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
916 self
.assertTrue(rc
, "Input Validation: Unexpected failure")
918 def test_edit_vnfd(self
):
919 vnfd_content
= deepcopy(db_vnfd_content
)
920 did
= vnfd_content
["_id"]
921 self
.fs
.file_exists
.return_value
= True
922 self
.fs
.dir_ls
.return_value
= True
923 with self
.subTest(i
=1, t
="Normal Edition"):
925 self
.db
.get_one
.side_effect
= [deepcopy(vnfd_content
), None]
926 data
= {"product-name": "new-vnfd-name"}
927 self
.topic
.edit(fake_session
, did
, data
)
928 db_args
= self
.db
.replace
.call_args
[0]
929 msg_args
= self
.msg
.write
.call_args
[0]
931 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
932 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
933 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
934 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
935 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
937 db_args
[2]["_admin"]["created"],
938 vnfd_content
["_admin"]["created"],
939 "Wrong creation time",
942 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
945 db_args
[2]["_admin"]["projects_read"],
946 vnfd_content
["_admin"]["projects_read"],
947 "Wrong read-only project list",
950 db_args
[2]["_admin"]["projects_write"],
951 vnfd_content
["_admin"]["projects_write"],
952 "Wrong read-write project list",
955 db_args
[2]["product-name"], data
["product-name"], "Wrong VNFD Name"
957 with self
.subTest(i
=2, t
="Conflict on Edit"):
958 data
= {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
959 self
.db
.get_one
.side_effect
= [
960 deepcopy(vnfd_content
),
961 {"_id": str(uuid4()), "id": data
["id"]},
963 with self
.assertRaises(
964 EngineException
, msg
="Accepted existing VNFD ID"
966 self
.topic
.edit(fake_session
, did
, data
)
968 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
972 "{} with id '{}' already exists for this project".format(
976 norm(str(e
.exception
)),
977 "Wrong exception text",
979 with self
.subTest(i
=3, t
="Check Envelope"):
980 data
= {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
981 with self
.assertRaises(
982 EngineException
, msg
="Accepted VNFD with wrong envelope"
984 self
.topic
.edit(fake_session
, did
, data
, content
=vnfd_content
)
986 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
989 "'vnfd' must be dict", norm(str(e
.exception
)), "Wrong exception text"
993 def test_delete_vnfd(self
):
994 did
= db_vnfd_content
["_id"]
995 self
.db
.get_one
.return_value
= db_vnfd_content
996 p_id
= db_vnfd_content
["_admin"]["projects_read"][0]
997 with self
.subTest(i
=1, t
="Normal Deletion"):
998 self
.db
.get_list
.return_value
= []
999 self
.db
.del_one
.return_value
= {"deleted": 1}
1000 self
.topic
.delete(fake_session
, did
)
1001 db_args
= self
.db
.del_one
.call_args
[0]
1002 msg_args
= self
.msg
.write
.call_args
[0]
1003 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1004 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
1005 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1006 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1007 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
1009 db_args
[1]["_admin.projects_write.cont"],
1013 db_g1_args
= self
.db
.get_one
.call_args
[0]
1014 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1015 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
1016 db_gl_calls
= self
.db
.get_list
.call_args_list
1017 self
.assertEqual(db_gl_calls
[0][0][0], "vnfrs", "Wrong DB topic")
1018 # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID") # Filter changed after call
1019 self
.assertEqual(db_gl_calls
[1][0][0], "nsds", "Wrong DB topic")
1021 db_gl_calls
[1][0][1]["vnfd-id"],
1022 db_vnfd_content
["id"],
1023 "Wrong DB NSD vnfd-id",
1027 self
.db
.del_list
.call_args
[0][0],
1028 self
.topic
.topic
+ "_revisions",
1033 self
.db
.del_list
.call_args
[0][1]["_id"]["$regex"],
1035 "Wrong ID for rexep delete",
1038 self
.db
.set_one
.assert_not_called()
1039 fs_del_calls
= self
.fs
.file_delete
.call_args_list
1040 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
1041 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
1042 with self
.subTest(i
=2, t
="Conflict on Delete - VNFD in use by VNFR"):
1043 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-vnfr"}]
1044 with self
.assertRaises(
1045 EngineException
, msg
="Accepted VNFD in use by VNFR"
1047 self
.topic
.delete(fake_session
, did
)
1049 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1052 "there is at least one vnf instance using this descriptor",
1053 norm(str(e
.exception
)),
1054 "Wrong exception text",
1056 with self
.subTest(i
=3, t
="Conflict on Delete - VNFD in use by NSD"):
1057 self
.db
.get_list
.side_effect
= [
1059 [{"_id": str(uuid4()), "name": "fake-nsd"}],
1061 with self
.assertRaises(
1062 EngineException
, msg
="Accepted VNFD in use by NSD"
1064 self
.topic
.delete(fake_session
, did
)
1066 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1069 "there is at least one ns package referencing this descriptor",
1070 norm(str(e
.exception
)),
1071 "Wrong exception text",
1073 with self
.subTest(i
=4, t
="Non-existent VNFD"):
1074 excp_msg
= "Not found any {} with filter='{}'".format("VNFD", {"_id": did
})
1075 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
1076 with self
.assertRaises(
1077 DbException
, msg
="Accepted non-existent VNFD ID"
1079 self
.topic
.delete(fake_session
, did
)
1081 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1084 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
1086 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
1087 db_vnfd_content
["_admin"]["projects_read"].append("other_project")
1088 self
.db
.get_one
= Mock(return_value
=db_vnfd_content
)
1089 self
.db
.get_list
= Mock(return_value
=[])
1090 self
.msg
.write
.reset_mock()
1091 self
.db
.del_one
.reset_mock()
1092 self
.fs
.file_delete
.reset_mock()
1094 self
.topic
.delete(fake_session
, did
)
1095 self
.db
.del_one
.assert_not_called()
1096 self
.msg
.write
.assert_not_called()
1097 db_g1_args
= self
.db
.get_one
.call_args
[0]
1098 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1099 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
1100 db_s1_args
= self
.db
.set_one
.call_args
1101 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
1102 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
1104 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1107 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
1110 db_s1_args
[1]["pull_list"],
1111 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
1112 "Wrong DB pull_list dictionary",
1114 self
.fs
.file_delete
.assert_not_called()
1117 def prepare_vnfd_validation(self
):
1118 descriptor_name
= "test_descriptor"
1119 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1120 "/tmp/" + str(uuid4()), "a+b"
1122 old_vnfd
, new_vnfd
= self
.create_desc_temp(db_vnfd_content
)
1123 return descriptor_name
, old_vnfd
, new_vnfd
1125 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1126 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1127 def test_validate_vnfd_changes_day12_config_primitive_changed(
1128 self
, mock_safe_load
, mock_detect_usage
1130 """Validating VNFD for VNFD updates, day1-2 config primitive has changed"""
1131 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
1132 did
= old_vnfd
["_id"]
1133 new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1135 ][0]["config-primitive"][0]["name"] = "new_action"
1136 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
1137 mock_detect_usage
.return_value
= True
1138 self
.db
.get_one
.return_value
= old_vnfd
1140 with self
.assertNotRaises(EngineException
):
1141 self
.topic
._validate
_descriptor
_changes
(
1142 did
, descriptor_name
, "/tmp/", "/tmp:1/"
1144 self
.db
.get_one
.assert_called_once()
1145 mock_detect_usage
.assert_called_once()
1146 self
.assertEqual(mock_safe_load
.call_count
, 2)
1148 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1149 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1150 def test_validate_vnfd_changes_sw_version_changed(
1151 self
, mock_safe_load
, mock_detect_usage
1153 """Validating VNFD for updates, software version has changed"""
1154 # old vnfd uses the default software version: 1.0
1155 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
1156 did
= old_vnfd
["_id"]
1157 new_vnfd
["software-version"] = "1.3"
1158 new_vnfd
["sw-image-desc"][0]["name"] = "new-image"
1159 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
1160 mock_detect_usage
.return_value
= True
1161 self
.db
.get_one
.return_value
= old_vnfd
1163 with self
.assertNotRaises(EngineException
):
1164 self
.topic
._validate
_descriptor
_changes
(
1165 did
, descriptor_name
, "/tmp/", "/tmp:1/"
1167 self
.db
.get_one
.assert_called_once()
1168 mock_detect_usage
.assert_called_once()
1169 self
.assertEqual(mock_safe_load
.call_count
, 2)
1171 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1172 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1173 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed(
1174 self
, mock_safe_load
, mock_detect_usage
1176 """Validating VNFD for updates, software version has not
1177 changed, mgmt-cp has changed."""
1178 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
1179 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
1180 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
1181 did
= old_vnfd
["_id"]
1182 mock_detect_usage
.return_value
= True
1183 self
.db
.get_one
.return_value
= old_vnfd
1185 with self
.assertRaises(
1186 EngineException
, msg
="there are disallowed changes in the vnf descriptor"
1188 self
.topic
._validate
_descriptor
_changes
(
1189 did
, descriptor_name
, "/tmp/", "/tmp:1/"
1193 e
.exception
.http_code
,
1194 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1195 "Wrong HTTP status code",
1198 norm("there are disallowed changes in the vnf descriptor"),
1199 norm(str(e
.exception
)),
1200 "Wrong exception text",
1202 self
.db
.get_one
.assert_called_once()
1203 mock_detect_usage
.assert_called_once()
1204 self
.assertEqual(mock_safe_load
.call_count
, 2)
1206 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1207 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1208 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed_vnfd_not_in_use(
1209 self
, mock_safe_load
, mock_detect_usage
1211 """Validating VNFD for updates, software version has not
1212 changed, mgmt-cp has changed, vnfd is not in use."""
1213 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
1214 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
1215 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
1216 did
= old_vnfd
["_id"]
1217 mock_detect_usage
.return_value
= None
1218 self
.db
.get_one
.return_value
= old_vnfd
1220 with self
.assertNotRaises(EngineException
):
1221 self
.topic
._validate
_descriptor
_changes
(
1222 did
, descriptor_name
, "/tmp/", "/tmp:1/"
1225 self
.db
.get_one
.assert_called_once()
1226 mock_detect_usage
.assert_called_once()
1227 mock_safe_load
.assert_not_called()
1229 def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self
):
1230 indata
= deepcopy(db_vnfd_content
)
1231 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1233 def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
1236 indata
= deepcopy(db_vnfd_content
)
1237 indata
["ext-cpd"] = []
1238 with self
.assertRaises(EngineException
) as e
:
1239 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1241 e
.exception
.http_code
,
1242 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1243 "Wrong HTTP status code",
1247 "mgmt-cp='{}' must match an existing ext-cpd".format(indata
["mgmt-cp"])
1249 norm(str(e
.exception
)),
1250 "Wrong exception text",
1253 def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self
):
1254 indata
= deepcopy(db_vnfd_content
)
1255 indata
.pop("mgmt-cp")
1256 with self
.assertRaises(EngineException
) as e
:
1257 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1259 e
.exception
.http_code
,
1260 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1261 "Wrong HTTP status code",
1264 norm("'mgmt-cp' is a mandatory field and it is not defined"),
1265 norm(str(e
.exception
)),
1266 "Wrong exception text",
1269 def test_validate_vdu_internal_connection_points_on_valid_descriptor(self
):
1270 indata
= db_vnfd_content
1271 vdu
= indata
["vdu"][0]
1272 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1274 def test_validate_external_connection_points_on_valid_descriptor(self
):
1275 indata
= db_vnfd_content
1276 self
.topic
.validate_external_connection_points(indata
)
1278 def test_validate_external_connection_points_when_missing_internal_connection_point(
1281 indata
= deepcopy(db_vnfd_content
)
1282 vdu
= indata
["vdu"][0]
1284 affected_ext_cpd
= indata
["ext-cpd"][0]
1285 with self
.assertRaises(EngineException
) as e
:
1286 self
.topic
.validate_external_connection_points(indata
)
1288 e
.exception
.http_code
,
1289 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1290 "Wrong HTTP status code",
1294 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
1295 affected_ext_cpd
["id"]
1298 norm(str(e
.exception
)),
1299 "Wrong exception text",
1302 def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
1305 indata
= deepcopy(db_vnfd_content
)
1306 vdu
= indata
["vdu"][0]
1310 "virtual-network-interface-requirement": [{"name": "duplicated"}],
1312 vdu
["int-cpd"].insert(0, duplicated_cpd
)
1313 with self
.assertRaises(EngineException
) as e
:
1314 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1316 e
.exception
.http_code
,
1317 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1318 "Wrong HTTP status code",
1322 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
1323 vdu
["id"], duplicated_cpd
["id"]
1326 norm(str(e
.exception
)),
1327 "Wrong exception text",
1330 def test_validate_external_connection_points_on_duplicated_external_connection_point(
1333 indata
= deepcopy(db_vnfd_content
)
1335 "id": "vnf-mgmt-ext",
1336 "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
1338 indata
["ext-cpd"].insert(0, duplicated_cpd
)
1339 with self
.assertRaises(EngineException
) as e
:
1340 self
.topic
.validate_external_connection_points(indata
)
1342 e
.exception
.http_code
,
1343 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1344 "Wrong HTTP status code",
1348 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
1349 duplicated_cpd
["id"]
1352 norm(str(e
.exception
)),
1353 "Wrong exception text",
1356 def test_validate_internal_virtual_links_on_valid_descriptor(self
):
1357 indata
= db_vnfd_content
1358 self
.topic
.validate_internal_virtual_links(indata
)
1360 def test_validate_internal_virtual_links_on_duplicated_ivld(self
):
1361 indata
= deepcopy(db_vnfd_content
)
1362 duplicated_vld
= {"id": "internal"}
1363 indata
["int-virtual-link-desc"].insert(0, duplicated_vld
)
1364 with self
.assertRaises(EngineException
) as e
:
1365 self
.topic
.validate_internal_virtual_links(indata
)
1367 e
.exception
.http_code
,
1368 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1369 "Wrong HTTP status code",
1373 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
1374 duplicated_vld
["id"]
1377 norm(str(e
.exception
)),
1378 "Wrong exception text",
1381 def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
1384 indata
= deepcopy(db_vnfd_content
)
1385 vdu
= indata
["vdu"][0]
1386 affected_int_cpd
= vdu
["int-cpd"][0]
1387 affected_int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
1388 with self
.assertRaises(EngineException
) as e
:
1389 self
.topic
.validate_internal_virtual_links(indata
)
1391 e
.exception
.http_code
,
1392 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1393 "Wrong HTTP status code",
1397 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
1398 "int-virtual-link-desc".format(
1400 affected_int_cpd
["id"],
1401 affected_int_cpd
["int-virtual-link-desc"],
1404 norm(str(e
.exception
)),
1405 "Wrong exception text",
1408 def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self
):
1409 indata
= deepcopy(db_vnfd_content
)
1410 affected_ivld_profile
= {"id": "non-existing-int-virtual-link-desc"}
1411 df
= indata
["df"][0]
1412 df
["virtual-link-profile"] = [affected_ivld_profile
]
1413 with self
.assertRaises(EngineException
) as e
:
1414 self
.topic
.validate_internal_virtual_links(indata
)
1416 e
.exception
.http_code
,
1417 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1418 "Wrong HTTP status code",
1422 "df[id='{}']:virtual-link-profile='{}' must match an existing "
1423 "int-virtual-link-desc".format(df
["id"], affected_ivld_profile
["id"])
1425 norm(str(e
.exception
)),
1426 "Wrong exception text",
1429 def test_validate_monitoring_params_on_valid_descriptor(self
):
1430 indata
= db_vnfd_content
1431 self
.topic
.validate_monitoring_params(indata
)
1433 def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self
):
1434 indata
= deepcopy(db_vnfd_content
)
1435 duplicated_mp
= {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1436 affected_ivld
= indata
["int-virtual-link-desc"][0]
1437 affected_ivld
["monitoring-parameters"] = [duplicated_mp
, duplicated_mp
]
1438 with self
.assertRaises(EngineException
) as e
:
1439 self
.topic
.validate_monitoring_params(indata
)
1441 e
.exception
.http_code
,
1442 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1443 "Wrong HTTP status code",
1447 "Duplicated monitoring-parameter id in "
1448 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1449 affected_ivld
["id"], duplicated_mp
["id"]
1452 norm(str(e
.exception
)),
1453 "Wrong exception text",
1456 def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self
):
1457 indata
= deepcopy(db_vnfd_content
)
1459 "id": "dataVM_cpu_util",
1460 "name": "dataVM_cpu_util",
1461 "performance_metric": "cpu",
1463 affected_vdu
= indata
["vdu"][1]
1464 affected_vdu
["monitoring-parameter"].insert(0, duplicated_mp
)
1465 with self
.assertRaises(EngineException
) as e
:
1466 self
.topic
.validate_monitoring_params(indata
)
1468 e
.exception
.http_code
,
1469 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1470 "Wrong HTTP status code",
1474 "Duplicated monitoring-parameter id in "
1475 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1476 affected_vdu
["id"], duplicated_mp
["id"]
1479 norm(str(e
.exception
)),
1480 "Wrong exception text",
1483 def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self
):
1484 indata
= deepcopy(db_vnfd_content
)
1488 "performance_metric": "memory",
1490 affected_df
= indata
["df"][0]
1491 affected_df
["monitoring-parameter"] = [duplicated_mp
, duplicated_mp
]
1492 with self
.assertRaises(EngineException
) as e
:
1493 self
.topic
.validate_monitoring_params(indata
)
1495 e
.exception
.http_code
,
1496 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1497 "Wrong HTTP status code",
1501 "Duplicated monitoring-parameter id in "
1502 "df[id='{}']:monitoring-parameter[id='{}']".format(
1503 affected_df
["id"], duplicated_mp
["id"]
1506 norm(str(e
.exception
)),
1507 "Wrong exception text",
1510 def test_validate_scaling_group_descriptor_on_valid_descriptor(self
):
1511 indata
= db_vnfd_content
1512 self
.topic
.validate_scaling_group_descriptor(indata
)
1514 def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self
):
1515 indata
= deepcopy(db_vnfd_content
)
1516 vdu
= indata
["vdu"][1]
1517 affected_df
= indata
["df"][0]
1518 affected_sa
= affected_df
["scaling-aspect"][0]
1519 affected_sp
= affected_sa
["scaling-policy"][0]
1520 affected_sc
= affected_sp
["scaling-criteria"][0]
1521 vdu
.pop("monitoring-parameter")
1522 with self
.assertRaises(EngineException
) as e
:
1523 self
.topic
.validate_scaling_group_descriptor(indata
)
1525 e
.exception
.http_code
,
1526 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1527 "Wrong HTTP status code",
1531 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1532 "[name='{}']:scaling-criteria[name='{}']: "
1533 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1536 affected_sp
["name"],
1537 affected_sc
["name"],
1538 affected_sc
["vnf-monitoring-param-ref"],
1541 norm(str(e
.exception
)),
1542 "Wrong exception text",
1545 def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self
):
1546 indata
= deepcopy(db_vnfd_content
)
1547 df
= indata
["df"][0]
1548 affected_sa
= df
["scaling-aspect"][0]
1549 indata
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1552 with self
.assertRaises(EngineException
) as e
:
1553 self
.topic
.validate_scaling_group_descriptor(indata
)
1555 e
.exception
.http_code
,
1556 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1557 "Wrong HTTP status code",
1561 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1562 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1563 df
["id"], affected_sa
["id"]
1566 norm(str(e
.exception
)),
1567 "Wrong exception text",
1570 def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1573 indata
= deepcopy(db_vnfd_content
)
1574 df
= indata
["df"][0]
1575 affected_sa
= df
["scaling-aspect"][0]
1576 affected_sca_primitive
= affected_sa
["scaling-config-action"][0][
1577 "vnf-config-primitive-name-ref"
1579 df
["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1582 with self
.assertRaises(EngineException
) as e
:
1583 self
.topic
.validate_scaling_group_descriptor(indata
)
1585 e
.exception
.http_code
,
1586 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1587 "Wrong HTTP status code",
1591 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1592 "config-primitive-name-ref='{}' does not match any "
1593 "day1-2 configuration:config-primitive:name".format(
1594 df
["id"], affected_sa
["id"], affected_sca_primitive
1597 norm(str(e
.exception
)),
1598 "Wrong exception text",
1601 def test_new_vnfd_revision(self
):
1602 did
= db_vnfd_content
["_id"]
1603 self
.fs
.get_params
.return_value
= {}
1604 self
.fs
.file_exists
.return_value
= False
1605 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1606 "/tmp/" + str(uuid4()), "a+b"
1608 test_vnfd
= deepcopy(db_vnfd_content
)
1609 del test_vnfd
["_id"]
1610 del test_vnfd
["_admin"]
1611 self
.db
.create
.return_value
= did
1613 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1614 db_args
= self
.db
.create
.call_args
[0]
1616 db_args
[1]["_admin"]["revision"], 0, "New package should be at revision 0"
1619 @patch("osm_nbi.descriptor_topics.shutil")
1620 @patch("osm_nbi.descriptor_topics.os.rename")
1621 def test_update_vnfd(self
, mock_rename
, mock_shutil
):
1623 did
= db_vnfd_content
["_id"]
1625 self
.fs
.get_params
.return_value
= {}
1626 self
.fs
.file_exists
.return_value
= False
1627 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1628 "/tmp/" + str(uuid4()), "a+b"
1630 new_vnfd
= deepcopy(db_vnfd_content
)
1632 self
.db
.create
.return_value
= did
1634 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1635 del new_vnfd
["vdu"][0]["cloud-init-file"]
1636 del new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1638 ][0]["execution-environment-list"][0]["juju"]
1640 old_vnfd
= {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])}
1641 old_vnfd
["_admin"]["revision"] = old_revision
1643 self
.db
.get_one
.side_effect
= [old_vnfd
, old_vnfd
, None]
1644 self
.topic
.upload_content(fake_session
, did
, new_vnfd
, {}, {"Content-Type": []})
1646 db_args
= self
.db
.replace
.call_args
[0]
1648 db_args
[2]["_admin"]["revision"],
1650 "Revision should increment",
1654 class Test_NsdTopic(TestCase
):
1656 def setUpClass(cls
):
1657 cls
.test_name
= "test-nsd-topic"
1660 def tearDownClass(cls
):
1664 self
.db
= Mock(dbbase
.DbBase())
1665 self
.fs
= Mock(fsbase
.FsBase())
1666 self
.msg
= Mock(msgbase
.MsgBase())
1667 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1668 self
.topic
= NsdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1669 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1672 def assertNotRaises(self
, exception_type
):
1675 except exception_type
:
1676 raise self
.failureException("{} raised".format(exception_type
.__name
__))
1678 def create_desc_temp(self
, template
):
1679 old_desc
= deepcopy(template
)
1680 new_desc
= deepcopy(template
)
1681 return old_desc
, new_desc
1683 def prepare_nsd_creation(self
):
1685 did
= db_nsd_content
["_id"]
1686 self
.fs
.get_params
.return_value
= {}
1687 self
.fs
.file_exists
.return_value
= False
1688 self
.fs
.file_open
.side_effect
= lambda path
, mode
: tempfile
.TemporaryFile(
1691 self
.db
.get_one
.side_effect
= [
1692 {"_id": did
, "_admin": deepcopy(db_nsd_content
["_admin"])},
1695 test_nsd
= deepcopy(db_nsd_content
)
1697 del test_nsd
["_admin"]
1698 return did
, test_nsd
1700 @patch("osm_nbi.descriptor_topics.shutil")
1701 @patch("osm_nbi.descriptor_topics.os.rename")
1702 def test_new_nsd_normal_creation(self
, mock_rename
, mock_shutil
):
1703 did
, test_nsd
= self
.prepare_nsd_creation()
1704 self
.db
.create
.return_value
= did
1707 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1708 db_args
= self
.db
.create
.call_args
[0]
1709 msg_args
= self
.msg
.write
.call_args
[0]
1710 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1711 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1712 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
1713 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1714 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1715 self
.assertEqual(did2
, did
, "Wrong DB NSD id")
1716 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
1718 db_args
[1]["_admin"]["modified"],
1719 db_args
[1]["_admin"]["created"],
1720 "Wrong modification time",
1723 db_args
[1]["_admin"]["projects_read"],
1725 "Wrong read-only project list",
1728 db_args
[1]["_admin"]["projects_write"],
1730 "Wrong read-write project list",
1733 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1735 self
.topic
.upload_content(fake_session
, did
, test_nsd
, {}, {"Content-Type": []})
1736 msg_args
= self
.msg
.write
.call_args
[0]
1737 test_nsd
["_id"] = did
1738 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1739 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1740 self
.assertEqual(msg_args
[2], test_nsd
, "Wrong message content")
1742 db_args
= self
.db
.get_one
.mock_calls
[0][1]
1743 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1744 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB NSD id")
1746 db_args
= self
.db
.replace
.call_args
[0]
1747 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1748 self
.assertEqual(db_args
[1], did
, "Wrong DB NSD id")
1750 admin
= db_args
[2]["_admin"]
1751 db_admin
= db_nsd_content
["_admin"]
1752 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
1754 admin
["modified"], db_admin
["created"], "Wrong modification time"
1757 admin
["projects_read"],
1758 db_admin
["projects_read"],
1759 "Wrong read-only project list",
1762 admin
["projects_write"],
1763 db_admin
["projects_write"],
1764 "Wrong read-write project list",
1767 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1770 admin
["operationalState"], "ENABLED", "Wrong operational state"
1772 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
1774 storage
= admin
["storage"]
1775 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
1776 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
1778 compare_desc(self
, test_nsd
, db_args
[2], "NSD")
1779 revision_args
= self
.db
.create
.call_args
[0]
1781 revision_args
[0], self
.topic
.topic
+ "_revisions", "Wrong topic"
1783 self
.assertEqual(revision_args
[1]["id"], db_args
[2]["id"], "Wrong revision id")
1785 revision_args
[1]["_id"], db_args
[2]["_id"] + ":1", "Wrong revision _id"
1788 @patch("osm_nbi.descriptor_topics.shutil")
1789 @patch("osm_nbi.descriptor_topics.os.rename")
1790 def test_new_nsd_check_pyangbind_validation_required_properties(
1791 self
, mock_rename
, mock_shutil
1793 did
, test_nsd
= self
.prepare_nsd_creation()
1796 with self
.assertRaises(
1797 EngineException
, msg
="Accepted NSD with a missing required property"
1799 self
.topic
.upload_content(
1800 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1803 e
.exception
.http_code
,
1804 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1805 "Wrong HTTP status code",
1808 norm("Error in pyangbind validation: '{}'".format("id")),
1809 norm(str(e
.exception
)),
1810 "Wrong exception text",
1813 @patch("osm_nbi.descriptor_topics.shutil")
1814 @patch("osm_nbi.descriptor_topics.os.rename")
1815 def test_new_nsd_check_pyangbind_validation_additional_properties(
1816 self
, mock_rename
, mock_shutil
1818 did
, test_nsd
= self
.prepare_nsd_creation()
1819 test_nsd
["extra-property"] = 0
1821 with self
.assertRaises(
1822 EngineException
, msg
="Accepted NSD with an additional property"
1824 self
.topic
.upload_content(
1825 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1828 e
.exception
.http_code
,
1829 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1830 "Wrong HTTP status code",
1834 "Error in pyangbind validation: {} ({})".format(
1835 "json object contained a key that did not exist", "extra-property"
1838 norm(str(e
.exception
)),
1839 "Wrong exception text",
1842 @patch("osm_nbi.descriptor_topics.shutil")
1843 @patch("osm_nbi.descriptor_topics.os.rename")
1844 def test_new_nsd_check_pyangbind_validation_property_types(
1845 self
, mock_rename
, mock_shutil
1847 did
, test_nsd
= self
.prepare_nsd_creation()
1848 test_nsd
["designer"] = {"key": 0}
1850 with self
.assertRaises(
1851 EngineException
, msg
="Accepted NSD with a wrongly typed property"
1853 self
.topic
.upload_content(
1854 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1857 e
.exception
.http_code
,
1858 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1859 "Wrong HTTP status code",
1863 "Error in pyangbind validation: {} ({})".format(
1864 "json object contained a key that did not exist", "key"
1867 norm(str(e
.exception
)),
1868 "Wrong exception text",
1871 @patch("osm_nbi.descriptor_topics.shutil")
1872 @patch("osm_nbi.descriptor_topics.os.rename")
1873 def test_new_nsd_check_input_validation_mgmt_network_virtual_link_protocol_data(
1874 self
, mock_rename
, mock_shutil
1876 did
, test_nsd
= self
.prepare_nsd_creation()
1877 df
= test_nsd
["df"][0]
1880 "virtual-link-desc-id": "mgmt",
1881 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1883 df
["virtual-link-profile"] = [mgmt_profile
]
1885 with self
.assertRaises(
1886 EngineException
, msg
="Accepted VLD with mgmt-network+ip-profile"
1888 self
.topic
.upload_content(
1889 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1892 e
.exception
.http_code
,
1893 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1894 "Wrong HTTP status code",
1898 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1899 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1900 df
["id"], mgmt_profile
["id"]
1903 norm(str(e
.exception
)),
1904 "Wrong exception text",
1907 @patch("osm_nbi.descriptor_topics.shutil")
1908 @patch("osm_nbi.descriptor_topics.os.rename")
1909 def test_new_nsd_check_descriptor_dependencies_vnfd_id(
1910 self
, mock_rename
, mock_shutil
1912 did
, test_nsd
= self
.prepare_nsd_creation()
1913 self
.db
.get_list
.return_value
= []
1915 with self
.assertRaises(
1916 EngineException
, msg
="Accepted wrong VNFD ID reference"
1918 self
.topic
.upload_content(
1919 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1922 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1926 "'vnfd-id'='{}' references a non existing vnfd".format(
1927 test_nsd
["vnfd-id"][0]
1930 norm(str(e
.exception
)),
1931 "Wrong exception text",
1934 @patch("osm_nbi.descriptor_topics.shutil")
1935 @patch("osm_nbi.descriptor_topics.os.rename")
1936 def test_new_nsd_check_descriptor_dependencies_vld_vnfd_connection_point_ref(
1937 self
, mock_rename
, mock_shutil
1939 # Check Descriptor Dependencies: "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]
1940 did
, test_nsd
= self
.prepare_nsd_creation()
1941 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1942 df
= test_nsd
["df"][0]
1943 affected_vnf_profile
= df
["vnf-profile"][0]
1944 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
1945 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
1946 self
.db
.get_list
.return_value
= [vnfd_descriptor
]
1948 with self
.assertRaises(
1949 EngineException
, msg
="Accepted wrong VLD CP reference"
1951 self
.topic
.upload_content(
1952 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1955 e
.exception
.http_code
,
1956 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1957 "Wrong HTTP status code",
1961 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1962 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1963 "non existing ext-cpd:id inside vnfd '{}'".format(
1965 affected_vnf_profile
["id"],
1966 affected_virtual_link
["virtual-link-profile-id"],
1968 vnfd_descriptor
["id"],
1971 norm(str(e
.exception
)),
1972 "Wrong exception text",
1975 def test_edit_nsd(self
):
1976 nsd_content
= deepcopy(db_nsd_content
)
1977 did
= nsd_content
["_id"]
1978 self
.fs
.file_exists
.return_value
= True
1979 self
.fs
.dir_ls
.return_value
= True
1980 with self
.subTest(i
=1, t
="Normal Edition"):
1982 self
.db
.get_one
.side_effect
= [deepcopy(nsd_content
), None]
1983 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1984 data
= {"id": "new-nsd-id", "name": "new-nsd-name"}
1985 self
.topic
.edit(fake_session
, did
, data
)
1986 db_args
= self
.db
.replace
.call_args
[0]
1987 msg_args
= self
.msg
.write
.call_args
[0]
1989 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1990 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1991 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
1992 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1993 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
1995 db_args
[2]["_admin"]["created"],
1996 nsd_content
["_admin"]["created"],
1997 "Wrong creation time",
2000 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
2003 db_args
[2]["_admin"]["projects_read"],
2004 nsd_content
["_admin"]["projects_read"],
2005 "Wrong read-only project list",
2008 db_args
[2]["_admin"]["projects_write"],
2009 nsd_content
["_admin"]["projects_write"],
2010 "Wrong read-write project list",
2012 self
.assertEqual(db_args
[2]["id"], data
["id"], "Wrong NSD ID")
2013 self
.assertEqual(db_args
[2]["name"], data
["name"], "Wrong NSD Name")
2014 with self
.subTest(i
=2, t
="Conflict on Edit"):
2015 data
= {"id": "fake-nsd-id", "name": "new-nsd-name"}
2016 self
.db
.get_one
.side_effect
= [
2018 {"_id": str(uuid4()), "id": data
["id"]},
2020 with self
.assertRaises(
2021 EngineException
, msg
="Accepted existing NSD ID"
2023 self
.topic
.edit(fake_session
, did
, data
)
2025 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2029 "{} with id '{}' already exists for this project".format(
2033 norm(str(e
.exception
)),
2034 "Wrong exception text",
2036 with self
.subTest(i
=3, t
="Check Envelope"):
2037 data
= {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
2038 self
.db
.get_one
.side_effect
= [nsd_content
, None]
2039 with self
.assertRaises(
2040 EngineException
, msg
="Accepted NSD with wrong envelope"
2042 self
.topic
.edit(fake_session
, did
, data
, content
=nsd_content
)
2044 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
2047 "'nsd' must be a list of only one element",
2048 norm(str(e
.exception
)),
2049 "Wrong exception text",
2051 self
.db
.reset_mock()
2054 def test_delete_nsd(self
):
2055 did
= db_nsd_content
["_id"]
2056 self
.db
.get_one
.return_value
= db_nsd_content
2057 p_id
= db_nsd_content
["_admin"]["projects_read"][0]
2058 with self
.subTest(i
=1, t
="Normal Deletion"):
2059 self
.db
.get_list
.return_value
= []
2060 self
.db
.del_one
.return_value
= {"deleted": 1}
2061 self
.topic
.delete(fake_session
, did
)
2062 db_args
= self
.db
.del_one
.call_args
[0]
2063 msg_args
= self
.msg
.write
.call_args
[0]
2064 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
2065 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
2066 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
2067 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
2068 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
2070 db_args
[1]["_admin.projects_write.cont"],
2074 db_g1_args
= self
.db
.get_one
.call_args
[0]
2075 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
2076 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB NSD ID")
2077 db_gl_calls
= self
.db
.get_list
.call_args_list
2078 self
.assertEqual(db_gl_calls
[0][0][0], "nsrs", "Wrong DB topic")
2079 # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID") # Filter changed after call
2080 self
.assertEqual(db_gl_calls
[1][0][0], "nsts", "Wrong DB topic")
2082 db_gl_calls
[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
2083 db_nsd_content
["id"],
2084 "Wrong DB NSD netslice-subnet nsd-ref",
2086 self
.db
.set_one
.assert_not_called()
2087 fs_del_calls
= self
.fs
.file_delete
.call_args_list
2088 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
2089 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
2090 with self
.subTest(i
=2, t
="Conflict on Delete - NSD in use by nsr"):
2091 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-nsr"}]
2092 with self
.assertRaises(
2093 EngineException
, msg
="Accepted NSD in use by NSR"
2095 self
.topic
.delete(fake_session
, did
)
2097 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2100 "there is at least one ns instance using this descriptor",
2101 norm(str(e
.exception
)),
2102 "Wrong exception text",
2104 with self
.subTest(i
=3, t
="Conflict on Delete - NSD in use by NST"):
2105 self
.db
.get_list
.side_effect
= [
2107 [{"_id": str(uuid4()), "name": "fake-nst"}],
2109 with self
.assertRaises(
2110 EngineException
, msg
="Accepted NSD in use by NST"
2112 self
.topic
.delete(fake_session
, did
)
2114 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2117 "there is at least one netslice template referencing this descriptor",
2118 norm(str(e
.exception
)),
2119 "Wrong exception text",
2121 with self
.subTest(i
=4, t
="Non-existent NSD"):
2122 excp_msg
= "Not found any {} with filter='{}'".format("NSD", {"_id": did
})
2123 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
2124 with self
.assertRaises(
2125 DbException
, msg
="Accepted non-existent NSD ID"
2127 self
.topic
.delete(fake_session
, did
)
2129 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
2132 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
2134 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
2135 db_nsd_content
["_admin"]["projects_read"].append("other_project")
2136 self
.db
.get_one
= Mock(return_value
=db_nsd_content
)
2137 self
.db
.get_list
= Mock(return_value
=[])
2138 self
.msg
.write
.reset_mock()
2139 self
.db
.del_one
.reset_mock()
2140 self
.fs
.file_delete
.reset_mock()
2142 self
.topic
.delete(fake_session
, did
)
2143 self
.db
.del_one
.assert_not_called()
2144 self
.msg
.write
.assert_not_called()
2145 db_g1_args
= self
.db
.get_one
.call_args
[0]
2146 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
2147 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
2148 db_s1_args
= self
.db
.set_one
.call_args
2149 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
2150 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
2152 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
2155 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
2158 db_s1_args
[1]["pull_list"],
2159 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
2160 "Wrong DB pull_list dictionary",
2162 self
.fs
.file_delete
.assert_not_called()
2163 self
.db
.reset_mock()
2166 def prepare_nsd_validation(self
):
2167 descriptor_name
= "test_ns_descriptor"
2168 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
2169 "/tmp/" + str(uuid4()), "a+b"
2171 old_nsd
, new_nsd
= self
.create_desc_temp(db_nsd_content
)
2172 return descriptor_name
, old_nsd
, new_nsd
2174 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2175 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2176 def test_validate_descriptor_ns_configuration_changed(
2177 self
, mock_safe_load
, mock_detect_usage
2179 """Validating NSD and NSD has changes in ns-configuration:config-primitive"""
2180 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
2181 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
2182 mock_detect_usage
.return_value
= True
2183 self
.db
.get_one
.return_value
= old_nsd
2185 {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
2188 {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
2191 with self
.assertNotRaises(EngineException
):
2192 self
.topic
._validate
_descriptor
_changes
(
2193 old_nsd
["_id"], descriptor_name
, "/tmp", "/tmp:1"
2195 self
.db
.get_one
.assert_called_once()
2196 mock_detect_usage
.assert_called_once()
2197 self
.assertEqual(mock_safe_load
.call_count
, 2)
2199 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2200 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2201 def test_validate_descriptor_nsd_name_changed(
2202 self
, mock_safe_load
, mock_detect_usage
2204 """Validating NSD, NSD name has changed."""
2205 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
2206 did
= old_nsd
["_id"]
2207 new_nsd
["name"] = "nscharm-ns2"
2208 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
2209 mock_detect_usage
.return_value
= True
2210 self
.db
.get_one
.return_value
= old_nsd
2212 with self
.assertRaises(
2213 EngineException
, msg
="there are disallowed changes in the ns descriptor"
2215 self
.topic
._validate
_descriptor
_changes
(
2216 did
, descriptor_name
, "/tmp", "/tmp:1"
2219 e
.exception
.http_code
,
2220 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2221 "Wrong HTTP status code",
2224 norm("there are disallowed changes in the ns descriptor"),
2225 norm(str(e
.exception
)),
2226 "Wrong exception text",
2229 self
.db
.get_one
.assert_called_once()
2230 mock_detect_usage
.assert_called_once()
2231 self
.assertEqual(mock_safe_load
.call_count
, 2)
2233 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2234 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2235 def test_validate_descriptor_nsd_name_changed_nsd_not_in_use(
2236 self
, mock_safe_load
, mock_detect_usage
2238 """Validating NSD, NSD name has changed, NSD is not in use."""
2239 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
2240 did
= old_nsd
["_id"]
2241 new_nsd
["name"] = "nscharm-ns2"
2242 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
2243 mock_detect_usage
.return_value
= None
2244 self
.db
.get_one
.return_value
= old_nsd
2246 with self
.assertNotRaises(Exception):
2247 self
.topic
._validate
_descriptor
_changes
(
2248 did
, descriptor_name
, "/tmp", "/tmp:1"
2251 self
.db
.get_one
.assert_called_once()
2252 mock_detect_usage
.assert_called_once()
2253 mock_safe_load
.assert_not_called()
2255 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
2258 indata
= deepcopy(db_nsd_content
)
2259 vld
= indata
["virtual-link-desc"][0]
2260 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2264 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
2267 indata
= deepcopy(db_nsd_content
)
2268 vld
= indata
["virtual-link-desc"][0]
2269 df
= indata
["df"][0]
2272 "virtual-link-desc-id": "mgmt",
2273 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
2275 df
["virtual-link-profile"] = [affected_vlp
]
2276 with self
.assertRaises(EngineException
) as e
:
2277 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2281 e
.exception
.http_code
,
2282 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2283 "Wrong HTTP status code",
2287 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
2288 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
2289 df
["id"], affected_vlp
["id"]
2292 norm(str(e
.exception
)),
2293 "Wrong exception text",
2296 def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self
):
2297 indata
= deepcopy(db_nsd_content
)
2298 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2300 def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self
):
2301 indata
= deepcopy(db_nsd_content
)
2302 df
= indata
["df"][0]
2303 affected_vnf_profile
= df
["vnf-profile"][0]
2304 indata
["vnfd-id"] = ["non-existing-vnfd"]
2305 with self
.assertRaises(EngineException
) as e
:
2306 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2308 e
.exception
.http_code
,
2309 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2310 "Wrong HTTP status code",
2314 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
2315 "does not match any vnfd-id".format(
2317 affected_vnf_profile
["id"],
2318 affected_vnf_profile
["vnfd-id"],
2321 norm(str(e
.exception
)),
2322 "Wrong exception text",
2325 def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
2328 nsd_descriptor
= deepcopy(db_nsd_content
)
2329 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2330 df
= nsd_descriptor
["df"][0]
2331 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2332 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2336 def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
2339 nsd_descriptor
= deepcopy(db_nsd_content
)
2340 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2341 df
= nsd_descriptor
["df"][0]
2342 affected_vnf_profile
= df
["vnf-profile"][0]
2343 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
2344 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2345 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
2346 with self
.assertRaises(EngineException
) as e
:
2347 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2351 e
.exception
.http_code
,
2352 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2353 "Wrong HTTP status code",
2357 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
2358 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
2359 "non existing ext-cpd:id inside vnfd '{}'".format(
2361 affected_vnf_profile
["id"],
2362 affected_virtual_link
["virtual-link-profile-id"],
2364 vnfd_descriptor
["id"],
2367 norm(str(e
.exception
)),
2368 "Wrong exception text",
2371 def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self
):
2372 nsd_descriptor
= deepcopy(db_nsd_content
)
2373 invalid_vnfd_id
= "invalid-vnfd-id"
2374 nsd_descriptor
["id"] = "invalid-vnfd-id-ns"
2375 nsd_descriptor
["vnfd-id"][0] = invalid_vnfd_id
2376 nsd_descriptor
["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
2377 nsd_descriptor
["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
2378 with self
.assertRaises(EngineException
) as e
:
2379 self
.db
.get_list
.return_value
= []
2380 nsd_descriptor
= self
.topic
.check_conflict_on_edit(
2381 fake_session
, nsd_descriptor
, [], "id"
2384 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2388 "Descriptor error at 'vnfd-id'='{}' references a non "
2389 "existing vnfd".format(invalid_vnfd_id
)
2391 norm(str(e
.exception
)),
2392 "Wrong exception text",
2395 def test_validate_vnffgd_descriptor_on_valid_descriptor(self
):
2396 indata
= yaml
.safe_load(db_sfc_nsds_text
)[0]
2397 vnffgd
= indata
.get("vnffgd")
2399 self
.topic
.validate_vnffgd_data(fg
, indata
)
2401 def test_validate_vnffgd_descriptor_not_matching_nfp_position_element(self
):
2402 indata
= yaml
.safe_load(db_sfc_nsds_text
)[0]
2403 vnffgd
= indata
.get("vnffgd")
2405 nfpd
= fg
.get("nfpd")[0]
2406 with self
.assertRaises(EngineException
) as e
:
2407 fg
.update({"nfp-position-element": [{"id": "test1"}]})
2408 self
.topic
.validate_vnffgd_data(fg
, indata
)
2410 e
.exception
.http_code
,
2411 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2412 "Wrong HTTP status code",
2416 "Error at vnffgd nfpd[id='{}']:nfp-position-element-id='{}' "
2417 "does not match any nfp-position-element".format(nfpd
["id"], "test")
2419 norm(str(e
.exception
)),
2420 "Wrong exception text",
2423 def test_validate_vnffgd_descriptor_not_matching_constituent_base_element_id(
2426 indata
= yaml
.safe_load(db_sfc_nsds_text
)[0]
2427 vnffgd
= indata
.get("vnffgd")
2429 fg
["nfpd"][0]["position-desc-id"][0]["cp-profile-id"][0][
2430 "constituent-profile-elements"
2431 ][0]["constituent-base-element-id"] = "error_vnf"
2432 with self
.assertRaises(EngineException
) as e
:
2433 self
.topic
.validate_vnffgd_data(fg
, indata
)
2435 e
.exception
.http_code
,
2436 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2437 "Wrong HTTP status code",
2441 "Error at vnffgd constituent_profile[id='{}']:vnfd-id='{}' "
2442 "does not match any constituent-base-element-id".format(
2446 norm(str(e
.exception
)),
2447 "Wrong exception text",
2451 if __name__
== "__main__":