08bececdaf742d03caba76ae16a0fe3e597758b7
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
,
36 from osm_nbi
.descriptor_topics
import VnfdTopic
, NsdTopic
37 from osm_nbi
.engine
import EngineException
38 from osm_common
.dbbase
import DbException
42 import collections
.abc
44 collections
.MutableSequence
= collections
.abc
.MutableSequence
46 test_name
= "test-user"
47 db_vnfd_content
= yaml
.safe_load(db_vnfds_text
)[0]
48 db_nsd_content
= yaml
.safe_load(db_nsds_text
)[0]
49 test_pid
= db_vnfd_content
["_admin"]["projects_read"][0]
51 "username": test_name
,
52 "project_id": (test_pid
,),
57 "allow_show_user_project_role": True,
59 UUID
= "00000000-0000-0000-0000-000000000000"
63 return {"projects_read": []}
66 def setup_mock_fs(fs
):
68 fs
.get_params
.return_value
= {}
69 fs
.file_exists
.return_value
= False
70 fs
.file_open
.side_effect
= lambda path
, mode
: tempfile
.TemporaryFile(mode
="a+b")
74 """Normalize string for checking"""
75 return " ".join(s
.strip().split()).lower()
78 def compare_desc(tc
, d1
, d2
, k
):
80 Compare two descriptors
81 We need this function because some methods are adding/removing items to/from the descriptors
82 before they are stored in the database, so the original and stored versions will differ
83 What we check is that COMMON LEAF ITEMS are equal
84 Lists of different length are not compared
85 :param tc: Test Case wich provides context (in particular the assert* methods)
86 :param d1,d2: Descriptors to be compared
87 :param k: key/item being compared
90 if isinstance(d1
, dict) and isinstance(d2
, dict):
93 compare_desc(tc
, d1
[key
], d2
[key
], k
+ "[{}]".format(key
))
94 elif isinstance(d1
, list) and isinstance(d2
, list) and len(d1
) == len(d2
):
95 for i
in range(len(d1
)):
96 compare_desc(tc
, d1
[i
], d2
[i
], k
+ "[{}]".format(i
))
98 tc
.assertEqual(d1
, d2
, "Wrong descriptor content: {}".format(k
))
101 class Test_VnfdTopic(TestCase
):
104 cls
.test_name
= "test-vnfd-topic"
107 def tearDownClass(cls
):
111 self
.db
= Mock(dbbase
.DbBase())
112 self
.fs
= Mock(fsbase
.FsBase())
113 self
.msg
= Mock(msgbase
.MsgBase())
114 self
.auth
= Mock(authconn
.Authconn(None, None, None))
115 self
.topic
= VnfdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
116 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
119 def assertNotRaises(self
, exception_type
=Exception):
122 except exception_type
:
123 raise self
.failureException("{} raised".format(exception_type
.__name
__))
125 def create_desc_temp(self
, template
):
126 old_desc
= deepcopy(template
)
127 new_desc
= deepcopy(template
)
128 return old_desc
, new_desc
130 def prepare_vnfd_creation(self
):
131 setup_mock_fs(self
.fs
)
132 test_vnfd
= deepcopy(db_vnfd_content
)
133 did
= db_vnfd_content
["_id"]
134 self
.db
.create
.return_value
= did
135 self
.db
.get_one
.side_effect
= [
136 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
139 return did
, test_vnfd
141 def prepare_vnfd(self
, vnfd_text
):
142 setup_mock_fs(self
.fs
)
143 test_vnfd
= yaml
.safe_load(vnfd_text
)
144 self
.db
.create
.return_value
= UUID
145 self
.db
.get_one
.side_effect
= [
146 {"_id": UUID
, "_admin": admin_value()},
149 return UUID
, test_vnfd
151 def prepare_test_vnfd(self
, test_vnfd
):
153 del test_vnfd
["_admin"]
154 del test_vnfd
["vdu"][0]["cloud-init-file"]
155 del test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
157 ][0]["execution-environment-list"][0]["juju"]
160 @patch("osm_nbi.descriptor_topics.shutil")
161 @patch("osm_nbi.descriptor_topics.os.rename")
162 def test_new_vnfd_normal_creation(self
, mock_rename
, mock_shutil
):
163 did
, test_vnfd
= self
.prepare_vnfd_creation()
164 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
166 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
167 db_args
= self
.db
.create
.call_args
[0]
168 msg_args
= self
.msg
.write
.call_args
[0]
170 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
171 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
172 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
173 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
174 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
175 self
.assertEqual(did2
, did
, "Wrong DB VNFD id")
176 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
178 db_args
[1]["_admin"]["modified"],
179 db_args
[1]["_admin"]["created"],
180 "Wrong modification time",
183 db_args
[1]["_admin"]["projects_read"],
185 "Wrong read-only project list",
188 db_args
[1]["_admin"]["projects_write"],
190 "Wrong read-write project list",
193 self
.db
.get_one
.side_effect
= [
194 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
198 self
.topic
.upload_content(
199 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
201 msg_args
= self
.msg
.write
.call_args
[0]
202 test_vnfd
["_id"] = did
203 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
204 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
205 self
.assertEqual(msg_args
[2], test_vnfd
, "Wrong message content")
207 db_args
= self
.db
.get_one
.mock_calls
[0][1]
208 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
209 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB VNFD id")
211 db_args
= self
.db
.replace
.call_args
[0]
212 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
213 self
.assertEqual(db_args
[1], did
, "Wrong DB VNFD id")
215 admin
= db_args
[2]["_admin"]
216 db_admin
= deepcopy(db_vnfd_content
["_admin"])
217 self
.assertEqual(admin
["type"], "vnfd", "Wrong descriptor type")
218 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
220 admin
["modified"], db_admin
["created"], "Wrong modification time"
223 admin
["projects_read"],
224 db_admin
["projects_read"],
225 "Wrong read-only project list",
228 admin
["projects_write"],
229 db_admin
["projects_write"],
230 "Wrong read-write project list",
233 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
236 admin
["operationalState"], "ENABLED", "Wrong operational state"
238 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
240 storage
= admin
["storage"]
241 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
242 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
243 self
.assertEqual(admin
["revision"], 1, "Wrong revision number")
244 compare_desc(self
, test_vnfd
, db_args
[2], "VNFD")
246 @patch("osm_nbi.descriptor_topics.shutil")
247 @patch("osm_nbi.descriptor_topics.os.rename")
248 def test_new_vnfd_exploit(self
, mock_rename
, mock_shutil
):
249 id, test_vnfd
= self
.prepare_vnfd(vnfd_exploit_text
)
251 with self
.assertRaises(EngineException
):
252 self
.topic
.upload_content(
253 fake_session
, id, test_vnfd
, {}, {"Content-Type": []}
256 @patch("osm_nbi.descriptor_topics.shutil")
257 @patch("osm_nbi.descriptor_topics.os.rename")
258 def test_new_vnfd_valid_helm_chart(self
, mock_rename
, mock_shutil
):
259 id, test_vnfd
= self
.prepare_vnfd(vnfd_exploit_fixed_text
)
261 with self
.assertNotRaises():
262 self
.topic
.upload_content(
263 fake_session
, id, test_vnfd
, {}, {"Content-Type": []}
266 @patch("osm_nbi.descriptor_topics.shutil")
267 @patch("osm_nbi.descriptor_topics.os.rename")
268 def test_new_vnfd_check_pyangbind_validation_additional_properties(
269 self
, mock_rename
, mock_shutil
271 did
, test_vnfd
= self
.prepare_vnfd_creation()
272 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
273 self
.topic
.upload_content(
274 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
276 test_vnfd
["_id"] = did
277 test_vnfd
["extra-property"] = 0
278 self
.db
.get_one
.side_effect
= (
279 lambda table
, filter, fail_on_empty
=None, fail_on_more
=None: {
281 "_admin": deepcopy(db_vnfd_content
["_admin"]),
285 with self
.assertRaises(
286 EngineException
, msg
="Accepted VNFD with an additional property"
288 self
.topic
.upload_content(
289 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
292 e
.exception
.http_code
,
293 HTTPStatus
.UNPROCESSABLE_ENTITY
,
294 "Wrong HTTP status code",
298 "Error in pyangbind validation: {} ({})".format(
299 "json object contained a key that did not exist", "extra-property"
302 norm(str(e
.exception
)),
303 "Wrong exception text",
305 db_args
= self
.db
.replace
.call_args
[0]
306 admin
= db_args
[2]["_admin"]
307 self
.assertEqual(admin
["revision"], 1, "Wrong revision number")
309 @patch("osm_nbi.descriptor_topics.shutil")
310 @patch("osm_nbi.descriptor_topics.os.rename")
311 def test_new_vnfd_check_pyangbind_validation_property_types(
312 self
, mock_rename
, mock_shutil
314 did
, test_vnfd
= self
.prepare_vnfd_creation()
315 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
316 test_vnfd
["_id"] = did
317 test_vnfd
["product-name"] = {"key": 0}
319 with self
.assertRaises(
320 EngineException
, msg
="Accepted VNFD with a wrongly typed property"
322 self
.topic
.upload_content(
323 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
326 e
.exception
.http_code
,
327 HTTPStatus
.UNPROCESSABLE_ENTITY
,
328 "Wrong HTTP status code",
332 "Error in pyangbind validation: {} ({})".format(
333 "json object contained a key that did not exist", "key"
336 norm(str(e
.exception
)),
337 "Wrong exception text",
340 @patch("osm_nbi.descriptor_topics.shutil")
341 @patch("osm_nbi.descriptor_topics.os.rename")
342 def test_new_vnfd_check_input_validation_cloud_init(self
, mock_rename
, mock_shutil
):
343 did
, test_vnfd
= self
.prepare_vnfd_creation()
344 del test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
346 ][0]["execution-environment-list"][0]["juju"]
348 with self
.assertRaises(
349 EngineException
, msg
="Accepted non-existent cloud_init file"
351 self
.topic
.upload_content(
352 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
355 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
359 "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format(
360 "cloud-init", test_vnfd
["id"], test_vnfd
["vdu"][0]["id"]
363 norm(str(e
.exception
)),
364 "Wrong exception text",
367 @patch("osm_nbi.descriptor_topics.shutil")
368 @patch("osm_nbi.descriptor_topics.os.rename")
369 def test_new_vnfd_check_input_validation_day12_configuration(
370 self
, mock_rename
, mock_shutil
372 did
, test_vnfd
= self
.prepare_vnfd_creation()
373 del test_vnfd
["vdu"][0]["cloud-init-file"]
375 with self
.assertRaises(
376 EngineException
, msg
="Accepted non-existent charm in VNF configuration"
378 self
.topic
.upload_content(
379 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
382 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
386 "{} defined in vnf[id={}] but not present in package".format(
387 "charm", test_vnfd
["id"]
390 norm(str(e
.exception
)),
391 "Wrong exception text",
394 @patch("osm_nbi.descriptor_topics.shutil")
395 @patch("osm_nbi.descriptor_topics.os.rename")
396 def test_new_vnfd_check_input_validation_mgmt_cp(self
, mock_rename
, mock_shutil
):
397 did
, test_vnfd
= self
.prepare_vnfd_creation()
398 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
399 del test_vnfd
["mgmt-cp"]
401 with self
.assertRaises(
402 EngineException
, msg
="Accepted VNFD without management interface"
404 self
.topic
.upload_content(
405 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
408 e
.exception
.http_code
,
409 HTTPStatus
.UNPROCESSABLE_ENTITY
,
410 "Wrong HTTP status code",
413 norm("'{}' is a mandatory field and it is not defined".format("mgmt-cp")),
414 norm(str(e
.exception
)),
415 "Wrong exception text",
418 @patch("osm_nbi.descriptor_topics.shutil")
419 @patch("osm_nbi.descriptor_topics.os.rename")
420 def test_new_vnfd_check_input_validation_mgmt_cp_connection_point(
421 self
, mock_rename
, mock_shutil
423 did
, test_vnfd
= self
.prepare_vnfd_creation()
424 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
425 test_vnfd
["mgmt-cp"] = "wrong-cp"
427 with self
.assertRaises(
428 EngineException
, msg
="Accepted wrong mgmt-cp connection point"
430 self
.topic
.upload_content(
431 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
434 e
.exception
.http_code
,
435 HTTPStatus
.UNPROCESSABLE_ENTITY
,
436 "Wrong HTTP status code",
440 "mgmt-cp='{}' must match an existing ext-cpd".format(
444 norm(str(e
.exception
)),
445 "Wrong exception text",
448 @patch("osm_nbi.descriptor_topics.shutil")
449 @patch("osm_nbi.descriptor_topics.os.rename")
450 def test_new_vnfd_check_input_validation_vdu_int_cpd(
451 self
, mock_rename
, mock_shutil
453 """Testing input validation during new vnfd creation
454 for vdu internal connection point"""
455 did
, test_vnfd
= self
.prepare_vnfd_creation()
456 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
457 ext_cpd
= test_vnfd
["ext-cpd"][1]
458 ext_cpd
["int-cpd"]["cpd"] = "wrong-cpd"
460 with self
.assertRaises(
461 EngineException
, msg
="Accepted wrong ext-cpd internal connection point"
463 self
.topic
.upload_content(
464 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
467 e
.exception
.http_code
,
468 HTTPStatus
.UNPROCESSABLE_ENTITY
,
469 "Wrong HTTP status code",
473 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
477 norm(str(e
.exception
)),
478 "Wrong exception text",
481 @patch("osm_nbi.descriptor_topics.shutil")
482 @patch("osm_nbi.descriptor_topics.os.rename")
483 def test_new_vnfd_check_input_validation_duplicated_vld(
484 self
, mock_rename
, mock_shutil
486 """Testing input validation during new vnfd creation
487 for dublicated virtual link description"""
488 did
, test_vnfd
= self
.prepare_vnfd_creation()
489 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
490 test_vnfd
["int-virtual-link-desc"].insert(0, {"id": "internal"})
492 with self
.assertRaises(
493 EngineException
, msg
="Accepted duplicated VLD name"
495 self
.topic
.upload_content(
496 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
499 e
.exception
.http_code
,
500 HTTPStatus
.UNPROCESSABLE_ENTITY
,
501 "Wrong HTTP status code",
505 "identifier id '{}' is not unique".format(
506 test_vnfd
["int-virtual-link-desc"][0]["id"]
509 norm(str(e
.exception
)),
510 "Wrong exception text",
513 @patch("osm_nbi.descriptor_topics.shutil")
514 @patch("osm_nbi.descriptor_topics.os.rename")
515 def test_new_vnfd_check_input_validation_vdu_int_virtual_link_desc(
516 self
, mock_rename
, mock_shutil
518 """Testing input validation during new vnfd creation
519 for vdu internal virtual link description"""
520 did
, test_vnfd
= self
.prepare_vnfd_creation()
521 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
522 vdu
= test_vnfd
["vdu"][0]
523 int_cpd
= vdu
["int-cpd"][1]
524 int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
526 with self
.assertRaises(
527 EngineException
, msg
="Accepted int-virtual-link-desc"
529 self
.topic
.upload_content(
530 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
533 e
.exception
.http_code
,
534 HTTPStatus
.UNPROCESSABLE_ENTITY
,
535 "Wrong HTTP status code",
539 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
540 "int-virtual-link-desc".format(
541 vdu
["id"], int_cpd
["id"], int_cpd
["int-virtual-link-desc"]
544 norm(str(e
.exception
)),
545 "Wrong exception text",
548 @patch("osm_nbi.descriptor_topics.shutil")
549 @patch("osm_nbi.descriptor_topics.os.rename")
550 def test_new_vnfd_check_input_validation_virtual_link_profile(
551 self
, mock_rename
, mock_shutil
553 """Testing input validation during new vnfd creation
554 for virtual link profile"""
555 did
, test_vnfd
= self
.prepare_vnfd_creation()
556 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
557 fake_ivld_profile
= {"id": "fake-profile-ref", "flavour": "fake-flavour"}
558 df
= test_vnfd
["df"][0]
559 df
["virtual-link-profile"] = [fake_ivld_profile
]
561 with self
.assertRaises(
562 EngineException
, msg
="Accepted non-existent Profile Ref"
564 self
.topic
.upload_content(
565 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
568 e
.exception
.http_code
,
569 HTTPStatus
.UNPROCESSABLE_ENTITY
,
570 "Wrong HTTP status code",
574 "df[id='{}']:virtual-link-profile='{}' must match an existing "
575 "int-virtual-link-desc".format(df
["id"], fake_ivld_profile
["id"])
577 norm(str(e
.exception
)),
578 "Wrong exception text",
581 @patch("osm_nbi.descriptor_topics.shutil")
582 @patch("osm_nbi.descriptor_topics.os.rename")
583 def test_new_vnfd_check_input_validation_scaling_criteria_vdu_id(
584 self
, mock_rename
, mock_shutil
586 """Testing input validation during new vnfd creation
587 for scaling criteria with invalid vdu-id"""
588 did
, test_vnfd
= self
.prepare_vnfd_creation()
589 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
590 test_vnfd
["df"][0]["scaling-aspect"][0]["aspect-delta-details"]["deltas"][0][
592 ][0]["id"] = "vdudelta1"
593 affected_df
= test_vnfd
["df"][0]
594 sa
= affected_df
["scaling-aspect"][0]
595 delta
= sa
["aspect-delta-details"]["deltas"][0]
596 vdu_delta
= delta
["vdu-delta"][0]
598 with self
.assertRaises(
599 EngineException
, msg
="Accepted invalid 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='{}']:aspect-delta-details"
613 "vdu-id='{}' not defined in vdu".format(
620 norm(str(e
.exception
)),
621 "Wrong exception text",
624 @patch("osm_nbi.descriptor_topics.shutil")
625 @patch("osm_nbi.descriptor_topics.os.rename")
626 def test_new_vnfd_check_input_validation_scaling_criteria_monitoring_param_ref(
627 self
, mock_rename
, mock_shutil
629 """Testing input validation during new vnfd creation
630 for scaling criteria without monitoring parameter"""
631 did
, test_vnfd
= self
.prepare_vnfd_creation()
632 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
633 vdu
= test_vnfd
["vdu"][1]
634 affected_df
= test_vnfd
["df"][0]
635 sa
= affected_df
["scaling-aspect"][0]
636 sp
= sa
["scaling-policy"][0]
637 sc
= sp
["scaling-criteria"][0]
638 vdu
.pop("monitoring-parameter")
640 with self
.assertRaises(
641 EngineException
, msg
="Accepted non-existent Scaling Group Policy Criteria"
643 self
.topic
.upload_content(
644 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
647 e
.exception
.http_code
,
648 HTTPStatus
.UNPROCESSABLE_ENTITY
,
649 "Wrong HTTP status code",
653 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
654 "[name='{}']:scaling-criteria[name='{}']: "
655 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
660 sc
["vnf-monitoring-param-ref"],
663 norm(str(e
.exception
)),
664 "Wrong exception text",
667 @patch("osm_nbi.descriptor_topics.shutil")
668 @patch("osm_nbi.descriptor_topics.os.rename")
669 def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration(
670 self
, mock_rename
, mock_shutil
672 """Testing input validation during new vnfd creation
673 for scaling criteria without day12 configuration"""
674 did
, test_vnfd
= self
.prepare_vnfd_creation()
675 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
676 test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
679 df
= test_vnfd
["df"][0]
681 with self
.assertRaises(
682 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
684 self
.topic
.upload_content(
685 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
688 e
.exception
.http_code
,
689 HTTPStatus
.UNPROCESSABLE_ENTITY
,
690 "Wrong HTTP status code",
694 "'day1-2 configuration' not defined in the descriptor but it is referenced "
695 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
696 df
["id"], df
["scaling-aspect"][0]["id"]
699 norm(str(e
.exception
)),
700 "Wrong exception text",
703 @patch("osm_nbi.descriptor_topics.shutil")
704 @patch("osm_nbi.descriptor_topics.os.rename")
705 def test_new_vnfd_check_input_validation_scaling_config_action(
706 self
, mock_rename
, mock_shutil
708 """Testing input validation during new vnfd creation
709 for scaling criteria wrong config primitive"""
710 did
, test_vnfd
= self
.prepare_vnfd_creation()
711 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
712 df
= test_vnfd
["df"][0]
713 affected_df
= test_vnfd
["df"][0]
714 sa
= affected_df
["scaling-aspect"][0]
715 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
716 "operate-vnf-op-config"
717 )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
719 with self
.assertRaises(
720 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
722 self
.topic
.upload_content(
723 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
726 e
.exception
.http_code
,
727 HTTPStatus
.UNPROCESSABLE_ENTITY
,
728 "Wrong HTTP status code",
732 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
733 "config-primitive-name-ref='{}' does not match any "
734 "day1-2 configuration:config-primitive:name".format(
736 df
["scaling-aspect"][0]["id"],
737 sa
["scaling-config-action"][0]["vnf-config-primitive-name-ref"],
740 norm(str(e
.exception
)),
741 "Wrong exception text",
744 @patch("osm_nbi.descriptor_topics.shutil")
745 @patch("osm_nbi.descriptor_topics.os.rename")
746 def test_new_vnfd_check_input_validation_healing_criteria_vdu_id(
747 self
, mock_rename
, mock_shutil
749 """Testing input validation during new vnfd creation
750 for healing criteria with invalid vdu-id"""
751 did
, test_vnfd
= self
.prepare_vnfd_creation()
752 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
753 test_vnfd
["df"][0]["healing-aspect"][0]["healing-policy"][0][
756 affected_df
= test_vnfd
["df"][0]
757 ha
= affected_df
["healing-aspect"][0]
758 hp
= ha
["healing-policy"][0]
759 hp_vdu_id
= hp
["vdu-id"]
761 with self
.assertRaises(
762 EngineException
, msg
="Accepted invalid Healing Group Policy Criteria"
764 self
.topic
.upload_content(
765 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
768 e
.exception
.http_code
,
769 HTTPStatus
.UNPROCESSABLE_ENTITY
,
770 "Wrong HTTP status code",
774 "df[id='{}']:healing-aspect[id='{}']:healing-policy"
776 "vdu-id='{}' not defined in vdu".format(
783 norm(str(e
.exception
)),
784 "Wrong exception text",
787 @patch("osm_nbi.descriptor_topics.shutil")
788 @patch("osm_nbi.descriptor_topics.os.rename")
789 def test_new_vnfd_check_input_validation_alarm_criteria_monitoring_param_ref(
790 self
, mock_rename
, mock_shutil
792 """Testing input validation during new vnfd creation
793 for alarm with invalid monitoring parameter reference"""
794 did
, test_vnfd
= self
.prepare_vnfd_creation()
795 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
796 test_vnfd
["vdu"][1]["alarm"][0]["vnf-monitoring-param-ref"] = "unit_test_alarm"
797 vdu
= test_vnfd
["vdu"][1]
798 alarm
= vdu
["alarm"][0]
799 alarm_monitoring_param
= alarm
["vnf-monitoring-param-ref"]
801 with self
.assertRaises(
802 EngineException
, msg
="Accepted invalid Alarm Criteria"
804 self
.topic
.upload_content(
805 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
808 e
.exception
.http_code
,
809 HTTPStatus
.UNPROCESSABLE_ENTITY
,
810 "Wrong HTTP status code",
814 "vdu[id='{}']:alarm[id='{}']:"
815 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
818 alarm_monitoring_param
,
821 norm(str(e
.exception
)),
822 "Wrong exception text",
825 @patch("osm_nbi.descriptor_topics.shutil")
826 @patch("osm_nbi.descriptor_topics.os.rename")
827 def test_new_vnfd_check_input_validation_storage_reference_criteria(
828 self
, mock_rename
, mock_shutil
830 """Testing input validation during new vnfd creation
831 for invalid virtual-storge-desc reference"""
832 did
, test_vnfd
= self
.prepare_vnfd_creation()
833 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
834 test_vnfd
["vdu"][1]["virtual-storage-desc"] = "unit_test_storage"
835 vdu
= test_vnfd
["vdu"][1]
836 vsd_ref
= vdu
["virtual-storage-desc"]
838 with self
.assertRaises(
839 EngineException
, msg
="Accepted invalid virtual-storage-desc"
841 self
.topic
.upload_content(
842 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
845 e
.exception
.http_code
,
846 HTTPStatus
.UNPROCESSABLE_ENTITY
,
847 "Wrong HTTP status code",
851 "vdu[virtual-storage-desc='{}']"
852 "not defined in vnfd".format(
856 norm(str(e
.exception
)),
857 "Wrong exception text",
860 @patch("osm_nbi.descriptor_topics.shutil")
861 @patch("osm_nbi.descriptor_topics.os.rename")
862 def test_new_vnfd_check_input_validation_compute_reference_criteria(
863 self
, mock_rename
, mock_shutil
865 """Testing input validation during new vnfd creation
866 for invalid virtual-compute-desc reference"""
867 did
, test_vnfd
= self
.prepare_vnfd_creation()
868 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
869 test_vnfd
["vdu"][1]["virtual-compute-desc"] = "unit_test_compute"
870 vdu
= test_vnfd
["vdu"][1]
871 vcd_ref
= vdu
["virtual-compute-desc"]
873 with self
.assertRaises(
874 EngineException
, msg
="Accepted invalid virtual-compute-desc"
876 self
.topic
.upload_content(
877 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
880 e
.exception
.http_code
,
881 HTTPStatus
.UNPROCESSABLE_ENTITY
,
882 "Wrong HTTP status code",
886 "vdu[virtual-compute-desc='{}']"
887 "not defined in vnfd".format(
891 norm(str(e
.exception
)),
892 "Wrong exception text",
895 @patch("osm_nbi.descriptor_topics.shutil")
896 @patch("osm_nbi.descriptor_topics.os.rename")
897 def test_new_vnfd_check_input_validation_everything_right(
898 self
, mock_rename
, mock_shutil
900 """Testing input validation during new vnfd creation
901 everything correct"""
902 did
, test_vnfd
= self
.prepare_vnfd_creation()
903 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
904 test_vnfd
["id"] = "fake-vnfd-id"
905 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
906 "operate-vnf-op-config"
907 )["day1-2"][0]["id"] = "fake-vnfd-id"
908 self
.db
.get_one
.side_effect
= [
909 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
912 rc
= self
.topic
.upload_content(
913 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
915 self
.assertTrue(rc
, "Input Validation: Unexpected failure")
917 def test_edit_vnfd(self
):
918 vnfd_content
= deepcopy(db_vnfd_content
)
919 did
= vnfd_content
["_id"]
920 self
.fs
.file_exists
.return_value
= True
921 self
.fs
.dir_ls
.return_value
= True
922 with self
.subTest(i
=1, t
="Normal Edition"):
924 self
.db
.get_one
.side_effect
= [deepcopy(vnfd_content
), None]
925 data
= {"product-name": "new-vnfd-name"}
926 self
.topic
.edit(fake_session
, did
, data
)
927 db_args
= self
.db
.replace
.call_args
[0]
928 msg_args
= self
.msg
.write
.call_args
[0]
930 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
931 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
932 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
933 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
934 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
936 db_args
[2]["_admin"]["created"],
937 vnfd_content
["_admin"]["created"],
938 "Wrong creation time",
941 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
944 db_args
[2]["_admin"]["projects_read"],
945 vnfd_content
["_admin"]["projects_read"],
946 "Wrong read-only project list",
949 db_args
[2]["_admin"]["projects_write"],
950 vnfd_content
["_admin"]["projects_write"],
951 "Wrong read-write project list",
954 db_args
[2]["product-name"], data
["product-name"], "Wrong VNFD Name"
956 with self
.subTest(i
=2, t
="Conflict on Edit"):
957 data
= {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
958 self
.db
.get_one
.side_effect
= [
959 deepcopy(vnfd_content
),
960 {"_id": str(uuid4()), "id": data
["id"]},
962 with self
.assertRaises(
963 EngineException
, msg
="Accepted existing VNFD ID"
965 self
.topic
.edit(fake_session
, did
, data
)
967 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
971 "{} with id '{}' already exists for this project".format(
975 norm(str(e
.exception
)),
976 "Wrong exception text",
978 with self
.subTest(i
=3, t
="Check Envelope"):
979 data
= {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
980 with self
.assertRaises(
981 EngineException
, msg
="Accepted VNFD with wrong envelope"
983 self
.topic
.edit(fake_session
, did
, data
, content
=vnfd_content
)
985 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
988 "'vnfd' must be dict", norm(str(e
.exception
)), "Wrong exception text"
992 def test_delete_vnfd(self
):
993 did
= db_vnfd_content
["_id"]
994 self
.db
.get_one
.return_value
= db_vnfd_content
995 p_id
= db_vnfd_content
["_admin"]["projects_read"][0]
996 with self
.subTest(i
=1, t
="Normal Deletion"):
997 self
.db
.get_list
.return_value
= []
998 self
.db
.del_one
.return_value
= {"deleted": 1}
999 self
.topic
.delete(fake_session
, did
)
1000 db_args
= self
.db
.del_one
.call_args
[0]
1001 msg_args
= self
.msg
.write
.call_args
[0]
1002 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1003 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
1004 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1005 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1006 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
1008 db_args
[1]["_admin.projects_write.cont"],
1012 db_g1_args
= self
.db
.get_one
.call_args
[0]
1013 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1014 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
1015 db_gl_calls
= self
.db
.get_list
.call_args_list
1016 self
.assertEqual(db_gl_calls
[0][0][0], "vnfrs", "Wrong DB topic")
1017 # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID") # Filter changed after call
1018 self
.assertEqual(db_gl_calls
[1][0][0], "nsds", "Wrong DB topic")
1020 db_gl_calls
[1][0][1]["vnfd-id"],
1021 db_vnfd_content
["id"],
1022 "Wrong DB NSD vnfd-id",
1026 self
.db
.del_list
.call_args
[0][0],
1027 self
.topic
.topic
+ "_revisions",
1032 self
.db
.del_list
.call_args
[0][1]["_id"]["$regex"],
1034 "Wrong ID for rexep delete",
1037 self
.db
.set_one
.assert_not_called()
1038 fs_del_calls
= self
.fs
.file_delete
.call_args_list
1039 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
1040 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
1041 with self
.subTest(i
=2, t
="Conflict on Delete - VNFD in use by VNFR"):
1042 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-vnfr"}]
1043 with self
.assertRaises(
1044 EngineException
, msg
="Accepted VNFD in use by VNFR"
1046 self
.topic
.delete(fake_session
, did
)
1048 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1051 "there is at least one vnf instance using this descriptor",
1052 norm(str(e
.exception
)),
1053 "Wrong exception text",
1055 with self
.subTest(i
=3, t
="Conflict on Delete - VNFD in use by NSD"):
1056 self
.db
.get_list
.side_effect
= [
1058 [{"_id": str(uuid4()), "name": "fake-nsd"}],
1060 with self
.assertRaises(
1061 EngineException
, msg
="Accepted VNFD in use by NSD"
1063 self
.topic
.delete(fake_session
, did
)
1065 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1068 "there is at least one ns package referencing this descriptor",
1069 norm(str(e
.exception
)),
1070 "Wrong exception text",
1072 with self
.subTest(i
=4, t
="Non-existent VNFD"):
1073 excp_msg
= "Not found any {} with filter='{}'".format("VNFD", {"_id": did
})
1074 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
1075 with self
.assertRaises(
1076 DbException
, msg
="Accepted non-existent VNFD ID"
1078 self
.topic
.delete(fake_session
, did
)
1080 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1083 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
1085 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
1086 db_vnfd_content
["_admin"]["projects_read"].append("other_project")
1087 self
.db
.get_one
= Mock(return_value
=db_vnfd_content
)
1088 self
.db
.get_list
= Mock(return_value
=[])
1089 self
.msg
.write
.reset_mock()
1090 self
.db
.del_one
.reset_mock()
1091 self
.fs
.file_delete
.reset_mock()
1093 self
.topic
.delete(fake_session
, did
)
1094 self
.db
.del_one
.assert_not_called()
1095 self
.msg
.write
.assert_not_called()
1096 db_g1_args
= self
.db
.get_one
.call_args
[0]
1097 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1098 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
1099 db_s1_args
= self
.db
.set_one
.call_args
1100 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
1101 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
1103 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1106 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
1109 db_s1_args
[1]["pull_list"],
1110 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
1111 "Wrong DB pull_list dictionary",
1113 self
.fs
.file_delete
.assert_not_called()
1116 def prepare_vnfd_validation(self
):
1117 descriptor_name
= "test_descriptor"
1118 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1119 "/tmp/" + str(uuid4()), "a+b"
1121 old_vnfd
, new_vnfd
= self
.create_desc_temp(db_vnfd_content
)
1122 return descriptor_name
, old_vnfd
, new_vnfd
1124 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1125 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1126 def test_validate_vnfd_changes_day12_config_primitive_changed(
1127 self
, mock_safe_load
, mock_detect_usage
1129 """Validating VNFD for VNFD updates, day1-2 config primitive has changed"""
1130 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
1131 did
= old_vnfd
["_id"]
1132 new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1134 ][0]["config-primitive"][0]["name"] = "new_action"
1135 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
1136 mock_detect_usage
.return_value
= True
1137 self
.db
.get_one
.return_value
= old_vnfd
1139 with self
.assertNotRaises(EngineException
):
1140 self
.topic
._validate
_descriptor
_changes
(
1141 did
, descriptor_name
, "/tmp/", "/tmp:1/"
1143 self
.db
.get_one
.assert_called_once()
1144 mock_detect_usage
.assert_called_once()
1145 self
.assertEqual(mock_safe_load
.call_count
, 2)
1147 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1148 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1149 def test_validate_vnfd_changes_sw_version_changed(
1150 self
, mock_safe_load
, mock_detect_usage
1152 """Validating VNFD for updates, software version has changed"""
1153 # old vnfd uses the default software version: 1.0
1154 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
1155 did
= old_vnfd
["_id"]
1156 new_vnfd
["software-version"] = "1.3"
1157 new_vnfd
["sw-image-desc"][0]["name"] = "new-image"
1158 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
1159 mock_detect_usage
.return_value
= True
1160 self
.db
.get_one
.return_value
= old_vnfd
1162 with self
.assertNotRaises(EngineException
):
1163 self
.topic
._validate
_descriptor
_changes
(
1164 did
, descriptor_name
, "/tmp/", "/tmp:1/"
1166 self
.db
.get_one
.assert_called_once()
1167 mock_detect_usage
.assert_called_once()
1168 self
.assertEqual(mock_safe_load
.call_count
, 2)
1170 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1171 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1172 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed(
1173 self
, mock_safe_load
, mock_detect_usage
1175 """Validating VNFD for updates, software version has not
1176 changed, mgmt-cp has changed."""
1177 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
1178 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
1179 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
1180 did
= old_vnfd
["_id"]
1181 mock_detect_usage
.return_value
= True
1182 self
.db
.get_one
.return_value
= old_vnfd
1184 with self
.assertRaises(
1185 EngineException
, msg
="there are disallowed changes in the vnf descriptor"
1187 self
.topic
._validate
_descriptor
_changes
(
1188 did
, descriptor_name
, "/tmp/", "/tmp:1/"
1192 e
.exception
.http_code
,
1193 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1194 "Wrong HTTP status code",
1197 norm("there are disallowed changes in the vnf descriptor"),
1198 norm(str(e
.exception
)),
1199 "Wrong exception text",
1201 self
.db
.get_one
.assert_called_once()
1202 mock_detect_usage
.assert_called_once()
1203 self
.assertEqual(mock_safe_load
.call_count
, 2)
1205 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1206 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1207 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed_vnfd_not_in_use(
1208 self
, mock_safe_load
, mock_detect_usage
1210 """Validating VNFD for updates, software version has not
1211 changed, mgmt-cp has changed, vnfd is not in use."""
1212 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
1213 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
1214 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
1215 did
= old_vnfd
["_id"]
1216 mock_detect_usage
.return_value
= None
1217 self
.db
.get_one
.return_value
= old_vnfd
1219 with self
.assertNotRaises(EngineException
):
1220 self
.topic
._validate
_descriptor
_changes
(
1221 did
, descriptor_name
, "/tmp/", "/tmp:1/"
1224 self
.db
.get_one
.assert_called_once()
1225 mock_detect_usage
.assert_called_once()
1226 mock_safe_load
.assert_not_called()
1228 def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self
):
1229 indata
= deepcopy(db_vnfd_content
)
1230 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1232 def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
1235 indata
= deepcopy(db_vnfd_content
)
1236 indata
["ext-cpd"] = []
1237 with self
.assertRaises(EngineException
) as e
:
1238 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1240 e
.exception
.http_code
,
1241 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1242 "Wrong HTTP status code",
1246 "mgmt-cp='{}' must match an existing ext-cpd".format(indata
["mgmt-cp"])
1248 norm(str(e
.exception
)),
1249 "Wrong exception text",
1252 def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self
):
1253 indata
= deepcopy(db_vnfd_content
)
1254 indata
.pop("mgmt-cp")
1255 with self
.assertRaises(EngineException
) as e
:
1256 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1258 e
.exception
.http_code
,
1259 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1260 "Wrong HTTP status code",
1263 norm("'mgmt-cp' is a mandatory field and it is not defined"),
1264 norm(str(e
.exception
)),
1265 "Wrong exception text",
1268 def test_validate_vdu_internal_connection_points_on_valid_descriptor(self
):
1269 indata
= db_vnfd_content
1270 vdu
= indata
["vdu"][0]
1271 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1273 def test_validate_external_connection_points_on_valid_descriptor(self
):
1274 indata
= db_vnfd_content
1275 self
.topic
.validate_external_connection_points(indata
)
1277 def test_validate_external_connection_points_when_missing_internal_connection_point(
1280 indata
= deepcopy(db_vnfd_content
)
1281 vdu
= indata
["vdu"][0]
1283 affected_ext_cpd
= indata
["ext-cpd"][0]
1284 with self
.assertRaises(EngineException
) as e
:
1285 self
.topic
.validate_external_connection_points(indata
)
1287 e
.exception
.http_code
,
1288 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1289 "Wrong HTTP status code",
1293 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
1294 affected_ext_cpd
["id"]
1297 norm(str(e
.exception
)),
1298 "Wrong exception text",
1301 def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
1304 indata
= deepcopy(db_vnfd_content
)
1305 vdu
= indata
["vdu"][0]
1309 "virtual-network-interface-requirement": [{"name": "duplicated"}],
1311 vdu
["int-cpd"].insert(0, duplicated_cpd
)
1312 with self
.assertRaises(EngineException
) as e
:
1313 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1315 e
.exception
.http_code
,
1316 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1317 "Wrong HTTP status code",
1321 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
1322 vdu
["id"], duplicated_cpd
["id"]
1325 norm(str(e
.exception
)),
1326 "Wrong exception text",
1329 def test_validate_external_connection_points_on_duplicated_external_connection_point(
1332 indata
= deepcopy(db_vnfd_content
)
1334 "id": "vnf-mgmt-ext",
1335 "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
1337 indata
["ext-cpd"].insert(0, duplicated_cpd
)
1338 with self
.assertRaises(EngineException
) as e
:
1339 self
.topic
.validate_external_connection_points(indata
)
1341 e
.exception
.http_code
,
1342 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1343 "Wrong HTTP status code",
1347 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
1348 duplicated_cpd
["id"]
1351 norm(str(e
.exception
)),
1352 "Wrong exception text",
1355 def test_validate_internal_virtual_links_on_valid_descriptor(self
):
1356 indata
= db_vnfd_content
1357 self
.topic
.validate_internal_virtual_links(indata
)
1359 def test_validate_internal_virtual_links_on_duplicated_ivld(self
):
1360 indata
= deepcopy(db_vnfd_content
)
1361 duplicated_vld
= {"id": "internal"}
1362 indata
["int-virtual-link-desc"].insert(0, duplicated_vld
)
1363 with self
.assertRaises(EngineException
) as e
:
1364 self
.topic
.validate_internal_virtual_links(indata
)
1366 e
.exception
.http_code
,
1367 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1368 "Wrong HTTP status code",
1372 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
1373 duplicated_vld
["id"]
1376 norm(str(e
.exception
)),
1377 "Wrong exception text",
1380 def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
1383 indata
= deepcopy(db_vnfd_content
)
1384 vdu
= indata
["vdu"][0]
1385 affected_int_cpd
= vdu
["int-cpd"][0]
1386 affected_int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
1387 with self
.assertRaises(EngineException
) as e
:
1388 self
.topic
.validate_internal_virtual_links(indata
)
1390 e
.exception
.http_code
,
1391 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1392 "Wrong HTTP status code",
1396 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
1397 "int-virtual-link-desc".format(
1399 affected_int_cpd
["id"],
1400 affected_int_cpd
["int-virtual-link-desc"],
1403 norm(str(e
.exception
)),
1404 "Wrong exception text",
1407 def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self
):
1408 indata
= deepcopy(db_vnfd_content
)
1409 affected_ivld_profile
= {"id": "non-existing-int-virtual-link-desc"}
1410 df
= indata
["df"][0]
1411 df
["virtual-link-profile"] = [affected_ivld_profile
]
1412 with self
.assertRaises(EngineException
) as e
:
1413 self
.topic
.validate_internal_virtual_links(indata
)
1415 e
.exception
.http_code
,
1416 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1417 "Wrong HTTP status code",
1421 "df[id='{}']:virtual-link-profile='{}' must match an existing "
1422 "int-virtual-link-desc".format(df
["id"], affected_ivld_profile
["id"])
1424 norm(str(e
.exception
)),
1425 "Wrong exception text",
1428 def test_validate_monitoring_params_on_valid_descriptor(self
):
1429 indata
= db_vnfd_content
1430 self
.topic
.validate_monitoring_params(indata
)
1432 def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self
):
1433 indata
= deepcopy(db_vnfd_content
)
1434 duplicated_mp
= {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1435 affected_ivld
= indata
["int-virtual-link-desc"][0]
1436 affected_ivld
["monitoring-parameters"] = [duplicated_mp
, duplicated_mp
]
1437 with self
.assertRaises(EngineException
) as e
:
1438 self
.topic
.validate_monitoring_params(indata
)
1440 e
.exception
.http_code
,
1441 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1442 "Wrong HTTP status code",
1446 "Duplicated monitoring-parameter id in "
1447 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1448 affected_ivld
["id"], duplicated_mp
["id"]
1451 norm(str(e
.exception
)),
1452 "Wrong exception text",
1455 def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self
):
1456 indata
= deepcopy(db_vnfd_content
)
1458 "id": "dataVM_cpu_util",
1459 "name": "dataVM_cpu_util",
1460 "performance_metric": "cpu",
1462 affected_vdu
= indata
["vdu"][1]
1463 affected_vdu
["monitoring-parameter"].insert(0, duplicated_mp
)
1464 with self
.assertRaises(EngineException
) as e
:
1465 self
.topic
.validate_monitoring_params(indata
)
1467 e
.exception
.http_code
,
1468 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1469 "Wrong HTTP status code",
1473 "Duplicated monitoring-parameter id in "
1474 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1475 affected_vdu
["id"], duplicated_mp
["id"]
1478 norm(str(e
.exception
)),
1479 "Wrong exception text",
1482 def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self
):
1483 indata
= deepcopy(db_vnfd_content
)
1487 "performance_metric": "memory",
1489 affected_df
= indata
["df"][0]
1490 affected_df
["monitoring-parameter"] = [duplicated_mp
, duplicated_mp
]
1491 with self
.assertRaises(EngineException
) as e
:
1492 self
.topic
.validate_monitoring_params(indata
)
1494 e
.exception
.http_code
,
1495 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1496 "Wrong HTTP status code",
1500 "Duplicated monitoring-parameter id in "
1501 "df[id='{}']:monitoring-parameter[id='{}']".format(
1502 affected_df
["id"], duplicated_mp
["id"]
1505 norm(str(e
.exception
)),
1506 "Wrong exception text",
1509 def test_validate_scaling_group_descriptor_on_valid_descriptor(self
):
1510 indata
= db_vnfd_content
1511 self
.topic
.validate_scaling_group_descriptor(indata
)
1513 def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self
):
1514 indata
= deepcopy(db_vnfd_content
)
1515 vdu
= indata
["vdu"][1]
1516 affected_df
= indata
["df"][0]
1517 affected_sa
= affected_df
["scaling-aspect"][0]
1518 affected_sp
= affected_sa
["scaling-policy"][0]
1519 affected_sc
= affected_sp
["scaling-criteria"][0]
1520 vdu
.pop("monitoring-parameter")
1521 with self
.assertRaises(EngineException
) as e
:
1522 self
.topic
.validate_scaling_group_descriptor(indata
)
1524 e
.exception
.http_code
,
1525 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1526 "Wrong HTTP status code",
1530 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1531 "[name='{}']:scaling-criteria[name='{}']: "
1532 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1535 affected_sp
["name"],
1536 affected_sc
["name"],
1537 affected_sc
["vnf-monitoring-param-ref"],
1540 norm(str(e
.exception
)),
1541 "Wrong exception text",
1544 def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self
):
1545 indata
= deepcopy(db_vnfd_content
)
1546 df
= indata
["df"][0]
1547 affected_sa
= df
["scaling-aspect"][0]
1548 indata
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1551 with self
.assertRaises(EngineException
) as e
:
1552 self
.topic
.validate_scaling_group_descriptor(indata
)
1554 e
.exception
.http_code
,
1555 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1556 "Wrong HTTP status code",
1560 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1561 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1562 df
["id"], affected_sa
["id"]
1565 norm(str(e
.exception
)),
1566 "Wrong exception text",
1569 def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1572 indata
= deepcopy(db_vnfd_content
)
1573 df
= indata
["df"][0]
1574 affected_sa
= df
["scaling-aspect"][0]
1575 affected_sca_primitive
= affected_sa
["scaling-config-action"][0][
1576 "vnf-config-primitive-name-ref"
1578 df
["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1581 with self
.assertRaises(EngineException
) as e
:
1582 self
.topic
.validate_scaling_group_descriptor(indata
)
1584 e
.exception
.http_code
,
1585 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1586 "Wrong HTTP status code",
1590 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1591 "config-primitive-name-ref='{}' does not match any "
1592 "day1-2 configuration:config-primitive:name".format(
1593 df
["id"], affected_sa
["id"], affected_sca_primitive
1596 norm(str(e
.exception
)),
1597 "Wrong exception text",
1600 def test_new_vnfd_revision(self
):
1601 did
= db_vnfd_content
["_id"]
1602 self
.fs
.get_params
.return_value
= {}
1603 self
.fs
.file_exists
.return_value
= False
1604 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1605 "/tmp/" + str(uuid4()), "a+b"
1607 test_vnfd
= deepcopy(db_vnfd_content
)
1608 del test_vnfd
["_id"]
1609 del test_vnfd
["_admin"]
1610 self
.db
.create
.return_value
= did
1612 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1613 db_args
= self
.db
.create
.call_args
[0]
1615 db_args
[1]["_admin"]["revision"], 0, "New package should be at revision 0"
1618 @patch("osm_nbi.descriptor_topics.shutil")
1619 @patch("osm_nbi.descriptor_topics.os.rename")
1620 def test_update_vnfd(self
, mock_rename
, mock_shutil
):
1622 did
= db_vnfd_content
["_id"]
1624 self
.fs
.get_params
.return_value
= {}
1625 self
.fs
.file_exists
.return_value
= False
1626 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1627 "/tmp/" + str(uuid4()), "a+b"
1629 new_vnfd
= deepcopy(db_vnfd_content
)
1631 self
.db
.create
.return_value
= did
1633 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1634 del new_vnfd
["vdu"][0]["cloud-init-file"]
1635 del new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1637 ][0]["execution-environment-list"][0]["juju"]
1639 old_vnfd
= {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])}
1640 old_vnfd
["_admin"]["revision"] = old_revision
1642 self
.db
.get_one
.side_effect
= [old_vnfd
, old_vnfd
, None]
1643 self
.topic
.upload_content(fake_session
, did
, new_vnfd
, {}, {"Content-Type": []})
1645 db_args
= self
.db
.replace
.call_args
[0]
1647 db_args
[2]["_admin"]["revision"],
1649 "Revision should increment",
1653 class Test_NsdTopic(TestCase
):
1655 def setUpClass(cls
):
1656 cls
.test_name
= "test-nsd-topic"
1659 def tearDownClass(cls
):
1663 self
.db
= Mock(dbbase
.DbBase())
1664 self
.fs
= Mock(fsbase
.FsBase())
1665 self
.msg
= Mock(msgbase
.MsgBase())
1666 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1667 self
.topic
= NsdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1668 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1671 def assertNotRaises(self
, exception_type
):
1674 except exception_type
:
1675 raise self
.failureException("{} raised".format(exception_type
.__name
__))
1677 def create_desc_temp(self
, template
):
1678 old_desc
= deepcopy(template
)
1679 new_desc
= deepcopy(template
)
1680 return old_desc
, new_desc
1682 def prepare_nsd_creation(self
):
1684 did
= db_nsd_content
["_id"]
1685 self
.fs
.get_params
.return_value
= {}
1686 self
.fs
.file_exists
.return_value
= False
1687 self
.fs
.file_open
.side_effect
= lambda path
, mode
: tempfile
.TemporaryFile(
1690 self
.db
.get_one
.side_effect
= [
1691 {"_id": did
, "_admin": deepcopy(db_nsd_content
["_admin"])},
1694 test_nsd
= deepcopy(db_nsd_content
)
1696 del test_nsd
["_admin"]
1697 return did
, test_nsd
1699 @patch("osm_nbi.descriptor_topics.shutil")
1700 @patch("osm_nbi.descriptor_topics.os.rename")
1701 def test_new_nsd_normal_creation(self
, mock_rename
, mock_shutil
):
1702 did
, test_nsd
= self
.prepare_nsd_creation()
1703 self
.db
.create
.return_value
= did
1706 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1707 db_args
= self
.db
.create
.call_args
[0]
1708 msg_args
= self
.msg
.write
.call_args
[0]
1709 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1710 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1711 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
1712 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1713 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1714 self
.assertEqual(did2
, did
, "Wrong DB NSD id")
1715 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
1717 db_args
[1]["_admin"]["modified"],
1718 db_args
[1]["_admin"]["created"],
1719 "Wrong modification time",
1722 db_args
[1]["_admin"]["projects_read"],
1724 "Wrong read-only project list",
1727 db_args
[1]["_admin"]["projects_write"],
1729 "Wrong read-write project list",
1732 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1734 self
.topic
.upload_content(fake_session
, did
, test_nsd
, {}, {"Content-Type": []})
1735 msg_args
= self
.msg
.write
.call_args
[0]
1736 test_nsd
["_id"] = did
1737 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1738 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1739 self
.assertEqual(msg_args
[2], test_nsd
, "Wrong message content")
1741 db_args
= self
.db
.get_one
.mock_calls
[0][1]
1742 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1743 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB NSD id")
1745 db_args
= self
.db
.replace
.call_args
[0]
1746 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1747 self
.assertEqual(db_args
[1], did
, "Wrong DB NSD id")
1749 admin
= db_args
[2]["_admin"]
1750 db_admin
= db_nsd_content
["_admin"]
1751 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
1753 admin
["modified"], db_admin
["created"], "Wrong modification time"
1756 admin
["projects_read"],
1757 db_admin
["projects_read"],
1758 "Wrong read-only project list",
1761 admin
["projects_write"],
1762 db_admin
["projects_write"],
1763 "Wrong read-write project list",
1766 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1769 admin
["operationalState"], "ENABLED", "Wrong operational state"
1771 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
1773 storage
= admin
["storage"]
1774 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
1775 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
1777 compare_desc(self
, test_nsd
, db_args
[2], "NSD")
1778 revision_args
= self
.db
.create
.call_args
[0]
1780 revision_args
[0], self
.topic
.topic
+ "_revisions", "Wrong topic"
1782 self
.assertEqual(revision_args
[1]["id"], db_args
[2]["id"], "Wrong revision id")
1784 revision_args
[1]["_id"], db_args
[2]["_id"] + ":1", "Wrong revision _id"
1787 @patch("osm_nbi.descriptor_topics.shutil")
1788 @patch("osm_nbi.descriptor_topics.os.rename")
1789 def test_new_nsd_check_pyangbind_validation_required_properties(
1790 self
, mock_rename
, mock_shutil
1792 did
, test_nsd
= self
.prepare_nsd_creation()
1795 with self
.assertRaises(
1796 EngineException
, msg
="Accepted NSD with a missing required property"
1798 self
.topic
.upload_content(
1799 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1802 e
.exception
.http_code
,
1803 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1804 "Wrong HTTP status code",
1807 norm("Error in pyangbind validation: '{}'".format("id")),
1808 norm(str(e
.exception
)),
1809 "Wrong exception text",
1812 @patch("osm_nbi.descriptor_topics.shutil")
1813 @patch("osm_nbi.descriptor_topics.os.rename")
1814 def test_new_nsd_check_pyangbind_validation_additional_properties(
1815 self
, mock_rename
, mock_shutil
1817 did
, test_nsd
= self
.prepare_nsd_creation()
1818 test_nsd
["extra-property"] = 0
1820 with self
.assertRaises(
1821 EngineException
, msg
="Accepted NSD with an additional property"
1823 self
.topic
.upload_content(
1824 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1827 e
.exception
.http_code
,
1828 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1829 "Wrong HTTP status code",
1833 "Error in pyangbind validation: {} ({})".format(
1834 "json object contained a key that did not exist", "extra-property"
1837 norm(str(e
.exception
)),
1838 "Wrong exception text",
1841 @patch("osm_nbi.descriptor_topics.shutil")
1842 @patch("osm_nbi.descriptor_topics.os.rename")
1843 def test_new_nsd_check_pyangbind_validation_property_types(
1844 self
, mock_rename
, mock_shutil
1846 did
, test_nsd
= self
.prepare_nsd_creation()
1847 test_nsd
["designer"] = {"key": 0}
1849 with self
.assertRaises(
1850 EngineException
, msg
="Accepted NSD with a wrongly typed property"
1852 self
.topic
.upload_content(
1853 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1856 e
.exception
.http_code
,
1857 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1858 "Wrong HTTP status code",
1862 "Error in pyangbind validation: {} ({})".format(
1863 "json object contained a key that did not exist", "key"
1866 norm(str(e
.exception
)),
1867 "Wrong exception text",
1870 @patch("osm_nbi.descriptor_topics.shutil")
1871 @patch("osm_nbi.descriptor_topics.os.rename")
1872 def test_new_nsd_check_input_validation_mgmt_network_virtual_link_protocol_data(
1873 self
, mock_rename
, mock_shutil
1875 did
, test_nsd
= self
.prepare_nsd_creation()
1876 df
= test_nsd
["df"][0]
1879 "virtual-link-desc-id": "mgmt",
1880 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1882 df
["virtual-link-profile"] = [mgmt_profile
]
1884 with self
.assertRaises(
1885 EngineException
, msg
="Accepted VLD with mgmt-network+ip-profile"
1887 self
.topic
.upload_content(
1888 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1891 e
.exception
.http_code
,
1892 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1893 "Wrong HTTP status code",
1897 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1898 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1899 df
["id"], mgmt_profile
["id"]
1902 norm(str(e
.exception
)),
1903 "Wrong exception text",
1906 @patch("osm_nbi.descriptor_topics.shutil")
1907 @patch("osm_nbi.descriptor_topics.os.rename")
1908 def test_new_nsd_check_descriptor_dependencies_vnfd_id(
1909 self
, mock_rename
, mock_shutil
1911 did
, test_nsd
= self
.prepare_nsd_creation()
1912 self
.db
.get_list
.return_value
= []
1914 with self
.assertRaises(
1915 EngineException
, msg
="Accepted wrong VNFD ID reference"
1917 self
.topic
.upload_content(
1918 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1921 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1925 "'vnfd-id'='{}' references a non existing vnfd".format(
1926 test_nsd
["vnfd-id"][0]
1929 norm(str(e
.exception
)),
1930 "Wrong exception text",
1933 @patch("osm_nbi.descriptor_topics.shutil")
1934 @patch("osm_nbi.descriptor_topics.os.rename")
1935 def test_new_nsd_check_descriptor_dependencies_vld_vnfd_connection_point_ref(
1936 self
, mock_rename
, mock_shutil
1938 # Check Descriptor Dependencies: "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]
1939 did
, test_nsd
= self
.prepare_nsd_creation()
1940 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1941 df
= test_nsd
["df"][0]
1942 affected_vnf_profile
= df
["vnf-profile"][0]
1943 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
1944 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
1945 self
.db
.get_list
.return_value
= [vnfd_descriptor
]
1947 with self
.assertRaises(
1948 EngineException
, msg
="Accepted wrong VLD CP reference"
1950 self
.topic
.upload_content(
1951 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1954 e
.exception
.http_code
,
1955 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1956 "Wrong HTTP status code",
1960 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1961 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1962 "non existing ext-cpd:id inside vnfd '{}'".format(
1964 affected_vnf_profile
["id"],
1965 affected_virtual_link
["virtual-link-profile-id"],
1967 vnfd_descriptor
["id"],
1970 norm(str(e
.exception
)),
1971 "Wrong exception text",
1974 def test_edit_nsd(self
):
1975 nsd_content
= deepcopy(db_nsd_content
)
1976 did
= nsd_content
["_id"]
1977 self
.fs
.file_exists
.return_value
= True
1978 self
.fs
.dir_ls
.return_value
= True
1979 with self
.subTest(i
=1, t
="Normal Edition"):
1981 self
.db
.get_one
.side_effect
= [deepcopy(nsd_content
), None]
1982 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1983 data
= {"id": "new-nsd-id", "name": "new-nsd-name"}
1984 self
.topic
.edit(fake_session
, did
, data
)
1985 db_args
= self
.db
.replace
.call_args
[0]
1986 msg_args
= self
.msg
.write
.call_args
[0]
1988 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1989 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1990 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
1991 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1992 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
1994 db_args
[2]["_admin"]["created"],
1995 nsd_content
["_admin"]["created"],
1996 "Wrong creation time",
1999 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
2002 db_args
[2]["_admin"]["projects_read"],
2003 nsd_content
["_admin"]["projects_read"],
2004 "Wrong read-only project list",
2007 db_args
[2]["_admin"]["projects_write"],
2008 nsd_content
["_admin"]["projects_write"],
2009 "Wrong read-write project list",
2011 self
.assertEqual(db_args
[2]["id"], data
["id"], "Wrong NSD ID")
2012 self
.assertEqual(db_args
[2]["name"], data
["name"], "Wrong NSD Name")
2013 with self
.subTest(i
=2, t
="Conflict on Edit"):
2014 data
= {"id": "fake-nsd-id", "name": "new-nsd-name"}
2015 self
.db
.get_one
.side_effect
= [
2017 {"_id": str(uuid4()), "id": data
["id"]},
2019 with self
.assertRaises(
2020 EngineException
, msg
="Accepted existing NSD ID"
2022 self
.topic
.edit(fake_session
, did
, data
)
2024 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2028 "{} with id '{}' already exists for this project".format(
2032 norm(str(e
.exception
)),
2033 "Wrong exception text",
2035 with self
.subTest(i
=3, t
="Check Envelope"):
2036 data
= {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
2037 self
.db
.get_one
.side_effect
= [nsd_content
, None]
2038 with self
.assertRaises(
2039 EngineException
, msg
="Accepted NSD with wrong envelope"
2041 self
.topic
.edit(fake_session
, did
, data
, content
=nsd_content
)
2043 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
2046 "'nsd' must be a list of only one element",
2047 norm(str(e
.exception
)),
2048 "Wrong exception text",
2050 self
.db
.reset_mock()
2053 def test_delete_nsd(self
):
2054 did
= db_nsd_content
["_id"]
2055 self
.db
.get_one
.return_value
= db_nsd_content
2056 p_id
= db_nsd_content
["_admin"]["projects_read"][0]
2057 with self
.subTest(i
=1, t
="Normal Deletion"):
2058 self
.db
.get_list
.return_value
= []
2059 self
.db
.del_one
.return_value
= {"deleted": 1}
2060 self
.topic
.delete(fake_session
, did
)
2061 db_args
= self
.db
.del_one
.call_args
[0]
2062 msg_args
= self
.msg
.write
.call_args
[0]
2063 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
2064 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
2065 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
2066 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
2067 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
2069 db_args
[1]["_admin.projects_write.cont"],
2073 db_g1_args
= self
.db
.get_one
.call_args
[0]
2074 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
2075 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB NSD ID")
2076 db_gl_calls
= self
.db
.get_list
.call_args_list
2077 self
.assertEqual(db_gl_calls
[0][0][0], "nsrs", "Wrong DB topic")
2078 # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID") # Filter changed after call
2079 self
.assertEqual(db_gl_calls
[1][0][0], "nsts", "Wrong DB topic")
2081 db_gl_calls
[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
2082 db_nsd_content
["id"],
2083 "Wrong DB NSD netslice-subnet nsd-ref",
2085 self
.db
.set_one
.assert_not_called()
2086 fs_del_calls
= self
.fs
.file_delete
.call_args_list
2087 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
2088 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
2089 with self
.subTest(i
=2, t
="Conflict on Delete - NSD in use by nsr"):
2090 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-nsr"}]
2091 with self
.assertRaises(
2092 EngineException
, msg
="Accepted NSD in use by NSR"
2094 self
.topic
.delete(fake_session
, did
)
2096 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2099 "there is at least one ns instance using this descriptor",
2100 norm(str(e
.exception
)),
2101 "Wrong exception text",
2103 with self
.subTest(i
=3, t
="Conflict on Delete - NSD in use by NST"):
2104 self
.db
.get_list
.side_effect
= [
2106 [{"_id": str(uuid4()), "name": "fake-nst"}],
2108 with self
.assertRaises(
2109 EngineException
, msg
="Accepted NSD in use by NST"
2111 self
.topic
.delete(fake_session
, did
)
2113 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2116 "there is at least one netslice template referencing this descriptor",
2117 norm(str(e
.exception
)),
2118 "Wrong exception text",
2120 with self
.subTest(i
=4, t
="Non-existent NSD"):
2121 excp_msg
= "Not found any {} with filter='{}'".format("NSD", {"_id": did
})
2122 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
2123 with self
.assertRaises(
2124 DbException
, msg
="Accepted non-existent NSD ID"
2126 self
.topic
.delete(fake_session
, did
)
2128 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
2131 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
2133 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
2134 db_nsd_content
["_admin"]["projects_read"].append("other_project")
2135 self
.db
.get_one
= Mock(return_value
=db_nsd_content
)
2136 self
.db
.get_list
= Mock(return_value
=[])
2137 self
.msg
.write
.reset_mock()
2138 self
.db
.del_one
.reset_mock()
2139 self
.fs
.file_delete
.reset_mock()
2141 self
.topic
.delete(fake_session
, did
)
2142 self
.db
.del_one
.assert_not_called()
2143 self
.msg
.write
.assert_not_called()
2144 db_g1_args
= self
.db
.get_one
.call_args
[0]
2145 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
2146 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
2147 db_s1_args
= self
.db
.set_one
.call_args
2148 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
2149 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
2151 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
2154 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
2157 db_s1_args
[1]["pull_list"],
2158 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
2159 "Wrong DB pull_list dictionary",
2161 self
.fs
.file_delete
.assert_not_called()
2162 self
.db
.reset_mock()
2165 def prepare_nsd_validation(self
):
2166 descriptor_name
= "test_ns_descriptor"
2167 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
2168 "/tmp/" + str(uuid4()), "a+b"
2170 old_nsd
, new_nsd
= self
.create_desc_temp(db_nsd_content
)
2171 return descriptor_name
, old_nsd
, new_nsd
2173 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2174 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2175 def test_validate_descriptor_ns_configuration_changed(
2176 self
, mock_safe_load
, mock_detect_usage
2178 """Validating NSD and NSD has changes in ns-configuration:config-primitive"""
2179 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
2180 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
2181 mock_detect_usage
.return_value
= True
2182 self
.db
.get_one
.return_value
= old_nsd
2184 {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
2187 {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
2190 with self
.assertNotRaises(EngineException
):
2191 self
.topic
._validate
_descriptor
_changes
(
2192 old_nsd
["_id"], descriptor_name
, "/tmp", "/tmp:1"
2194 self
.db
.get_one
.assert_called_once()
2195 mock_detect_usage
.assert_called_once()
2196 self
.assertEqual(mock_safe_load
.call_count
, 2)
2198 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2199 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2200 def test_validate_descriptor_nsd_name_changed(
2201 self
, mock_safe_load
, mock_detect_usage
2203 """Validating NSD, NSD name has changed."""
2204 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
2205 did
= old_nsd
["_id"]
2206 new_nsd
["name"] = "nscharm-ns2"
2207 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
2208 mock_detect_usage
.return_value
= True
2209 self
.db
.get_one
.return_value
= old_nsd
2211 with self
.assertRaises(
2212 EngineException
, msg
="there are disallowed changes in the ns descriptor"
2214 self
.topic
._validate
_descriptor
_changes
(
2215 did
, descriptor_name
, "/tmp", "/tmp:1"
2218 e
.exception
.http_code
,
2219 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2220 "Wrong HTTP status code",
2223 norm("there are disallowed changes in the ns descriptor"),
2224 norm(str(e
.exception
)),
2225 "Wrong exception text",
2228 self
.db
.get_one
.assert_called_once()
2229 mock_detect_usage
.assert_called_once()
2230 self
.assertEqual(mock_safe_load
.call_count
, 2)
2232 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
2233 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
2234 def test_validate_descriptor_nsd_name_changed_nsd_not_in_use(
2235 self
, mock_safe_load
, mock_detect_usage
2237 """Validating NSD, NSD name has changed, NSD is not in use."""
2238 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
2239 did
= old_nsd
["_id"]
2240 new_nsd
["name"] = "nscharm-ns2"
2241 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
2242 mock_detect_usage
.return_value
= None
2243 self
.db
.get_one
.return_value
= old_nsd
2245 with self
.assertNotRaises(Exception):
2246 self
.topic
._validate
_descriptor
_changes
(
2247 did
, descriptor_name
, "/tmp", "/tmp:1"
2250 self
.db
.get_one
.assert_called_once()
2251 mock_detect_usage
.assert_called_once()
2252 mock_safe_load
.assert_not_called()
2254 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
2257 indata
= deepcopy(db_nsd_content
)
2258 vld
= indata
["virtual-link-desc"][0]
2259 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2263 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
2266 indata
= deepcopy(db_nsd_content
)
2267 vld
= indata
["virtual-link-desc"][0]
2268 df
= indata
["df"][0]
2271 "virtual-link-desc-id": "mgmt",
2272 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
2274 df
["virtual-link-profile"] = [affected_vlp
]
2275 with self
.assertRaises(EngineException
) as e
:
2276 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2280 e
.exception
.http_code
,
2281 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2282 "Wrong HTTP status code",
2286 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
2287 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
2288 df
["id"], affected_vlp
["id"]
2291 norm(str(e
.exception
)),
2292 "Wrong exception text",
2295 def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self
):
2296 indata
= deepcopy(db_nsd_content
)
2297 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2299 def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self
):
2300 indata
= deepcopy(db_nsd_content
)
2301 df
= indata
["df"][0]
2302 affected_vnf_profile
= df
["vnf-profile"][0]
2303 indata
["vnfd-id"] = ["non-existing-vnfd"]
2304 with self
.assertRaises(EngineException
) as e
:
2305 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2307 e
.exception
.http_code
,
2308 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2309 "Wrong HTTP status code",
2313 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
2314 "does not match any vnfd-id".format(
2316 affected_vnf_profile
["id"],
2317 affected_vnf_profile
["vnfd-id"],
2320 norm(str(e
.exception
)),
2321 "Wrong exception text",
2324 def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
2327 nsd_descriptor
= deepcopy(db_nsd_content
)
2328 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2329 df
= nsd_descriptor
["df"][0]
2330 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2331 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2335 def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
2338 nsd_descriptor
= deepcopy(db_nsd_content
)
2339 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2340 df
= nsd_descriptor
["df"][0]
2341 affected_vnf_profile
= df
["vnf-profile"][0]
2342 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
2343 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2344 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
2345 with self
.assertRaises(EngineException
) as e
:
2346 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2350 e
.exception
.http_code
,
2351 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2352 "Wrong HTTP status code",
2356 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
2357 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
2358 "non existing ext-cpd:id inside vnfd '{}'".format(
2360 affected_vnf_profile
["id"],
2361 affected_virtual_link
["virtual-link-profile-id"],
2363 vnfd_descriptor
["id"],
2366 norm(str(e
.exception
)),
2367 "Wrong exception text",
2370 def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self
):
2371 nsd_descriptor
= deepcopy(db_nsd_content
)
2372 invalid_vnfd_id
= "invalid-vnfd-id"
2373 nsd_descriptor
["id"] = "invalid-vnfd-id-ns"
2374 nsd_descriptor
["vnfd-id"][0] = invalid_vnfd_id
2375 nsd_descriptor
["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
2376 nsd_descriptor
["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
2377 with self
.assertRaises(EngineException
) as e
:
2378 self
.db
.get_list
.return_value
= []
2379 nsd_descriptor
= self
.topic
.check_conflict_on_edit(
2380 fake_session
, nsd_descriptor
, [], "id"
2383 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2387 "Descriptor error at 'vnfd-id'='{}' references a non "
2388 "existing vnfd".format(invalid_vnfd_id
)
2390 norm(str(e
.exception
)),
2391 "Wrong exception text",
2395 if __name__
== "__main__":