00b18a58fa4f055343c6db48b13f13ffd953b4aa
2 # -*- coding: utf-8 -*-
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
17 __author__
= "Pedro de la Cruz Ramos, pedro.delacruzramos@altran.com"
18 __date__
= "2019-11-20"
20 from contextlib
import contextmanager
22 from unittest
import TestCase
23 from unittest
.mock
import Mock
, patch
24 from uuid
import uuid4
25 from http
import HTTPStatus
26 from copy
import deepcopy
28 from osm_common
import dbbase
, fsbase
, msgbase
29 from osm_nbi
import authconn
30 from osm_nbi
.tests
.test_pkg_descriptors
import (
34 vnfd_exploit_fixed_text
,
36 from osm_nbi
.descriptor_topics
import VnfdTopic
, NsdTopic
37 from osm_nbi
.engine
import EngineException
38 from osm_common
.dbbase
import DbException
42 import collections
.abc
44 collections
.MutableSequence
= collections
.abc
.MutableSequence
46 test_name
= "test-user"
47 db_vnfd_content
= yaml
.load(db_vnfds_text
, Loader
=yaml
.Loader
)[0]
48 db_nsd_content
= yaml
.load(db_nsds_text
, Loader
=yaml
.Loader
)[0]
49 test_pid
= db_vnfd_content
["_admin"]["projects_read"][0]
51 "username": test_name
,
52 "project_id": (test_pid
,),
57 "allow_show_user_project_role": True,
59 UUID
= "00000000-0000-0000-0000-000000000000"
63 return {"projects_read": []}
66 def setup_mock_fs(fs
):
68 fs
.get_params
.return_value
= {}
69 fs
.file_exists
.return_value
= False
70 fs
.file_open
.side_effect
= lambda path
, mode
: tempfile
.TemporaryFile(mode
="a+b")
74 """Normalize string for checking"""
75 return " ".join(s
.strip().split()).lower()
78 def compare_desc(tc
, d1
, d2
, k
):
80 Compare two descriptors
81 We need this function because some methods are adding/removing items to/from the descriptors
82 before they are stored in the database, so the original and stored versions will differ
83 What we check is that COMMON LEAF ITEMS are equal
84 Lists of different length are not compared
85 :param tc: Test Case wich provides context (in particular the assert* methods)
86 :param d1,d2: Descriptors to be compared
87 :param key/item being compared
90 if isinstance(d1
, dict) and isinstance(d2
, dict):
93 compare_desc(tc
, d1
[key
], d2
[key
], k
+ "[{}]".format(key
))
94 elif isinstance(d1
, list) and isinstance(d2
, list) and len(d1
) == len(d2
):
95 for i
in range(len(d1
)):
96 compare_desc(tc
, d1
[i
], d2
[i
], k
+ "[{}]".format(i
))
98 tc
.assertEqual(d1
, d2
, "Wrong descriptor content: {}".format(k
))
101 class Test_VnfdTopic(TestCase
):
104 cls
.test_name
= "test-vnfd-topic"
107 def tearDownClass(cls
):
111 self
.db
= Mock(dbbase
.DbBase())
112 self
.fs
= Mock(fsbase
.FsBase())
113 self
.msg
= Mock(msgbase
.MsgBase())
114 self
.auth
= Mock(authconn
.Authconn(None, None, None))
115 self
.topic
= VnfdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
116 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
119 def assertNotRaises(self
, exception_type
=Exception):
122 except exception_type
:
123 raise self
.failureException("{} raised".format(exception_type
.__name
__))
125 def create_desc_temp(self
, template
):
126 old_desc
= deepcopy(template
)
127 new_desc
= deepcopy(template
)
128 return old_desc
, new_desc
130 def prepare_vnfd_creation(self
):
131 setup_mock_fs(self
.fs
)
132 test_vnfd
= deepcopy(db_vnfd_content
)
133 did
= db_vnfd_content
["_id"]
134 self
.db
.create
.return_value
= did
135 self
.db
.get_one
.side_effect
= [
136 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
139 return did
, test_vnfd
141 def prepare_vnfd(self
, vnfd_text
):
142 setup_mock_fs(self
.fs
)
143 test_vnfd
= yaml
.safe_load(vnfd_text
)
144 self
.db
.create
.return_value
= UUID
145 self
.db
.get_one
.side_effect
= [
146 {"_id": UUID
, "_admin": admin_value()},
149 return UUID
, test_vnfd
151 def prepare_test_vnfd(self
, test_vnfd
):
153 del test_vnfd
["_admin"]
154 del test_vnfd
["vdu"][0]["cloud-init-file"]
155 del test_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
157 ][0]["execution-environment-list"][0]["juju"]
160 @patch("osm_nbi.descriptor_topics.shutil")
161 @patch("osm_nbi.descriptor_topics.os.rename")
162 def test_new_vnfd_normal_creation(self
, mock_rename
, mock_shutil
):
163 did
, test_vnfd
= self
.prepare_vnfd_creation()
164 test_vnfd
= self
.prepare_test_vnfd(test_vnfd
)
166 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
167 db_args
= self
.db
.create
.call_args
[0]
168 msg_args
= self
.msg
.write
.call_args
[0]
170 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
171 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
172 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
173 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
174 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
175 self
.assertEqual(did2
, did
, "Wrong DB VNFD id")
176 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
178 db_args
[1]["_admin"]["modified"],
179 db_args
[1]["_admin"]["created"],
180 "Wrong modification time",
183 db_args
[1]["_admin"]["projects_read"],
185 "Wrong read-only project list",
188 db_args
[1]["_admin"]["projects_write"],
190 "Wrong read-write project list",
193 self
.db
.get_one
.side_effect
= [
194 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
198 self
.topic
.upload_content(
199 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
201 msg_args
= self
.msg
.write
.call_args
[0]
202 test_vnfd
["_id"] = did
203 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
204 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
205 self
.assertEqual(msg_args
[2], test_vnfd
, "Wrong message content")
207 db_args
= self
.db
.get_one
.mock_calls
[0][1]
208 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
209 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB VNFD id")
211 db_args
= self
.db
.replace
.call_args
[0]
212 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
213 self
.assertEqual(db_args
[1], did
, "Wrong DB VNFD id")
215 admin
= db_args
[2]["_admin"]
216 db_admin
= deepcopy(db_vnfd_content
["_admin"])
217 self
.assertEqual(admin
["type"], "vnfd", "Wrong descriptor type")
218 self
.assertEqual(admin
["created"], db_admin
["created"], "Wrong creation time")
220 admin
["modified"], db_admin
["created"], "Wrong modification time"
223 admin
["projects_read"],
224 db_admin
["projects_read"],
225 "Wrong read-only project list",
228 admin
["projects_write"],
229 db_admin
["projects_write"],
230 "Wrong read-write project list",
233 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
236 admin
["operationalState"], "ENABLED", "Wrong operational state"
238 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
240 storage
= admin
["storage"]
241 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
242 self
.assertEqual(storage
["descriptor"], "package", "Wrong storage descriptor")
243 self
.assertEqual(admin
["revision"], 1, "Wrong revision number")
244 compare_desc(self
, test_vnfd
, db_args
[2], "VNFD")
246 @patch("osm_nbi.descriptor_topics.shutil")
247 @patch("osm_nbi.descriptor_topics.os.rename")
248 def test_new_vnfd_exploit(self
, mock_rename
, mock_shutil
):
249 id, test_vnfd
= self
.prepare_vnfd(vnfd_exploit_text
)
251 with self
.assertRaises(EngineException
):
252 self
.topic
.upload_content(
253 fake_session
, id, test_vnfd
, {}, {"Content-Type": []}
256 @patch("osm_nbi.descriptor_topics.shutil")
257 @patch("osm_nbi.descriptor_topics.os.rename")
258 def test_new_vnfd_valid_helm_chart(self
, mock_rename
, mock_shutil
):
259 id, test_vnfd
= self
.prepare_vnfd(vnfd_exploit_fixed_text
)
261 with self
.assertNotRaises():
262 self
.topic
.upload_content(
263 fake_session
, id, test_vnfd
, {}, {"Content-Type": []}
266 @patch("osm_nbi.descriptor_topics.shutil")
267 @patch("osm_nbi.descriptor_topics.os.rename")
268 def test_new_vnfd(self
, mock_rename
, mock_shutil
):
269 did
= db_vnfd_content
["_id"]
271 self
.fs
.get_params
.return_value
= {}
272 self
.fs
.file_exists
.return_value
= False
273 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
274 "/tmp/" + str(uuid4()), "a+b"
276 test_vnfd
= deepcopy(db_vnfd_content
)
278 del test_vnfd
["_admin"]
279 with self
.subTest(i
=1, t
="Normal Creation"):
280 self
.db
.create
.return_value
= did
282 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
283 db_args
= self
.db
.create
.call_args
[0]
284 msg_args
= self
.msg
.write
.call_args
[0]
285 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
286 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
287 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
288 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
289 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
290 self
.assertEqual(did2
, did
, "Wrong DB VNFD id")
291 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
293 db_args
[1]["_admin"]["modified"],
294 db_args
[1]["_admin"]["created"],
295 "Wrong modification time",
298 db_args
[1]["_admin"]["projects_read"],
300 "Wrong read-only project list",
303 db_args
[1]["_admin"]["projects_write"],
305 "Wrong read-write project list",
307 tmp1
= test_vnfd
["vdu"][0]["cloud-init-file"]
308 tmp2
= test_vnfd
["df"][0]["lcm-operations-configuration"][
309 "operate-vnf-op-config"
310 ]["day1-2"][0]["execution-environment-list"][0]["juju"]
311 del test_vnfd
["vdu"][0]["cloud-init-file"]
312 del test_vnfd
["df"][0]["lcm-operations-configuration"][
313 "operate-vnf-op-config"
314 ]["day1-2"][0]["execution-environment-list"][0]["juju"]
316 self
.db
.get_one
.side_effect
= [
317 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
320 self
.topic
.upload_content(
321 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
323 msg_args
= self
.msg
.write
.call_args
[0]
324 test_vnfd
["_id"] = did
326 msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic"
328 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
329 self
.assertEqual(msg_args
[2], test_vnfd
, "Wrong message content")
330 db_args
= self
.db
.get_one
.mock_calls
[0][1]
331 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
332 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB VNFD id")
333 db_args
= self
.db
.replace
.call_args
[0]
334 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
335 self
.assertEqual(db_args
[1], did
, "Wrong DB VNFD id")
336 admin
= db_args
[2]["_admin"]
337 db_admin
= deepcopy(db_vnfd_content
["_admin"])
338 self
.assertEqual(admin
["type"], "vnfd", "Wrong descriptor type")
340 admin
["created"], db_admin
["created"], "Wrong creation time"
343 admin
["modified"], db_admin
["created"], "Wrong modification time"
346 admin
["projects_read"],
347 db_admin
["projects_read"],
348 "Wrong read-only project list",
351 admin
["projects_write"],
352 db_admin
["projects_write"],
353 "Wrong read-write project list",
356 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
359 admin
["operationalState"], "ENABLED", "Wrong operational state"
361 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
362 storage
= admin
["storage"]
363 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
365 storage
["descriptor"], "package", "Wrong storage descriptor"
368 admin
["revision"], 1, "Wrong revision number"
371 compare_desc(self
, test_vnfd
, db_args
[2], "VNFD")
373 test_vnfd
["vdu"][0]["cloud-init-file"] = tmp1
374 test_vnfd
["df"][0]["lcm-operations-configuration"][
375 "operate-vnf-op-config"
376 ]["day1-2"][0]["execution-environment-list"][0]["juju"] = tmp2
377 self
.db
.get_one
.side_effect
= (
378 lambda table
, filter, fail_on_empty
=None, fail_on_more
=None: {
380 "_admin": deepcopy(db_vnfd_content
["_admin"]),
383 with self
.subTest(i
=2, t
="Check Pyangbind Validation: additional properties"):
384 test_vnfd
["extra-property"] = 0
386 with self
.assertRaises(
387 EngineException
, msg
="Accepted VNFD with an additional property"
389 self
.topic
.upload_content(
390 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
393 e
.exception
.http_code
,
394 HTTPStatus
.UNPROCESSABLE_ENTITY
,
395 "Wrong HTTP status code",
399 "Error in pyangbind validation: {} ({})".format(
400 "json object contained a key that did not exist",
404 norm(str(e
.exception
)),
405 "Wrong exception text",
407 db_args
= self
.db
.replace
.call_args
[0]
408 admin
= db_args
[2]["_admin"]
410 admin
["revision"], 1, "Wrong revision number"
414 del test_vnfd
["extra-property"]
415 with self
.subTest(i
=3, t
="Check Pyangbind Validation: property types"):
416 tmp
= test_vnfd
["product-name"]
417 test_vnfd
["product-name"] = {"key": 0}
419 with self
.assertRaises(
420 EngineException
, msg
="Accepted VNFD with a wrongly typed property"
422 self
.topic
.upload_content(
423 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
426 e
.exception
.http_code
,
427 HTTPStatus
.UNPROCESSABLE_ENTITY
,
428 "Wrong HTTP status code",
432 "Error in pyangbind validation: {} ({})".format(
433 "json object contained a key that did not exist", "key"
436 norm(str(e
.exception
)),
437 "Wrong exception text",
440 test_vnfd
["product-name"] = tmp
441 with self
.subTest(i
=4, t
="Check Input Validation: cloud-init"):
442 with self
.assertRaises(
443 EngineException
, msg
="Accepted non-existent cloud_init file"
445 self
.topic
.upload_content(
446 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
449 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
453 "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format(
454 "cloud-init", test_vnfd
["id"], test_vnfd
["vdu"][0]["id"]
457 norm(str(e
.exception
)),
458 "Wrong exception text",
460 with self
.subTest(i
=5, t
="Check Input Validation: day1-2 configuration[juju]"):
461 del test_vnfd
["vdu"][0]["cloud-init-file"]
462 with self
.assertRaises(
463 EngineException
, msg
="Accepted non-existent charm in VNF configuration"
465 self
.topic
.upload_content(
466 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
469 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
473 "{} defined in vnf[id={}] but not present in package".format(
474 "charm", test_vnfd
["id"]
477 norm(str(e
.exception
)),
478 "Wrong exception text",
480 del test_vnfd
["df"][0]["lcm-operations-configuration"][
481 "operate-vnf-op-config"
482 ]["day1-2"][0]["execution-environment-list"][0]["juju"]
483 with self
.subTest(i
=6, t
="Check Input Validation: mgmt-cp"):
484 tmp
= test_vnfd
["mgmt-cp"]
485 del test_vnfd
["mgmt-cp"]
487 with self
.assertRaises(
488 EngineException
, msg
="Accepted VNFD without management interface"
490 self
.topic
.upload_content(
491 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
494 e
.exception
.http_code
,
495 HTTPStatus
.UNPROCESSABLE_ENTITY
,
496 "Wrong HTTP status code",
500 "'{}' is a mandatory field and it is not defined".format(
504 norm(str(e
.exception
)),
505 "Wrong exception text",
508 test_vnfd
["mgmt-cp"] = tmp
509 with self
.subTest(i
=7, t
="Check Input Validation: mgmt-cp connection point"):
510 tmp
= test_vnfd
["mgmt-cp"]
511 test_vnfd
["mgmt-cp"] = "wrong-cp"
513 with self
.assertRaises(
514 EngineException
, msg
="Accepted wrong mgmt-cp connection point"
516 self
.topic
.upload_content(
517 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
520 e
.exception
.http_code
,
521 HTTPStatus
.UNPROCESSABLE_ENTITY
,
522 "Wrong HTTP status code",
526 "mgmt-cp='{}' must match an existing ext-cpd".format(
530 norm(str(e
.exception
)),
531 "Wrong exception text",
534 test_vnfd
["mgmt-cp"] = tmp
535 with self
.subTest(i
=8, t
="Check Input Validation: vdu int-cpd"):
536 ext_cpd
= test_vnfd
["ext-cpd"][1]
537 tmp
= ext_cpd
["int-cpd"]["cpd"]
538 ext_cpd
["int-cpd"]["cpd"] = "wrong-cpd"
540 with self
.assertRaises(
542 msg
="Accepted wrong ext-cpd internal connection point",
544 self
.topic
.upload_content(
545 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
548 e
.exception
.http_code
,
549 HTTPStatus
.UNPROCESSABLE_ENTITY
,
550 "Wrong HTTP status code",
554 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
558 norm(str(e
.exception
)),
559 "Wrong exception text",
562 ext_cpd
["int-cpd"]["cpd"] = tmp
563 with self
.subTest(i
=9, t
="Check Input Validation: Duplicated VLD"):
564 test_vnfd
["int-virtual-link-desc"].insert(0, {"id": "internal"})
566 with self
.assertRaises(
567 EngineException
, msg
="Accepted duplicated VLD name"
569 self
.topic
.upload_content(
570 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
573 e
.exception
.http_code
,
574 HTTPStatus
.UNPROCESSABLE_ENTITY
,
575 "Wrong HTTP status code",
579 "identifier id '{}' is not unique".format(
580 test_vnfd
["int-virtual-link-desc"][0]["id"]
583 norm(str(e
.exception
)),
584 "Wrong exception text",
587 del test_vnfd
["int-virtual-link-desc"][0]
588 with self
.subTest(i
=10, t
="Check Input Validation: vdu int-virtual-link-desc"):
589 vdu
= test_vnfd
["vdu"][0]
590 int_cpd
= vdu
["int-cpd"][1]
591 tmp
= int_cpd
["int-virtual-link-desc"]
592 int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
594 with self
.assertRaises(
595 EngineException
, msg
="Accepted int-virtual-link-desc"
597 self
.topic
.upload_content(
598 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
601 e
.exception
.http_code
,
602 HTTPStatus
.UNPROCESSABLE_ENTITY
,
603 "Wrong HTTP status code",
607 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
608 "int-virtual-link-desc".format(
609 vdu
["id"], int_cpd
["id"], int_cpd
["int-virtual-link-desc"]
612 norm(str(e
.exception
)),
613 "Wrong exception text",
616 int_cpd
["int-virtual-link-desc"] = tmp
617 with self
.subTest(i
=11, t
="Check Input Validation: virtual-link-profile)"):
618 fake_ivld_profile
= {"id": "fake-profile-ref", "flavour": "fake-flavour"}
619 df
= test_vnfd
["df"][0]
620 df
["virtual-link-profile"] = [fake_ivld_profile
]
622 with self
.assertRaises(
623 EngineException
, msg
="Accepted non-existent Profile Ref"
625 self
.topic
.upload_content(
626 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
629 e
.exception
.http_code
,
630 HTTPStatus
.UNPROCESSABLE_ENTITY
,
631 "Wrong HTTP status code",
635 "df[id='{}']:virtual-link-profile='{}' must match an existing "
636 "int-virtual-link-desc".format(
637 df
["id"], fake_ivld_profile
["id"]
640 norm(str(e
.exception
)),
641 "Wrong exception text",
644 del df
["virtual-link-profile"]
646 i
=12, t
="Check Input Validation: scaling-criteria monitoring-param-ref"
648 vdu
= test_vnfd
["vdu"][1]
649 affected_df
= test_vnfd
["df"][0]
650 sa
= affected_df
["scaling-aspect"][0]
651 sp
= sa
["scaling-policy"][0]
652 sc
= sp
["scaling-criteria"][0]
653 tmp
= vdu
.pop("monitoring-parameter")
655 with self
.assertRaises(
657 msg
="Accepted non-existent Scaling Group Policy Criteria",
659 self
.topic
.upload_content(
660 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
663 e
.exception
.http_code
,
664 HTTPStatus
.UNPROCESSABLE_ENTITY
,
665 "Wrong HTTP status code",
669 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
670 "[name='{}']:scaling-criteria[name='{}']: "
671 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
676 sc
["vnf-monitoring-param-ref"],
679 norm(str(e
.exception
)),
680 "Wrong exception text",
683 vdu
["monitoring-parameter"] = tmp
685 i
=13, t
="Check Input Validation: scaling-aspect vnf-configuration"
687 df
= test_vnfd
["df"][0]
688 tmp
= test_vnfd
["df"][0]["lcm-operations-configuration"][
689 "operate-vnf-op-config"
692 with self
.assertRaises(
694 msg
="Accepted non-existent Scaling Group VDU ID Reference",
696 self
.topic
.upload_content(
697 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
700 e
.exception
.http_code
,
701 HTTPStatus
.UNPROCESSABLE_ENTITY
,
702 "Wrong HTTP status code",
706 "'day1-2 configuration' not defined in the descriptor but it is referenced "
707 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
708 df
["id"], df
["scaling-aspect"][0]["id"]
711 norm(str(e
.exception
)),
712 "Wrong exception text",
715 test_vnfd
["df"][0]["lcm-operations-configuration"][
716 "operate-vnf-op-config"
717 ]["day1-2"].append(tmp
)
718 with self
.subTest(i
=14, t
="Check Input Validation: scaling-config-action"):
719 df
= test_vnfd
["df"][0]
722 .get("lcm-operations-configuration")
723 .get("operate-vnf-op-config")["day1-2"][0]["config-primitive"]
725 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
726 "operate-vnf-op-config"
727 )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
729 with self
.assertRaises(
731 msg
="Accepted non-existent Scaling Group VDU ID Reference",
733 self
.topic
.upload_content(
734 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
737 e
.exception
.http_code
,
738 HTTPStatus
.UNPROCESSABLE_ENTITY
,
739 "Wrong HTTP status code",
743 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
744 "config-primitive-name-ref='{}' does not match any "
745 "day1-2 configuration:config-primitive:name".format(
747 df
["scaling-aspect"][0]["id"],
748 sa
["scaling-config-action"][0][
749 "vnf-config-primitive-name-ref"
753 norm(str(e
.exception
)),
754 "Wrong exception text",
757 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
758 "operate-vnf-op-config"
759 )["day1-2"][0]["config-primitive"] = tmp
760 with self
.subTest(i
=15, t
="Check Input Validation: everything right"):
761 test_vnfd
["id"] = "fake-vnfd-id"
762 test_vnfd
["df"][0].get("lcm-operations-configuration").get(
763 "operate-vnf-op-config"
764 )["day1-2"][0]["id"] = "fake-vnfd-id"
765 self
.db
.get_one
.side_effect
= [
766 {"_id": did
, "_admin": deepcopy(db_vnfd_content
["_admin"])},
769 rc
= self
.topic
.upload_content(
770 fake_session
, did
, test_vnfd
, {}, {"Content-Type": []}
772 self
.assertTrue(rc
, "Input Validation: Unexpected failure")
775 def test_edit_vnfd(self
):
776 vnfd_content
= deepcopy(db_vnfd_content
)
777 did
= vnfd_content
["_id"]
778 self
.fs
.file_exists
.return_value
= True
779 self
.fs
.dir_ls
.return_value
= True
780 with self
.subTest(i
=1, t
="Normal Edition"):
782 self
.db
.get_one
.side_effect
= [deepcopy(vnfd_content
), None]
783 data
= {"product-name": "new-vnfd-name"}
784 self
.topic
.edit(fake_session
, did
, data
)
785 db_args
= self
.db
.replace
.call_args
[0]
786 msg_args
= self
.msg
.write
.call_args
[0]
788 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
789 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
790 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
791 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
792 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
794 db_args
[2]["_admin"]["created"],
795 vnfd_content
["_admin"]["created"],
796 "Wrong creation time",
799 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
802 db_args
[2]["_admin"]["projects_read"],
803 vnfd_content
["_admin"]["projects_read"],
804 "Wrong read-only project list",
807 db_args
[2]["_admin"]["projects_write"],
808 vnfd_content
["_admin"]["projects_write"],
809 "Wrong read-write project list",
812 db_args
[2]["product-name"], data
["product-name"], "Wrong VNFD Name"
814 with self
.subTest(i
=2, t
="Conflict on Edit"):
815 data
= {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
816 self
.db
.get_one
.side_effect
= [
817 deepcopy(vnfd_content
),
818 {"_id": str(uuid4()), "id": data
["id"]},
820 with self
.assertRaises(
821 EngineException
, msg
="Accepted existing VNFD ID"
823 self
.topic
.edit(fake_session
, did
, data
)
825 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
829 "{} with id '{}' already exists for this project".format(
833 norm(str(e
.exception
)),
834 "Wrong exception text",
836 with self
.subTest(i
=3, t
="Check Envelope"):
837 data
= {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
838 with self
.assertRaises(
839 EngineException
, msg
="Accepted VNFD with wrong envelope"
841 self
.topic
.edit(fake_session
, did
, data
, content
=vnfd_content
)
843 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
846 "'vnfd' must be dict", norm(str(e
.exception
)), "Wrong exception text"
850 def test_delete_vnfd(self
):
851 did
= db_vnfd_content
["_id"]
852 self
.db
.get_one
.return_value
= db_vnfd_content
853 p_id
= db_vnfd_content
["_admin"]["projects_read"][0]
854 with self
.subTest(i
=1, t
="Normal Deletion"):
855 self
.db
.get_list
.return_value
= []
856 self
.db
.del_one
.return_value
= {"deleted": 1}
857 self
.topic
.delete(fake_session
, did
)
858 db_args
= self
.db
.del_one
.call_args
[0]
859 msg_args
= self
.msg
.write
.call_args
[0]
860 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
861 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
862 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
863 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
864 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
866 db_args
[1]["_admin.projects_write.cont"],
870 db_g1_args
= self
.db
.get_one
.call_args
[0]
871 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
872 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
873 db_gl_calls
= self
.db
.get_list
.call_args_list
874 self
.assertEqual(db_gl_calls
[0][0][0], "vnfrs", "Wrong DB topic")
875 # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID") # Filter changed after call
876 self
.assertEqual(db_gl_calls
[1][0][0], "nsds", "Wrong DB topic")
878 db_gl_calls
[1][0][1]["vnfd-id"],
879 db_vnfd_content
["id"],
880 "Wrong DB NSD vnfd-id",
883 db_del_args
= self
.db
.del_list
.call_args
[0]
885 self
.db
.del_list
.call_args
[0][0],
886 self
.topic
.topic
+"_revisions",
891 self
.db
.del_list
.call_args
[0][1]['_id']['$regex'],
893 "Wrong ID for rexep delete",
896 self
.db
.set_one
.assert_not_called()
897 fs_del_calls
= self
.fs
.file_delete
.call_args_list
898 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
899 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
900 with self
.subTest(i
=2, t
="Conflict on Delete - VNFD in use by VNFR"):
901 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-vnfr"}]
902 with self
.assertRaises(
903 EngineException
, msg
="Accepted VNFD in use by VNFR"
905 self
.topic
.delete(fake_session
, did
)
907 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
910 "there is at least one vnf instance using this descriptor",
911 norm(str(e
.exception
)),
912 "Wrong exception text",
914 with self
.subTest(i
=3, t
="Conflict on Delete - VNFD in use by NSD"):
915 self
.db
.get_list
.side_effect
= [
917 [{"_id": str(uuid4()), "name": "fake-nsd"}],
919 with self
.assertRaises(
920 EngineException
, msg
="Accepted VNFD in use by NSD"
922 self
.topic
.delete(fake_session
, did
)
924 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
927 "there is at least one ns package referencing this descriptor",
928 norm(str(e
.exception
)),
929 "Wrong exception text",
931 with self
.subTest(i
=4, t
="Non-existent VNFD"):
932 excp_msg
= "Not found any {} with filter='{}'".format("VNFD", {"_id": did
})
933 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
934 with self
.assertRaises(
935 DbException
, msg
="Accepted non-existent VNFD ID"
937 self
.topic
.delete(fake_session
, did
)
939 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
942 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
944 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
945 db_vnfd_content
["_admin"]["projects_read"].append("other_project")
946 self
.db
.get_one
= Mock(return_value
=db_vnfd_content
)
947 self
.db
.get_list
= Mock(return_value
=[])
948 self
.msg
.write
.reset_mock()
949 self
.db
.del_one
.reset_mock()
950 self
.fs
.file_delete
.reset_mock()
952 self
.topic
.delete(fake_session
, did
)
953 self
.db
.del_one
.assert_not_called()
954 self
.msg
.write
.assert_not_called()
955 db_g1_args
= self
.db
.get_one
.call_args
[0]
956 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
957 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
958 db_s1_args
= self
.db
.set_one
.call_args
959 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
960 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
962 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
965 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
968 db_s1_args
[1]["pull_list"],
969 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
970 "Wrong DB pull_list dictionary",
972 self
.fs
.file_delete
.assert_not_called()
975 @patch("osm_nbi.descriptor_topics.yaml")
976 def test_validate_descriptor_changes(self
, mock_yaml
):
977 descriptor_name
= "test_descriptor"
978 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
979 "/tmp/" + str(uuid4()), "a+b"
981 with self
.subTest(i
=1, t
="VNFD has changes in day1-2 config primitive"):
982 old_vnfd
, new_vnfd
= self
.create_desc_temp(db_vnfd_content
)
983 new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
985 ][0]["config-primitive"][0]["name"] = "new_action"
986 mock_yaml
.load
.side_effect
= [old_vnfd
, new_vnfd
]
987 with self
.assertNotRaises(EngineException
):
988 self
.topic
._validate
_descriptor
_changes
(
989 descriptor_name
, "/tmp/", "/tmp:1/"
991 with self
.subTest(i
=2, t
="VNFD sw version changed"):
992 # old vnfd uses the default software version: 1.0
993 new_vnfd
["software-version"] = "1.3"
994 new_vnfd
["sw-image-desc"][0]["name"] = "new-image"
995 mock_yaml
.load
.side_effect
= [old_vnfd
, new_vnfd
]
996 with self
.assertNotRaises(EngineException
):
997 self
.topic
._validate
_descriptor
_changes
(
998 descriptor_name
, "/tmp/", "/tmp:1/"
1001 i
=3, t
="VNFD sw version is not changed and mgmt-cp has changed"
1003 old_vnfd
, new_vnfd
= self
.create_desc_temp(db_vnfd_content
)
1004 new_vnfd
["mgmt-cp"] = "new-mgmt-cp"
1005 mock_yaml
.load
.side_effect
= [old_vnfd
, new_vnfd
]
1006 with self
.assertRaises(EngineException
) as e
:
1007 self
.topic
._validate
_descriptor
_changes
(
1008 descriptor_name
, "/tmp/", "/tmp:1/"
1011 e
.exception
.http_code
,
1012 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1013 "Wrong HTTP status code",
1016 norm("there are disallowed changes in the vnf descriptor"),
1017 norm(str(e
.exception
)),
1018 "Wrong exception text",
1021 def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self
):
1022 indata
= deepcopy(db_vnfd_content
)
1023 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1025 def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
1028 indata
= deepcopy(db_vnfd_content
)
1029 indata
["ext-cpd"] = []
1030 with self
.assertRaises(EngineException
) as e
:
1031 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1033 e
.exception
.http_code
,
1034 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1035 "Wrong HTTP status code",
1039 "mgmt-cp='{}' must match an existing ext-cpd".format(indata
["mgmt-cp"])
1041 norm(str(e
.exception
)),
1042 "Wrong exception text",
1045 def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self
):
1046 indata
= deepcopy(db_vnfd_content
)
1047 indata
.pop("mgmt-cp")
1048 with self
.assertRaises(EngineException
) as e
:
1049 self
.topic
.validate_mgmt_interface_connection_point(indata
)
1051 e
.exception
.http_code
,
1052 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1053 "Wrong HTTP status code",
1056 norm("'mgmt-cp' is a mandatory field and it is not defined"),
1057 norm(str(e
.exception
)),
1058 "Wrong exception text",
1061 def test_validate_vdu_internal_connection_points_on_valid_descriptor(self
):
1062 indata
= db_vnfd_content
1063 vdu
= indata
["vdu"][0]
1064 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1066 def test_validate_external_connection_points_on_valid_descriptor(self
):
1067 indata
= db_vnfd_content
1068 self
.topic
.validate_external_connection_points(indata
)
1070 def test_validate_external_connection_points_when_missing_internal_connection_point(
1073 indata
= deepcopy(db_vnfd_content
)
1074 vdu
= indata
["vdu"][0]
1076 affected_ext_cpd
= indata
["ext-cpd"][0]
1077 with self
.assertRaises(EngineException
) as e
:
1078 self
.topic
.validate_external_connection_points(indata
)
1080 e
.exception
.http_code
,
1081 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1082 "Wrong HTTP status code",
1086 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
1087 affected_ext_cpd
["id"]
1090 norm(str(e
.exception
)),
1091 "Wrong exception text",
1094 def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
1097 indata
= deepcopy(db_vnfd_content
)
1098 vdu
= indata
["vdu"][0]
1102 "virtual-network-interface-requirement": [{"name": "duplicated"}],
1104 vdu
["int-cpd"].insert(0, duplicated_cpd
)
1105 with self
.assertRaises(EngineException
) as e
:
1106 self
.topic
.validate_vdu_internal_connection_points(vdu
)
1108 e
.exception
.http_code
,
1109 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1110 "Wrong HTTP status code",
1114 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
1115 vdu
["id"], duplicated_cpd
["id"]
1118 norm(str(e
.exception
)),
1119 "Wrong exception text",
1122 def test_validate_external_connection_points_on_duplicated_external_connection_point(
1125 indata
= deepcopy(db_vnfd_content
)
1127 "id": "vnf-mgmt-ext",
1128 "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
1130 indata
["ext-cpd"].insert(0, duplicated_cpd
)
1131 with self
.assertRaises(EngineException
) as e
:
1132 self
.topic
.validate_external_connection_points(indata
)
1134 e
.exception
.http_code
,
1135 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1136 "Wrong HTTP status code",
1140 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
1141 duplicated_cpd
["id"]
1144 norm(str(e
.exception
)),
1145 "Wrong exception text",
1148 def test_validate_internal_virtual_links_on_valid_descriptor(self
):
1149 indata
= db_vnfd_content
1150 self
.topic
.validate_internal_virtual_links(indata
)
1152 def test_validate_internal_virtual_links_on_duplicated_ivld(self
):
1153 indata
= deepcopy(db_vnfd_content
)
1154 duplicated_vld
= {"id": "internal"}
1155 indata
["int-virtual-link-desc"].insert(0, duplicated_vld
)
1156 with self
.assertRaises(EngineException
) as e
:
1157 self
.topic
.validate_internal_virtual_links(indata
)
1159 e
.exception
.http_code
,
1160 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1161 "Wrong HTTP status code",
1165 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
1166 duplicated_vld
["id"]
1169 norm(str(e
.exception
)),
1170 "Wrong exception text",
1173 def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
1176 indata
= deepcopy(db_vnfd_content
)
1177 vdu
= indata
["vdu"][0]
1178 affected_int_cpd
= vdu
["int-cpd"][0]
1179 affected_int_cpd
["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
1180 with self
.assertRaises(EngineException
) as e
:
1181 self
.topic
.validate_internal_virtual_links(indata
)
1183 e
.exception
.http_code
,
1184 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1185 "Wrong HTTP status code",
1189 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
1190 "int-virtual-link-desc".format(
1192 affected_int_cpd
["id"],
1193 affected_int_cpd
["int-virtual-link-desc"],
1196 norm(str(e
.exception
)),
1197 "Wrong exception text",
1200 def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self
):
1201 indata
= deepcopy(db_vnfd_content
)
1202 affected_ivld_profile
= {"id": "non-existing-int-virtual-link-desc"}
1203 df
= indata
["df"][0]
1204 df
["virtual-link-profile"] = [affected_ivld_profile
]
1205 with self
.assertRaises(EngineException
) as e
:
1206 self
.topic
.validate_internal_virtual_links(indata
)
1208 e
.exception
.http_code
,
1209 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1210 "Wrong HTTP status code",
1214 "df[id='{}']:virtual-link-profile='{}' must match an existing "
1215 "int-virtual-link-desc".format(df
["id"], affected_ivld_profile
["id"])
1217 norm(str(e
.exception
)),
1218 "Wrong exception text",
1221 def test_validate_monitoring_params_on_valid_descriptor(self
):
1222 indata
= db_vnfd_content
1223 self
.topic
.validate_monitoring_params(indata
)
1225 def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self
):
1226 indata
= deepcopy(db_vnfd_content
)
1227 duplicated_mp
= {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
1228 affected_ivld
= indata
["int-virtual-link-desc"][0]
1229 affected_ivld
["monitoring-parameters"] = [duplicated_mp
, duplicated_mp
]
1230 with self
.assertRaises(EngineException
) as e
:
1231 self
.topic
.validate_monitoring_params(indata
)
1233 e
.exception
.http_code
,
1234 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1235 "Wrong HTTP status code",
1239 "Duplicated monitoring-parameter id in "
1240 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
1241 affected_ivld
["id"], duplicated_mp
["id"]
1244 norm(str(e
.exception
)),
1245 "Wrong exception text",
1248 def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self
):
1249 indata
= deepcopy(db_vnfd_content
)
1251 "id": "dataVM_cpu_util",
1252 "name": "dataVM_cpu_util",
1253 "performance_metric": "cpu",
1255 affected_vdu
= indata
["vdu"][1]
1256 affected_vdu
["monitoring-parameter"].insert(0, duplicated_mp
)
1257 with self
.assertRaises(EngineException
) as e
:
1258 self
.topic
.validate_monitoring_params(indata
)
1260 e
.exception
.http_code
,
1261 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1262 "Wrong HTTP status code",
1266 "Duplicated monitoring-parameter id in "
1267 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1268 affected_vdu
["id"], duplicated_mp
["id"]
1271 norm(str(e
.exception
)),
1272 "Wrong exception text",
1275 def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self
):
1276 indata
= deepcopy(db_vnfd_content
)
1280 "performance_metric": "memory",
1282 affected_df
= indata
["df"][0]
1283 affected_df
["monitoring-parameter"] = [duplicated_mp
, duplicated_mp
]
1284 with self
.assertRaises(EngineException
) as e
:
1285 self
.topic
.validate_monitoring_params(indata
)
1287 e
.exception
.http_code
,
1288 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1289 "Wrong HTTP status code",
1293 "Duplicated monitoring-parameter id in "
1294 "df[id='{}']:monitoring-parameter[id='{}']".format(
1295 affected_df
["id"], duplicated_mp
["id"]
1298 norm(str(e
.exception
)),
1299 "Wrong exception text",
1302 def test_validate_scaling_group_descriptor_on_valid_descriptor(self
):
1303 indata
= db_vnfd_content
1304 self
.topic
.validate_scaling_group_descriptor(indata
)
1306 def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self
):
1307 indata
= deepcopy(db_vnfd_content
)
1308 vdu
= indata
["vdu"][1]
1309 affected_df
= indata
["df"][0]
1310 affected_sa
= affected_df
["scaling-aspect"][0]
1311 affected_sp
= affected_sa
["scaling-policy"][0]
1312 affected_sc
= affected_sp
["scaling-criteria"][0]
1313 vdu
.pop("monitoring-parameter")
1314 with self
.assertRaises(EngineException
) as e
:
1315 self
.topic
.validate_scaling_group_descriptor(indata
)
1317 e
.exception
.http_code
,
1318 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1319 "Wrong HTTP status code",
1323 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1324 "[name='{}']:scaling-criteria[name='{}']: "
1325 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1328 affected_sp
["name"],
1329 affected_sc
["name"],
1330 affected_sc
["vnf-monitoring-param-ref"],
1333 norm(str(e
.exception
)),
1334 "Wrong exception text",
1337 def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self
):
1338 indata
= deepcopy(db_vnfd_content
)
1339 df
= indata
["df"][0]
1340 affected_sa
= df
["scaling-aspect"][0]
1341 indata
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1344 with self
.assertRaises(EngineException
) as e
:
1345 self
.topic
.validate_scaling_group_descriptor(indata
)
1347 e
.exception
.http_code
,
1348 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1349 "Wrong HTTP status code",
1353 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1354 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1355 df
["id"], affected_sa
["id"]
1358 norm(str(e
.exception
)),
1359 "Wrong exception text",
1362 def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1365 indata
= deepcopy(db_vnfd_content
)
1366 df
= indata
["df"][0]
1367 affected_sa
= df
["scaling-aspect"][0]
1368 affected_sca_primitive
= affected_sa
["scaling-config-action"][0][
1369 "vnf-config-primitive-name-ref"
1371 df
["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1374 with self
.assertRaises(EngineException
) as e
:
1375 self
.topic
.validate_scaling_group_descriptor(indata
)
1377 e
.exception
.http_code
,
1378 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1379 "Wrong HTTP status code",
1383 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1384 "config-primitive-name-ref='{}' does not match any "
1385 "day1-2 configuration:config-primitive:name".format(
1386 df
["id"], affected_sa
["id"], affected_sca_primitive
1389 norm(str(e
.exception
)),
1390 "Wrong exception text",
1393 def test_new_vnfd_revision(self
):
1394 did
= db_vnfd_content
["_id"]
1395 self
.fs
.get_params
.return_value
= {}
1396 self
.fs
.file_exists
.return_value
= False
1397 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1398 "/tmp/" + str(uuid4()), "a+b"
1400 test_vnfd
= deepcopy(db_vnfd_content
)
1401 del test_vnfd
["_id"]
1402 del test_vnfd
["_admin"]
1403 self
.db
.create
.return_value
= did
1405 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1406 db_args
= self
.db
.create
.call_args
[0]
1407 self
.assertEqual(db_args
[1]['_admin']['revision'], 0,
1408 "New package should be at revision 0")
1410 @patch("osm_nbi.descriptor_topics.shutil")
1411 @patch("osm_nbi.descriptor_topics.os.rename")
1412 def test_update_vnfd(self
, mock_rename
, mock_shutil
):
1414 did
= db_vnfd_content
["_id"]
1416 self
.fs
.get_params
.return_value
= {}
1417 self
.fs
.file_exists
.return_value
= False
1418 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1419 "/tmp/" + str(uuid4()), "a+b"
1421 new_vnfd
= deepcopy(db_vnfd_content
)
1423 self
.db
.create
.return_value
= did
1425 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1426 del new_vnfd
["vdu"][0]["cloud-init-file"]
1427 del new_vnfd
["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1428 "day1-2"][0]["execution-environment-list"][0]["juju"]
1433 "_admin": deepcopy(db_vnfd_content
["_admin"])
1435 old_vnfd
["_admin"]["revision"] = old_revision
1436 self
.db
.get_one
.side_effect
= [
1440 self
.topic
.upload_content(
1441 fake_session
, did
, new_vnfd
, {}, {"Content-Type": []}
1444 db_args
= self
.db
.replace
.call_args
[0]
1445 self
.assertEqual(db_args
[2]['_admin']['revision'], old_revision
+ 1,
1446 "Revision should increment")
1449 class Test_NsdTopic(TestCase
):
1451 def setUpClass(cls
):
1452 cls
.test_name
= "test-nsd-topic"
1455 def tearDownClass(cls
):
1459 self
.db
= Mock(dbbase
.DbBase())
1460 self
.fs
= Mock(fsbase
.FsBase())
1461 self
.msg
= Mock(msgbase
.MsgBase())
1462 self
.auth
= Mock(authconn
.Authconn(None, None, None))
1463 self
.topic
= NsdTopic(self
.db
, self
.fs
, self
.msg
, self
.auth
)
1464 self
.topic
.check_quota
= Mock(return_value
=None) # skip quota
1467 def assertNotRaises(self
, exception_type
):
1470 except exception_type
:
1471 raise self
.failureException("{} raised".format(exception_type
.__name
__))
1473 def create_desc_temp(self
, template
):
1474 old_desc
= deepcopy(template
)
1475 new_desc
= deepcopy(template
)
1476 return old_desc
, new_desc
1478 @patch("osm_nbi.descriptor_topics.shutil")
1479 @patch("osm_nbi.descriptor_topics.os.rename")
1480 def test_new_nsd(self
, mock_rename
, mock_shutil
):
1481 did
= db_nsd_content
["_id"]
1482 self
.fs
.get_params
.return_value
= {}
1483 self
.fs
.file_exists
.return_value
= False
1484 self
.fs
.file_open
.side_effect
= lambda path
, mode
: tempfile
.TemporaryFile(
1487 test_nsd
= deepcopy(db_nsd_content
)
1489 del test_nsd
["_admin"]
1490 with self
.subTest(i
=1, t
="Normal Creation"):
1491 self
.db
.create
.return_value
= did
1493 did2
, oid
= self
.topic
.new(rollback
, fake_session
, {})
1494 db_args
= self
.db
.create
.call_args
[0]
1495 msg_args
= self
.msg
.write
.call_args
[0]
1496 self
.assertEqual(len(rollback
), 1, "Wrong rollback length")
1497 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1498 self
.assertEqual(msg_args
[1], "created", "Wrong message action")
1499 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1500 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1501 self
.assertEqual(did2
, did
, "Wrong DB NSD id")
1502 self
.assertIsNotNone(db_args
[1]["_admin"]["created"], "Wrong creation time")
1504 db_args
[1]["_admin"]["modified"],
1505 db_args
[1]["_admin"]["created"],
1506 "Wrong modification time",
1509 db_args
[1]["_admin"]["projects_read"],
1511 "Wrong read-only project list",
1514 db_args
[1]["_admin"]["projects_write"],
1516 "Wrong read-write project list",
1520 self
.db
.get_one
.side_effect
= [
1521 {"_id": did
, "_admin": db_nsd_content
["_admin"]},
1524 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1525 self
.topic
.upload_content(
1526 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1528 msg_args
= self
.msg
.write
.call_args
[0]
1529 test_nsd
["_id"] = did
1531 msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic"
1533 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1534 self
.assertEqual(msg_args
[2], test_nsd
, "Wrong message content")
1535 db_args
= self
.db
.get_one
.mock_calls
[0][1]
1536 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1537 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB NSD id")
1538 db_args
= self
.db
.replace
.call_args
[0]
1539 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1540 self
.assertEqual(db_args
[1], did
, "Wrong DB NSD id")
1541 admin
= db_args
[2]["_admin"]
1542 db_admin
= db_nsd_content
["_admin"]
1544 admin
["created"], db_admin
["created"], "Wrong creation time"
1547 admin
["modified"], db_admin
["created"], "Wrong modification time"
1550 admin
["projects_read"],
1551 db_admin
["projects_read"],
1552 "Wrong read-only project list",
1555 admin
["projects_write"],
1556 db_admin
["projects_write"],
1557 "Wrong read-write project list",
1560 admin
["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1563 admin
["operationalState"], "ENABLED", "Wrong operational state"
1565 self
.assertEqual(admin
["usageState"], "NOT_IN_USE", "Wrong usage state")
1566 storage
= admin
["storage"]
1567 self
.assertEqual(storage
["folder"], did
+ ":1", "Wrong storage folder")
1569 storage
["descriptor"], "package", "Wrong storage descriptor"
1571 compare_desc(self
, test_nsd
, db_args
[2], "NSD")
1572 revision_args
= self
.db
.create
.call_args
[0]
1574 revision_args
[0], self
.topic
.topic
+ "_revisions", "Wrong topic"
1577 revision_args
[1]["id"], db_args
[2]["id"], "Wrong revision id"
1580 revision_args
[1]["_id"],
1581 db_args
[2]["_id"] + ":1",
1582 "Wrong revision _id"
1587 self
.db
.get_one
.side_effect
= (
1588 lambda table
, filter, fail_on_empty
=None, fail_on_more
=None: {
1590 "_admin": db_nsd_content
["_admin"],
1593 with self
.subTest(i
=2, t
="Check Pyangbind Validation: required properties"):
1594 tmp
= test_nsd
["id"]
1597 with self
.assertRaises(
1598 EngineException
, msg
="Accepted NSD with a missing required property"
1600 self
.topic
.upload_content(
1601 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1604 e
.exception
.http_code
,
1605 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1606 "Wrong HTTP status code",
1609 norm("Error in pyangbind validation: '{}'".format("id")),
1610 norm(str(e
.exception
)),
1611 "Wrong exception text",
1614 test_nsd
["id"] = tmp
1615 with self
.subTest(i
=3, t
="Check Pyangbind Validation: additional properties"):
1616 test_nsd
["extra-property"] = 0
1618 with self
.assertRaises(
1619 EngineException
, msg
="Accepted NSD with an additional property"
1621 self
.topic
.upload_content(
1622 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1625 e
.exception
.http_code
,
1626 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1627 "Wrong HTTP status code",
1631 "Error in pyangbind validation: {} ({})".format(
1632 "json object contained a key that did not exist",
1636 norm(str(e
.exception
)),
1637 "Wrong exception text",
1640 del test_nsd
["extra-property"]
1641 with self
.subTest(i
=4, t
="Check Pyangbind Validation: property types"):
1642 tmp
= test_nsd
["designer"]
1643 test_nsd
["designer"] = {"key": 0}
1645 with self
.assertRaises(
1646 EngineException
, msg
="Accepted NSD with a wrongly typed property"
1648 self
.topic
.upload_content(
1649 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1652 e
.exception
.http_code
,
1653 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1654 "Wrong HTTP status code",
1658 "Error in pyangbind validation: {} ({})".format(
1659 "json object contained a key that did not exist", "key"
1662 norm(str(e
.exception
)),
1663 "Wrong exception text",
1666 test_nsd
["designer"] = tmp
1668 i
=5, t
="Check Input Validation: mgmt-network+virtual-link-protocol-data"
1670 df
= test_nsd
["df"][0]
1673 "virtual-link-desc-id": "mgmt",
1674 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1676 df
["virtual-link-profile"] = [mgmt_profile
]
1678 with self
.assertRaises(
1679 EngineException
, msg
="Accepted VLD with mgmt-network+ip-profile"
1681 self
.topic
.upload_content(
1682 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1685 e
.exception
.http_code
,
1686 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1687 "Wrong HTTP status code",
1691 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1692 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1693 df
["id"], mgmt_profile
["id"]
1696 norm(str(e
.exception
)),
1697 "Wrong exception text",
1700 del df
["virtual-link-profile"]
1701 with self
.subTest(i
=6, t
="Check Descriptor Dependencies: vnfd-id[]"):
1702 self
.db
.get_one
.side_effect
= [
1703 {"_id": did
, "_admin": db_nsd_content
["_admin"]},
1706 self
.db
.get_list
.return_value
= []
1708 with self
.assertRaises(
1709 EngineException
, msg
="Accepted wrong VNFD ID reference"
1711 self
.topic
.upload_content(
1712 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1715 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1719 "'vnfd-id'='{}' references a non existing vnfd".format(
1720 test_nsd
["vnfd-id"][0]
1723 norm(str(e
.exception
)),
1724 "Wrong exception text",
1730 t
="Check Descriptor Dependencies: "
1731 "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]",
1733 vnfd_descriptor
= deepcopy(db_vnfd_content
)
1734 df
= test_nsd
["df"][0]
1735 affected_vnf_profile
= df
["vnf-profile"][0]
1736 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
1737 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
1738 self
.db
.get_one
.side_effect
= [
1739 {"_id": did
, "_admin": db_nsd_content
["_admin"]},
1742 self
.db
.get_list
.return_value
= [vnfd_descriptor
]
1744 with self
.assertRaises(
1745 EngineException
, msg
="Accepted wrong VLD CP reference"
1747 self
.topic
.upload_content(
1748 fake_session
, did
, test_nsd
, {}, {"Content-Type": []}
1751 e
.exception
.http_code
,
1752 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1753 "Wrong HTTP status code",
1757 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1758 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1759 "non existing ext-cpd:id inside vnfd '{}'".format(
1761 affected_vnf_profile
["id"],
1762 affected_virtual_link
["virtual-link-profile-id"],
1764 vnfd_descriptor
["id"],
1767 norm(str(e
.exception
)),
1768 "Wrong exception text",
1774 def test_edit_nsd(self
):
1775 nsd_content
= deepcopy(db_nsd_content
)
1776 did
= nsd_content
["_id"]
1777 self
.fs
.file_exists
.return_value
= True
1778 self
.fs
.dir_ls
.return_value
= True
1779 with self
.subTest(i
=1, t
="Normal Edition"):
1781 self
.db
.get_one
.side_effect
= [deepcopy(nsd_content
), None]
1782 self
.db
.get_list
.return_value
= [db_vnfd_content
]
1783 data
= {"id": "new-nsd-id", "name": "new-nsd-name"}
1784 self
.topic
.edit(fake_session
, did
, data
)
1785 db_args
= self
.db
.replace
.call_args
[0]
1786 msg_args
= self
.msg
.write
.call_args
[0]
1788 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1789 self
.assertEqual(msg_args
[1], "edited", "Wrong message action")
1790 self
.assertEqual(msg_args
[2], data
, "Wrong message content")
1791 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1792 self
.assertEqual(db_args
[1], did
, "Wrong DB ID")
1794 db_args
[2]["_admin"]["created"],
1795 nsd_content
["_admin"]["created"],
1796 "Wrong creation time",
1799 db_args
[2]["_admin"]["modified"], now
, "Wrong modification time"
1802 db_args
[2]["_admin"]["projects_read"],
1803 nsd_content
["_admin"]["projects_read"],
1804 "Wrong read-only project list",
1807 db_args
[2]["_admin"]["projects_write"],
1808 nsd_content
["_admin"]["projects_write"],
1809 "Wrong read-write project list",
1811 self
.assertEqual(db_args
[2]["id"], data
["id"], "Wrong NSD ID")
1812 self
.assertEqual(db_args
[2]["name"], data
["name"], "Wrong NSD Name")
1813 with self
.subTest(i
=2, t
="Conflict on Edit"):
1814 data
= {"id": "fake-nsd-id", "name": "new-nsd-name"}
1815 self
.db
.get_one
.side_effect
= [
1817 {"_id": str(uuid4()), "id": data
["id"]},
1819 with self
.assertRaises(
1820 EngineException
, msg
="Accepted existing NSD ID"
1822 self
.topic
.edit(fake_session
, did
, data
)
1824 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1828 "{} with id '{}' already exists for this project".format(
1832 norm(str(e
.exception
)),
1833 "Wrong exception text",
1835 with self
.subTest(i
=3, t
="Check Envelope"):
1836 data
= {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
1837 self
.db
.get_one
.side_effect
= [nsd_content
, None]
1838 with self
.assertRaises(
1839 EngineException
, msg
="Accepted NSD with wrong envelope"
1841 self
.topic
.edit(fake_session
, did
, data
, content
=nsd_content
)
1843 e
.exception
.http_code
, HTTPStatus
.BAD_REQUEST
, "Wrong HTTP status code"
1846 "'nsd' must be a list of only one element",
1847 norm(str(e
.exception
)),
1848 "Wrong exception text",
1852 def test_delete_nsd(self
):
1853 did
= db_nsd_content
["_id"]
1854 self
.db
.get_one
.return_value
= db_nsd_content
1855 p_id
= db_nsd_content
["_admin"]["projects_read"][0]
1856 with self
.subTest(i
=1, t
="Normal Deletion"):
1857 self
.db
.get_list
.return_value
= []
1858 self
.db
.del_one
.return_value
= {"deleted": 1}
1859 self
.topic
.delete(fake_session
, did
)
1860 db_args
= self
.db
.del_one
.call_args
[0]
1861 msg_args
= self
.msg
.write
.call_args
[0]
1862 self
.assertEqual(msg_args
[0], self
.topic
.topic_msg
, "Wrong message topic")
1863 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
1864 self
.assertEqual(msg_args
[2], {"_id": did
}, "Wrong message content")
1865 self
.assertEqual(db_args
[0], self
.topic
.topic
, "Wrong DB topic")
1866 self
.assertEqual(db_args
[1]["_id"], did
, "Wrong DB ID")
1868 db_args
[1]["_admin.projects_write.cont"],
1872 db_g1_args
= self
.db
.get_one
.call_args
[0]
1873 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1874 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB NSD ID")
1875 db_gl_calls
= self
.db
.get_list
.call_args_list
1876 self
.assertEqual(db_gl_calls
[0][0][0], "nsrs", "Wrong DB topic")
1877 # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID") # Filter changed after call
1878 self
.assertEqual(db_gl_calls
[1][0][0], "nsts", "Wrong DB topic")
1880 db_gl_calls
[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
1881 db_nsd_content
["id"],
1882 "Wrong DB NSD netslice-subnet nsd-ref",
1884 self
.db
.set_one
.assert_not_called()
1885 fs_del_calls
= self
.fs
.file_delete
.call_args_list
1886 self
.assertEqual(fs_del_calls
[0][0][0], did
, "Wrong FS file id")
1887 self
.assertEqual(fs_del_calls
[1][0][0], did
+ "_", "Wrong FS folder id")
1888 with self
.subTest(i
=2, t
="Conflict on Delete - NSD in use by nsr"):
1889 self
.db
.get_list
.return_value
= [{"_id": str(uuid4()), "name": "fake-nsr"}]
1890 with self
.assertRaises(
1891 EngineException
, msg
="Accepted NSD in use by NSR"
1893 self
.topic
.delete(fake_session
, did
)
1895 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1898 "there is at least one ns instance using this descriptor",
1899 norm(str(e
.exception
)),
1900 "Wrong exception text",
1902 with self
.subTest(i
=3, t
="Conflict on Delete - NSD in use by NST"):
1903 self
.db
.get_list
.side_effect
= [
1905 [{"_id": str(uuid4()), "name": "fake-nst"}],
1907 with self
.assertRaises(
1908 EngineException
, msg
="Accepted NSD in use by NST"
1910 self
.topic
.delete(fake_session
, did
)
1912 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
1915 "there is at least one netslice template referencing this descriptor",
1916 norm(str(e
.exception
)),
1917 "Wrong exception text",
1919 with self
.subTest(i
=4, t
="Non-existent NSD"):
1920 excp_msg
= "Not found any {} with filter='{}'".format("NSD", {"_id": did
})
1921 self
.db
.get_one
.side_effect
= DbException(excp_msg
, HTTPStatus
.NOT_FOUND
)
1922 with self
.assertRaises(
1923 DbException
, msg
="Accepted non-existent NSD ID"
1925 self
.topic
.delete(fake_session
, did
)
1927 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
1930 norm(excp_msg
), norm(str(e
.exception
)), "Wrong exception text"
1932 with self
.subTest(i
=5, t
="No delete because referenced by other project"):
1933 db_nsd_content
["_admin"]["projects_read"].append("other_project")
1934 self
.db
.get_one
= Mock(return_value
=db_nsd_content
)
1935 self
.db
.get_list
= Mock(return_value
=[])
1936 self
.msg
.write
.reset_mock()
1937 self
.db
.del_one
.reset_mock()
1938 self
.fs
.file_delete
.reset_mock()
1940 self
.topic
.delete(fake_session
, did
)
1941 self
.db
.del_one
.assert_not_called()
1942 self
.msg
.write
.assert_not_called()
1943 db_g1_args
= self
.db
.get_one
.call_args
[0]
1944 self
.assertEqual(db_g1_args
[0], self
.topic
.topic
, "Wrong DB topic")
1945 self
.assertEqual(db_g1_args
[1]["_id"], did
, "Wrong DB VNFD ID")
1946 db_s1_args
= self
.db
.set_one
.call_args
1947 self
.assertEqual(db_s1_args
[0][0], self
.topic
.topic
, "Wrong DB topic")
1948 self
.assertEqual(db_s1_args
[0][1]["_id"], did
, "Wrong DB ID")
1950 p_id
, db_s1_args
[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1953 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
1956 db_s1_args
[1]["pull_list"],
1957 {"_admin.projects_read": (p_id
,), "_admin.projects_write": (p_id
,)},
1958 "Wrong DB pull_list dictionary",
1960 self
.fs
.file_delete
.assert_not_called()
1963 @patch("osm_nbi.descriptor_topics.yaml")
1964 def test_validate_descriptor_changes(self
, mock_yaml
):
1965 descriptor_name
= "test_ns_descriptor"
1966 self
.fs
.file_open
.side_effect
= lambda path
, mode
: open(
1967 "/tmp/" + str(uuid4()), "a+b"
1970 i
=1, t
="NSD has changes in ns-configuration:config-primitive"
1972 old_nsd
, new_nsd
= self
.create_desc_temp(db_nsd_content
)
1974 {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}}
1977 {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}}
1979 mock_yaml
.load
.side_effect
= [old_nsd
, new_nsd
]
1980 with self
.assertNotRaises(EngineException
):
1981 self
.topic
._validate
_descriptor
_changes
(
1982 descriptor_name
, "/tmp", "/tmp:1"
1984 with self
.subTest(i
=2, t
="NSD name has changed"):
1985 old_nsd
, new_nsd
= self
.create_desc_temp(db_nsd_content
)
1986 new_nsd
["name"] = "nscharm-ns2"
1987 mock_yaml
.load
.side_effect
= [old_nsd
, new_nsd
]
1988 with self
.assertRaises(EngineException
) as e
:
1989 self
.topic
._validate
_descriptor
_changes
(
1990 descriptor_name
, "/tmp", "/tmp:1"
1993 e
.exception
.http_code
,
1994 HTTPStatus
.UNPROCESSABLE_ENTITY
,
1995 "Wrong HTTP status code",
1998 norm("there are disallowed changes in the ns descriptor"),
1999 norm(str(e
.exception
)),
2000 "Wrong exception text",
2003 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
2006 indata
= deepcopy(db_nsd_content
)
2007 vld
= indata
["virtual-link-desc"][0]
2008 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2012 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
2015 indata
= deepcopy(db_nsd_content
)
2016 vld
= indata
["virtual-link-desc"][0]
2017 df
= indata
["df"][0]
2020 "virtual-link-desc-id": "mgmt",
2021 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
2023 df
["virtual-link-profile"] = [affected_vlp
]
2024 with self
.assertRaises(EngineException
) as e
:
2025 self
.topic
.validate_vld_mgmt_network_with_virtual_link_protocol_data(
2029 e
.exception
.http_code
,
2030 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2031 "Wrong HTTP status code",
2035 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
2036 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
2037 df
["id"], affected_vlp
["id"]
2040 norm(str(e
.exception
)),
2041 "Wrong exception text",
2044 def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self
):
2045 indata
= deepcopy(db_nsd_content
)
2046 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2048 def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self
):
2049 indata
= deepcopy(db_nsd_content
)
2050 df
= indata
["df"][0]
2051 affected_vnf_profile
= df
["vnf-profile"][0]
2052 indata
["vnfd-id"] = ["non-existing-vnfd"]
2053 with self
.assertRaises(EngineException
) as e
:
2054 self
.topic
.validate_vnf_profiles_vnfd_id(indata
)
2056 e
.exception
.http_code
,
2057 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2058 "Wrong HTTP status code",
2062 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
2063 "does not match any vnfd-id".format(
2065 affected_vnf_profile
["id"],
2066 affected_vnf_profile
["vnfd-id"],
2069 norm(str(e
.exception
)),
2070 "Wrong exception text",
2073 def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
2076 nsd_descriptor
= deepcopy(db_nsd_content
)
2077 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2078 df
= nsd_descriptor
["df"][0]
2079 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2080 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2084 def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
2087 nsd_descriptor
= deepcopy(db_nsd_content
)
2088 vnfd_descriptor
= deepcopy(db_vnfd_content
)
2089 df
= nsd_descriptor
["df"][0]
2090 affected_vnf_profile
= df
["vnf-profile"][0]
2091 affected_virtual_link
= affected_vnf_profile
["virtual-link-connectivity"][1]
2092 vnfds_index
= {vnfd_descriptor
["id"]: vnfd_descriptor
}
2093 affected_cpd
= vnfd_descriptor
["ext-cpd"].pop()
2094 with self
.assertRaises(EngineException
) as e
:
2095 self
.topic
.validate_df_vnf_profiles_constituent_connection_points(
2099 e
.exception
.http_code
,
2100 HTTPStatus
.UNPROCESSABLE_ENTITY
,
2101 "Wrong HTTP status code",
2105 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
2106 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
2107 "non existing ext-cpd:id inside vnfd '{}'".format(
2109 affected_vnf_profile
["id"],
2110 affected_virtual_link
["virtual-link-profile-id"],
2112 vnfd_descriptor
["id"],
2115 norm(str(e
.exception
)),
2116 "Wrong exception text",
2119 def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self
):
2120 nsd_descriptor
= deepcopy(db_nsd_content
)
2121 invalid_vnfd_id
= "invalid-vnfd-id"
2122 nsd_descriptor
["id"] = "invalid-vnfd-id-ns"
2123 nsd_descriptor
["vnfd-id"][0] = invalid_vnfd_id
2124 nsd_descriptor
["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
2125 nsd_descriptor
["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
2126 with self
.assertRaises(EngineException
) as e
:
2127 self
.db
.get_list
.return_value
= []
2128 nsd_descriptor
= self
.topic
.check_conflict_on_edit(
2129 fake_session
, nsd_descriptor
, [], "id"
2132 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
2136 "Descriptor error at 'vnfd-id'='{}' references a non "
2137 "existing vnfd".format(invalid_vnfd_id
)
2139 norm(str(e
.exception
)),
2140 "Wrong exception text",
2144 if __name__
== "__main__":