Bug 1538 Fixed
[osm/NBI.git] / osm_nbi / tests / test_descriptor_topics.py
1 #! /usr/bin/python3
2 # -*- coding: utf-8 -*-
3
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
7 #
8 # http://www.apache.org/licenses/LICENSE-2.0
9 #
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
13 # implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16
17 __author__ = "Pedro de la Cruz Ramos, pedro.delacruzramos@altran.com"
18 __date__ = "2019-11-20"
19
20 import unittest
21 from unittest import TestCase
22 from unittest.mock import Mock
23 from uuid import uuid4
24 from http import HTTPStatus
25 from copy import deepcopy
26 from time import time
27 from osm_common import dbbase, fsbase, msgbase
28 from osm_nbi import authconn
29 from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text
30 from osm_nbi.descriptor_topics import VnfdTopic, NsdTopic
31 from osm_nbi.engine import EngineException
32 from osm_common.dbbase import DbException
33 import yaml
34
35 test_name = "test-user"
36 db_vnfd_content = yaml.load(db_vnfds_text, Loader=yaml.Loader)[0]
37 db_nsd_content = yaml.load(db_nsds_text, Loader=yaml.Loader)[0]
38 test_pid = db_vnfd_content["_admin"]["projects_read"][0]
39 fake_session = {
40 "username": test_name,
41 "project_id": (test_pid,),
42 "method": None,
43 "admin": True,
44 "force": False,
45 "public": False,
46 "allow_show_user_project_role": True,
47 }
48
49
50 def norm(str):
51 """Normalize string for checking"""
52 return " ".join(str.strip().split()).lower()
53
54
55 def compare_desc(tc, d1, d2, k):
56 """
57 Compare two descriptors
58 We need this function because some methods are adding/removing items to/from the descriptors
59 before they are stored in the database, so the original and stored versions will differ
60 What we check is that COMMON LEAF ITEMS are equal
61 Lists of different length are not compared
62 :param tc: Test Case wich provides context (in particular the assert* methods)
63 :param d1,d2: Descriptors to be compared
64 :param key/item being compared
65 :return: Nothing
66 """
67 if isinstance(d1, dict) and isinstance(d2, dict):
68 for key in d1.keys():
69 if key in d2:
70 compare_desc(tc, d1[key], d2[key], k + "[{}]".format(key))
71 elif isinstance(d1, list) and isinstance(d2, list) and len(d1) == len(d2):
72 for i in range(len(d1)):
73 compare_desc(tc, d1[i], d2[i], k + "[{}]".format(i))
74 else:
75 tc.assertEqual(d1, d2, "Wrong descriptor content: {}".format(k))
76
77
78 class Test_VnfdTopic(TestCase):
79 @classmethod
80 def setUpClass(cls):
81 cls.test_name = "test-vnfd-topic"
82
83 @classmethod
84 def tearDownClass(cls):
85 pass
86
87 def setUp(self):
88 self.db = Mock(dbbase.DbBase())
89 self.fs = Mock(fsbase.FsBase())
90 self.msg = Mock(msgbase.MsgBase())
91 self.auth = Mock(authconn.Authconn(None, None, None))
92 self.topic = VnfdTopic(self.db, self.fs, self.msg, self.auth)
93 self.topic.check_quota = Mock(return_value=None) # skip quota
94
95 def test_new_vnfd(self):
96 did = db_vnfd_content["_id"]
97 self.fs.get_params.return_value = {}
98 self.fs.file_exists.return_value = False
99 self.fs.file_open.side_effect = lambda path, mode: open(
100 "/tmp/" + str(uuid4()), "a+b"
101 )
102 test_vnfd = deepcopy(db_vnfd_content)
103 del test_vnfd["_id"]
104 del test_vnfd["_admin"]
105 with self.subTest(i=1, t="Normal Creation"):
106 self.db.create.return_value = did
107 rollback = []
108 did2, oid = self.topic.new(rollback, fake_session, {})
109 db_args = self.db.create.call_args[0]
110 msg_args = self.msg.write.call_args[0]
111 self.assertEqual(len(rollback), 1, "Wrong rollback length")
112 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
113 self.assertEqual(msg_args[1], "created", "Wrong message action")
114 self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
115 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
116 self.assertEqual(did2, did, "Wrong DB VNFD id")
117 self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
118 self.assertEqual(
119 db_args[1]["_admin"]["modified"],
120 db_args[1]["_admin"]["created"],
121 "Wrong modification time",
122 )
123 self.assertEqual(
124 db_args[1]["_admin"]["projects_read"],
125 [test_pid],
126 "Wrong read-only project list",
127 )
128 self.assertEqual(
129 db_args[1]["_admin"]["projects_write"],
130 [test_pid],
131 "Wrong read-write project list",
132 )
133 tmp1 = test_vnfd["vdu"][0]["cloud-init-file"]
134 tmp2 = test_vnfd["df"][0]["lcm-operations-configuration"][
135 "operate-vnf-op-config"
136 ]["day1-2"][0]["execution-environment-list"][0]["juju"]
137 del test_vnfd["vdu"][0]["cloud-init-file"]
138 del test_vnfd["df"][0]["lcm-operations-configuration"][
139 "operate-vnf-op-config"
140 ]["day1-2"][0]["execution-environment-list"][0]["juju"]
141 try:
142 self.db.get_one.side_effect = [
143 {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])},
144 None,
145 ]
146 self.topic.upload_content(
147 fake_session, did, test_vnfd, {}, {"Content-Type": []}
148 )
149 msg_args = self.msg.write.call_args[0]
150 test_vnfd["_id"] = did
151 self.assertEqual(
152 msg_args[0], self.topic.topic_msg, "Wrong message topic"
153 )
154 self.assertEqual(msg_args[1], "edited", "Wrong message action")
155 self.assertEqual(msg_args[2], test_vnfd, "Wrong message content")
156 db_args = self.db.get_one.mock_calls[0][1]
157 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
158 self.assertEqual(db_args[1]["_id"], did, "Wrong DB VNFD id")
159 db_args = self.db.replace.call_args[0]
160 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
161 self.assertEqual(db_args[1], did, "Wrong DB VNFD id")
162 admin = db_args[2]["_admin"]
163 db_admin = deepcopy(db_vnfd_content["_admin"])
164 self.assertEqual(admin["type"], "vnfd", "Wrong descriptor type")
165 self.assertEqual(
166 admin["created"], db_admin["created"], "Wrong creation time"
167 )
168 self.assertGreater(
169 admin["modified"], db_admin["created"], "Wrong modification time"
170 )
171 self.assertEqual(
172 admin["projects_read"],
173 db_admin["projects_read"],
174 "Wrong read-only project list",
175 )
176 self.assertEqual(
177 admin["projects_write"],
178 db_admin["projects_write"],
179 "Wrong read-write project list",
180 )
181 self.assertEqual(
182 admin["onboardingState"], "ONBOARDED", "Wrong onboarding state"
183 )
184 self.assertEqual(
185 admin["operationalState"], "ENABLED", "Wrong operational state"
186 )
187 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
188 storage = admin["storage"]
189 self.assertEqual(storage["folder"], did, "Wrong storage folder")
190 self.assertEqual(
191 storage["descriptor"], "package", "Wrong storage descriptor"
192 )
193 compare_desc(self, test_vnfd, db_args[2], "VNFD")
194 finally:
195 test_vnfd["vdu"][0]["cloud-init-file"] = tmp1
196 test_vnfd["df"][0]["lcm-operations-configuration"][
197 "operate-vnf-op-config"
198 ]["day1-2"][0]["execution-environment-list"][0]["juju"] = tmp2
199 self.db.get_one.side_effect = (
200 lambda table, filter, fail_on_empty=None, fail_on_more=None: {
201 "_id": did,
202 "_admin": deepcopy(db_vnfd_content["_admin"]),
203 }
204 )
205 with self.subTest(i=2, t="Check Pyangbind Validation: additional properties"):
206 test_vnfd["extra-property"] = 0
207 try:
208 with self.assertRaises(
209 EngineException, msg="Accepted VNFD with an additional property"
210 ) as e:
211 self.topic.upload_content(
212 fake_session, did, test_vnfd, {}, {"Content-Type": []}
213 )
214 self.assertEqual(
215 e.exception.http_code,
216 HTTPStatus.UNPROCESSABLE_ENTITY,
217 "Wrong HTTP status code",
218 )
219 self.assertIn(
220 norm(
221 "Error in pyangbind validation: {} ({})".format(
222 "json object contained a key that did not exist",
223 "extra-property",
224 )
225 ),
226 norm(str(e.exception)),
227 "Wrong exception text",
228 )
229 finally:
230 del test_vnfd["extra-property"]
231 with self.subTest(i=3, t="Check Pyangbind Validation: property types"):
232 tmp = test_vnfd["product-name"]
233 test_vnfd["product-name"] = {"key": 0}
234 try:
235 with self.assertRaises(
236 EngineException, msg="Accepted VNFD with a wrongly typed property"
237 ) as e:
238 self.topic.upload_content(
239 fake_session, did, test_vnfd, {}, {"Content-Type": []}
240 )
241 self.assertEqual(
242 e.exception.http_code,
243 HTTPStatus.UNPROCESSABLE_ENTITY,
244 "Wrong HTTP status code",
245 )
246 self.assertIn(
247 norm(
248 "Error in pyangbind validation: {} ({})".format(
249 "json object contained a key that did not exist", "key"
250 )
251 ),
252 norm(str(e.exception)),
253 "Wrong exception text",
254 )
255 finally:
256 test_vnfd["product-name"] = tmp
257 with self.subTest(i=4, t="Check Input Validation: cloud-init"):
258 with self.assertRaises(
259 EngineException, msg="Accepted non-existent cloud_init file"
260 ) as e:
261 self.topic.upload_content(
262 fake_session, did, test_vnfd, {}, {"Content-Type": []}
263 )
264 self.assertEqual(
265 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
266 )
267 self.assertIn(
268 norm(
269 "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format(
270 "cloud-init", test_vnfd["id"], test_vnfd["vdu"][0]["id"]
271 )
272 ),
273 norm(str(e.exception)),
274 "Wrong exception text",
275 )
276 with self.subTest(i=5, t="Check Input Validation: day1-2 configuration[juju]"):
277 del test_vnfd["vdu"][0]["cloud-init-file"]
278 with self.assertRaises(
279 EngineException, msg="Accepted non-existent charm in VNF configuration"
280 ) as e:
281 self.topic.upload_content(
282 fake_session, did, test_vnfd, {}, {"Content-Type": []}
283 )
284 print(str(e.exception))
285 self.assertEqual(
286 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
287 )
288 self.assertIn(
289 norm(
290 "{} defined in vnf[id={}] but not present in package".format(
291 "charm", test_vnfd["id"]
292 )
293 ),
294 norm(str(e.exception)),
295 "Wrong exception text",
296 )
297 del test_vnfd["df"][0]["lcm-operations-configuration"][
298 "operate-vnf-op-config"
299 ]["day1-2"][0]["execution-environment-list"][0]["juju"]
300 with self.subTest(i=6, t="Check Input Validation: mgmt-cp"):
301 tmp = test_vnfd["mgmt-cp"]
302 del test_vnfd["mgmt-cp"]
303 try:
304 with self.assertRaises(
305 EngineException, msg="Accepted VNFD without management interface"
306 ) as e:
307 self.topic.upload_content(
308 fake_session, did, test_vnfd, {}, {"Content-Type": []}
309 )
310 self.assertEqual(
311 e.exception.http_code,
312 HTTPStatus.UNPROCESSABLE_ENTITY,
313 "Wrong HTTP status code",
314 )
315 self.assertIn(
316 norm(
317 "'{}' is a mandatory field and it is not defined".format(
318 "mgmt-cp"
319 )
320 ),
321 norm(str(e.exception)),
322 "Wrong exception text",
323 )
324 finally:
325 test_vnfd["mgmt-cp"] = tmp
326 with self.subTest(i=7, t="Check Input Validation: mgmt-cp connection point"):
327 tmp = test_vnfd["mgmt-cp"]
328 test_vnfd["mgmt-cp"] = "wrong-cp"
329 try:
330 with self.assertRaises(
331 EngineException, msg="Accepted wrong mgmt-cp connection point"
332 ) as e:
333 self.topic.upload_content(
334 fake_session, did, test_vnfd, {}, {"Content-Type": []}
335 )
336 self.assertEqual(
337 e.exception.http_code,
338 HTTPStatus.UNPROCESSABLE_ENTITY,
339 "Wrong HTTP status code",
340 )
341 self.assertIn(
342 norm(
343 "mgmt-cp='{}' must match an existing ext-cpd".format(
344 test_vnfd["mgmt-cp"]
345 )
346 ),
347 norm(str(e.exception)),
348 "Wrong exception text",
349 )
350 finally:
351 test_vnfd["mgmt-cp"] = tmp
352 with self.subTest(i=8, t="Check Input Validation: vdu int-cpd"):
353 ext_cpd = test_vnfd["ext-cpd"][1]
354 tmp = ext_cpd["int-cpd"]["cpd"]
355 ext_cpd["int-cpd"]["cpd"] = "wrong-cpd"
356 try:
357 with self.assertRaises(
358 EngineException,
359 msg="Accepted wrong ext-cpd internal connection point",
360 ) as e:
361 self.topic.upload_content(
362 fake_session, did, test_vnfd, {}, {"Content-Type": []}
363 )
364 self.assertEqual(
365 e.exception.http_code,
366 HTTPStatus.UNPROCESSABLE_ENTITY,
367 "Wrong HTTP status code",
368 )
369 self.assertIn(
370 norm(
371 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
372 ext_cpd["id"]
373 )
374 ),
375 norm(str(e.exception)),
376 "Wrong exception text",
377 )
378 finally:
379 ext_cpd["int-cpd"]["cpd"] = tmp
380 with self.subTest(i=9, t="Check Input Validation: Duplicated VLD"):
381 test_vnfd["int-virtual-link-desc"].insert(0, {"id": "internal"})
382 try:
383 with self.assertRaises(
384 EngineException, msg="Accepted duplicated VLD name"
385 ) as e:
386 self.topic.upload_content(
387 fake_session, did, test_vnfd, {}, {"Content-Type": []}
388 )
389 self.assertEqual(
390 e.exception.http_code,
391 HTTPStatus.UNPROCESSABLE_ENTITY,
392 "Wrong HTTP status code",
393 )
394 self.assertIn(
395 norm(
396 "identifier id '{}' is not unique".format(
397 test_vnfd["int-virtual-link-desc"][0]["id"]
398 )
399 ),
400 norm(str(e.exception)),
401 "Wrong exception text",
402 )
403 finally:
404 del test_vnfd["int-virtual-link-desc"][0]
405 with self.subTest(i=10, t="Check Input Validation: vdu int-virtual-link-desc"):
406 vdu = test_vnfd["vdu"][0]
407 int_cpd = vdu["int-cpd"][1]
408 tmp = int_cpd["int-virtual-link-desc"]
409 int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
410 try:
411 with self.assertRaises(
412 EngineException, msg="Accepted int-virtual-link-desc"
413 ) as e:
414 self.topic.upload_content(
415 fake_session, did, test_vnfd, {}, {"Content-Type": []}
416 )
417 self.assertEqual(
418 e.exception.http_code,
419 HTTPStatus.UNPROCESSABLE_ENTITY,
420 "Wrong HTTP status code",
421 )
422 self.assertIn(
423 norm(
424 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
425 "int-virtual-link-desc".format(
426 vdu["id"], int_cpd["id"], int_cpd["int-virtual-link-desc"]
427 )
428 ),
429 norm(str(e.exception)),
430 "Wrong exception text",
431 )
432 finally:
433 int_cpd["int-virtual-link-desc"] = tmp
434 with self.subTest(i=11, t="Check Input Validation: virtual-link-profile)"):
435 fake_ivld_profile = {"id": "fake-profile-ref", "flavour": "fake-flavour"}
436 df = test_vnfd["df"][0]
437 df["virtual-link-profile"] = [fake_ivld_profile]
438 try:
439 with self.assertRaises(
440 EngineException, msg="Accepted non-existent Profile Ref"
441 ) as e:
442 self.topic.upload_content(
443 fake_session, did, test_vnfd, {}, {"Content-Type": []}
444 )
445 self.assertEqual(
446 e.exception.http_code,
447 HTTPStatus.UNPROCESSABLE_ENTITY,
448 "Wrong HTTP status code",
449 )
450 self.assertIn(
451 norm(
452 "df[id='{}']:virtual-link-profile='{}' must match an existing "
453 "int-virtual-link-desc".format(
454 df["id"], fake_ivld_profile["id"]
455 )
456 ),
457 norm(str(e.exception)),
458 "Wrong exception text",
459 )
460 finally:
461 del df["virtual-link-profile"]
462 with self.subTest(
463 i=12, t="Check Input Validation: scaling-criteria monitoring-param-ref"
464 ):
465 vdu = test_vnfd["vdu"][1]
466 affected_df = test_vnfd["df"][0]
467 sa = affected_df["scaling-aspect"][0]
468 sp = sa["scaling-policy"][0]
469 sc = sp["scaling-criteria"][0]
470 tmp = vdu.pop("monitoring-parameter")
471 try:
472 with self.assertRaises(
473 EngineException,
474 msg="Accepted non-existent Scaling Group Policy Criteria",
475 ) as e:
476 self.topic.upload_content(
477 fake_session, did, test_vnfd, {}, {"Content-Type": []}
478 )
479 self.assertEqual(
480 e.exception.http_code,
481 HTTPStatus.UNPROCESSABLE_ENTITY,
482 "Wrong HTTP status code",
483 )
484 self.assertIn(
485 norm(
486 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
487 "[name='{}']:scaling-criteria[name='{}']: "
488 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
489 affected_df["id"],
490 sa["id"],
491 sp["name"],
492 sc["name"],
493 sc["vnf-monitoring-param-ref"],
494 )
495 ),
496 norm(str(e.exception)),
497 "Wrong exception text",
498 )
499 finally:
500 vdu["monitoring-parameter"] = tmp
501 with self.subTest(
502 i=13, t="Check Input Validation: scaling-aspect vnf-configuration"
503 ):
504 df = test_vnfd["df"][0]
505 tmp = test_vnfd["df"][0]["lcm-operations-configuration"][
506 "operate-vnf-op-config"
507 ]["day1-2"].pop()
508 try:
509 with self.assertRaises(
510 EngineException,
511 msg="Accepted non-existent Scaling Group VDU ID Reference",
512 ) as e:
513 self.topic.upload_content(
514 fake_session, did, test_vnfd, {}, {"Content-Type": []}
515 )
516 self.assertEqual(
517 e.exception.http_code,
518 HTTPStatus.UNPROCESSABLE_ENTITY,
519 "Wrong HTTP status code",
520 )
521 self.assertIn(
522 norm(
523 "'day1-2 configuration' not defined in the descriptor but it is referenced "
524 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
525 df["id"], df["scaling-aspect"][0]["id"]
526 )
527 ),
528 norm(str(e.exception)),
529 "Wrong exception text",
530 )
531 finally:
532 test_vnfd["df"][0]["lcm-operations-configuration"][
533 "operate-vnf-op-config"
534 ]["day1-2"].append(tmp)
535 with self.subTest(i=14, t="Check Input Validation: scaling-config-action"):
536 df = test_vnfd["df"][0]
537 tmp = (
538 test_vnfd["df"][0]
539 .get("lcm-operations-configuration")
540 .get("operate-vnf-op-config")["day1-2"][0]["config-primitive"]
541 )
542 test_vnfd["df"][0].get("lcm-operations-configuration").get(
543 "operate-vnf-op-config"
544 )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}]
545 try:
546 with self.assertRaises(
547 EngineException,
548 msg="Accepted non-existent Scaling Group VDU ID Reference",
549 ) as e:
550 self.topic.upload_content(
551 fake_session, did, test_vnfd, {}, {"Content-Type": []}
552 )
553 self.assertEqual(
554 e.exception.http_code,
555 HTTPStatus.UNPROCESSABLE_ENTITY,
556 "Wrong HTTP status code",
557 )
558 self.assertIn(
559 norm(
560 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
561 "config-primitive-name-ref='{}' does not match any "
562 "day1-2 configuration:config-primitive:name".format(
563 df["id"],
564 df["scaling-aspect"][0]["id"],
565 sa["scaling-config-action"][0][
566 "vnf-config-primitive-name-ref"
567 ],
568 )
569 ),
570 norm(str(e.exception)),
571 "Wrong exception text",
572 )
573 finally:
574 test_vnfd["df"][0].get("lcm-operations-configuration").get(
575 "operate-vnf-op-config"
576 )["day1-2"][0]["config-primitive"] = tmp
577 with self.subTest(i=15, t="Check Input Validation: everything right"):
578 test_vnfd["id"] = "fake-vnfd-id"
579 test_vnfd["df"][0].get("lcm-operations-configuration").get(
580 "operate-vnf-op-config"
581 )["day1-2"][0]["id"] = "fake-vnfd-id"
582 self.db.get_one.side_effect = [
583 {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])},
584 None,
585 ]
586 rc = self.topic.upload_content(
587 fake_session, did, test_vnfd, {}, {"Content-Type": []}
588 )
589 self.assertTrue(rc, "Input Validation: Unexpected failure")
590 return
591
592 def test_edit_vnfd(self):
593 vnfd_content = deepcopy(db_vnfd_content)
594 did = vnfd_content["_id"]
595 self.fs.file_exists.return_value = True
596 self.fs.dir_ls.return_value = True
597 with self.subTest(i=1, t="Normal Edition"):
598 now = time()
599 self.db.get_one.side_effect = [deepcopy(vnfd_content), None]
600 data = {"product-name": "new-vnfd-name"}
601 self.topic.edit(fake_session, did, data)
602 db_args = self.db.replace.call_args[0]
603 msg_args = self.msg.write.call_args[0]
604 data["_id"] = did
605 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
606 self.assertEqual(msg_args[1], "edited", "Wrong message action")
607 self.assertEqual(msg_args[2], data, "Wrong message content")
608 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
609 self.assertEqual(db_args[1], did, "Wrong DB ID")
610 self.assertEqual(
611 db_args[2]["_admin"]["created"],
612 vnfd_content["_admin"]["created"],
613 "Wrong creation time",
614 )
615 self.assertGreater(
616 db_args[2]["_admin"]["modified"], now, "Wrong modification time"
617 )
618 self.assertEqual(
619 db_args[2]["_admin"]["projects_read"],
620 vnfd_content["_admin"]["projects_read"],
621 "Wrong read-only project list",
622 )
623 self.assertEqual(
624 db_args[2]["_admin"]["projects_write"],
625 vnfd_content["_admin"]["projects_write"],
626 "Wrong read-write project list",
627 )
628 self.assertEqual(
629 db_args[2]["product-name"], data["product-name"], "Wrong VNFD Name"
630 )
631 with self.subTest(i=2, t="Conflict on Edit"):
632 data = {"id": "hackfest3charmed-vnf", "product-name": "new-vnfd-name"}
633 self.db.get_one.side_effect = [
634 deepcopy(vnfd_content),
635 {"_id": str(uuid4()), "id": data["id"]},
636 ]
637 with self.assertRaises(
638 EngineException, msg="Accepted existing VNFD ID"
639 ) as e:
640 self.topic.edit(fake_session, did, data)
641 self.assertEqual(
642 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
643 )
644 self.assertIn(
645 norm(
646 "{} with id '{}' already exists for this project".format(
647 "vnfd", data["id"]
648 )
649 ),
650 norm(str(e.exception)),
651 "Wrong exception text",
652 )
653 with self.subTest(i=3, t="Check Envelope"):
654 data = {"vnfd": [{"id": "new-vnfd-id-1", "product-name": "new-vnfd-name"}]}
655 with self.assertRaises(
656 EngineException, msg="Accepted VNFD with wrong envelope"
657 ) as e:
658 self.topic.edit(fake_session, did, data, content=vnfd_content)
659 self.assertEqual(
660 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
661 )
662 self.assertIn(
663 "'vnfd' must be dict", norm(str(e.exception)), "Wrong exception text"
664 )
665 return
666
667 def test_delete_vnfd(self):
668 did = db_vnfd_content["_id"]
669 self.db.get_one.return_value = db_vnfd_content
670 p_id = db_vnfd_content["_admin"]["projects_read"][0]
671 with self.subTest(i=1, t="Normal Deletion"):
672 self.db.get_list.return_value = []
673 self.db.del_one.return_value = {"deleted": 1}
674 self.topic.delete(fake_session, did)
675 db_args = self.db.del_one.call_args[0]
676 msg_args = self.msg.write.call_args[0]
677 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
678 self.assertEqual(msg_args[1], "deleted", "Wrong message action")
679 self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
680 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
681 self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
682 self.assertEqual(
683 db_args[1]["_admin.projects_write.cont"],
684 [p_id, "ANY"],
685 "Wrong DB filter",
686 )
687 db_g1_args = self.db.get_one.call_args[0]
688 self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
689 self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
690 db_gl_calls = self.db.get_list.call_args_list
691 self.assertEqual(db_gl_calls[0][0][0], "vnfrs", "Wrong DB topic")
692 # self.assertEqual(db_gl_calls[0][0][1]["vnfd-id"], did, "Wrong DB VNFD ID") # Filter changed after call
693 self.assertEqual(db_gl_calls[1][0][0], "nsds", "Wrong DB topic")
694 self.assertEqual(
695 db_gl_calls[1][0][1]["vnfd-id"],
696 db_vnfd_content["id"],
697 "Wrong DB NSD vnfd-id",
698 )
699
700 self.db.set_one.assert_not_called()
701 fs_del_calls = self.fs.file_delete.call_args_list
702 self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
703 self.assertEqual(fs_del_calls[1][0][0], did + "_", "Wrong FS folder id")
704 with self.subTest(i=2, t="Conflict on Delete - VNFD in use by VNFR"):
705 self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-vnfr"}]
706 with self.assertRaises(
707 EngineException, msg="Accepted VNFD in use by VNFR"
708 ) as e:
709 self.topic.delete(fake_session, did)
710 self.assertEqual(
711 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
712 )
713 self.assertIn(
714 "there is at least one vnf instance using this descriptor",
715 norm(str(e.exception)),
716 "Wrong exception text",
717 )
718 with self.subTest(i=3, t="Conflict on Delete - VNFD in use by NSD"):
719 self.db.get_list.side_effect = [
720 [],
721 [{"_id": str(uuid4()), "name": "fake-nsd"}],
722 ]
723 with self.assertRaises(
724 EngineException, msg="Accepted VNFD in use by NSD"
725 ) as e:
726 self.topic.delete(fake_session, did)
727 self.assertEqual(
728 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
729 )
730 self.assertIn(
731 "there is at least one ns package referencing this descriptor",
732 norm(str(e.exception)),
733 "Wrong exception text",
734 )
735 with self.subTest(i=4, t="Non-existent VNFD"):
736 excp_msg = "Not found any {} with filter='{}'".format("VNFD", {"_id": did})
737 self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
738 with self.assertRaises(
739 DbException, msg="Accepted non-existent VNFD ID"
740 ) as e:
741 self.topic.delete(fake_session, did)
742 self.assertEqual(
743 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
744 )
745 self.assertIn(
746 norm(excp_msg), norm(str(e.exception)), "Wrong exception text"
747 )
748 with self.subTest(i=5, t="No delete because referenced by other project"):
749 db_vnfd_content["_admin"]["projects_read"].append("other_project")
750 self.db.get_one = Mock(return_value=db_vnfd_content)
751 self.db.get_list = Mock(return_value=[])
752 self.msg.write.reset_mock()
753 self.db.del_one.reset_mock()
754 self.fs.file_delete.reset_mock()
755
756 self.topic.delete(fake_session, did)
757 self.db.del_one.assert_not_called()
758 self.msg.write.assert_not_called()
759 db_g1_args = self.db.get_one.call_args[0]
760 self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
761 self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
762 db_s1_args = self.db.set_one.call_args
763 self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
764 self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
765 self.assertIn(
766 p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
767 )
768 self.assertIsNone(
769 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
770 )
771 self.assertEqual(
772 db_s1_args[1]["pull_list"],
773 {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
774 "Wrong DB pull_list dictionary",
775 )
776 self.fs.file_delete.assert_not_called()
777 return
778
779 def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self):
780 indata = deepcopy(db_vnfd_content)
781 self.topic.validate_mgmt_interface_connection_point(indata)
782
783 def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
784 self,
785 ):
786 indata = deepcopy(db_vnfd_content)
787 indata["ext-cpd"] = []
788 with self.assertRaises(EngineException) as e:
789 self.topic.validate_mgmt_interface_connection_point(indata)
790 self.assertEqual(
791 e.exception.http_code,
792 HTTPStatus.UNPROCESSABLE_ENTITY,
793 "Wrong HTTP status code",
794 )
795 self.assertIn(
796 norm(
797 "mgmt-cp='{}' must match an existing ext-cpd".format(indata["mgmt-cp"])
798 ),
799 norm(str(e.exception)),
800 "Wrong exception text",
801 )
802
803 def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self):
804 indata = deepcopy(db_vnfd_content)
805 indata.pop("mgmt-cp")
806 with self.assertRaises(EngineException) as e:
807 self.topic.validate_mgmt_interface_connection_point(indata)
808 self.assertEqual(
809 e.exception.http_code,
810 HTTPStatus.UNPROCESSABLE_ENTITY,
811 "Wrong HTTP status code",
812 )
813 self.assertIn(
814 norm("'mgmt-cp' is a mandatory field and it is not defined"),
815 norm(str(e.exception)),
816 "Wrong exception text",
817 )
818
819 def test_validate_vdu_internal_connection_points_on_valid_descriptor(self):
820 indata = db_vnfd_content
821 vdu = indata["vdu"][0]
822 self.topic.validate_vdu_internal_connection_points(vdu)
823
824 def test_validate_external_connection_points_on_valid_descriptor(self):
825 indata = db_vnfd_content
826 self.topic.validate_external_connection_points(indata)
827
828 def test_validate_external_connection_points_when_missing_internal_connection_point(
829 self,
830 ):
831 indata = deepcopy(db_vnfd_content)
832 vdu = indata["vdu"][0]
833 vdu.pop("int-cpd")
834 affected_ext_cpd = indata["ext-cpd"][0]
835 with self.assertRaises(EngineException) as e:
836 self.topic.validate_external_connection_points(indata)
837 self.assertEqual(
838 e.exception.http_code,
839 HTTPStatus.UNPROCESSABLE_ENTITY,
840 "Wrong HTTP status code",
841 )
842 self.assertIn(
843 norm(
844 "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
845 affected_ext_cpd["id"]
846 )
847 ),
848 norm(str(e.exception)),
849 "Wrong exception text",
850 )
851
852 def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
853 self,
854 ):
855 indata = deepcopy(db_vnfd_content)
856 vdu = indata["vdu"][0]
857 duplicated_cpd = {
858 "id": "vnf-mgmt",
859 "order": 3,
860 "virtual-network-interface-requirement": [{"name": "duplicated"}],
861 }
862 vdu["int-cpd"].insert(0, duplicated_cpd)
863 with self.assertRaises(EngineException) as e:
864 self.topic.validate_vdu_internal_connection_points(vdu)
865 self.assertEqual(
866 e.exception.http_code,
867 HTTPStatus.UNPROCESSABLE_ENTITY,
868 "Wrong HTTP status code",
869 )
870 self.assertIn(
871 norm(
872 "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
873 vdu["id"], duplicated_cpd["id"]
874 )
875 ),
876 norm(str(e.exception)),
877 "Wrong exception text",
878 )
879
880 def test_validate_external_connection_points_on_duplicated_external_connection_point(
881 self,
882 ):
883 indata = deepcopy(db_vnfd_content)
884 duplicated_cpd = {
885 "id": "vnf-mgmt-ext",
886 "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
887 }
888 indata["ext-cpd"].insert(0, duplicated_cpd)
889 with self.assertRaises(EngineException) as e:
890 self.topic.validate_external_connection_points(indata)
891 self.assertEqual(
892 e.exception.http_code,
893 HTTPStatus.UNPROCESSABLE_ENTITY,
894 "Wrong HTTP status code",
895 )
896 self.assertIn(
897 norm(
898 "ext-cpd[id='{}'] is already used by other ext-cpd".format(
899 duplicated_cpd["id"]
900 )
901 ),
902 norm(str(e.exception)),
903 "Wrong exception text",
904 )
905
906 def test_validate_internal_virtual_links_on_valid_descriptor(self):
907 indata = db_vnfd_content
908 self.topic.validate_internal_virtual_links(indata)
909
910 def test_validate_internal_virtual_links_on_duplicated_ivld(self):
911 indata = deepcopy(db_vnfd_content)
912 duplicated_vld = {"id": "internal"}
913 indata["int-virtual-link-desc"].insert(0, duplicated_vld)
914 with self.assertRaises(EngineException) as e:
915 self.topic.validate_internal_virtual_links(indata)
916 self.assertEqual(
917 e.exception.http_code,
918 HTTPStatus.UNPROCESSABLE_ENTITY,
919 "Wrong HTTP status code",
920 )
921 self.assertIn(
922 norm(
923 "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
924 duplicated_vld["id"]
925 )
926 ),
927 norm(str(e.exception)),
928 "Wrong exception text",
929 )
930
931 def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
932 self,
933 ):
934 indata = deepcopy(db_vnfd_content)
935 vdu = indata["vdu"][0]
936 affected_int_cpd = vdu["int-cpd"][0]
937 affected_int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
938 with self.assertRaises(EngineException) as e:
939 self.topic.validate_internal_virtual_links(indata)
940 self.assertEqual(
941 e.exception.http_code,
942 HTTPStatus.UNPROCESSABLE_ENTITY,
943 "Wrong HTTP status code",
944 )
945 self.assertIn(
946 norm(
947 "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
948 "int-virtual-link-desc".format(
949 vdu["id"],
950 affected_int_cpd["id"],
951 affected_int_cpd["int-virtual-link-desc"],
952 )
953 ),
954 norm(str(e.exception)),
955 "Wrong exception text",
956 )
957
958 def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self):
959 indata = deepcopy(db_vnfd_content)
960 affected_ivld_profile = {"id": "non-existing-int-virtual-link-desc"}
961 df = indata["df"][0]
962 df["virtual-link-profile"] = [affected_ivld_profile]
963 with self.assertRaises(EngineException) as e:
964 self.topic.validate_internal_virtual_links(indata)
965 self.assertEqual(
966 e.exception.http_code,
967 HTTPStatus.UNPROCESSABLE_ENTITY,
968 "Wrong HTTP status code",
969 )
970 self.assertIn(
971 norm(
972 "df[id='{}']:virtual-link-profile='{}' must match an existing "
973 "int-virtual-link-desc".format(df["id"], affected_ivld_profile["id"])
974 ),
975 norm(str(e.exception)),
976 "Wrong exception text",
977 )
978
979 def test_validate_monitoring_params_on_valid_descriptor(self):
980 indata = db_vnfd_content
981 self.topic.validate_monitoring_params(indata)
982
983 def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self):
984 indata = deepcopy(db_vnfd_content)
985 duplicated_mp = {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
986 affected_ivld = indata["int-virtual-link-desc"][0]
987 affected_ivld["monitoring-parameters"] = [duplicated_mp, duplicated_mp]
988 with self.assertRaises(EngineException) as e:
989 self.topic.validate_monitoring_params(indata)
990 self.assertEqual(
991 e.exception.http_code,
992 HTTPStatus.UNPROCESSABLE_ENTITY,
993 "Wrong HTTP status code",
994 )
995 self.assertIn(
996 norm(
997 "Duplicated monitoring-parameter id in "
998 "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
999 affected_ivld["id"], duplicated_mp["id"]
1000 )
1001 ),
1002 norm(str(e.exception)),
1003 "Wrong exception text",
1004 )
1005
1006 def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self):
1007 indata = deepcopy(db_vnfd_content)
1008 duplicated_mp = {
1009 "id": "dataVM_cpu_util",
1010 "name": "dataVM_cpu_util",
1011 "performance_metric": "cpu",
1012 }
1013 affected_vdu = indata["vdu"][1]
1014 affected_vdu["monitoring-parameter"].insert(0, duplicated_mp)
1015 with self.assertRaises(EngineException) as e:
1016 self.topic.validate_monitoring_params(indata)
1017 self.assertEqual(
1018 e.exception.http_code,
1019 HTTPStatus.UNPROCESSABLE_ENTITY,
1020 "Wrong HTTP status code",
1021 )
1022 self.assertIn(
1023 norm(
1024 "Duplicated monitoring-parameter id in "
1025 "vdu[id='{}']:monitoring-parameter[id='{}']".format(
1026 affected_vdu["id"], duplicated_mp["id"]
1027 )
1028 ),
1029 norm(str(e.exception)),
1030 "Wrong exception text",
1031 )
1032
1033 def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self):
1034 indata = deepcopy(db_vnfd_content)
1035 duplicated_mp = {
1036 "id": "memory",
1037 "name": "memory",
1038 "performance_metric": "memory",
1039 }
1040 affected_df = indata["df"][0]
1041 affected_df["monitoring-parameter"] = [duplicated_mp, duplicated_mp]
1042 with self.assertRaises(EngineException) as e:
1043 self.topic.validate_monitoring_params(indata)
1044 self.assertEqual(
1045 e.exception.http_code,
1046 HTTPStatus.UNPROCESSABLE_ENTITY,
1047 "Wrong HTTP status code",
1048 )
1049 self.assertIn(
1050 norm(
1051 "Duplicated monitoring-parameter id in "
1052 "df[id='{}']:monitoring-parameter[id='{}']".format(
1053 affected_df["id"], duplicated_mp["id"]
1054 )
1055 ),
1056 norm(str(e.exception)),
1057 "Wrong exception text",
1058 )
1059
1060 def test_validate_scaling_group_descriptor_on_valid_descriptor(self):
1061 indata = db_vnfd_content
1062 self.topic.validate_scaling_group_descriptor(indata)
1063
1064 def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self):
1065 indata = deepcopy(db_vnfd_content)
1066 vdu = indata["vdu"][1]
1067 affected_df = indata["df"][0]
1068 affected_sa = affected_df["scaling-aspect"][0]
1069 affected_sp = affected_sa["scaling-policy"][0]
1070 affected_sc = affected_sp["scaling-criteria"][0]
1071 vdu.pop("monitoring-parameter")
1072 with self.assertRaises(EngineException) as e:
1073 self.topic.validate_scaling_group_descriptor(indata)
1074 self.assertEqual(
1075 e.exception.http_code,
1076 HTTPStatus.UNPROCESSABLE_ENTITY,
1077 "Wrong HTTP status code",
1078 )
1079 self.assertIn(
1080 norm(
1081 "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
1082 "[name='{}']:scaling-criteria[name='{}']: "
1083 "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
1084 affected_df["id"],
1085 affected_sa["id"],
1086 affected_sp["name"],
1087 affected_sc["name"],
1088 affected_sc["vnf-monitoring-param-ref"],
1089 )
1090 ),
1091 norm(str(e.exception)),
1092 "Wrong exception text",
1093 )
1094
1095 def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self):
1096 indata = deepcopy(db_vnfd_content)
1097 df = indata["df"][0]
1098 affected_sa = df["scaling-aspect"][0]
1099 indata["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][
1100 "day1-2"
1101 ].pop()
1102 with self.assertRaises(EngineException) as e:
1103 self.topic.validate_scaling_group_descriptor(indata)
1104 self.assertEqual(
1105 e.exception.http_code,
1106 HTTPStatus.UNPROCESSABLE_ENTITY,
1107 "Wrong HTTP status code",
1108 )
1109 self.assertIn(
1110 norm(
1111 "'day1-2 configuration' not defined in the descriptor but it is referenced "
1112 "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
1113 df["id"], affected_sa["id"]
1114 )
1115 ),
1116 norm(str(e.exception)),
1117 "Wrong exception text",
1118 )
1119
1120 def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(
1121 self,
1122 ):
1123 indata = deepcopy(db_vnfd_content)
1124 df = indata["df"][0]
1125 affected_sa = df["scaling-aspect"][0]
1126 affected_sca_primitive = affected_sa["scaling-config-action"][0][
1127 "vnf-config-primitive-name-ref"
1128 ]
1129 df["lcm-operations-configuration"]["operate-vnf-op-config"]["day1-2"][0][
1130 "config-primitive"
1131 ] = []
1132 with self.assertRaises(EngineException) as e:
1133 self.topic.validate_scaling_group_descriptor(indata)
1134 self.assertEqual(
1135 e.exception.http_code,
1136 HTTPStatus.UNPROCESSABLE_ENTITY,
1137 "Wrong HTTP status code",
1138 )
1139 self.assertIn(
1140 norm(
1141 "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
1142 "config-primitive-name-ref='{}' does not match any "
1143 "day1-2 configuration:config-primitive:name".format(
1144 df["id"], affected_sa["id"], affected_sca_primitive
1145 )
1146 ),
1147 norm(str(e.exception)),
1148 "Wrong exception text",
1149 )
1150
1151
1152 class Test_NsdTopic(TestCase):
1153 @classmethod
1154 def setUpClass(cls):
1155 cls.test_name = "test-nsd-topic"
1156
1157 @classmethod
1158 def tearDownClass(cls):
1159 pass
1160
1161 def setUp(self):
1162 self.db = Mock(dbbase.DbBase())
1163 self.fs = Mock(fsbase.FsBase())
1164 self.msg = Mock(msgbase.MsgBase())
1165 self.auth = Mock(authconn.Authconn(None, None, None))
1166 self.topic = NsdTopic(self.db, self.fs, self.msg, self.auth)
1167 self.topic.check_quota = Mock(return_value=None) # skip quota
1168
1169 def test_new_nsd(self):
1170 did = db_nsd_content["_id"]
1171 self.fs.get_params.return_value = {}
1172 self.fs.file_exists.return_value = False
1173 self.fs.file_open.side_effect = lambda path, mode: open(
1174 "/tmp/" + str(uuid4()), "a+b"
1175 )
1176 test_nsd = deepcopy(db_nsd_content)
1177 del test_nsd["_id"]
1178 del test_nsd["_admin"]
1179 with self.subTest(i=1, t="Normal Creation"):
1180 self.db.create.return_value = did
1181 rollback = []
1182 did2, oid = self.topic.new(rollback, fake_session, {})
1183 db_args = self.db.create.call_args[0]
1184 msg_args = self.msg.write.call_args[0]
1185 self.assertEqual(len(rollback), 1, "Wrong rollback length")
1186 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1187 self.assertEqual(msg_args[1], "created", "Wrong message action")
1188 self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
1189 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1190 self.assertEqual(did2, did, "Wrong DB NSD id")
1191 self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time")
1192 self.assertEqual(
1193 db_args[1]["_admin"]["modified"],
1194 db_args[1]["_admin"]["created"],
1195 "Wrong modification time",
1196 )
1197 self.assertEqual(
1198 db_args[1]["_admin"]["projects_read"],
1199 [test_pid],
1200 "Wrong read-only project list",
1201 )
1202 self.assertEqual(
1203 db_args[1]["_admin"]["projects_write"],
1204 [test_pid],
1205 "Wrong read-write project list",
1206 )
1207 try:
1208 self.db.get_one.side_effect = [
1209 {"_id": did, "_admin": db_nsd_content["_admin"]},
1210 None,
1211 ]
1212 self.db.get_list.return_value = [db_vnfd_content]
1213 self.topic.upload_content(
1214 fake_session, did, test_nsd, {}, {"Content-Type": []}
1215 )
1216 msg_args = self.msg.write.call_args[0]
1217 test_nsd["_id"] = did
1218 self.assertEqual(
1219 msg_args[0], self.topic.topic_msg, "Wrong message topic"
1220 )
1221 self.assertEqual(msg_args[1], "edited", "Wrong message action")
1222 self.assertEqual(msg_args[2], test_nsd, "Wrong message content")
1223 db_args = self.db.get_one.mock_calls[0][1]
1224 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1225 self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id")
1226 db_args = self.db.replace.call_args[0]
1227 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1228 self.assertEqual(db_args[1], did, "Wrong DB NSD id")
1229 admin = db_args[2]["_admin"]
1230 db_admin = db_nsd_content["_admin"]
1231 self.assertEqual(
1232 admin["created"], db_admin["created"], "Wrong creation time"
1233 )
1234 self.assertGreater(
1235 admin["modified"], db_admin["created"], "Wrong modification time"
1236 )
1237 self.assertEqual(
1238 admin["projects_read"],
1239 db_admin["projects_read"],
1240 "Wrong read-only project list",
1241 )
1242 self.assertEqual(
1243 admin["projects_write"],
1244 db_admin["projects_write"],
1245 "Wrong read-write project list",
1246 )
1247 self.assertEqual(
1248 admin["onboardingState"], "ONBOARDED", "Wrong onboarding state"
1249 )
1250 self.assertEqual(
1251 admin["operationalState"], "ENABLED", "Wrong operational state"
1252 )
1253 self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
1254 storage = admin["storage"]
1255 self.assertEqual(storage["folder"], did, "Wrong storage folder")
1256 self.assertEqual(
1257 storage["descriptor"], "package", "Wrong storage descriptor"
1258 )
1259 compare_desc(self, test_nsd, db_args[2], "NSD")
1260 finally:
1261 pass
1262 self.db.get_one.side_effect = (
1263 lambda table, filter, fail_on_empty=None, fail_on_more=None: {
1264 "_id": did,
1265 "_admin": db_nsd_content["_admin"],
1266 }
1267 )
1268 with self.subTest(i=2, t="Check Pyangbind Validation: required properties"):
1269 tmp = test_nsd["id"]
1270 del test_nsd["id"]
1271 try:
1272 with self.assertRaises(
1273 EngineException, msg="Accepted NSD with a missing required property"
1274 ) as e:
1275 self.topic.upload_content(
1276 fake_session, did, test_nsd, {}, {"Content-Type": []}
1277 )
1278 self.assertEqual(
1279 e.exception.http_code,
1280 HTTPStatus.UNPROCESSABLE_ENTITY,
1281 "Wrong HTTP status code",
1282 )
1283 self.assertIn(
1284 norm("Error in pyangbind validation: '{}'".format("id")),
1285 norm(str(e.exception)),
1286 "Wrong exception text",
1287 )
1288 finally:
1289 test_nsd["id"] = tmp
1290 with self.subTest(i=3, t="Check Pyangbind Validation: additional properties"):
1291 test_nsd["extra-property"] = 0
1292 try:
1293 with self.assertRaises(
1294 EngineException, msg="Accepted NSD with an additional property"
1295 ) as e:
1296 self.topic.upload_content(
1297 fake_session, did, test_nsd, {}, {"Content-Type": []}
1298 )
1299 self.assertEqual(
1300 e.exception.http_code,
1301 HTTPStatus.UNPROCESSABLE_ENTITY,
1302 "Wrong HTTP status code",
1303 )
1304 self.assertIn(
1305 norm(
1306 "Error in pyangbind validation: {} ({})".format(
1307 "json object contained a key that did not exist",
1308 "extra-property",
1309 )
1310 ),
1311 norm(str(e.exception)),
1312 "Wrong exception text",
1313 )
1314 finally:
1315 del test_nsd["extra-property"]
1316 with self.subTest(i=4, t="Check Pyangbind Validation: property types"):
1317 tmp = test_nsd["designer"]
1318 test_nsd["designer"] = {"key": 0}
1319 try:
1320 with self.assertRaises(
1321 EngineException, msg="Accepted NSD with a wrongly typed property"
1322 ) as e:
1323 self.topic.upload_content(
1324 fake_session, did, test_nsd, {}, {"Content-Type": []}
1325 )
1326 self.assertEqual(
1327 e.exception.http_code,
1328 HTTPStatus.UNPROCESSABLE_ENTITY,
1329 "Wrong HTTP status code",
1330 )
1331 self.assertIn(
1332 norm(
1333 "Error in pyangbind validation: {} ({})".format(
1334 "json object contained a key that did not exist", "key"
1335 )
1336 ),
1337 norm(str(e.exception)),
1338 "Wrong exception text",
1339 )
1340 finally:
1341 test_nsd["designer"] = tmp
1342 with self.subTest(
1343 i=5, t="Check Input Validation: mgmt-network+virtual-link-protocol-data"
1344 ):
1345 df = test_nsd["df"][0]
1346 mgmt_profile = {
1347 "id": "id",
1348 "virtual-link-desc-id": "mgmt",
1349 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1350 }
1351 df["virtual-link-profile"] = [mgmt_profile]
1352 try:
1353 with self.assertRaises(
1354 EngineException, msg="Accepted VLD with mgmt-network+ip-profile"
1355 ) as e:
1356 self.topic.upload_content(
1357 fake_session, did, test_nsd, {}, {"Content-Type": []}
1358 )
1359 self.assertEqual(
1360 e.exception.http_code,
1361 HTTPStatus.UNPROCESSABLE_ENTITY,
1362 "Wrong HTTP status code",
1363 )
1364 self.assertIn(
1365 norm(
1366 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1367 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1368 df["id"], mgmt_profile["id"]
1369 )
1370 ),
1371 norm(str(e.exception)),
1372 "Wrong exception text",
1373 )
1374 finally:
1375 del df["virtual-link-profile"]
1376 with self.subTest(i=6, t="Check Descriptor Dependencies: vnfd-id[]"):
1377 self.db.get_one.side_effect = [
1378 {"_id": did, "_admin": db_nsd_content["_admin"]},
1379 None,
1380 ]
1381 self.db.get_list.return_value = []
1382 try:
1383 with self.assertRaises(
1384 EngineException, msg="Accepted wrong VNFD ID reference"
1385 ) as e:
1386 self.topic.upload_content(
1387 fake_session, did, test_nsd, {}, {"Content-Type": []}
1388 )
1389 self.assertEqual(
1390 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1391 )
1392 self.assertIn(
1393 norm(
1394 "'vnfd-id'='{}' references a non existing vnfd".format(
1395 test_nsd["vnfd-id"][0]
1396 )
1397 ),
1398 norm(str(e.exception)),
1399 "Wrong exception text",
1400 )
1401 finally:
1402 pass
1403 with self.subTest(
1404 i=7,
1405 t="Check Descriptor Dependencies: "
1406 "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]",
1407 ):
1408 vnfd_descriptor = deepcopy(db_vnfd_content)
1409 df = test_nsd["df"][0]
1410 affected_vnf_profile = df["vnf-profile"][0]
1411 affected_virtual_link = affected_vnf_profile["virtual-link-connectivity"][1]
1412 affected_cpd = vnfd_descriptor["ext-cpd"].pop()
1413 self.db.get_one.side_effect = [
1414 {"_id": did, "_admin": db_nsd_content["_admin"]},
1415 None,
1416 ]
1417 self.db.get_list.return_value = [vnfd_descriptor]
1418 try:
1419 with self.assertRaises(
1420 EngineException, msg="Accepted wrong VLD CP reference"
1421 ) as e:
1422 self.topic.upload_content(
1423 fake_session, did, test_nsd, {}, {"Content-Type": []}
1424 )
1425 self.assertEqual(
1426 e.exception.http_code,
1427 HTTPStatus.UNPROCESSABLE_ENTITY,
1428 "Wrong HTTP status code",
1429 )
1430 self.assertIn(
1431 norm(
1432 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1433 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1434 "non existing ext-cpd:id inside vnfd '{}'".format(
1435 df["id"],
1436 affected_vnf_profile["id"],
1437 affected_virtual_link["virtual-link-profile-id"],
1438 affected_cpd["id"],
1439 vnfd_descriptor["id"],
1440 )
1441 ),
1442 norm(str(e.exception)),
1443 "Wrong exception text",
1444 )
1445 finally:
1446 pass
1447 return
1448
1449 def test_edit_nsd(self):
1450 nsd_content = deepcopy(db_nsd_content)
1451 did = nsd_content["_id"]
1452 self.fs.file_exists.return_value = True
1453 self.fs.dir_ls.return_value = True
1454 with self.subTest(i=1, t="Normal Edition"):
1455 now = time()
1456 self.db.get_one.side_effect = [deepcopy(nsd_content), None]
1457 self.db.get_list.return_value = [db_vnfd_content]
1458 data = {"id": "new-nsd-id", "name": "new-nsd-name"}
1459 self.topic.edit(fake_session, did, data)
1460 db_args = self.db.replace.call_args[0]
1461 msg_args = self.msg.write.call_args[0]
1462 data["_id"] = did
1463 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1464 self.assertEqual(msg_args[1], "edited", "Wrong message action")
1465 self.assertEqual(msg_args[2], data, "Wrong message content")
1466 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1467 self.assertEqual(db_args[1], did, "Wrong DB ID")
1468 self.assertEqual(
1469 db_args[2]["_admin"]["created"],
1470 nsd_content["_admin"]["created"],
1471 "Wrong creation time",
1472 )
1473 self.assertGreater(
1474 db_args[2]["_admin"]["modified"], now, "Wrong modification time"
1475 )
1476 self.assertEqual(
1477 db_args[2]["_admin"]["projects_read"],
1478 nsd_content["_admin"]["projects_read"],
1479 "Wrong read-only project list",
1480 )
1481 self.assertEqual(
1482 db_args[2]["_admin"]["projects_write"],
1483 nsd_content["_admin"]["projects_write"],
1484 "Wrong read-write project list",
1485 )
1486 self.assertEqual(db_args[2]["id"], data["id"], "Wrong NSD ID")
1487 self.assertEqual(db_args[2]["name"], data["name"], "Wrong NSD Name")
1488 with self.subTest(i=2, t="Conflict on Edit"):
1489 data = {"id": "fake-nsd-id", "name": "new-nsd-name"}
1490 self.db.get_one.side_effect = [
1491 nsd_content,
1492 {"_id": str(uuid4()), "id": data["id"]},
1493 ]
1494 with self.assertRaises(
1495 EngineException, msg="Accepted existing NSD ID"
1496 ) as e:
1497 self.topic.edit(fake_session, did, data)
1498 self.assertEqual(
1499 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1500 )
1501 self.assertIn(
1502 norm(
1503 "{} with id '{}' already exists for this project".format(
1504 "nsd", data["id"]
1505 )
1506 ),
1507 norm(str(e.exception)),
1508 "Wrong exception text",
1509 )
1510 with self.subTest(i=3, t="Check Envelope"):
1511 data = {"nsd": {"nsd": {"id": "new-nsd-id", "name": "new-nsd-name"}}}
1512 self.db.get_one.side_effect = [nsd_content, None]
1513 with self.assertRaises(
1514 EngineException, msg="Accepted NSD with wrong envelope"
1515 ) as e:
1516 self.topic.edit(fake_session, did, data, content=nsd_content)
1517 self.assertEqual(
1518 e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code"
1519 )
1520 self.assertIn(
1521 "'nsd' must be a list of only one element",
1522 norm(str(e.exception)),
1523 "Wrong exception text",
1524 )
1525 return
1526
1527 def test_delete_nsd(self):
1528 did = db_nsd_content["_id"]
1529 self.db.get_one.return_value = db_nsd_content
1530 p_id = db_nsd_content["_admin"]["projects_read"][0]
1531 with self.subTest(i=1, t="Normal Deletion"):
1532 self.db.get_list.return_value = []
1533 self.db.del_one.return_value = {"deleted": 1}
1534 self.topic.delete(fake_session, did)
1535 db_args = self.db.del_one.call_args[0]
1536 msg_args = self.msg.write.call_args[0]
1537 self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic")
1538 self.assertEqual(msg_args[1], "deleted", "Wrong message action")
1539 self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content")
1540 self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
1541 self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID")
1542 self.assertEqual(
1543 db_args[1]["_admin.projects_write.cont"],
1544 [p_id, "ANY"],
1545 "Wrong DB filter",
1546 )
1547 db_g1_args = self.db.get_one.call_args[0]
1548 self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
1549 self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB NSD ID")
1550 db_gl_calls = self.db.get_list.call_args_list
1551 self.assertEqual(db_gl_calls[0][0][0], "nsrs", "Wrong DB topic")
1552 # self.assertEqual(db_gl_calls[0][0][1]["nsd-id"], did, "Wrong DB NSD ID") # Filter changed after call
1553 self.assertEqual(db_gl_calls[1][0][0], "nsts", "Wrong DB topic")
1554 self.assertEqual(
1555 db_gl_calls[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"],
1556 db_nsd_content["id"],
1557 "Wrong DB NSD netslice-subnet nsd-ref",
1558 )
1559 self.db.set_one.assert_not_called()
1560 fs_del_calls = self.fs.file_delete.call_args_list
1561 self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
1562 self.assertEqual(fs_del_calls[1][0][0], did + "_", "Wrong FS folder id")
1563 with self.subTest(i=2, t="Conflict on Delete - NSD in use by nsr"):
1564 self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-nsr"}]
1565 with self.assertRaises(
1566 EngineException, msg="Accepted NSD in use by NSR"
1567 ) as e:
1568 self.topic.delete(fake_session, did)
1569 self.assertEqual(
1570 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1571 )
1572 self.assertIn(
1573 "there is at least one ns instance using this descriptor",
1574 norm(str(e.exception)),
1575 "Wrong exception text",
1576 )
1577 with self.subTest(i=3, t="Conflict on Delete - NSD in use by NST"):
1578 self.db.get_list.side_effect = [
1579 [],
1580 [{"_id": str(uuid4()), "name": "fake-nst"}],
1581 ]
1582 with self.assertRaises(
1583 EngineException, msg="Accepted NSD in use by NST"
1584 ) as e:
1585 self.topic.delete(fake_session, did)
1586 self.assertEqual(
1587 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1588 )
1589 self.assertIn(
1590 "there is at least one netslice template referencing this descriptor",
1591 norm(str(e.exception)),
1592 "Wrong exception text",
1593 )
1594 with self.subTest(i=4, t="Non-existent NSD"):
1595 excp_msg = "Not found any {} with filter='{}'".format("NSD", {"_id": did})
1596 self.db.get_one.side_effect = DbException(excp_msg, HTTPStatus.NOT_FOUND)
1597 with self.assertRaises(
1598 DbException, msg="Accepted non-existent NSD ID"
1599 ) as e:
1600 self.topic.delete(fake_session, did)
1601 self.assertEqual(
1602 e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code"
1603 )
1604 self.assertIn(
1605 norm(excp_msg), norm(str(e.exception)), "Wrong exception text"
1606 )
1607 with self.subTest(i=5, t="No delete because referenced by other project"):
1608 db_nsd_content["_admin"]["projects_read"].append("other_project")
1609 self.db.get_one = Mock(return_value=db_nsd_content)
1610 self.db.get_list = Mock(return_value=[])
1611 self.msg.write.reset_mock()
1612 self.db.del_one.reset_mock()
1613 self.fs.file_delete.reset_mock()
1614
1615 self.topic.delete(fake_session, did)
1616 self.db.del_one.assert_not_called()
1617 self.msg.write.assert_not_called()
1618 db_g1_args = self.db.get_one.call_args[0]
1619 self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic")
1620 self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID")
1621 db_s1_args = self.db.set_one.call_args
1622 self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic")
1623 self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID")
1624 self.assertIn(
1625 p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter"
1626 )
1627 self.assertIsNone(
1628 db_s1_args[1]["update_dict"], "Wrong DB update dictionary"
1629 )
1630 self.assertEqual(
1631 db_s1_args[1]["pull_list"],
1632 {"_admin.projects_read": (p_id,), "_admin.projects_write": (p_id,)},
1633 "Wrong DB pull_list dictionary",
1634 )
1635 self.fs.file_delete.assert_not_called()
1636 return
1637
1638 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
1639 self,
1640 ):
1641 indata = deepcopy(db_nsd_content)
1642 vld = indata["virtual-link-desc"][0]
1643 self.topic.validate_vld_mgmt_network_with_virtual_link_protocol_data(
1644 vld, indata
1645 )
1646
1647 def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_when_both_defined(
1648 self,
1649 ):
1650 indata = deepcopy(db_nsd_content)
1651 vld = indata["virtual-link-desc"][0]
1652 df = indata["df"][0]
1653 affected_vlp = {
1654 "id": "id",
1655 "virtual-link-desc-id": "mgmt",
1656 "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
1657 }
1658 df["virtual-link-profile"] = [affected_vlp]
1659 with self.assertRaises(EngineException) as e:
1660 self.topic.validate_vld_mgmt_network_with_virtual_link_protocol_data(
1661 vld, indata
1662 )
1663 self.assertEqual(
1664 e.exception.http_code,
1665 HTTPStatus.UNPROCESSABLE_ENTITY,
1666 "Wrong HTTP status code",
1667 )
1668 self.assertIn(
1669 norm(
1670 "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
1671 " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
1672 df["id"], affected_vlp["id"]
1673 )
1674 ),
1675 norm(str(e.exception)),
1676 "Wrong exception text",
1677 )
1678
1679 def test_validate_vnf_profiles_vnfd_id_on_valid_descriptor(self):
1680 indata = deepcopy(db_nsd_content)
1681 self.topic.validate_vnf_profiles_vnfd_id(indata)
1682
1683 def test_validate_vnf_profiles_vnfd_id_when_missing_vnfd(self):
1684 indata = deepcopy(db_nsd_content)
1685 df = indata["df"][0]
1686 affected_vnf_profile = df["vnf-profile"][0]
1687 indata["vnfd-id"] = ["non-existing-vnfd"]
1688 with self.assertRaises(EngineException) as e:
1689 self.topic.validate_vnf_profiles_vnfd_id(indata)
1690 self.assertEqual(
1691 e.exception.http_code,
1692 HTTPStatus.UNPROCESSABLE_ENTITY,
1693 "Wrong HTTP status code",
1694 )
1695 self.assertIn(
1696 norm(
1697 "Error at df[id='{}']:vnf_profile[id='{}']:vnfd-id='{}' "
1698 "does not match any vnfd-id".format(
1699 df["id"],
1700 affected_vnf_profile["id"],
1701 affected_vnf_profile["vnfd-id"],
1702 )
1703 ),
1704 norm(str(e.exception)),
1705 "Wrong exception text",
1706 )
1707
1708 def test_validate_df_vnf_profiles_constituent_connection_points_on_valid_descriptor(
1709 self,
1710 ):
1711 nsd_descriptor = deepcopy(db_nsd_content)
1712 vnfd_descriptor = deepcopy(db_vnfd_content)
1713 df = nsd_descriptor["df"][0]
1714 vnfds_index = {vnfd_descriptor["id"]: vnfd_descriptor}
1715 self.topic.validate_df_vnf_profiles_constituent_connection_points(
1716 df, vnfds_index
1717 )
1718
1719 def test_validate_df_vnf_profiles_constituent_connection_points_when_missing_connection_point(
1720 self,
1721 ):
1722 nsd_descriptor = deepcopy(db_nsd_content)
1723 vnfd_descriptor = deepcopy(db_vnfd_content)
1724 df = nsd_descriptor["df"][0]
1725 affected_vnf_profile = df["vnf-profile"][0]
1726 affected_virtual_link = affected_vnf_profile["virtual-link-connectivity"][1]
1727 vnfds_index = {vnfd_descriptor["id"]: vnfd_descriptor}
1728 affected_cpd = vnfd_descriptor["ext-cpd"].pop()
1729 with self.assertRaises(EngineException) as e:
1730 self.topic.validate_df_vnf_profiles_constituent_connection_points(
1731 df, vnfds_index
1732 )
1733 self.assertEqual(
1734 e.exception.http_code,
1735 HTTPStatus.UNPROCESSABLE_ENTITY,
1736 "Wrong HTTP status code",
1737 )
1738 self.assertIn(
1739 norm(
1740 "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity"
1741 "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a "
1742 "non existing ext-cpd:id inside vnfd '{}'".format(
1743 df["id"],
1744 affected_vnf_profile["id"],
1745 affected_virtual_link["virtual-link-profile-id"],
1746 affected_cpd["id"],
1747 vnfd_descriptor["id"],
1748 )
1749 ),
1750 norm(str(e.exception)),
1751 "Wrong exception text",
1752 )
1753
1754 def test_check_conflict_on_edit_when_missing_constituent_vnfd_id(self):
1755 nsd_descriptor = deepcopy(db_nsd_content)
1756 invalid_vnfd_id = "invalid-vnfd-id"
1757 nsd_descriptor["id"] = "invalid-vnfd-id-ns"
1758 nsd_descriptor["vnfd-id"][0] = invalid_vnfd_id
1759 nsd_descriptor["df"][0]["vnf-profile"][0]["vnfd-id"] = invalid_vnfd_id
1760 nsd_descriptor["df"][0]["vnf-profile"][1]["vnfd-id"] = invalid_vnfd_id
1761 with self.assertRaises(EngineException) as e:
1762 self.db.get_list.return_value = []
1763 nsd_descriptor = self.topic.check_conflict_on_edit(
1764 fake_session, nsd_descriptor, [], "id"
1765 )
1766 self.assertEqual(
1767 e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code"
1768 )
1769 self.assertIn(
1770 norm(
1771 "Descriptor error at 'vnfd-id'='{}' references a non "
1772 "existing vnfd".format(invalid_vnfd_id)
1773 ),
1774 norm(str(e.exception)),
1775 "Wrong exception text",
1776 )
1777
1778
1779 if __name__ == "__main__":
1780 unittest.main()