2 # -*- coding: utf-8 -*-
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
17 __author__
= "Pedro de la Cruz Ramos, pedro.delacruzramos@altran.com"
18 __date__
= "2019-11-20"
20 from contextlib
import contextmanager
22 from unittest
import TestCase
23 from unittest
.mock
import Mock
, patch
24 from uuid
import uuid4
25 from http
import HTTPStatus
26 from copy
import deepcopy
28 from osm_common
import dbbase
, fsbase
, msgbase
29 from osm_nbi
import authconn
30 from osm_nbi
.tests
.test_pkg_descriptors
import db_vnfds_text
, db_nsds_text
31 from osm_nbi
.descriptor_topics
import VnfdTopic
, NsdTopic
32 from osm_nbi
.engine
import EngineException
33 from osm_common
.dbbase
import DbException
36 test_name
= "test-user"
37 db_vnfd_content
= yaml
.safe_load(db_vnfds_text
)[0]
38 db_nsd_content
= yaml
.safe_load(db_nsds_text
)[0]
39 test_pid
= db_vnfd_content
["_admin"]["projects_read"][0]
41 "username": test_name
,
42 "project_id": (test_pid
,),
47 "allow_show_user_project_role": True,
52 """Normalize string for checking"""
53 return " ".join(str.strip().split()).lower()
56 def compare_desc(tc
, d1
, d2
, k
):
58 Compare two descriptors
59 We need this function because some methods are adding/removing items to/from the descriptors
60 before they are stored in the database, so the original and stored versions will differ
61 What we check is that COMMON LEAF ITEMS are equal
62 Lists of different length are not compared
63 :param tc: Test Case wich provides context (in particular the assert* methods)
64 :param d1,d2: Descriptors to be compared
65 :param k: key/item being compared
68 if isinstance(d1
, dict) and isinstance(d2
, dict):
71 compare_desc(tc
, d1
[key
], d2
[key
], k
+ "[{}]".format(key
))
72 elif isinstance(d1
, list) and isinstance(d2
, list) and len(d1
) == len(d2
):
73 for i
in range(len(d1
)):
74 compare_desc(tc
, d1
[i
], d2
[i
], k
+ "[{}]".format(i
))
76 tc
.assertEqual(d1
, d2
, "Wrong descriptor content: {}".format(k
))
79 class Test_VnfdTopic(TestCase
):
82 cls
.test_name
= "test-vnfd-topic"
85 def tearDownClass(cls
):
89 self
.db
= Mock(dbbase
.DbBase())
90 self
.fs
= Mock(fsbase
.FsBase())
91 self
.msg
= Mock(msgbase
.MsgBase())
92 self
.auth
= Mock(authconn
.Authconn(None, None, None))
93 self
.topic
= VnfdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
94 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
97 def assertNotRaises(self
, exception_type
):
100 except exception_type
:
101 raise self
.failureException("{} raised".format(exception_type
.__name
__))
103 def create_desc_temp(self
, template
):
104 old_desc
= deepcopy(template
)
105 new_desc
= deepcopy(template
)
106 return old_desc
, new_desc
108 def prepare_vnfd_creation(self
):
110 self
.fs
.get_params
.return_value
= {}
111 self
.fs
.file_exists
.return_value
= False
112 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
113 "/tmp/" + str(uuid4()), "a+b"
115 test_vnfd
= deepcopy(db_vnfd_content
)
116 did
= db_vnfd_content
["_id"]
117 self
.db
.create
.return_value
= did
118 self
.db
.get_one
.side_effect
= [
119 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
122 return did
, test_vnfd
124 def prepare_test_vnfd(self
, test_vnfd
):
126 del test_vnfd
["_admin"]
127 del test_vnfd
["vdu"][0]["cloud-init-file"]
128 del test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
130 ][0]["execution-environment-list"][0]["juju"]
133 @patch("osm_nbi.descriptor_topics.shutil")
134 @patch("osm_nbi.descriptor_topics.os.rename")
135 def test_new_vnfd_normal_creation(self
, mock_rename
, mock_shutil
):
136 did
, test_vnfd
= self
.prepare_vnfd_creation()
137 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
139 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
140 db_args
= self
.db
.create
.call_args
[0]
141 msg_args
= self
.msg
.write
.call_args
[0]
143 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
144 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
145 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
146 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
147 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
148 self
.assertEqual(did2
, did
, "Wrong DB VNFD id")
149 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
151 db_args
[1]["_admin"]["modified"],
152 db_args
[1]["_admin"]["created"],
153 "Wrong modification time",
156 db_args
[1]["_admin"]["projects_read"],
158 "Wrong read-only project list",
161 db_args
[1]["_admin"]["projects_write"],
163 "Wrong read-write project list",
166 self
.db
.get_one
.side_effect
= [
167 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
171 self
.topic
.upload_content(
172 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
174 msg_args
= self
.msg
.write
.call_args
[0]
175 test_vnfd
["_id"] = did
176 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
177 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
178 self
.assertEqual(msg_args
[2], test_vnfd
, "Wrong message content")
180 db_args
= self
.db
.get_one
.mock_calls
[0][1]
181 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
182 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB VNFD id")
184 db_args
= self
.db
.replace
.call_args
[0]
185 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
186 self
.assertEqual(db_args
[1], did
, "Wrong DB VNFD id")
188 admin
= db_args
[2]["_admin"]
189 db_admin
= deepcopy(db_vnfd_content
["_admin"])
190 self
.assertEqual(admin
["type"], "vnfd", "Wrong descriptor type")
191 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
193 admin
["modified"], db_admin
["created"], "Wrong modification time"
196 admin
["projects_read"],
197 db_admin
["projects_read"],
198 "Wrong read-only project list",
201 admin
["projects_write"],
202 db_admin
["projects_write"],
203 "Wrong read-write project list",
206 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
209 admin
["operationalState"], "ENABLED", "Wrong operational state"
211 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
213 storage
= admin
["storage"]
214 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
215 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
216 self
.assertEqual(admin
["revision"], 1, "Wrong revision number")
217 compare_desc(self
, test_vnfd
, db_args
[2], "VNFD")
219 @patch("osm_nbi.descriptor_topics.shutil")
220 @patch("osm_nbi.descriptor_topics.os.rename")
221 def test_new_vnfd_check_pyangbind_validation_additional_properties(
222 self
, mock_rename
, mock_shutil
224 did
, test_vnfd
= self
.prepare_vnfd_creation()
225 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
226 self
.topic
.upload_content(
227 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
229 test_vnfd
["_id"] = did
230 test_vnfd
["extra-property"] = 0
231 self
.db
.get_one
.side_effect
= (
232 lambda table
, filter, fail_on_empty
=None, fail_on_more
=None: {
234 "_admin": deepcopy(db_vnfd_content
["_admin"]),
238 with self
.assertRaises(
239 EngineException
, msg
="Accepted VNFD with an additional property"
241 self
.topic
.upload_content(
242 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
245 e
.exception
.http_code
,
246 HTTPStatus
.UNPROCESSABLE_ENTITY
,
247 "Wrong HTTP status code",
251 "Error in pyangbind validation: {} ({})".format(
252 "json object contained a key that did not exist", "extra-property"
255 norm(str(e
.exception
)),
256 "Wrong exception text",
258 db_args
= self
.db
.replace
.call_args
[0]
259 admin
= db_args
[2]["_admin"]
260 self
.assertEqual(admin
["revision"], 1, "Wrong revision number")
262 @patch("osm_nbi.descriptor_topics.shutil")
263 @patch("osm_nbi.descriptor_topics.os.rename")
264 def test_new_vnfd_check_pyangbind_validation_property_types(
265 self
, mock_rename
, mock_shutil
267 did
, test_vnfd
= self
.prepare_vnfd_creation()
268 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
269 test_vnfd
["_id"] = did
270 test_vnfd
["product-name"] = {"key": 0}
272 with self
.assertRaises(
273 EngineException
, msg
="Accepted VNFD with a wrongly typed property"
275 self
.topic
.upload_content(
276 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
279 e
.exception
.http_code
,
280 HTTPStatus
.UNPROCESSABLE_ENTITY
,
281 "Wrong HTTP status code",
285 "Error in pyangbind validation: {} ({})".format(
286 "json object contained a key that did not exist", "key"
289 norm(str(e
.exception
)),
290 "Wrong exception text",
293 @patch("osm_nbi.descriptor_topics.shutil")
294 @patch("osm_nbi.descriptor_topics.os.rename")
295 def test_new_vnfd_check_input_validation_cloud_init(self
, mock_rename
, mock_shutil
):
296 did
, test_vnfd
= self
.prepare_vnfd_creation()
297 del test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
299 ][0]["execution-environment-list"][0]["juju"]
301 with self
.assertRaises(
302 EngineException
, msg
="Accepted non-existent cloud_init file"
304 self
.topic
.upload_content(
305 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
308 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
312 "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format(
313 "cloud-init", test_vnfd
["id"], test_vnfd
["vdu"][0]["id"]
316 norm(str(e
.exception
)),
317 "Wrong exception text",
320 @patch("osm_nbi.descriptor_topics.shutil")
321 @patch("osm_nbi.descriptor_topics.os.rename")
322 def test_new_vnfd_check_input_validation_day12_configuration(
323 self
, mock_rename
, mock_shutil
325 did
, test_vnfd
= self
.prepare_vnfd_creation()
326 del test_vnfd
["vdu"][0]["cloud-init-file"]
328 with self
.assertRaises(
329 EngineException
, msg
="Accepted non-existent charm in VNF configuration"
331 self
.topic
.upload_content(
332 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
335 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
339 "{} defined in vnf[id={}] but not present in package".format(
340 "charm", test_vnfd
["id"]
343 norm(str(e
.exception
)),
344 "Wrong exception text",
347 @patch("osm_nbi.descriptor_topics.shutil")
348 @patch("osm_nbi.descriptor_topics.os.rename")
349 def test_new_vnfd_check_input_validation_mgmt_cp(self
, mock_rename
, mock_shutil
):
350 did
, test_vnfd
= self
.prepare_vnfd_creation()
351 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
352 del test_vnfd
["mgmt-cp"]
354 with self
.assertRaises(
355 EngineException
, msg
="Accepted VNFD without management interface"
357 self
.topic
.upload_content(
358 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
361 e
.exception
.http_code
,
362 HTTPStatus
.UNPROCESSABLE_ENTITY
,
363 "Wrong HTTP status code",
366 norm("'{}' is a mandatory field and it is not defined".format("mgmt-cp")),
367 norm(str(e
.exception
)),
368 "Wrong exception text",
371 @patch("osm_nbi.descriptor_topics.shutil")
372 @patch("osm_nbi.descriptor_topics.os.rename")
373 def test_new_vnfd_check_input_validation_mgmt_cp_connection_point(
374 self
, mock_rename
, mock_shutil
376 did
, test_vnfd
= self
.prepare_vnfd_creation()
377 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
378 test_vnfd
["mgmt-cp"] = "wrong-cp"
380 with self
.assertRaises(
381 EngineException
, msg
="Accepted wrong mgmt-cp connection point"
383 self
.topic
.upload_content(
384 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
387 e
.exception
.http_code
,
388 HTTPStatus
.UNPROCESSABLE_ENTITY
,
389 "Wrong HTTP status code",
393 "mgmt-cp='{}' must match an existing ext-cpd".format(
397 norm(str(e
.exception
)),
398 "Wrong exception text",
401 @patch("osm_nbi.descriptor_topics.shutil")
402 @patch("osm_nbi.descriptor_topics.os.rename")
403 def test_new_vnfd_check_input_validation_vdu_int_cpd(
404 self
, mock_rename
, mock_shutil
406 """Testing input validation during new vnfd creation
407 for vdu internal connection point"""
408 did
, test_vnfd
= self
.prepare_vnfd_creation()
409 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
410 ext_cpd
= test_vnfd
["ext-cpd"][1]
411 ext_cpd
["int-cpd"]["cpd"] = "wrong-cpd"
413 with self
.assertRaises(
414 EngineException
, msg
="Accepted wrong ext-cpd internal connection point"
416 self
.topic
.upload_content(
417 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
420 e
.exception
.http_code
,
421 HTTPStatus
.UNPROCESSABLE_ENTITY
,
422 "Wrong HTTP status code",
426 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
430 norm(str(e
.exception
)),
431 "Wrong exception text",
434 @patch("osm_nbi.descriptor_topics.shutil")
435 @patch("osm_nbi.descriptor_topics.os.rename")
436 def test_new_vnfd_check_input_validation_duplicated_vld(
437 self
, mock_rename
, mock_shutil
439 """Testing input validation during new vnfd creation
440 for dublicated virtual link description"""
441 did
, test_vnfd
= self
.prepare_vnfd_creation()
442 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
443 test_vnfd
["int-virtual-link-desc"].insert(0, {"id": "internal"})
445 with self
.assertRaises(
446 EngineException
, msg
="Accepted duplicated VLD name"
448 self
.topic
.upload_content(
449 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
452 e
.exception
.http_code
,
453 HTTPStatus
.UNPROCESSABLE_ENTITY
,
454 "Wrong HTTP status code",
458 "identifier id '{}' is not unique".format(
459 test_vnfd
["int-virtual-link-desc"][0]["id"]
462 norm(str(e
.exception
)),
463 "Wrong exception text",
466 @patch("osm_nbi.descriptor_topics.shutil")
467 @patch("osm_nbi.descriptor_topics.os.rename")
468 def test_new_vnfd_check_input_validation_vdu_int_virtual_link_desc(
469 self
, mock_rename
, mock_shutil
471 """Testing input validation during new vnfd creation
472 for vdu internal virtual link description"""
473 did
, test_vnfd
= self
.prepare_vnfd_creation()
474 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
475 vdu
= test_vnfd
["vdu"][0]
476 int_cpd
= vdu
["int-cpd"][1]
477 int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
479 with self
.assertRaises(
480 EngineException
, msg
="Accepted int-virtual-link-desc"
482 self
.topic
.upload_content(
483 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
486 e
.exception
.http_code
,
487 HTTPStatus
.UNPROCESSABLE_ENTITY
,
488 "Wrong HTTP status code",
492 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
493 "int-virtual-link-desc".format(
494 vdu
["id"], int_cpd
["id"], int_cpd
["int-virtual-link-desc"]
497 norm(str(e
.exception
)),
498 "Wrong exception text",
501 @patch("osm_nbi.descriptor_topics.shutil")
502 @patch("osm_nbi.descriptor_topics.os.rename")
503 def test_new_vnfd_check_input_validation_virtual_link_profile(
504 self
, mock_rename
, mock_shutil
506 """Testing input validation during new vnfd creation
507 for virtual link profile"""
508 did
, test_vnfd
= self
.prepare_vnfd_creation()
509 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
510 fake_ivld_profile
= {"id": "fake-profile-ref", "flavour": "fake-flavour"}
511 df
= test_vnfd
["df"][0]
512 df
["virtual-link-profile"] = [fake_ivld_profile
]
514 with self
.assertRaises(
515 EngineException
, msg
="Accepted non-existent Profile Ref"
517 self
.topic
.upload_content(
518 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
521 e
.exception
.http_code
,
522 HTTPStatus
.UNPROCESSABLE_ENTITY
,
523 "Wrong HTTP status code",
527 "df[id='{}']:virtual-link-profile='{}' must match an existing "
528 "int-virtual-link-desc".format(df
["id"], fake_ivld_profile
["id"])
530 norm(str(e
.exception
)),
531 "Wrong exception text",
534 @patch("osm_nbi.descriptor_topics.shutil")
535 @patch("osm_nbi.descriptor_topics.os.rename")
536 def test_new_vnfd_check_input_validation_scaling_criteria_monitoring_param_ref(
537 self
, mock_rename
, mock_shutil
539 """Testing input validation during new vnfd creation
540 for scaling criteria without monitoring parameter"""
541 did
, test_vnfd
= self
.prepare_vnfd_creation()
542 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
543 vdu
= test_vnfd
["vdu"][1]
544 affected_df
= test_vnfd
["df"][0]
545 sa
= affected_df
["scaling-aspect"][0]
546 sp
= sa
["scaling-policy"][0]
547 sc
= sp
["scaling-criteria"][0]
548 vdu
.pop("monitoring-parameter")
550 with self
.assertRaises(
551 EngineException
, msg
="Accepted non-existent Scaling Group Policy Criteria"
553 self
.topic
.upload_content(
554 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
557 e
.exception
.http_code
,
558 HTTPStatus
.UNPROCESSABLE_ENTITY
,
559 "Wrong HTTP status code",
563 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
564 "[name='{}']:scaling-criteria[name='{}']: "
565 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
570 sc
["vnf-monitoring-param-ref"],
573 norm(str(e
.exception
)),
574 "Wrong exception text",
577 @patch("osm_nbi.descriptor_topics.shutil")
578 @patch("osm_nbi.descriptor_topics.os.rename")
579 def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration(
580 self
, mock_rename
, mock_shutil
582 """Testing input validation during new vnfd creation
583 for scaling criteria without day12 configuration"""
584 did
, test_vnfd
= self
.prepare_vnfd_creation()
585 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
586 test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
589 df
= test_vnfd
["df"][0]
591 with self
.assertRaises(
592 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
594 self
.topic
.upload_content(
595 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
598 e
.exception
.http_code
,
599 HTTPStatus
.UNPROCESSABLE_ENTITY
,
600 "Wrong HTTP status code",
604 "'day1-2 configuration' not defined in the descriptor but it is referenced "
605 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
606 df
["id"], df
["scaling-aspect"][0]["id"]
609 norm(str(e
.exception
)),
610 "Wrong exception text",
613 @patch("osm_nbi.descriptor_topics.shutil")
614 @patch("osm_nbi.descriptor_topics.os.rename")
615 def test_new_vnfd_check_input_validation_scaling_config_action(
616 self
, mock_rename
, mock_shutil
618 """Testing input validation during new vnfd creation
619 for scaling criteria wrong config primitive"""
620 did
, test_vnfd
= self
.prepare_vnfd_creation()
621 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
622 df
= test_vnfd
["df"][0]
623 affected_df
= test_vnfd
["df"][0]
624 sa
= affected_df
["scaling-aspect"][0]
625 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
626 "operate-vnf-op-config"
627 )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
629 with self
.assertRaises(
630 EngineException
, msg
="Accepted non-existent Scaling Group VDU ID Reference"
632 self
.topic
.upload_content(
633 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
636 e
.exception
.http_code
,
637 HTTPStatus
.UNPROCESSABLE_ENTITY
,
638 "Wrong HTTP status code",
642 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
643 "config-primitive-name-ref='{}' does not match any "
644 "day1-2 configuration:config-primitive:name".format(
646 df
["scaling-aspect"][0]["id"],
647 sa
["scaling-config-action"][0]["vnf-config-primitive-name-ref"],
650 norm(str(e
.exception
)),
651 "Wrong exception text",
654 @patch("osm_nbi.descriptor_topics.shutil")
655 @patch("osm_nbi.descriptor_topics.os.rename")
656 def test_new_vnfd_check_input_validation_everything_right(
657 self
, mock_rename
, mock_shutil
659 """Testing input validation during new vnfd creation
660 everything correct"""
661 did
, test_vnfd
= self
.prepare_vnfd_creation()
662 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
663 test_vnfd
["id"] = "fake-vnfd-id"
664 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
665 "operate-vnf-op-config"
666 )["day1-2"][0]["id"] = "fake-vnfd-id"
667 self
.db
.get_one
.side_effect
= [
668 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
671 rc
= self
.topic
.upload_content(
672 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
674 self
.assertTrue(rc
, "Input Validation: Unexpected failure")
676 def test_edit_vnfd(self
):
677 vnfd_content
= deepcopy(db_vnfd_content
)
678 did
= vnfd_content
["_id"]
679 self
.fs
.file_exists
.return_value
= True
680 self
.fs
.dir_ls
.return_value
= True
681 with self
.subTest(i
=1, t
="Normal Edition"):
683 self
.db
.get_one
.side_effect
= [deepcopy(vnfd_content
), None]
684 data
= {"product-name": "new-vnfd-name"}
685 self
.topic
.edit(fake_session
, did
, data
)
686 db_args
= self
.db
.replace
.call_args
[0]
687 msg_args
= self
.msg
.write
.call_args
[0]
689 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
690 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
691 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
692 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
693 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
695 db_args
[2]["_admin"]["created"],
696 vnfd_content
["_admin"]["created"],
697 "Wrong creation time",
700 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
703 db_args
[2]["_admin"]["projects_read"],
704 vnfd_content
["_admin"]["projects_read"],
705 "Wrong read-only project list",
708 db_args
[2]["_admin"]["projects_write"],
709 vnfd_content
["_admin"]["projects_write"],
710 "Wrong read-write project list",
713 db_args
[2]["product-name"], data
["product-name"], "Wrong VNFD Name"
715 with self
.subTest(i
=2, t
="Conflict on Edit"):
716 data
= {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
717 self
.db
.get_one
.side_effect
= [
718 deepcopy(vnfd_content
),
719 {"_id": str(uuid4()), "id": data
["id"]},
721 with self
.assertRaises(
722 EngineException
, msg
="Accepted existing VNFD ID"
724 self
.topic
.edit(fake_session
, did
, data
)
726 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
730 "{} with id '{}' already exists for this project".format(
734 norm(str(e
.exception
)),
735 "Wrong exception text",
737 with self
.subTest(i
=3, t
="Check Envelope"):
738 data
= {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
739 with self
.assertRaises(
740 EngineException
, msg
="Accepted VNFD with wrong envelope"
742 self
.topic
.edit(fake_session
, did
, data
, content
=vnfd_content
)
744 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
747 "'vnfd' must be dict", norm(str(e
.exception
)), "Wrong exception text"
751 def test_delete_vnfd(self
):
752 did
= db_vnfd_content
["_id"]
753 self
.db
.get_one
.return_value
= db_vnfd_content
754 p_id
= db_vnfd_content
["_admin"]["projects_read"][0]
755 with self
.subTest(i
=1, t
="Normal Deletion"):
756 self
.db
.get_list
.return_value
= []
757 self
.db
.del_one
.return_value
= {"deleted": 1}
758 self
.topic
.delete(fake_session
, did
)
759 db_args
= self
.db
.del_one
.call_args
[0]
760 msg_args
= self
.msg
.write
.call_args
[0]
761 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
762 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
763 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
764 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
765 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
767 db_args
[1]["_admin.projects_write.cont"],
771 db_g1_args
= self
.db
.get_one
.call_args
[0]
772 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
773 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
774 db_gl_calls
= self
.db
.get_list
.call_args_list
775 self
.assertEqual(db_gl_calls
[0][0][0], "vnfrs", "Wrong DB topic")
776 # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID") # Filter changed after call
777 self
.assertEqual(db_gl_calls
[1][0][0], "nsds", "Wrong DB topic")
779 db_gl_calls
[1][0][1]["vnfd-id"],
780 db_vnfd_content
["id"],
781 "Wrong DB NSD vnfd-id",
785 self
.db
.del_list
.call_args
[0][0],
786 self
.topic
.topic
+ "_revisions",
791 self
.db
.del_list
.call_args
[0][1]["_id"]["$regex"],
793 "Wrong ID for rexep delete",
796 self
.db
.set_one
.assert_not_called()
797 fs_del_calls
= self
.fs
.file_delete
.call_args_list
798 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
799 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
800 with self
.subTest(i
=2, t
="Conflict on Delete - VNFD in use by VNFR"):
801 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-vnfr"}]
802 with self
.assertRaises(
803 EngineException
, msg
="Accepted VNFD in use by VNFR"
805 self
.topic
.delete(fake_session
, did
)
807 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
810 "there is at least one vnf instance using this descriptor",
811 norm(str(e
.exception
)),
812 "Wrong exception text",
814 with self
.subTest(i
=3, t
="Conflict on Delete - VNFD in use by NSD"):
815 self
.db
.get_list
.side_effect
= [
817 [{"_id": str(uuid4()), "name": "fake-nsd"}],
819 with self
.assertRaises(
820 EngineException
, msg
="Accepted VNFD in use by NSD"
822 self
.topic
.delete(fake_session
, did
)
824 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
827 "there is at least one ns package referencing this descriptor",
828 norm(str(e
.exception
)),
829 "Wrong exception text",
831 with self
.subTest(i
=4, t
="Non-existent VNFD"):
832 excp_msg
= "Not found any {} with filter='{}'".format("VNFD", {"_id": did
})
833 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
834 with self
.assertRaises(
835 DbException
, msg
="Accepted non-existent VNFD ID"
837 self
.topic
.delete(fake_session
, did
)
839 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
842 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
844 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
845 db_vnfd_content
["_admin"]["projects_read"].append("other_project")
846 self
.db
.get_one
= Mock(return_value
=db_vnfd_content
)
847 self
.db
.get_list
= Mock(return_value
=[])
848 self
.msg
.write
.reset_mock()
849 self
.db
.del_one
.reset_mock()
850 self
.fs
.file_delete
.reset_mock()
852 self
.topic
.delete(fake_session
, did
)
853 self
.db
.del_one
.assert_not_called()
854 self
.msg
.write
.assert_not_called()
855 db_g1_args
= self
.db
.get_one
.call_args
[0]
856 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
857 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
858 db_s1_args
= self
.db
.set_one
.call_args
859 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
860 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
862 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
865 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
868 db_s1_args
[1]["pull_list"],
869 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
870 "Wrong DB pull_list dictionary",
872 self
.fs
.file_delete
.assert_not_called()
875 def prepare_vnfd_validation(self
):
876 descriptor_name
= "test_descriptor"
877 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
878 "/tmp/" + str(uuid4()), "a+b"
880 old_vnfd
, new_vnfd
= self
.create_desc_temp(db_vnfd_content
)
881 return descriptor_name
, old_vnfd
, new_vnfd
883 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
884 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
885 def test_validate_vnfd_changes_day12_config_primitive_changed(
886 self
, mock_safe_load
, mock_detect_usage
888 """Validating VNFD for VNFD updates, day1-2 config primitive has changed"""
889 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
890 did
= old_vnfd
["_id"]
891 new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
893 ][0]["config-primitive"][0]["name"] = "new_action"
894 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
895 mock_detect_usage
.return_value
= True
896 self
.db
.get_one
.return_value
= old_vnfd
898 with self
.assertNotRaises(EngineException
):
899 self
.topic
._validate
_descriptor
_changes
(
900 did
, descriptor_name
, "/tmp/", "/tmp:1/"
902 self
.db
.get_one
.assert_called_once()
903 mock_detect_usage
.assert_called_once()
904 self
.assertEqual(mock_safe_load
.call_count
, 2)
906 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
907 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
908 def test_validate_vnfd_changes_sw_version_changed(
909 self
, mock_safe_load
, mock_detect_usage
911 """Validating VNFD for updates, software version has changed"""
912 # old vnfd uses the default software version: 1.0
913 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
914 did
= old_vnfd
["_id"]
915 new_vnfd
["software-version"] = "1.3"
916 new_vnfd
["sw-image-desc"][0]["name"] = "new-image"
917 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
918 mock_detect_usage
.return_value
= True
919 self
.db
.get_one
.return_value
= old_vnfd
921 with self
.assertNotRaises(EngineException
):
922 self
.topic
._validate
_descriptor
_changes
(
923 did
, descriptor_name
, "/tmp/", "/tmp:1/"
925 self
.db
.get_one
.assert_called_once()
926 mock_detect_usage
.assert_called_once()
927 self
.assertEqual(mock_safe_load
.call_count
, 2)
929 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
930 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
931 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed(
932 self
, mock_safe_load
, mock_detect_usage
934 """Validating VNFD for updates, software version has not
935 changed, mgmt-cp has changed."""
936 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
937 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
938 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
939 did
= old_vnfd
["_id"]
940 mock_detect_usage
.return_value
= True
941 self
.db
.get_one
.return_value
= old_vnfd
943 with self
.assertRaises(
944 EngineException
, msg
="there are disallowed changes in the vnf descriptor"
946 self
.topic
._validate
_descriptor
_changes
(
947 did
, descriptor_name
, "/tmp/", "/tmp:1/"
951 e
.exception
.http_code
,
952 HTTPStatus
.UNPROCESSABLE_ENTITY
,
953 "Wrong HTTP status code",
956 norm("there are disallowed changes in the vnf descriptor"),
957 norm(str(e
.exception
)),
958 "Wrong exception text",
960 self
.db
.get_one
.assert_called_once()
961 mock_detect_usage
.assert_called_once()
962 self
.assertEqual(mock_safe_load
.call_count
, 2)
964 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
965 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
966 def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed_vnfd_not_in_use(
967 self
, mock_safe_load
, mock_detect_usage
969 """Validating VNFD for updates, software version has not
970 changed, mgmt-cp has changed, vnfd is not in use."""
971 descriptor_name
, old_vnfd
, new_vnfd
= self
.prepare_vnfd_validation()
972 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
973 mock_safe_load
.side_effect
= [old_vnfd
, new_vnfd
]
974 did
= old_vnfd
["_id"]
975 mock_detect_usage
.return_value
= None
976 self
.db
.get_one
.return_value
= old_vnfd
978 with self
.assertNotRaises(EngineException
):
979 self
.topic
._validate
_descriptor
_changes
(
980 did
, descriptor_name
, "/tmp/", "/tmp:1/"
983 self
.db
.get_one
.assert_called_once()
984 mock_detect_usage
.assert_called_once()
985 mock_safe_load
.assert_not_called()
987 def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self
):
988 indata
= deepcopy(db_vnfd_content
)
989 self
.topic
.validate_mgmt_interface_connection_point(indata
)
991 def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
994 indata
= deepcopy(db_vnfd_content
)
995 indata
["ext-cpd"] = []
996 with self
.assertRaises(EngineException
) as e
:
997 self
.topic
.validate_mgmt_interface_connection_point(indata
)
999 e
.exception
.http_code
,
1000 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1001 "Wrong HTTP status code",
1005 "mgmt-cp='{}' must match an existing ext-cpd".format(indata
["mgmt-cp"])
1007 norm(str(e
.exception
)),
1008 "Wrong exception text",
1011 def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self
):
1012 indata
= deepcopy(db_vnfd_content
)
1013 indata
.pop("mgmt-cp")
1014 with self
.assertRaises(EngineException
) as e
:
1015 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1017 e
.exception
.http_code
,
1018 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1019 "Wrong HTTP status code",
1022 norm("'mgmt-cp' is a mandatory field and it is not defined"),
1023 norm(str(e
.exception
)),
1024 "Wrong exception text",
1027 def test_validate_vdu_internal_connection_points_on_valid_descriptor(self
):
1028 indata
= db_vnfd_content
1029 vdu
= indata
["vdu"][0]
1030 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1032 def test_validate_external_connection_points_on_valid_descriptor(self
):
1033 indata
= db_vnfd_content
1034 self
.topic
.validate_external_connection_points(indata
)
1036 def test_validate_external_connection_points_when_missing_internal_connection_point(
1039 indata
= deepcopy(db_vnfd_content
)
1040 vdu
= indata
["vdu"][0]
1042 affected_ext_cpd
= indata
["ext-cpd"][0]
1043 with self
.assertRaises(EngineException
) as e
:
1044 self
.topic
.validate_external_connection_points(indata
)
1046 e
.exception
.http_code
,
1047 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1048 "Wrong HTTP status code",
1052 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
1053 affected_ext_cpd
["id"]
1056 norm(str(e
.exception
)),
1057 "Wrong exception text",
1060 def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
1063 indata
= deepcopy(db_vnfd_content
)
1064 vdu
= indata
["vdu"][0]
1068 "virtual-network-interface-requirement": [{"name": "duplicated"}],
1070 vdu
["int-cpd"].insert(0, duplicated_cpd
)
1071 with self
.assertRaises(EngineException
) as e
:
1072 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1074 e
.exception
.http_code
,
1075 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1076 "Wrong HTTP status code",
1080 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
1081 vdu
["id"], duplicated_cpd
["id"]
1084 norm(str(e
.exception
)),
1085 "Wrong exception text",
1088 def test_validate_external_connection_points_on_duplicated_external_connection_point(
1091 indata
= deepcopy(db_vnfd_content
)
1093 "id": "vnf-mgmt-ext",
1094 "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
1096 indata
["ext-cpd"].insert(0, duplicated_cpd
)
1097 with self
.assertRaises(EngineException
) as e
:
1098 self
.topic
.validate_external_connection_points(indata
)
1100 e
.exception
.http_code
,
1101 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1102 "Wrong HTTP status code",
1106 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
1107 duplicated_cpd
["id"]
1110 norm(str(e
.exception
)),
1111 "Wrong exception text",
1114 def test_validate_internal_virtual_links_on_valid_descriptor(self
):
1115 indata
= db_vnfd_content
1116 self
.topic
.validate_internal_virtual_links(indata
)
1118 def test_validate_internal_virtual_links_on_duplicated_ivld(self
):
1119 indata
= deepcopy(db_vnfd_content
)
1120 duplicated_vld
= {"id": "internal"}
1121 indata
["int-virtual-link-desc"].insert(0, duplicated_vld
)
1122 with self
.assertRaises(EngineException
) as e
:
1123 self
.topic
.validate_internal_virtual_links(indata
)
1125 e
.exception
.http_code
,
1126 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1127 "Wrong HTTP status code",
1131 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
1132 duplicated_vld
["id"]
1135 norm(str(e
.exception
)),
1136 "Wrong exception text",
1139 def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
1142 indata
= deepcopy(db_vnfd_content
)
1143 vdu
= indata
["vdu"][0]
1144 affected_int_cpd
= vdu
["int-cpd"][0]
1145 affected_int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
1146 with self
.assertRaises(EngineException
) as e
:
1147 self
.topic
.validate_internal_virtual_links(indata
)
1149 e
.exception
.http_code
,
1150 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1151 "Wrong HTTP status code",
1155 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
1156 "int-virtual-link-desc".format(
1158 affected_int_cpd
["id"],
1159 affected_int_cpd
["int-virtual-link-desc"],
1162 norm(str(e
.exception
)),
1163 "Wrong exception text",
1166 def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self
):
1167 indata
= deepcopy(db_vnfd_content
)
1168 affected_ivld_profile
= {"id": "non-existing-int-virtual-link-desc"}
1169 df
= indata
["df"][0]
1170 df
["virtual-link-profile"] = [affected_ivld_profile
]
1171 with self
.assertRaises(EngineException
) as e
:
1172 self
.topic
.validate_internal_virtual_links(indata
)
1174 e
.exception
.http_code
,
1175 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1176 "Wrong HTTP status code",
1180 "df[id='{}']:virtual-link-profile='{}' must match an existing "
1181 "int-virtual-link-desc".format(df
["id"], affected_ivld_profile
["id"])
1183 norm(str(e
.exception
)),
1184 "Wrong exception text",
1187 def test_validate_monitoring_params_on_valid_descriptor(self
):
1188 indata
= db_vnfd_content
1189 self
.topic
.validate_monitoring_params(indata
)
1191 def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self
):
1192 indata
= deepcopy(db_vnfd_content
)
1193 duplicated_mp
= {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1194 affected_ivld
= indata
["int-virtual-link-desc"][0]
1195 affected_ivld
["monitoring-parameters"] = [duplicated_mp
, duplicated_mp
]
1196 with self
.assertRaises(EngineException
) as e
:
1197 self
.topic
.validate_monitoring_params(indata
)
1199 e
.exception
.http_code
,
1200 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1201 "Wrong HTTP status code",
1205 "Duplicated monitoring-parameter id in "
1206 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1207 affected_ivld
["id"], duplicated_mp
["id"]
1210 norm(str(e
.exception
)),
1211 "Wrong exception text",
1214 def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self
):
1215 indata
= deepcopy(db_vnfd_content
)
1217 "id": "dataVM_cpu_util",
1218 "name": "dataVM_cpu_util",
1219 "performance_metric": "cpu",
1221 affected_vdu
= indata
["vdu"][1]
1222 affected_vdu
["monitoring-parameter"].insert(0, duplicated_mp
)
1223 with self
.assertRaises(EngineException
) as e
:
1224 self
.topic
.validate_monitoring_params(indata
)
1226 e
.exception
.http_code
,
1227 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1228 "Wrong HTTP status code",
1232 "Duplicated monitoring-parameter id in "
1233 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1234 affected_vdu
["id"], duplicated_mp
["id"]
1237 norm(str(e
.exception
)),
1238 "Wrong exception text",
1241 def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self
):
1242 indata
= deepcopy(db_vnfd_content
)
1246 "performance_metric": "memory",
1248 affected_df
= indata
["df"][0]
1249 affected_df
["monitoring-parameter"] = [duplicated_mp
, duplicated_mp
]
1250 with self
.assertRaises(EngineException
) as e
:
1251 self
.topic
.validate_monitoring_params(indata
)
1253 e
.exception
.http_code
,
1254 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1255 "Wrong HTTP status code",
1259 "Duplicated monitoring-parameter id in "
1260 "df[id='{}']:monitoring-parameter[id='{}']".format(
1261 affected_df
["id"], duplicated_mp
["id"]
1264 norm(str(e
.exception
)),
1265 "Wrong exception text",
1268 def test_validate_scaling_group_descriptor_on_valid_descriptor(self
):
1269 indata
= db_vnfd_content
1270 self
.topic
.validate_scaling_group_descriptor(indata
)
1272 def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self
):
1273 indata
= deepcopy(db_vnfd_content
)
1274 vdu
= indata
["vdu"][1]
1275 affected_df
= indata
["df"][0]
1276 affected_sa
= affected_df
["scaling-aspect"][0]
1277 affected_sp
= affected_sa
["scaling-policy"][0]
1278 affected_sc
= affected_sp
["scaling-criteria"][0]
1279 vdu
.pop("monitoring-parameter")
1280 with self
.assertRaises(EngineException
) as e
:
1281 self
.topic
.validate_scaling_group_descriptor(indata
)
1283 e
.exception
.http_code
,
1284 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1285 "Wrong HTTP status code",
1289 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1290 "[name='{}']:scaling-criteria[name='{}']: "
1291 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1294 affected_sp
["name"],
1295 affected_sc
["name"],
1296 affected_sc
["vnf-monitoring-param-ref"],
1299 norm(str(e
.exception
)),
1300 "Wrong exception text",
1303 def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self
):
1304 indata
= deepcopy(db_vnfd_content
)
1305 df
= indata
["df"][0]
1306 affected_sa
= df
["scaling-aspect"][0]
1307 indata
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1310 with self
.assertRaises(EngineException
) as e
:
1311 self
.topic
.validate_scaling_group_descriptor(indata
)
1313 e
.exception
.http_code
,
1314 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1315 "Wrong HTTP status code",
1319 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1320 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1321 df
["id"], affected_sa
["id"]
1324 norm(str(e
.exception
)),
1325 "Wrong exception text",
1328 def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1331 indata
= deepcopy(db_vnfd_content
)
1332 df
= indata
["df"][0]
1333 affected_sa
= df
["scaling-aspect"][0]
1334 affected_sca_primitive
= affected_sa
["scaling-config-action"][0][
1335 "vnf-config-primitive-name-ref"
1337 df
["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1340 with self
.assertRaises(EngineException
) as e
:
1341 self
.topic
.validate_scaling_group_descriptor(indata
)
1343 e
.exception
.http_code
,
1344 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1345 "Wrong HTTP status code",
1349 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1350 "config-primitive-name-ref='{}' does not match any "
1351 "day1-2 configuration:config-primitive:name".format(
1352 df
["id"], affected_sa
["id"], affected_sca_primitive
1355 norm(str(e
.exception
)),
1356 "Wrong exception text",
1359 def test_new_vnfd_revision(self
):
1360 did
= db_vnfd_content
["_id"]
1361 self
.fs
.get_params
.return_value
= {}
1362 self
.fs
.file_exists
.return_value
= False
1363 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1364 "/tmp/" + str(uuid4()), "a+b"
1366 test_vnfd
= deepcopy(db_vnfd_content
)
1367 del test_vnfd
["_id"]
1368 del test_vnfd
["_admin"]
1369 self
.db
.create
.return_value
= did
1371 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1372 db_args
= self
.db
.create
.call_args
[0]
1374 db_args
[1]["_admin"]["revision"], 0, "New package should be at revision 0"
1377 @patch("osm_nbi.descriptor_topics.shutil")
1378 @patch("osm_nbi.descriptor_topics.os.rename")
1379 def test_update_vnfd(self
, mock_rename
, mock_shutil
):
1381 did
= db_vnfd_content
["_id"]
1383 self
.fs
.get_params
.return_value
= {}
1384 self
.fs
.file_exists
.return_value
= False
1385 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1386 "/tmp/" + str(uuid4()), "a+b"
1388 new_vnfd
= deepcopy(db_vnfd_content
)
1390 self
.db
.create
.return_value
= did
1392 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1393 del new_vnfd
["vdu"][0]["cloud-init-file"]
1394 del new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1396 ][0]["execution-environment-list"][0]["juju"]
1398 old_vnfd
= {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])}
1399 old_vnfd
["_admin"]["revision"] = old_revision
1401 self
.db
.get_one
.side_effect
= [old_vnfd
, old_vnfd
, None]
1402 self
.topic
.upload_content(fake_session
, did
, new_vnfd
, {}, {"Content-Type": []})
1404 db_args
= self
.db
.replace
.call_args
[0]
1406 db_args
[2]["_admin"]["revision"],
1408 "Revision should increment",
1412 class Test_NsdTopic(TestCase
):
1414 def setUpClass(cls
):
1415 cls
.test_name
= "test-nsd-topic"
1418 def tearDownClass(cls
):
1422 self
.db
= Mock(dbbase
.DbBase())
1423 self
.fs
= Mock(fsbase
.FsBase())
1424 self
.msg
= Mock(msgbase
.MsgBase())
1425 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1426 self
.topic
= NsdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1427 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1430 def assertNotRaises(self
, exception_type
):
1433 except exception_type
:
1434 raise self
.failureException("{} raised".format(exception_type
.__name
__))
1436 def create_desc_temp(self
, template
):
1437 old_desc
= deepcopy(template
)
1438 new_desc
= deepcopy(template
)
1439 return old_desc
, new_desc
1441 def prepare_nsd_creation(self
):
1443 did
= db_nsd_content
["_id"]
1444 self
.fs
.get_params
.return_value
= {}
1445 self
.fs
.file_exists
.return_value
= False
1446 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1447 "/tmp/" + str(uuid4()), "a+b"
1449 self
.db
.get_one
.side_effect
= [
1450 {"_id": did
, "_admin": deepcopy(db_nsd_content
["_admin"])},
1453 test_nsd
= deepcopy(db_nsd_content
)
1455 del test_nsd
["_admin"]
1456 return did
, test_nsd
1458 @patch("osm_nbi.descriptor_topics.shutil")
1459 @patch("osm_nbi.descriptor_topics.os.rename")
1460 def test_new_nsd_normal_creation(self
, mock_rename
, mock_shutil
):
1461 did
, test_nsd
= self
.prepare_nsd_creation()
1462 self
.db
.create
.return_value
= did
1465 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1466 db_args
= self
.db
.create
.call_args
[0]
1467 msg_args
= self
.msg
.write
.call_args
[0]
1468 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1469 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1470 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
1471 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1472 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1473 self
.assertEqual(did2
, did
, "Wrong DB NSD id")
1474 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
1476 db_args
[1]["_admin"]["modified"],
1477 db_args
[1]["_admin"]["created"],
1478 "Wrong modification time",
1481 db_args
[1]["_admin"]["projects_read"],
1483 "Wrong read-only project list",
1486 db_args
[1]["_admin"]["projects_write"],
1488 "Wrong read-write project list",
1491 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1493 self
.topic
.upload_content(fake_session
, did
, test_nsd
, {}, {"Content-Type": []})
1494 msg_args
= self
.msg
.write
.call_args
[0]
1495 test_nsd
["_id"] = did
1496 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1497 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1498 self
.assertEqual(msg_args
[2], test_nsd
, "Wrong message content")
1500 db_args
= self
.db
.get_one
.mock_calls
[0][1]
1501 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1502 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB NSD id")
1504 db_args
= self
.db
.replace
.call_args
[0]
1505 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1506 self
.assertEqual(db_args
[1], did
, "Wrong DB NSD id")
1508 admin
= db_args
[2]["_admin"]
1509 db_admin
= db_nsd_content
["_admin"]
1510 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
1512 admin
["modified"], db_admin
["created"], "Wrong modification time"
1515 admin
["projects_read"],
1516 db_admin
["projects_read"],
1517 "Wrong read-only project list",
1520 admin
["projects_write"],
1521 db_admin
["projects_write"],
1522 "Wrong read-write project list",
1525 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1528 admin
["operationalState"], "ENABLED", "Wrong operational state"
1530 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
1532 storage
= admin
["storage"]
1533 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
1534 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
1536 compare_desc(self
, test_nsd
, db_args
[2], "NSD")
1537 revision_args
= self
.db
.create
.call_args
[0]
1539 revision_args
[0], self
.topic
.topic
+ "_revisions", "Wrong topic"
1541 self
.assertEqual(revision_args
[1]["id"], db_args
[2]["id"], "Wrong revision id")
1543 revision_args
[1]["_id"], db_args
[2]["_id"] + ":1", "Wrong revision _id"
1546 @patch("osm_nbi.descriptor_topics.shutil")
1547 @patch("osm_nbi.descriptor_topics.os.rename")
1548 def test_new_nsd_check_pyangbind_validation_required_properties(
1549 self
, mock_rename
, mock_shutil
1551 did
, test_nsd
= self
.prepare_nsd_creation()
1554 with self
.assertRaises(
1555 EngineException
, msg
="Accepted NSD with a missing required property"
1557 self
.topic
.upload_content(
1558 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1561 e
.exception
.http_code
,
1562 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1563 "Wrong HTTP status code",
1566 norm("Error in pyangbind validation: '{}'".format("id")),
1567 norm(str(e
.exception
)),
1568 "Wrong exception text",
1571 @patch("osm_nbi.descriptor_topics.shutil")
1572 @patch("osm_nbi.descriptor_topics.os.rename")
1573 def test_new_nsd_check_pyangbind_validation_additional_properties(
1574 self
, mock_rename
, mock_shutil
1576 did
, test_nsd
= self
.prepare_nsd_creation()
1577 test_nsd
["extra-property"] = 0
1579 with self
.assertRaises(
1580 EngineException
, msg
="Accepted NSD with an additional property"
1582 self
.topic
.upload_content(
1583 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1586 e
.exception
.http_code
,
1587 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1588 "Wrong HTTP status code",
1592 "Error in pyangbind validation: {} ({})".format(
1593 "json object contained a key that did not exist", "extra-property"
1596 norm(str(e
.exception
)),
1597 "Wrong exception text",
1600 @patch("osm_nbi.descriptor_topics.shutil")
1601 @patch("osm_nbi.descriptor_topics.os.rename")
1602 def test_new_nsd_check_pyangbind_validation_property_types(
1603 self
, mock_rename
, mock_shutil
1605 did
, test_nsd
= self
.prepare_nsd_creation()
1606 test_nsd
["designer"] = {"key": 0}
1608 with self
.assertRaises(
1609 EngineException
, msg
="Accepted NSD with a wrongly typed property"
1611 self
.topic
.upload_content(
1612 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1615 e
.exception
.http_code
,
1616 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1617 "Wrong HTTP status code",
1621 "Error in pyangbind validation: {} ({})".format(
1622 "json object contained a key that did not exist", "key"
1625 norm(str(e
.exception
)),
1626 "Wrong exception text",
1629 @patch("osm_nbi.descriptor_topics.shutil")
1630 @patch("osm_nbi.descriptor_topics.os.rename")
1631 def test_new_nsd_check_input_validation_mgmt_network_virtual_link_protocol_data(
1632 self
, mock_rename
, mock_shutil
1634 did
, test_nsd
= self
.prepare_nsd_creation()
1635 df
= test_nsd
["df"][0]
1638 "virtual-link-desc-id": "mgmt",
1639 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1641 df
["virtual-link-profile"] = [mgmt_profile
]
1643 with self
.assertRaises(
1644 EngineException
, msg
="Accepted VLD with mgmt-network+ip-profile"
1646 self
.topic
.upload_content(
1647 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1650 e
.exception
.http_code
,
1651 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1652 "Wrong HTTP status code",
1656 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1657 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1658 df
["id"], mgmt_profile
["id"]
1661 norm(str(e
.exception
)),
1662 "Wrong exception text",
1665 @patch("osm_nbi.descriptor_topics.shutil")
1666 @patch("osm_nbi.descriptor_topics.os.rename")
1667 def test_new_nsd_check_descriptor_dependencies_vnfd_id(
1668 self
, mock_rename
, mock_shutil
1670 did
, test_nsd
= self
.prepare_nsd_creation()
1671 self
.db
.get_list
.return_value
= []
1673 with self
.assertRaises(
1674 EngineException
, msg
="Accepted wrong VNFD ID reference"
1676 self
.topic
.upload_content(
1677 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1680 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1684 "'vnfd-id'='{}' references a non existing vnfd".format(
1685 test_nsd
["vnfd-id"][0]
1688 norm(str(e
.exception
)),
1689 "Wrong exception text",
1692 @patch("osm_nbi.descriptor_topics.shutil")
1693 @patch("osm_nbi.descriptor_topics.os.rename")
1694 def test_new_nsd_check_descriptor_dependencies_vld_vnfd_connection_point_ref(
1695 self
, mock_rename
, mock_shutil
1697 # Check Descriptor Dependencies: "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]
1698 did
, test_nsd
= self
.prepare_nsd_creation()
1699 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1700 df
= test_nsd
["df"][0]
1701 affected_vnf_profile
= df
["vnf-profile"][0]
1702 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
1703 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
1704 self
.db
.get_list
.return_value
= [vnfd_descriptor
]
1706 with self
.assertRaises(
1707 EngineException
, msg
="Accepted wrong VLD CP reference"
1709 self
.topic
.upload_content(
1710 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1713 e
.exception
.http_code
,
1714 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1715 "Wrong HTTP status code",
1719 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1720 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1721 "non existing ext-cpd:id inside vnfd '{}'".format(
1723 affected_vnf_profile
["id"],
1724 affected_virtual_link
["virtual-link-profile-id"],
1726 vnfd_descriptor
["id"],
1729 norm(str(e
.exception
)),
1730 "Wrong exception text",
1733 def test_edit_nsd(self
):
1734 nsd_content
= deepcopy(db_nsd_content
)
1735 did
= nsd_content
["_id"]
1736 self
.fs
.file_exists
.return_value
= True
1737 self
.fs
.dir_ls
.return_value
= True
1738 with self
.subTest(i
=1, t
="Normal Edition"):
1740 self
.db
.get_one
.side_effect
= [deepcopy(nsd_content
), None]
1741 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1742 data
= {"id": "new-nsd-id", "name": "new-nsd-name"}
1743 self
.topic
.edit(fake_session
, did
, data
)
1744 db_args
= self
.db
.replace
.call_args
[0]
1745 msg_args
= self
.msg
.write
.call_args
[0]
1747 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1748 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1749 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
1750 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1751 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
1753 db_args
[2]["_admin"]["created"],
1754 nsd_content
["_admin"]["created"],
1755 "Wrong creation time",
1758 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
1761 db_args
[2]["_admin"]["projects_read"],
1762 nsd_content
["_admin"]["projects_read"],
1763 "Wrong read-only project list",
1766 db_args
[2]["_admin"]["projects_write"],
1767 nsd_content
["_admin"]["projects_write"],
1768 "Wrong read-write project list",
1770 self
.assertEqual(db_args
[2]["id"], data
["id"], "Wrong NSD ID")
1771 self
.assertEqual(db_args
[2]["name"], data
["name"], "Wrong NSD Name")
1772 with self
.subTest(i
=2, t
="Conflict on Edit"):
1773 data
= {"id": "fake-nsd-id", "name": "new-nsd-name"}
1774 self
.db
.get_one
.side_effect
= [
1776 {"_id": str(uuid4()), "id": data
["id"]},
1778 with self
.assertRaises(
1779 EngineException
, msg
="Accepted existing NSD ID"
1781 self
.topic
.edit(fake_session
, did
, data
)
1783 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1787 "{} with id '{}' already exists for this project".format(
1791 norm(str(e
.exception
)),
1792 "Wrong exception text",
1794 with self
.subTest(i
=3, t
="Check Envelope"):
1795 data
= {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
1796 self
.db
.get_one
.side_effect
= [nsd_content
, None]
1797 with self
.assertRaises(
1798 EngineException
, msg
="Accepted NSD with wrong envelope"
1800 self
.topic
.edit(fake_session
, did
, data
, content
=nsd_content
)
1802 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
1805 "'nsd' must be a list of only one element",
1806 norm(str(e
.exception
)),
1807 "Wrong exception text",
1809 self
.db
.reset_mock()
1812 def test_delete_nsd(self
):
1813 did
= db_nsd_content
["_id"]
1814 self
.db
.get_one
.return_value
= db_nsd_content
1815 p_id
= db_nsd_content
["_admin"]["projects_read"][0]
1816 with self
.subTest(i
=1, t
="Normal Deletion"):
1817 self
.db
.get_list
.return_value
= []
1818 self
.db
.del_one
.return_value
= {"deleted": 1}
1819 self
.topic
.delete(fake_session
, did
)
1820 db_args
= self
.db
.del_one
.call_args
[0]
1821 msg_args
= self
.msg
.write
.call_args
[0]
1822 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1823 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
1824 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1825 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1826 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
1828 db_args
[1]["_admin.projects_write.cont"],
1832 db_g1_args
= self
.db
.get_one
.call_args
[0]
1833 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1834 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB NSD ID")
1835 db_gl_calls
= self
.db
.get_list
.call_args_list
1836 self
.assertEqual(db_gl_calls
[0][0][0], "nsrs", "Wrong DB topic")
1837 # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID") # Filter changed after call
1838 self
.assertEqual(db_gl_calls
[1][0][0], "nsts", "Wrong DB topic")
1840 db_gl_calls
[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
1841 db_nsd_content
["id"],
1842 "Wrong DB NSD netslice-subnet nsd-ref",
1844 self
.db
.set_one
.assert_not_called()
1845 fs_del_calls
= self
.fs
.file_delete
.call_args_list
1846 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
1847 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
1848 with self
.subTest(i
=2, t
="Conflict on Delete - NSD in use by nsr"):
1849 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-nsr"}]
1850 with self
.assertRaises(
1851 EngineException
, msg
="Accepted NSD in use by NSR"
1853 self
.topic
.delete(fake_session
, did
)
1855 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1858 "there is at least one ns instance using this descriptor",
1859 norm(str(e
.exception
)),
1860 "Wrong exception text",
1862 with self
.subTest(i
=3, t
="Conflict on Delete - NSD in use by NST"):
1863 self
.db
.get_list
.side_effect
= [
1865 [{"_id": str(uuid4()), "name": "fake-nst"}],
1867 with self
.assertRaises(
1868 EngineException
, msg
="Accepted NSD in use by NST"
1870 self
.topic
.delete(fake_session
, did
)
1872 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1875 "there is at least one netslice template referencing this descriptor",
1876 norm(str(e
.exception
)),
1877 "Wrong exception text",
1879 with self
.subTest(i
=4, t
="Non-existent NSD"):
1880 excp_msg
= "Not found any {} with filter='{}'".format("NSD", {"_id": did
})
1881 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
1882 with self
.assertRaises(
1883 DbException
, msg
="Accepted non-existent NSD ID"
1885 self
.topic
.delete(fake_session
, did
)
1887 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1890 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
1892 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
1893 db_nsd_content
["_admin"]["projects_read"].append("other_project")
1894 self
.db
.get_one
= Mock(return_value
=db_nsd_content
)
1895 self
.db
.get_list
= Mock(return_value
=[])
1896 self
.msg
.write
.reset_mock()
1897 self
.db
.del_one
.reset_mock()
1898 self
.fs
.file_delete
.reset_mock()
1900 self
.topic
.delete(fake_session
, did
)
1901 self
.db
.del_one
.assert_not_called()
1902 self
.msg
.write
.assert_not_called()
1903 db_g1_args
= self
.db
.get_one
.call_args
[0]
1904 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1905 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
1906 db_s1_args
= self
.db
.set_one
.call_args
1907 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
1908 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
1910 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1913 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
1916 db_s1_args
[1]["pull_list"],
1917 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
1918 "Wrong DB pull_list dictionary",
1920 self
.fs
.file_delete
.assert_not_called()
1921 self
.db
.reset_mock()
1924 def prepare_nsd_validation(self
):
1925 descriptor_name
= "test_ns_descriptor"
1926 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1927 "/tmp/" + str(uuid4()), "a+b"
1929 old_nsd
, new_nsd
= self
.create_desc_temp(db_nsd_content
)
1930 return descriptor_name
, old_nsd
, new_nsd
1932 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1933 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1934 def test_validate_descriptor_ns_configuration_changed(
1935 self
, mock_safe_load
, mock_detect_usage
1937 """Validating NSD and NSD has changes in ns-configuration:config-primitive"""
1938 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
1939 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
1940 mock_detect_usage
.return_value
= True
1941 self
.db
.get_one
.return_value
= old_nsd
1943 {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
1946 {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
1949 with self
.assertNotRaises(EngineException
):
1950 self
.topic
._validate
_descriptor
_changes
(
1951 old_nsd
["_id"], descriptor_name
, "/tmp", "/tmp:1"
1953 self
.db
.get_one
.assert_called_once()
1954 mock_detect_usage
.assert_called_once()
1955 self
.assertEqual(mock_safe_load
.call_count
, 2)
1957 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1958 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1959 def test_validate_descriptor_nsd_name_changed(
1960 self
, mock_safe_load
, mock_detect_usage
1962 """Validating NSD, NSD name has changed."""
1963 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
1964 did
= old_nsd
["_id"]
1965 new_nsd
["name"] = "nscharm-ns2"
1966 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
1967 mock_detect_usage
.return_value
= True
1968 self
.db
.get_one
.return_value
= old_nsd
1970 with self
.assertRaises(
1971 EngineException
, msg
="there are disallowed changes in the ns descriptor"
1973 self
.topic
._validate
_descriptor
_changes
(
1974 did
, descriptor_name
, "/tmp", "/tmp:1"
1977 e
.exception
.http_code
,
1978 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1979 "Wrong HTTP status code",
1982 norm("there are disallowed changes in the ns descriptor"),
1983 norm(str(e
.exception
)),
1984 "Wrong exception text",
1987 self
.db
.get_one
.assert_called_once()
1988 mock_detect_usage
.assert_called_once()
1989 self
.assertEqual(mock_safe_load
.call_count
, 2)
1991 @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
1992 @patch("osm_nbi.descriptor_topics.yaml.safe_load")
1993 def test_validate_descriptor_nsd_name_changed_nsd_not_in_use(
1994 self
, mock_safe_load
, mock_detect_usage
1996 """Validating NSD, NSD name has changed, NSD is not in use."""
1997 descriptor_name
, old_nsd
, new_nsd
= self
.prepare_nsd_validation()
1998 did
= old_nsd
["_id"]
1999 new_nsd
["name"] = "nscharm-ns2"
2000 mock_safe_load
.side_effect
= [old_nsd
, new_nsd
]
2001 mock_detect_usage
.return_value
= None
2002 self
.db
.get_one
.return_value
= old_nsd
2004 with self
.assertNotRaises(Exception):
2005 self
.topic
._validate
_descriptor
_changes
(
2006 did
, descriptor_name
, "/tmp", "/tmp:1"
2009 self
.db
.get_one
.assert_called_once()
2010 mock_detect_usage
.assert_called_once()
2011 mock_safe_load
.assert_not_called()
2013 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
2016 indata
= deepcopy(db_nsd_content
)
2017 vld
= indata
["virtual-link-desc"][0]
2018 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2022 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
2025 indata
= deepcopy(db_nsd_content
)
2026 vld
= indata
["virtual-link-desc"][0]
2027 df
= indata
["df"][0]
2030 "virtual-link-desc-id": "mgmt",
2031 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
2033 df
["virtual-link-profile"] = [affected_vlp
]
2034 with self
.assertRaises(EngineException
) as e
:
2035 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2039 e
.exception
.http_code
,
2040 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2041 "Wrong HTTP status code",
2045 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
2046 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
2047 df
["id"], affected_vlp
["id"]
2050 norm(str(e
.exception
)),
2051 "Wrong exception text",
2054 def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self
):
2055 indata
= deepcopy(db_nsd_content
)
2056 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2058 def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self
):
2059 indata
= deepcopy(db_nsd_content
)
2060 df
= indata
["df"][0]
2061 affected_vnf_profile
= df
["vnf-profile"][0]
2062 indata
["vnfd-id"] = ["non-existing-vnfd"]
2063 with self
.assertRaises(EngineException
) as e
:
2064 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2066 e
.exception
.http_code
,
2067 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2068 "Wrong HTTP status code",
2072 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
2073 "does not match any vnfd-id".format(
2075 affected_vnf_profile
["id"],
2076 affected_vnf_profile
["vnfd-id"],
2079 norm(str(e
.exception
)),
2080 "Wrong exception text",
2083 def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
2086 nsd_descriptor
= deepcopy(db_nsd_content
)
2087 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2088 df
= nsd_descriptor
["df"][0]
2089 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2090 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2094 def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
2097 nsd_descriptor
= deepcopy(db_nsd_content
)
2098 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2099 df
= nsd_descriptor
["df"][0]
2100 affected_vnf_profile
= df
["vnf-profile"][0]
2101 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
2102 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2103 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
2104 with self
.assertRaises(EngineException
) as e
:
2105 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2109 e
.exception
.http_code
,
2110 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2111 "Wrong HTTP status code",
2115 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
2116 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
2117 "non existing ext-cpd:id inside vnfd '{}'".format(
2119 affected_vnf_profile
["id"],
2120 affected_virtual_link
["virtual-link-profile-id"],
2122 vnfd_descriptor
["id"],
2125 norm(str(e
.exception
)),
2126 "Wrong exception text",
2129 def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self
):
2130 nsd_descriptor
= deepcopy(db_nsd_content
)
2131 invalid_vnfd_id
= "invalid-vnfd-id"
2132 nsd_descriptor
["id"] = "invalid-vnfd-id-ns"
2133 nsd_descriptor
["vnfd-id"][0] = invalid_vnfd_id
2134 nsd_descriptor
["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
2135 nsd_descriptor
["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
2136 with self
.assertRaises(EngineException
) as e
:
2137 self
.db
.get_list
.return_value
= []
2138 nsd_descriptor
= self
.topic
.check_conflict_on_edit(
2139 fake_session
, nsd_descriptor
, [], "id"
2142 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2146 "Descriptor error at 'vnfd-id'='{}' references a non "
2147 "existing vnfd".format(invalid_vnfd_id
)
2149 norm(str(e
.exception
)),
2150 "Wrong exception text",
2154 if __name__
== "__main__":