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
.load(db_vnfds_text
, Loader
=yaml
.Loader
)[0]
38 db_nsd_content
= yaml
.load(db_nsds_text
, Loader
=yaml
.Loader
)[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 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 @patch("osm_nbi.descriptor_topics.shutil")
109 @patch("osm_nbi.descriptor_topics.os.rename")
110 def test_new_vnfd(self
, mock_rename
, mock_shutil
):
111 did
= db_vnfd_content
["_id"]
113 self
.fs
.get_params
.return_value
= {}
114 self
.fs
.file_exists
.return_value
= False
115 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
116 "/tmp/" + str(uuid4()), "a+b"
118 test_vnfd
= deepcopy(db_vnfd_content
)
120 del test_vnfd
["_admin"]
121 with self
.subTest(i
=1, t
="Normal Creation"):
122 self
.db
.create
.return_value
= did
124 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
125 db_args
= self
.db
.create
.call_args
[0]
126 msg_args
= self
.msg
.write
.call_args
[0]
127 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
128 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
129 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
130 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
131 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
132 self
.assertEqual(did2
, did
, "Wrong DB VNFD id")
133 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
135 db_args
[1]["_admin"]["modified"],
136 db_args
[1]["_admin"]["created"],
137 "Wrong modification time",
140 db_args
[1]["_admin"]["projects_read"],
142 "Wrong read-only project list",
145 db_args
[1]["_admin"]["projects_write"],
147 "Wrong read-write project list",
149 tmp1
= test_vnfd
["vdu"][0]["cloud-init-file"]
150 tmp2
= test_vnfd
["df"][0]["lcm-operations-configuration"][
151 "operate-vnf-op-config"
152 ]["day1-2"][0]["execution-environment-list"][0]["juju"]
153 del test_vnfd
["vdu"][0]["cloud-init-file"]
154 del test_vnfd
["df"][0]["lcm-operations-configuration"][
155 "operate-vnf-op-config"
156 ]["day1-2"][0]["execution-environment-list"][0]["juju"]
158 self
.db
.get_one
.side_effect
= [
159 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
162 self
.topic
.upload_content(
163 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
165 msg_args
= self
.msg
.write
.call_args
[0]
166 test_vnfd
["_id"] = did
168 msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic"
170 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
171 self
.assertEqual(msg_args
[2], test_vnfd
, "Wrong message content")
172 db_args
= self
.db
.get_one
.mock_calls
[0][1]
173 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
174 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB VNFD id")
175 db_args
= self
.db
.replace
.call_args
[0]
176 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
177 self
.assertEqual(db_args
[1], did
, "Wrong DB VNFD id")
178 admin
= db_args
[2]["_admin"]
179 db_admin
= deepcopy(db_vnfd_content
["_admin"])
180 self
.assertEqual(admin
["type"], "vnfd", "Wrong descriptor type")
182 admin
["created"], db_admin
["created"], "Wrong creation time"
185 admin
["modified"], db_admin
["created"], "Wrong modification time"
188 admin
["projects_read"],
189 db_admin
["projects_read"],
190 "Wrong read-only project list",
193 admin
["projects_write"],
194 db_admin
["projects_write"],
195 "Wrong read-write project list",
198 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
201 admin
["operationalState"], "ENABLED", "Wrong operational state"
203 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
204 storage
= admin
["storage"]
205 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
207 storage
["descriptor"], "package", "Wrong storage descriptor"
210 admin
["revision"], 1, "Wrong revision number"
213 compare_desc(self
, test_vnfd
, db_args
[2], "VNFD")
215 test_vnfd
["vdu"][0]["cloud-init-file"] = tmp1
216 test_vnfd
["df"][0]["lcm-operations-configuration"][
217 "operate-vnf-op-config"
218 ]["day1-2"][0]["execution-environment-list"][0]["juju"] = tmp2
219 self
.db
.get_one
.side_effect
= (
220 lambda table
, filter, fail_on_empty
=None, fail_on_more
=None: {
222 "_admin": deepcopy(db_vnfd_content
["_admin"]),
225 with self
.subTest(i
=2, t
="Check Pyangbind Validation: additional properties"):
226 test_vnfd
["extra-property"] = 0
228 with self
.assertRaises(
229 EngineException
, msg
="Accepted VNFD with an additional property"
231 self
.topic
.upload_content(
232 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
235 e
.exception
.http_code
,
236 HTTPStatus
.UNPROCESSABLE_ENTITY
,
237 "Wrong HTTP status code",
241 "Error in pyangbind validation: {} ({})".format(
242 "json object contained a key that did not exist",
246 norm(str(e
.exception
)),
247 "Wrong exception text",
249 db_args
= self
.db
.replace
.call_args
[0]
250 admin
= db_args
[2]["_admin"]
252 admin
["revision"], 1, "Wrong revision number"
256 del test_vnfd
["extra-property"]
257 with self
.subTest(i
=3, t
="Check Pyangbind Validation: property types"):
258 tmp
= test_vnfd
["product-name"]
259 test_vnfd
["product-name"] = {"key": 0}
261 with self
.assertRaises(
262 EngineException
, msg
="Accepted VNFD with a wrongly typed property"
264 self
.topic
.upload_content(
265 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
268 e
.exception
.http_code
,
269 HTTPStatus
.UNPROCESSABLE_ENTITY
,
270 "Wrong HTTP status code",
274 "Error in pyangbind validation: {} ({})".format(
275 "json object contained a key that did not exist", "key"
278 norm(str(e
.exception
)),
279 "Wrong exception text",
282 test_vnfd
["product-name"] = tmp
283 with self
.subTest(i
=4, t
="Check Input Validation: cloud-init"):
284 with self
.assertRaises(
285 EngineException
, msg
="Accepted non-existent cloud_init file"
287 self
.topic
.upload_content(
288 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
291 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
295 "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format(
296 "cloud-init", test_vnfd
["id"], test_vnfd
["vdu"][0]["id"]
299 norm(str(e
.exception
)),
300 "Wrong exception text",
302 with self
.subTest(i
=5, t
="Check Input Validation: day1-2 configuration[juju]"):
303 del test_vnfd
["vdu"][0]["cloud-init-file"]
304 with self
.assertRaises(
305 EngineException
, msg
="Accepted non-existent charm in VNF configuration"
307 self
.topic
.upload_content(
308 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
311 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
315 "{} defined in vnf[id={}] but not present in package".format(
316 "charm", test_vnfd
["id"]
319 norm(str(e
.exception
)),
320 "Wrong exception text",
322 del test_vnfd
["df"][0]["lcm-operations-configuration"][
323 "operate-vnf-op-config"
324 ]["day1-2"][0]["execution-environment-list"][0]["juju"]
325 with self
.subTest(i
=6, t
="Check Input Validation: mgmt-cp"):
326 tmp
= test_vnfd
["mgmt-cp"]
327 del test_vnfd
["mgmt-cp"]
329 with self
.assertRaises(
330 EngineException
, msg
="Accepted VNFD without management interface"
332 self
.topic
.upload_content(
333 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
336 e
.exception
.http_code
,
337 HTTPStatus
.UNPROCESSABLE_ENTITY
,
338 "Wrong HTTP status code",
342 "'{}' is a mandatory field and it is not defined".format(
346 norm(str(e
.exception
)),
347 "Wrong exception text",
350 test_vnfd
["mgmt-cp"] = tmp
351 with self
.subTest(i
=7, t
="Check Input Validation: mgmt-cp connection point"):
352 tmp
= test_vnfd
["mgmt-cp"]
353 test_vnfd
["mgmt-cp"] = "wrong-cp"
355 with self
.assertRaises(
356 EngineException
, msg
="Accepted wrong mgmt-cp connection point"
358 self
.topic
.upload_content(
359 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
362 e
.exception
.http_code
,
363 HTTPStatus
.UNPROCESSABLE_ENTITY
,
364 "Wrong HTTP status code",
368 "mgmt-cp='{}' must match an existing ext-cpd".format(
372 norm(str(e
.exception
)),
373 "Wrong exception text",
376 test_vnfd
["mgmt-cp"] = tmp
377 with self
.subTest(i
=8, t
="Check Input Validation: vdu int-cpd"):
378 ext_cpd
= test_vnfd
["ext-cpd"][1]
379 tmp
= ext_cpd
["int-cpd"]["cpd"]
380 ext_cpd
["int-cpd"]["cpd"] = "wrong-cpd"
382 with self
.assertRaises(
384 msg
="Accepted wrong ext-cpd internal connection point",
386 self
.topic
.upload_content(
387 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
390 e
.exception
.http_code
,
391 HTTPStatus
.UNPROCESSABLE_ENTITY
,
392 "Wrong HTTP status code",
396 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
400 norm(str(e
.exception
)),
401 "Wrong exception text",
404 ext_cpd
["int-cpd"]["cpd"] = tmp
405 with self
.subTest(i
=9, t
="Check Input Validation: Duplicated VLD"):
406 test_vnfd
["int-virtual-link-desc"].insert(0, {"id": "internal"})
408 with self
.assertRaises(
409 EngineException
, msg
="Accepted duplicated VLD name"
411 self
.topic
.upload_content(
412 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
415 e
.exception
.http_code
,
416 HTTPStatus
.UNPROCESSABLE_ENTITY
,
417 "Wrong HTTP status code",
421 "identifier id '{}' is not unique".format(
422 test_vnfd
["int-virtual-link-desc"][0]["id"]
425 norm(str(e
.exception
)),
426 "Wrong exception text",
429 del test_vnfd
["int-virtual-link-desc"][0]
430 with self
.subTest(i
=10, t
="Check Input Validation: vdu int-virtual-link-desc"):
431 vdu
= test_vnfd
["vdu"][0]
432 int_cpd
= vdu
["int-cpd"][1]
433 tmp
= int_cpd
["int-virtual-link-desc"]
434 int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
436 with self
.assertRaises(
437 EngineException
, msg
="Accepted int-virtual-link-desc"
439 self
.topic
.upload_content(
440 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
443 e
.exception
.http_code
,
444 HTTPStatus
.UNPROCESSABLE_ENTITY
,
445 "Wrong HTTP status code",
449 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
450 "int-virtual-link-desc".format(
451 vdu
["id"], int_cpd
["id"], int_cpd
["int-virtual-link-desc"]
454 norm(str(e
.exception
)),
455 "Wrong exception text",
458 int_cpd
["int-virtual-link-desc"] = tmp
459 with self
.subTest(i
=11, t
="Check Input Validation: virtual-link-profile)"):
460 fake_ivld_profile
= {"id": "fake-profile-ref", "flavour": "fake-flavour"}
461 df
= test_vnfd
["df"][0]
462 df
["virtual-link-profile"] = [fake_ivld_profile
]
464 with self
.assertRaises(
465 EngineException
, msg
="Accepted non-existent Profile Ref"
467 self
.topic
.upload_content(
468 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
471 e
.exception
.http_code
,
472 HTTPStatus
.UNPROCESSABLE_ENTITY
,
473 "Wrong HTTP status code",
477 "df[id='{}']:virtual-link-profile='{}' must match an existing "
478 "int-virtual-link-desc".format(
479 df
["id"], fake_ivld_profile
["id"]
482 norm(str(e
.exception
)),
483 "Wrong exception text",
486 del df
["virtual-link-profile"]
488 i
=12, t
="Check Input Validation: scaling-criteria monitoring-param-ref"
490 vdu
= test_vnfd
["vdu"][1]
491 affected_df
= test_vnfd
["df"][0]
492 sa
= affected_df
["scaling-aspect"][0]
493 sp
= sa
["scaling-policy"][0]
494 sc
= sp
["scaling-criteria"][0]
495 tmp
= vdu
.pop("monitoring-parameter")
497 with self
.assertRaises(
499 msg
="Accepted non-existent Scaling Group Policy Criteria",
501 self
.topic
.upload_content(
502 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
505 e
.exception
.http_code
,
506 HTTPStatus
.UNPROCESSABLE_ENTITY
,
507 "Wrong HTTP status code",
511 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
512 "[name='{}']:scaling-criteria[name='{}']: "
513 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
518 sc
["vnf-monitoring-param-ref"],
521 norm(str(e
.exception
)),
522 "Wrong exception text",
525 vdu
["monitoring-parameter"] = tmp
527 i
=13, t
="Check Input Validation: scaling-aspect vnf-configuration"
529 df
= test_vnfd
["df"][0]
530 tmp
= test_vnfd
["df"][0]["lcm-operations-configuration"][
531 "operate-vnf-op-config"
534 with self
.assertRaises(
536 msg
="Accepted non-existent Scaling Group VDU ID Reference",
538 self
.topic
.upload_content(
539 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
542 e
.exception
.http_code
,
543 HTTPStatus
.UNPROCESSABLE_ENTITY
,
544 "Wrong HTTP status code",
548 "'day1-2 configuration' not defined in the descriptor but it is referenced "
549 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
550 df
["id"], df
["scaling-aspect"][0]["id"]
553 norm(str(e
.exception
)),
554 "Wrong exception text",
557 test_vnfd
["df"][0]["lcm-operations-configuration"][
558 "operate-vnf-op-config"
559 ]["day1-2"].append(tmp
)
560 with self
.subTest(i
=14, t
="Check Input Validation: scaling-config-action"):
561 df
= test_vnfd
["df"][0]
564 .get("lcm-operations-configuration")
565 .get("operate-vnf-op-config")["day1-2"][0]["config-primitive"]
567 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
568 "operate-vnf-op-config"
569 )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
571 with self
.assertRaises(
573 msg
="Accepted non-existent Scaling Group VDU ID Reference",
575 self
.topic
.upload_content(
576 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
579 e
.exception
.http_code
,
580 HTTPStatus
.UNPROCESSABLE_ENTITY
,
581 "Wrong HTTP status code",
585 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
586 "config-primitive-name-ref='{}' does not match any "
587 "day1-2 configuration:config-primitive:name".format(
589 df
["scaling-aspect"][0]["id"],
590 sa
["scaling-config-action"][0][
591 "vnf-config-primitive-name-ref"
595 norm(str(e
.exception
)),
596 "Wrong exception text",
599 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
600 "operate-vnf-op-config"
601 )["day1-2"][0]["config-primitive"] = tmp
602 with self
.subTest(i
=15, t
="Check Input Validation: everything right"):
603 test_vnfd
["id"] = "fake-vnfd-id"
604 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
605 "operate-vnf-op-config"
606 )["day1-2"][0]["id"] = "fake-vnfd-id"
607 self
.db
.get_one
.side_effect
= [
608 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
611 rc
= self
.topic
.upload_content(
612 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
614 self
.assertTrue(rc
, "Input Validation: Unexpected failure")
617 def test_edit_vnfd(self
):
618 vnfd_content
= deepcopy(db_vnfd_content
)
619 did
= vnfd_content
["_id"]
620 self
.fs
.file_exists
.return_value
= True
621 self
.fs
.dir_ls
.return_value
= True
622 with self
.subTest(i
=1, t
="Normal Edition"):
624 self
.db
.get_one
.side_effect
= [deepcopy(vnfd_content
), None]
625 data
= {"product-name": "new-vnfd-name"}
626 self
.topic
.edit(fake_session
, did
, data
)
627 db_args
= self
.db
.replace
.call_args
[0]
628 msg_args
= self
.msg
.write
.call_args
[0]
630 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
631 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
632 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
633 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
634 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
636 db_args
[2]["_admin"]["created"],
637 vnfd_content
["_admin"]["created"],
638 "Wrong creation time",
641 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
644 db_args
[2]["_admin"]["projects_read"],
645 vnfd_content
["_admin"]["projects_read"],
646 "Wrong read-only project list",
649 db_args
[2]["_admin"]["projects_write"],
650 vnfd_content
["_admin"]["projects_write"],
651 "Wrong read-write project list",
654 db_args
[2]["product-name"], data
["product-name"], "Wrong VNFD Name"
656 with self
.subTest(i
=2, t
="Conflict on Edit"):
657 data
= {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
658 self
.db
.get_one
.side_effect
= [
659 deepcopy(vnfd_content
),
660 {"_id": str(uuid4()), "id": data
["id"]},
662 with self
.assertRaises(
663 EngineException
, msg
="Accepted existing VNFD ID"
665 self
.topic
.edit(fake_session
, did
, data
)
667 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
671 "{} with id '{}' already exists for this project".format(
675 norm(str(e
.exception
)),
676 "Wrong exception text",
678 with self
.subTest(i
=3, t
="Check Envelope"):
679 data
= {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
680 with self
.assertRaises(
681 EngineException
, msg
="Accepted VNFD with wrong envelope"
683 self
.topic
.edit(fake_session
, did
, data
, content
=vnfd_content
)
685 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
688 "'vnfd' must be dict", norm(str(e
.exception
)), "Wrong exception text"
692 def test_delete_vnfd(self
):
693 did
= db_vnfd_content
["_id"]
694 self
.db
.get_one
.return_value
= db_vnfd_content
695 p_id
= db_vnfd_content
["_admin"]["projects_read"][0]
696 with self
.subTest(i
=1, t
="Normal Deletion"):
697 self
.db
.get_list
.return_value
= []
698 self
.db
.del_one
.return_value
= {"deleted": 1}
699 self
.topic
.delete(fake_session
, did
)
700 db_args
= self
.db
.del_one
.call_args
[0]
701 msg_args
= self
.msg
.write
.call_args
[0]
702 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
703 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
704 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
705 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
706 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
708 db_args
[1]["_admin.projects_write.cont"],
712 db_g1_args
= self
.db
.get_one
.call_args
[0]
713 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
714 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
715 db_gl_calls
= self
.db
.get_list
.call_args_list
716 self
.assertEqual(db_gl_calls
[0][0][0], "vnfrs", "Wrong DB topic")
717 # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID") # Filter changed after call
718 self
.assertEqual(db_gl_calls
[1][0][0], "nsds", "Wrong DB topic")
720 db_gl_calls
[1][0][1]["vnfd-id"],
721 db_vnfd_content
["id"],
722 "Wrong DB NSD vnfd-id",
725 db_del_args
= self
.db
.del_list
.call_args
[0]
727 self
.db
.del_list
.call_args
[0][0],
728 self
.topic
.topic
+"_revisions",
733 self
.db
.del_list
.call_args
[0][1]['_id']['$regex'],
735 "Wrong ID for rexep delete",
738 self
.db
.set_one
.assert_not_called()
739 fs_del_calls
= self
.fs
.file_delete
.call_args_list
740 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
741 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
742 with self
.subTest(i
=2, t
="Conflict on Delete - VNFD in use by VNFR"):
743 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-vnfr"}]
744 with self
.assertRaises(
745 EngineException
, msg
="Accepted VNFD in use by VNFR"
747 self
.topic
.delete(fake_session
, did
)
749 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
752 "there is at least one vnf instance using this descriptor",
753 norm(str(e
.exception
)),
754 "Wrong exception text",
756 with self
.subTest(i
=3, t
="Conflict on Delete - VNFD in use by NSD"):
757 self
.db
.get_list
.side_effect
= [
759 [{"_id": str(uuid4()), "name": "fake-nsd"}],
761 with self
.assertRaises(
762 EngineException
, msg
="Accepted VNFD in use by NSD"
764 self
.topic
.delete(fake_session
, did
)
766 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
769 "there is at least one ns package referencing this descriptor",
770 norm(str(e
.exception
)),
771 "Wrong exception text",
773 with self
.subTest(i
=4, t
="Non-existent VNFD"):
774 excp_msg
= "Not found any {} with filter='{}'".format("VNFD", {"_id": did
})
775 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
776 with self
.assertRaises(
777 DbException
, msg
="Accepted non-existent VNFD ID"
779 self
.topic
.delete(fake_session
, did
)
781 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
784 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
786 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
787 db_vnfd_content
["_admin"]["projects_read"].append("other_project")
788 self
.db
.get_one
= Mock(return_value
=db_vnfd_content
)
789 self
.db
.get_list
= Mock(return_value
=[])
790 self
.msg
.write
.reset_mock()
791 self
.db
.del_one
.reset_mock()
792 self
.fs
.file_delete
.reset_mock()
794 self
.topic
.delete(fake_session
, did
)
795 self
.db
.del_one
.assert_not_called()
796 self
.msg
.write
.assert_not_called()
797 db_g1_args
= self
.db
.get_one
.call_args
[0]
798 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
799 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
800 db_s1_args
= self
.db
.set_one
.call_args
801 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
802 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
804 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
807 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
810 db_s1_args
[1]["pull_list"],
811 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
812 "Wrong DB pull_list dictionary",
814 self
.fs
.file_delete
.assert_not_called()
817 @patch("osm_nbi.descriptor_topics.yaml")
818 def test_validate_descriptor_changes(self
, mock_yaml
):
819 descriptor_name
= "test_descriptor"
820 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
821 "/tmp/" + str(uuid4()), "a+b"
823 with self
.subTest(i
=1, t
="VNFD has changes in day1-2 config primitive"):
824 old_vnfd
, new_vnfd
= self
.create_desc_temp(db_vnfd_content
)
825 new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
827 ][0]["config-primitive"][0]["name"] = "new_action"
828 mock_yaml
.load
.side_effect
= [old_vnfd
, new_vnfd
]
829 with self
.assertNotRaises(EngineException
):
830 self
.topic
._validate
_descriptor
_changes
(
831 descriptor_name
, "/tmp/", "/tmp:1/"
833 with self
.subTest(i
=2, t
="VNFD sw version changed"):
834 # old vnfd uses the default software version: 1.0
835 new_vnfd
["software-version"] = "1.3"
836 new_vnfd
["sw-image-desc"][0]["name"] = "new-image"
837 mock_yaml
.load
.side_effect
= [old_vnfd
, new_vnfd
]
838 with self
.assertNotRaises(EngineException
):
839 self
.topic
._validate
_descriptor
_changes
(
840 descriptor_name
, "/tmp/", "/tmp:1/"
843 i
=3, t
="VNFD sw version is not changed and mgmt-cp has changed"
845 old_vnfd
, new_vnfd
= self
.create_desc_temp(db_vnfd_content
)
846 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
847 mock_yaml
.load
.side_effect
= [old_vnfd
, new_vnfd
]
848 with self
.assertRaises(EngineException
) as e
:
849 self
.topic
._validate
_descriptor
_changes
(
850 descriptor_name
, "/tmp/", "/tmp:1/"
853 e
.exception
.http_code
,
854 HTTPStatus
.UNPROCESSABLE_ENTITY
,
855 "Wrong HTTP status code",
858 norm("there are disallowed changes in the vnf descriptor"),
859 norm(str(e
.exception
)),
860 "Wrong exception text",
863 def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self
):
864 indata
= deepcopy(db_vnfd_content
)
865 self
.topic
.validate_mgmt_interface_connection_point(indata
)
867 def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
870 indata
= deepcopy(db_vnfd_content
)
871 indata
["ext-cpd"] = []
872 with self
.assertRaises(EngineException
) as e
:
873 self
.topic
.validate_mgmt_interface_connection_point(indata
)
875 e
.exception
.http_code
,
876 HTTPStatus
.UNPROCESSABLE_ENTITY
,
877 "Wrong HTTP status code",
881 "mgmt-cp='{}' must match an existing ext-cpd".format(indata
["mgmt-cp"])
883 norm(str(e
.exception
)),
884 "Wrong exception text",
887 def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self
):
888 indata
= deepcopy(db_vnfd_content
)
889 indata
.pop("mgmt-cp")
890 with self
.assertRaises(EngineException
) as e
:
891 self
.topic
.validate_mgmt_interface_connection_point(indata
)
893 e
.exception
.http_code
,
894 HTTPStatus
.UNPROCESSABLE_ENTITY
,
895 "Wrong HTTP status code",
898 norm("'mgmt-cp' is a mandatory field and it is not defined"),
899 norm(str(e
.exception
)),
900 "Wrong exception text",
903 def test_validate_vdu_internal_connection_points_on_valid_descriptor(self
):
904 indata
= db_vnfd_content
905 vdu
= indata
["vdu"][0]
906 self
.topic
.validate_vdu_internal_connection_points(vdu
)
908 def test_validate_external_connection_points_on_valid_descriptor(self
):
909 indata
= db_vnfd_content
910 self
.topic
.validate_external_connection_points(indata
)
912 def test_validate_external_connection_points_when_missing_internal_connection_point(
915 indata
= deepcopy(db_vnfd_content
)
916 vdu
= indata
["vdu"][0]
918 affected_ext_cpd
= indata
["ext-cpd"][0]
919 with self
.assertRaises(EngineException
) as e
:
920 self
.topic
.validate_external_connection_points(indata
)
922 e
.exception
.http_code
,
923 HTTPStatus
.UNPROCESSABLE_ENTITY
,
924 "Wrong HTTP status code",
928 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
929 affected_ext_cpd
["id"]
932 norm(str(e
.exception
)),
933 "Wrong exception text",
936 def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
939 indata
= deepcopy(db_vnfd_content
)
940 vdu
= indata
["vdu"][0]
944 "virtual-network-interface-requirement": [{"name": "duplicated"}],
946 vdu
["int-cpd"].insert(0, duplicated_cpd
)
947 with self
.assertRaises(EngineException
) as e
:
948 self
.topic
.validate_vdu_internal_connection_points(vdu
)
950 e
.exception
.http_code
,
951 HTTPStatus
.UNPROCESSABLE_ENTITY
,
952 "Wrong HTTP status code",
956 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
957 vdu
["id"], duplicated_cpd
["id"]
960 norm(str(e
.exception
)),
961 "Wrong exception text",
964 def test_validate_external_connection_points_on_duplicated_external_connection_point(
967 indata
= deepcopy(db_vnfd_content
)
969 "id": "vnf-mgmt-ext",
970 "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
972 indata
["ext-cpd"].insert(0, duplicated_cpd
)
973 with self
.assertRaises(EngineException
) as e
:
974 self
.topic
.validate_external_connection_points(indata
)
976 e
.exception
.http_code
,
977 HTTPStatus
.UNPROCESSABLE_ENTITY
,
978 "Wrong HTTP status code",
982 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
986 norm(str(e
.exception
)),
987 "Wrong exception text",
990 def test_validate_internal_virtual_links_on_valid_descriptor(self
):
991 indata
= db_vnfd_content
992 self
.topic
.validate_internal_virtual_links(indata
)
994 def test_validate_internal_virtual_links_on_duplicated_ivld(self
):
995 indata
= deepcopy(db_vnfd_content
)
996 duplicated_vld
= {"id": "internal"}
997 indata
["int-virtual-link-desc"].insert(0, duplicated_vld
)
998 with self
.assertRaises(EngineException
) as e
:
999 self
.topic
.validate_internal_virtual_links(indata
)
1001 e
.exception
.http_code
,
1002 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1003 "Wrong HTTP status code",
1007 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
1008 duplicated_vld
["id"]
1011 norm(str(e
.exception
)),
1012 "Wrong exception text",
1015 def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
1018 indata
= deepcopy(db_vnfd_content
)
1019 vdu
= indata
["vdu"][0]
1020 affected_int_cpd
= vdu
["int-cpd"][0]
1021 affected_int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
1022 with self
.assertRaises(EngineException
) as e
:
1023 self
.topic
.validate_internal_virtual_links(indata
)
1025 e
.exception
.http_code
,
1026 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1027 "Wrong HTTP status code",
1031 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
1032 "int-virtual-link-desc".format(
1034 affected_int_cpd
["id"],
1035 affected_int_cpd
["int-virtual-link-desc"],
1038 norm(str(e
.exception
)),
1039 "Wrong exception text",
1042 def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self
):
1043 indata
= deepcopy(db_vnfd_content
)
1044 affected_ivld_profile
= {"id": "non-existing-int-virtual-link-desc"}
1045 df
= indata
["df"][0]
1046 df
["virtual-link-profile"] = [affected_ivld_profile
]
1047 with self
.assertRaises(EngineException
) as e
:
1048 self
.topic
.validate_internal_virtual_links(indata
)
1050 e
.exception
.http_code
,
1051 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1052 "Wrong HTTP status code",
1056 "df[id='{}']:virtual-link-profile='{}' must match an existing "
1057 "int-virtual-link-desc".format(df
["id"], affected_ivld_profile
["id"])
1059 norm(str(e
.exception
)),
1060 "Wrong exception text",
1063 def test_validate_monitoring_params_on_valid_descriptor(self
):
1064 indata
= db_vnfd_content
1065 self
.topic
.validate_monitoring_params(indata
)
1067 def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self
):
1068 indata
= deepcopy(db_vnfd_content
)
1069 duplicated_mp
= {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1070 affected_ivld
= indata
["int-virtual-link-desc"][0]
1071 affected_ivld
["monitoring-parameters"] = [duplicated_mp
, duplicated_mp
]
1072 with self
.assertRaises(EngineException
) as e
:
1073 self
.topic
.validate_monitoring_params(indata
)
1075 e
.exception
.http_code
,
1076 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1077 "Wrong HTTP status code",
1081 "Duplicated monitoring-parameter id in "
1082 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1083 affected_ivld
["id"], duplicated_mp
["id"]
1086 norm(str(e
.exception
)),
1087 "Wrong exception text",
1090 def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self
):
1091 indata
= deepcopy(db_vnfd_content
)
1093 "id": "dataVM_cpu_util",
1094 "name": "dataVM_cpu_util",
1095 "performance_metric": "cpu",
1097 affected_vdu
= indata
["vdu"][1]
1098 affected_vdu
["monitoring-parameter"].insert(0, duplicated_mp
)
1099 with self
.assertRaises(EngineException
) as e
:
1100 self
.topic
.validate_monitoring_params(indata
)
1102 e
.exception
.http_code
,
1103 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1104 "Wrong HTTP status code",
1108 "Duplicated monitoring-parameter id in "
1109 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1110 affected_vdu
["id"], duplicated_mp
["id"]
1113 norm(str(e
.exception
)),
1114 "Wrong exception text",
1117 def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self
):
1118 indata
= deepcopy(db_vnfd_content
)
1122 "performance_metric": "memory",
1124 affected_df
= indata
["df"][0]
1125 affected_df
["monitoring-parameter"] = [duplicated_mp
, duplicated_mp
]
1126 with self
.assertRaises(EngineException
) as e
:
1127 self
.topic
.validate_monitoring_params(indata
)
1129 e
.exception
.http_code
,
1130 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1131 "Wrong HTTP status code",
1135 "Duplicated monitoring-parameter id in "
1136 "df[id='{}']:monitoring-parameter[id='{}']".format(
1137 affected_df
["id"], duplicated_mp
["id"]
1140 norm(str(e
.exception
)),
1141 "Wrong exception text",
1144 def test_validate_scaling_group_descriptor_on_valid_descriptor(self
):
1145 indata
= db_vnfd_content
1146 self
.topic
.validate_scaling_group_descriptor(indata
)
1148 def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self
):
1149 indata
= deepcopy(db_vnfd_content
)
1150 vdu
= indata
["vdu"][1]
1151 affected_df
= indata
["df"][0]
1152 affected_sa
= affected_df
["scaling-aspect"][0]
1153 affected_sp
= affected_sa
["scaling-policy"][0]
1154 affected_sc
= affected_sp
["scaling-criteria"][0]
1155 vdu
.pop("monitoring-parameter")
1156 with self
.assertRaises(EngineException
) as e
:
1157 self
.topic
.validate_scaling_group_descriptor(indata
)
1159 e
.exception
.http_code
,
1160 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1161 "Wrong HTTP status code",
1165 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1166 "[name='{}']:scaling-criteria[name='{}']: "
1167 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1170 affected_sp
["name"],
1171 affected_sc
["name"],
1172 affected_sc
["vnf-monitoring-param-ref"],
1175 norm(str(e
.exception
)),
1176 "Wrong exception text",
1179 def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self
):
1180 indata
= deepcopy(db_vnfd_content
)
1181 df
= indata
["df"][0]
1182 affected_sa
= df
["scaling-aspect"][0]
1183 indata
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1186 with self
.assertRaises(EngineException
) as e
:
1187 self
.topic
.validate_scaling_group_descriptor(indata
)
1189 e
.exception
.http_code
,
1190 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1191 "Wrong HTTP status code",
1195 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1196 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1197 df
["id"], affected_sa
["id"]
1200 norm(str(e
.exception
)),
1201 "Wrong exception text",
1204 def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1207 indata
= deepcopy(db_vnfd_content
)
1208 df
= indata
["df"][0]
1209 affected_sa
= df
["scaling-aspect"][0]
1210 affected_sca_primitive
= affected_sa
["scaling-config-action"][0][
1211 "vnf-config-primitive-name-ref"
1213 df
["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1216 with self
.assertRaises(EngineException
) as e
:
1217 self
.topic
.validate_scaling_group_descriptor(indata
)
1219 e
.exception
.http_code
,
1220 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1221 "Wrong HTTP status code",
1225 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1226 "config-primitive-name-ref='{}' does not match any "
1227 "day1-2 configuration:config-primitive:name".format(
1228 df
["id"], affected_sa
["id"], affected_sca_primitive
1231 norm(str(e
.exception
)),
1232 "Wrong exception text",
1235 def test_new_vnfd_revision(self
):
1236 did
= db_vnfd_content
["_id"]
1237 self
.fs
.get_params
.return_value
= {}
1238 self
.fs
.file_exists
.return_value
= False
1239 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1240 "/tmp/" + str(uuid4()), "a+b"
1242 test_vnfd
= deepcopy(db_vnfd_content
)
1243 del test_vnfd
["_id"]
1244 del test_vnfd
["_admin"]
1245 self
.db
.create
.return_value
= did
1247 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1248 db_args
= self
.db
.create
.call_args
[0]
1249 self
.assertEqual(db_args
[1]['_admin']['revision'], 0,
1250 "New package should be at revision 0")
1252 @patch("osm_nbi.descriptor_topics.shutil")
1253 @patch("osm_nbi.descriptor_topics.os.rename")
1254 def test_update_vnfd(self
, mock_rename
, mock_shutil
):
1256 did
= db_vnfd_content
["_id"]
1258 self
.fs
.get_params
.return_value
= {}
1259 self
.fs
.file_exists
.return_value
= False
1260 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1261 "/tmp/" + str(uuid4()), "a+b"
1263 new_vnfd
= deepcopy(db_vnfd_content
)
1265 self
.db
.create
.return_value
= did
1267 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1268 del new_vnfd
["vdu"][0]["cloud-init-file"]
1269 del new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1270 "day1-2"][0]["execution-environment-list"][0]["juju"]
1275 "_admin": deepcopy(db_vnfd_content
["_admin"])
1277 old_vnfd
["_admin"]["revision"] = old_revision
1278 self
.db
.get_one
.side_effect
= [
1282 self
.topic
.upload_content(
1283 fake_session
, did
, new_vnfd
, {}, {"Content-Type": []}
1286 db_args
= self
.db
.replace
.call_args
[0]
1287 self
.assertEqual(db_args
[2]['_admin']['revision'], old_revision
+ 1,
1288 "Revision should increment")
1291 class Test_NsdTopic(TestCase
):
1293 def setUpClass(cls
):
1294 cls
.test_name
= "test-nsd-topic"
1297 def tearDownClass(cls
):
1301 self
.db
= Mock(dbbase
.DbBase())
1302 self
.fs
= Mock(fsbase
.FsBase())
1303 self
.msg
= Mock(msgbase
.MsgBase())
1304 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1305 self
.topic
= NsdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1306 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1309 def assertNotRaises(self
, exception_type
):
1312 except exception_type
:
1313 raise self
.failureException("{} raised".format(exception_type
.__name
__))
1315 def create_desc_temp(self
, template
):
1316 old_desc
= deepcopy(template
)
1317 new_desc
= deepcopy(template
)
1318 return old_desc
, new_desc
1320 @patch("osm_nbi.descriptor_topics.shutil")
1321 @patch("osm_nbi.descriptor_topics.os.rename")
1322 def test_new_nsd(self
, mock_rename
, mock_shutil
):
1323 did
= db_nsd_content
["_id"]
1324 self
.fs
.get_params
.return_value
= {}
1325 self
.fs
.file_exists
.return_value
= False
1326 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1327 "/tmp/" + str(uuid4()), "a+b"
1329 test_nsd
= deepcopy(db_nsd_content
)
1331 del test_nsd
["_admin"]
1332 with self
.subTest(i
=1, t
="Normal Creation"):
1333 self
.db
.create
.return_value
= did
1335 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1336 db_args
= self
.db
.create
.call_args
[0]
1337 msg_args
= self
.msg
.write
.call_args
[0]
1338 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1339 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1340 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
1341 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1342 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1343 self
.assertEqual(did2
, did
, "Wrong DB NSD id")
1344 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
1346 db_args
[1]["_admin"]["modified"],
1347 db_args
[1]["_admin"]["created"],
1348 "Wrong modification time",
1351 db_args
[1]["_admin"]["projects_read"],
1353 "Wrong read-only project list",
1356 db_args
[1]["_admin"]["projects_write"],
1358 "Wrong read-write project list",
1362 self
.db
.get_one
.side_effect
= [
1363 {"_id": did
, "_admin": db_nsd_content
["_admin"]},
1366 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1367 self
.topic
.upload_content(
1368 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1370 msg_args
= self
.msg
.write
.call_args
[0]
1371 test_nsd
["_id"] = did
1373 msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic"
1375 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1376 self
.assertEqual(msg_args
[2], test_nsd
, "Wrong message content")
1377 db_args
= self
.db
.get_one
.mock_calls
[0][1]
1378 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1379 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB NSD id")
1380 db_args
= self
.db
.replace
.call_args
[0]
1381 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1382 self
.assertEqual(db_args
[1], did
, "Wrong DB NSD id")
1383 admin
= db_args
[2]["_admin"]
1384 db_admin
= db_nsd_content
["_admin"]
1386 admin
["created"], db_admin
["created"], "Wrong creation time"
1389 admin
["modified"], db_admin
["created"], "Wrong modification time"
1392 admin
["projects_read"],
1393 db_admin
["projects_read"],
1394 "Wrong read-only project list",
1397 admin
["projects_write"],
1398 db_admin
["projects_write"],
1399 "Wrong read-write project list",
1402 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1405 admin
["operationalState"], "ENABLED", "Wrong operational state"
1407 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
1408 storage
= admin
["storage"]
1409 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
1411 storage
["descriptor"], "package", "Wrong storage descriptor"
1413 compare_desc(self
, test_nsd
, db_args
[2], "NSD")
1414 revision_args
= self
.db
.create
.call_args
[0]
1416 revision_args
[0], self
.topic
.topic
+ "_revisions", "Wrong topic"
1419 revision_args
[1]["id"], db_args
[2]["id"], "Wrong revision id"
1422 revision_args
[1]["_id"],
1423 db_args
[2]["_id"] + ":1",
1424 "Wrong revision _id"
1429 self
.db
.get_one
.side_effect
= (
1430 lambda table
, filter, fail_on_empty
=None, fail_on_more
=None: {
1432 "_admin": db_nsd_content
["_admin"],
1435 with self
.subTest(i
=2, t
="Check Pyangbind Validation: required properties"):
1436 tmp
= test_nsd
["id"]
1439 with self
.assertRaises(
1440 EngineException
, msg
="Accepted NSD with a missing required property"
1442 self
.topic
.upload_content(
1443 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1446 e
.exception
.http_code
,
1447 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1448 "Wrong HTTP status code",
1451 norm("Error in pyangbind validation: '{}'".format("id")),
1452 norm(str(e
.exception
)),
1453 "Wrong exception text",
1456 test_nsd
["id"] = tmp
1457 with self
.subTest(i
=3, t
="Check Pyangbind Validation: additional properties"):
1458 test_nsd
["extra-property"] = 0
1460 with self
.assertRaises(
1461 EngineException
, msg
="Accepted NSD with an additional property"
1463 self
.topic
.upload_content(
1464 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1467 e
.exception
.http_code
,
1468 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1469 "Wrong HTTP status code",
1473 "Error in pyangbind validation: {} ({})".format(
1474 "json object contained a key that did not exist",
1478 norm(str(e
.exception
)),
1479 "Wrong exception text",
1482 del test_nsd
["extra-property"]
1483 with self
.subTest(i
=4, t
="Check Pyangbind Validation: property types"):
1484 tmp
= test_nsd
["designer"]
1485 test_nsd
["designer"] = {"key": 0}
1487 with self
.assertRaises(
1488 EngineException
, msg
="Accepted NSD with a wrongly typed property"
1490 self
.topic
.upload_content(
1491 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1494 e
.exception
.http_code
,
1495 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1496 "Wrong HTTP status code",
1500 "Error in pyangbind validation: {} ({})".format(
1501 "json object contained a key that did not exist", "key"
1504 norm(str(e
.exception
)),
1505 "Wrong exception text",
1508 test_nsd
["designer"] = tmp
1510 i
=5, t
="Check Input Validation: mgmt-network+virtual-link-protocol-data"
1512 df
= test_nsd
["df"][0]
1515 "virtual-link-desc-id": "mgmt",
1516 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1518 df
["virtual-link-profile"] = [mgmt_profile
]
1520 with self
.assertRaises(
1521 EngineException
, msg
="Accepted VLD with mgmt-network+ip-profile"
1523 self
.topic
.upload_content(
1524 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1527 e
.exception
.http_code
,
1528 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1529 "Wrong HTTP status code",
1533 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1534 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1535 df
["id"], mgmt_profile
["id"]
1538 norm(str(e
.exception
)),
1539 "Wrong exception text",
1542 del df
["virtual-link-profile"]
1543 with self
.subTest(i
=6, t
="Check Descriptor Dependencies: vnfd-id[]"):
1544 self
.db
.get_one
.side_effect
= [
1545 {"_id": did
, "_admin": db_nsd_content
["_admin"]},
1548 self
.db
.get_list
.return_value
= []
1550 with self
.assertRaises(
1551 EngineException
, msg
="Accepted wrong VNFD ID reference"
1553 self
.topic
.upload_content(
1554 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1557 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1561 "'vnfd-id'='{}' references a non existing vnfd".format(
1562 test_nsd
["vnfd-id"][0]
1565 norm(str(e
.exception
)),
1566 "Wrong exception text",
1572 t
="Check Descriptor Dependencies: "
1573 "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]",
1575 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1576 df
= test_nsd
["df"][0]
1577 affected_vnf_profile
= df
["vnf-profile"][0]
1578 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
1579 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
1580 self
.db
.get_one
.side_effect
= [
1581 {"_id": did
, "_admin": db_nsd_content
["_admin"]},
1584 self
.db
.get_list
.return_value
= [vnfd_descriptor
]
1586 with self
.assertRaises(
1587 EngineException
, msg
="Accepted wrong VLD CP reference"
1589 self
.topic
.upload_content(
1590 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1593 e
.exception
.http_code
,
1594 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1595 "Wrong HTTP status code",
1599 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1600 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1601 "non existing ext-cpd:id inside vnfd '{}'".format(
1603 affected_vnf_profile
["id"],
1604 affected_virtual_link
["virtual-link-profile-id"],
1606 vnfd_descriptor
["id"],
1609 norm(str(e
.exception
)),
1610 "Wrong exception text",
1616 def test_edit_nsd(self
):
1617 nsd_content
= deepcopy(db_nsd_content
)
1618 did
= nsd_content
["_id"]
1619 self
.fs
.file_exists
.return_value
= True
1620 self
.fs
.dir_ls
.return_value
= True
1621 with self
.subTest(i
=1, t
="Normal Edition"):
1623 self
.db
.get_one
.side_effect
= [deepcopy(nsd_content
), None]
1624 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1625 data
= {"id": "new-nsd-id", "name": "new-nsd-name"}
1626 self
.topic
.edit(fake_session
, did
, data
)
1627 db_args
= self
.db
.replace
.call_args
[0]
1628 msg_args
= self
.msg
.write
.call_args
[0]
1630 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1631 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1632 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
1633 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1634 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
1636 db_args
[2]["_admin"]["created"],
1637 nsd_content
["_admin"]["created"],
1638 "Wrong creation time",
1641 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
1644 db_args
[2]["_admin"]["projects_read"],
1645 nsd_content
["_admin"]["projects_read"],
1646 "Wrong read-only project list",
1649 db_args
[2]["_admin"]["projects_write"],
1650 nsd_content
["_admin"]["projects_write"],
1651 "Wrong read-write project list",
1653 self
.assertEqual(db_args
[2]["id"], data
["id"], "Wrong NSD ID")
1654 self
.assertEqual(db_args
[2]["name"], data
["name"], "Wrong NSD Name")
1655 with self
.subTest(i
=2, t
="Conflict on Edit"):
1656 data
= {"id": "fake-nsd-id", "name": "new-nsd-name"}
1657 self
.db
.get_one
.side_effect
= [
1659 {"_id": str(uuid4()), "id": data
["id"]},
1661 with self
.assertRaises(
1662 EngineException
, msg
="Accepted existing NSD ID"
1664 self
.topic
.edit(fake_session
, did
, data
)
1666 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1670 "{} with id '{}' already exists for this project".format(
1674 norm(str(e
.exception
)),
1675 "Wrong exception text",
1677 with self
.subTest(i
=3, t
="Check Envelope"):
1678 data
= {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
1679 self
.db
.get_one
.side_effect
= [nsd_content
, None]
1680 with self
.assertRaises(
1681 EngineException
, msg
="Accepted NSD with wrong envelope"
1683 self
.topic
.edit(fake_session
, did
, data
, content
=nsd_content
)
1685 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
1688 "'nsd' must be a list of only one element",
1689 norm(str(e
.exception
)),
1690 "Wrong exception text",
1694 def test_delete_nsd(self
):
1695 did
= db_nsd_content
["_id"]
1696 self
.db
.get_one
.return_value
= db_nsd_content
1697 p_id
= db_nsd_content
["_admin"]["projects_read"][0]
1698 with self
.subTest(i
=1, t
="Normal Deletion"):
1699 self
.db
.get_list
.return_value
= []
1700 self
.db
.del_one
.return_value
= {"deleted": 1}
1701 self
.topic
.delete(fake_session
, did
)
1702 db_args
= self
.db
.del_one
.call_args
[0]
1703 msg_args
= self
.msg
.write
.call_args
[0]
1704 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1705 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
1706 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1707 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1708 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
1710 db_args
[1]["_admin.projects_write.cont"],
1714 db_g1_args
= self
.db
.get_one
.call_args
[0]
1715 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1716 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB NSD ID")
1717 db_gl_calls
= self
.db
.get_list
.call_args_list
1718 self
.assertEqual(db_gl_calls
[0][0][0], "nsrs", "Wrong DB topic")
1719 # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID") # Filter changed after call
1720 self
.assertEqual(db_gl_calls
[1][0][0], "nsts", "Wrong DB topic")
1722 db_gl_calls
[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
1723 db_nsd_content
["id"],
1724 "Wrong DB NSD netslice-subnet nsd-ref",
1726 self
.db
.set_one
.assert_not_called()
1727 fs_del_calls
= self
.fs
.file_delete
.call_args_list
1728 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
1729 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
1730 with self
.subTest(i
=2, t
="Conflict on Delete - NSD in use by nsr"):
1731 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-nsr"}]
1732 with self
.assertRaises(
1733 EngineException
, msg
="Accepted NSD in use by NSR"
1735 self
.topic
.delete(fake_session
, did
)
1737 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1740 "there is at least one ns instance using this descriptor",
1741 norm(str(e
.exception
)),
1742 "Wrong exception text",
1744 with self
.subTest(i
=3, t
="Conflict on Delete - NSD in use by NST"):
1745 self
.db
.get_list
.side_effect
= [
1747 [{"_id": str(uuid4()), "name": "fake-nst"}],
1749 with self
.assertRaises(
1750 EngineException
, msg
="Accepted NSD in use by NST"
1752 self
.topic
.delete(fake_session
, did
)
1754 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1757 "there is at least one netslice template referencing this descriptor",
1758 norm(str(e
.exception
)),
1759 "Wrong exception text",
1761 with self
.subTest(i
=4, t
="Non-existent NSD"):
1762 excp_msg
= "Not found any {} with filter='{}'".format("NSD", {"_id": did
})
1763 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
1764 with self
.assertRaises(
1765 DbException
, msg
="Accepted non-existent NSD ID"
1767 self
.topic
.delete(fake_session
, did
)
1769 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1772 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
1774 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
1775 db_nsd_content
["_admin"]["projects_read"].append("other_project")
1776 self
.db
.get_one
= Mock(return_value
=db_nsd_content
)
1777 self
.db
.get_list
= Mock(return_value
=[])
1778 self
.msg
.write
.reset_mock()
1779 self
.db
.del_one
.reset_mock()
1780 self
.fs
.file_delete
.reset_mock()
1782 self
.topic
.delete(fake_session
, did
)
1783 self
.db
.del_one
.assert_not_called()
1784 self
.msg
.write
.assert_not_called()
1785 db_g1_args
= self
.db
.get_one
.call_args
[0]
1786 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1787 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
1788 db_s1_args
= self
.db
.set_one
.call_args
1789 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
1790 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
1792 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1795 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
1798 db_s1_args
[1]["pull_list"],
1799 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
1800 "Wrong DB pull_list dictionary",
1802 self
.fs
.file_delete
.assert_not_called()
1805 @patch("osm_nbi.descriptor_topics.yaml")
1806 def test_validate_descriptor_changes(self
, mock_yaml
):
1807 descriptor_name
= "test_ns_descriptor"
1808 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1809 "/tmp/" + str(uuid4()), "a+b"
1812 i
=1, t
="NSD has changes in ns-configuration:config-primitive"
1814 old_nsd
, new_nsd
= self
.create_desc_temp(db_nsd_content
)
1816 {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
1819 {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
1821 mock_yaml
.load
.side_effect
= [old_nsd
, new_nsd
]
1822 with self
.assertNotRaises(EngineException
):
1823 self
.topic
._validate
_descriptor
_changes
(
1824 descriptor_name
, "/tmp", "/tmp:1"
1826 with self
.subTest(i
=2, t
="NSD name has changed"):
1827 old_nsd
, new_nsd
= self
.create_desc_temp(db_nsd_content
)
1828 new_nsd
["name"] = "nscharm-ns2"
1829 mock_yaml
.load
.side_effect
= [old_nsd
, new_nsd
]
1830 with self
.assertRaises(EngineException
) as e
:
1831 self
.topic
._validate
_descriptor
_changes
(
1832 descriptor_name
, "/tmp", "/tmp:1"
1835 e
.exception
.http_code
,
1836 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1837 "Wrong HTTP status code",
1840 norm("there are disallowed changes in the ns descriptor"),
1841 norm(str(e
.exception
)),
1842 "Wrong exception text",
1845 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
1848 indata
= deepcopy(db_nsd_content
)
1849 vld
= indata
["virtual-link-desc"][0]
1850 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
1854 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
1857 indata
= deepcopy(db_nsd_content
)
1858 vld
= indata
["virtual-link-desc"][0]
1859 df
= indata
["df"][0]
1862 "virtual-link-desc-id": "mgmt",
1863 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1865 df
["virtual-link-profile"] = [affected_vlp
]
1866 with self
.assertRaises(EngineException
) as e
:
1867 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
1871 e
.exception
.http_code
,
1872 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1873 "Wrong HTTP status code",
1877 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1878 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1879 df
["id"], affected_vlp
["id"]
1882 norm(str(e
.exception
)),
1883 "Wrong exception text",
1886 def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self
):
1887 indata
= deepcopy(db_nsd_content
)
1888 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
1890 def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self
):
1891 indata
= deepcopy(db_nsd_content
)
1892 df
= indata
["df"][0]
1893 affected_vnf_profile
= df
["vnf-profile"][0]
1894 indata
["vnfd-id"] = ["non-existing-vnfd"]
1895 with self
.assertRaises(EngineException
) as e
:
1896 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
1898 e
.exception
.http_code
,
1899 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1900 "Wrong HTTP status code",
1904 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
1905 "does not match any vnfd-id".format(
1907 affected_vnf_profile
["id"],
1908 affected_vnf_profile
["vnfd-id"],
1911 norm(str(e
.exception
)),
1912 "Wrong exception text",
1915 def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
1918 nsd_descriptor
= deepcopy(db_nsd_content
)
1919 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1920 df
= nsd_descriptor
["df"][0]
1921 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
1922 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
1926 def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
1929 nsd_descriptor
= deepcopy(db_nsd_content
)
1930 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1931 df
= nsd_descriptor
["df"][0]
1932 affected_vnf_profile
= df
["vnf-profile"][0]
1933 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
1934 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
1935 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
1936 with self
.assertRaises(EngineException
) as e
:
1937 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
1941 e
.exception
.http_code
,
1942 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1943 "Wrong HTTP status code",
1947 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1948 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1949 "non existing ext-cpd:id inside vnfd '{}'".format(
1951 affected_vnf_profile
["id"],
1952 affected_virtual_link
["virtual-link-profile-id"],
1954 vnfd_descriptor
["id"],
1957 norm(str(e
.exception
)),
1958 "Wrong exception text",
1961 def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self
):
1962 nsd_descriptor
= deepcopy(db_nsd_content
)
1963 invalid_vnfd_id
= "invalid-vnfd-id"
1964 nsd_descriptor
["id"] = "invalid-vnfd-id-ns"
1965 nsd_descriptor
["vnfd-id"][0] = invalid_vnfd_id
1966 nsd_descriptor
["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
1967 nsd_descriptor
["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
1968 with self
.assertRaises(EngineException
) as e
:
1969 self
.db
.get_list
.return_value
= []
1970 nsd_descriptor
= self
.topic
.check_conflict_on_edit(
1971 fake_session
, nsd_descriptor
, [], "id"
1974 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1978 "Descriptor error at 'vnfd-id'='{}' references a non "
1979 "existing vnfd".format(invalid_vnfd_id
)
1981 norm(str(e
.exception
)),
1982 "Wrong exception text",
1986 if __name__
== "__main__":