2 # Licensed under the Apache License, Version 2.0 (the "License"); you may
3 # not use this file except in compliance with the License. You may obtain
4 # a copy of the License at
6 # http://www.apache.org/licenses/LICENSE-2.0
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
10 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
11 # License for the specific language governing permissions and limitations
14 # For those usages not covered by the Apache License, Version 2.0 please
15 # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
18 from contextlib
import contextmanager
21 from unittest
.mock
import Mock
, mock_open
# patch, MagicMock
22 from osm_common
.dbbase
import DbException
23 from osm_nbi
.engine
import EngineException
24 from osm_common
.dbmemory
import DbMemory
25 from osm_common
.fsbase
import FsBase
26 from osm_common
.msgbase
import MsgBase
27 from osm_common
import dbbase
28 from http
import HTTPStatus
29 from osm_nbi
.instance_topics
import NsLcmOpTopic
, NsrTopic
30 from osm_nbi
.tests
.test_db_descriptors
import (
37 from copy
import deepcopy
41 class TestNsLcmOpTopic(unittest
.TestCase
):
44 self
.fs
= Mock(FsBase())
45 self
.fs
.get_params
.return_value
= {"./fake/folder"}
46 self
.fs
.file_open
= mock_open()
47 self
.msg
= Mock(MsgBase())
49 self
.nslcmop_topic
= NsLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
50 self
.nslcmop_topic
.check_quota
= Mock(return_value
=None) # skip quota
52 self
.db
.create_list("vim_accounts", yaml
.safe_load(db_vim_accounts_text
))
53 self
.db
.create_list("nsds", yaml
.safe_load(db_nsds_text
))
54 self
.db
.create_list("vnfds", yaml
.safe_load(db_vnfds_text
))
55 self
.db
.create_list("vnfrs", yaml
.safe_load(db_vnfrs_text
))
56 self
.db
.create_list("nsrs", yaml
.safe_load(db_nsrs_text
))
57 self
.db
.create
= Mock(return_value
="created_id")
58 self
.nsd
= self
.db
.get_list("nsds")[0]
59 self
.nsd_id
= self
.nsd
["_id"]
60 self
.nsr
= self
.db
.get_list("nsrs")[0]
61 self
.nsr_id
= self
.nsr
["_id"]
62 self
.nsr_project
= self
.nsr
["_admin"]["projects_read"][0]
64 self
.vim
= self
.db
.get_list("vim_accounts")[0]
65 self
.vim_id
= self
.vim
["_id"]
67 def test_create_instantiate(self
):
68 self
.db
.set_one
= Mock(return_value
={"updated": 1})
73 "project_id": [self
.nsr_project
],
78 "nsInstanceId": self
.nsr_id
,
80 "vimAccountId": self
.vim_id
,
81 "additionalParamsForVnf": [
83 "member-vnf-index": "1",
84 "additionalParams": {"touch_filename": "file"},
87 "member-vnf-index": "2",
88 "additionalParams": {"touch_filename": "file"},
93 "member-vnf-index": "1",
99 "name": "dataVM-eth0",
100 "ip-address": "10.11.12.13",
101 "floating-ip-required": True,
107 {"name": "internal", "vim-network-id": "vim-net-id"}
111 "lcmOperationType": "instantiate",
116 nslcmop_id
, _
= self
.nslcmop_topic
.new(
117 rollback
, session
, indata
=deepcopy(indata
), kwargs
=None, headers
=headers
120 # check nslcmop is created at database
122 self
.db
.create
.call_count
,
124 "database create not called, or called more than once",
126 _call
= self
.db
.create
.call_args_list
[0]
128 _call
[0][0], "nslcmops", "must be create a nslcmops entry at database"
131 created_nslcmop
= _call
[0][1]
134 created_nslcmop
["_id"],
135 "mismatch between return id and database '_id'",
139 created_nslcmop
["nsInstanceId"],
140 "bad reference id from nslcmop to nsr",
143 created_nslcmop
["_admin"].get("projects_read"),
144 "Database record must contain '_amdin.projects_read'",
148 created_nslcmop
["_admin"],
149 "Database record must contain '_admin.created'",
152 created_nslcmop
["lcmOperationType"] == "instantiate",
153 "Database record must contain 'lcmOperationType=instantiate'",
158 len(self
.db
.set_one
.call_args_list
) + 1,
159 "rollback mismatch with created/set items at database",
162 # test parameters with error
163 bad_id
= "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
167 {"nsInstanceId": bad_id
},
169 HTTPStatus
.NOT_FOUND
,
170 ("not found", bad_id
),
173 # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)), # TODO add "vim"
175 "bad member-vnf-index",
176 {"vnf.0.member-vnf-index": "k"},
178 HTTPStatus
.BAD_REQUEST
,
182 for message
, kwargs_
, expect_exc
, expect_code
, expect_text_list
in test_set
:
183 with self
.assertRaises(expect_exc
, msg
=message
) as e
:
184 self
.nslcmop_topic
.new(
187 indata
=deepcopy(indata
),
192 self
.assertTrue(e
.exception
.http_code
== expect_code
)
194 for expect_text
in expect_text_list
:
197 str(e
.exception
).lower(),
198 "Expected '{}' at exception text".format(expect_text
),
201 def test_check_ns_operation_action(self
):
202 nsrs
= self
.db
.get_list("nsrs")[0]
206 "member_vnf_index": "1",
208 "primitive": "touch",
209 "primitive_params": {"filename": "file"},
212 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "action", indata
)
214 indata_copy
= indata
.copy()
215 if k
== "primitive_params":
217 indata_copy
[k
] = "non_existing"
218 with self
.assertRaises(EngineException
) as exc_manager
:
219 self
.nslcmop_topic
._check
_ns
_operation
(
220 session
, nsrs
, "action", indata_copy
222 exc
= exc_manager
.exception
225 HTTPStatus
.BAD_REQUEST
,
226 "Engine exception bad http_code with {}".format(indata_copy
),
229 def test_update_remove_vnf(self
):
230 vnfr_id
= self
.db
.get_list("vnfrs")[0]["_id"]
234 {"_id": self
.nsr_id
},
235 {"_admin.nsState": "INSTANTIATED"},
238 "lcmOperationType": "update",
239 "updateType": "REMOVE_VNF",
240 "nsInstanceId": self
.nsr_id
,
241 "removeVnfInstanceId": vnfr_id
,
248 "project_id": [self
.nsr_project
],
254 nslcmop_id
, _
= self
.nslcmop_topic
.new(
255 rollback
, session
, indata
, kwargs
=None, headers
=headers
259 self
.db
.create
.call_count
,
261 "database create not called, or called more than once",
263 _call
= self
.db
.create
.call_args_list
[0]
265 _call
[0][0], "nslcmops", "nslcmops entry must be created at database"
267 created_nslcmop
= _call
[0][1]
270 created_nslcmop
["nsInstanceId"],
271 "mismatch between nsId '_id' in created nslcmop and database nsr",
274 created_nslcmop
["lcmOperationType"] == "update",
275 "Database record must contain 'lcmOperationType=update'",
278 created_nslcmop
["operationParams"]["updateType"] == "REMOVE_VNF",
279 "Database record must contain 'updateType=REMOVE_VNF'",
282 def test_migrate(self
):
283 _
= self
.db
.get_list("vnfrs")[0]["_id"]
287 {"_id": self
.nsr_id
},
288 {"_admin.nsState": "INSTANTIATED"},
294 "project_id": [self
.nsr_project
],
300 with self
.subTest(i
=1, t
="Migration for Specific VM"):
302 "lcmOperationType": "migrate",
303 "nsInstanceId": self
.nsr_id
,
304 "migrateToHost": "sample02",
305 "vdu": {"vduCountIndex": 0, "vduId": "mgmtVM"},
306 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
308 nslcmop_id
, _
= self
.nslcmop_topic
.new(
309 rollback
, session
, indata
, kwargs
=None, headers
=headers
313 self
.db
.create
.call_count
,
315 "database create not called, or called more than once",
317 _call
= self
.db
.create
.call_args_list
[0]
319 _call
[0][0], "nslcmops", "nslcmops entry must be created at database"
321 created_nslcmop
= _call
[0][1]
324 created_nslcmop
["nsInstanceId"],
325 "mismatch between nsId '_id' in created nslcmop and database nsr",
328 created_nslcmop
["lcmOperationType"] == "migrate",
329 "Database record must contain 'lcmOperationType=migrate'",
331 with self
.subTest(i
=2, t
="Migration of all VDUs in a VNF"):
333 "lcmOperationType": "migrate",
334 "nsInstanceId": self
.nsr_id
,
335 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
337 nslcmop_id
, _
= self
.nslcmop_topic
.new(
338 rollback
, session
, indata
, kwargs
=None, headers
=headers
342 self
.db
.create
.call_count
,
344 "database create not called, or called more than once",
346 _call
= self
.db
.create
.call_args_list
[0]
348 _call
[0][0], "nslcmops", "nslcmops entry must be created at database"
350 created_nslcmop
= _call
[0][1]
353 created_nslcmop
["nsInstanceId"],
354 "mismatch between nsId '_id' in created nslcmop and database nsr",
357 created_nslcmop
["lcmOperationType"] == "migrate",
358 "Database record must contain 'lcmOperationType=migrate'",
360 with self
.subTest(i
=3, t
="Migration failure - vduId not provided in vdu "):
362 "lcmOperationType": "migrate",
363 "nsInstanceId": self
.nsr_id
,
364 "migrateToHost": "sample02",
365 "vdu": {"vduCountIndex": 0},
366 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f",
369 with self
.assertRaises(Exception) as e
:
370 nslcmop_id
, _
= self
.nslcmop_topic
.new(
371 rollback
, session
, indata
, kwargs
=None, headers
=headers
374 "Format error at 'vdu' ''vduId' is a required property'"
379 class TestNsLcmOpTopicWithMock(unittest
.TestCase
):
381 self
.db
= Mock(dbbase
.DbBase())
382 self
.fs
= Mock(FsBase())
383 self
.fs
.get_params
.return_value
= {"./fake/folder"}
384 self
.fs
.file_open
= mock_open()
385 self
.msg
= Mock(MsgBase())
387 self
.nslcmop_topic
= NsLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
389 def test_get_vnfd_from_vnf_member_revision(self
):
390 test_vnfr
= yaml
.safe_load(db_vnfrs_text
)[0]
391 test_vnfd
= yaml
.safe_load(db_vnfds_text
)
392 self
.db
.get_one
.side_effect
= [test_vnfr
, test_vnfd
]
393 _
= self
.nslcmop_topic
._get
_vnfd
_from
_vnf
_member
_index
("1", test_vnfr
["_id"])
395 self
.db
.get_one
.call_args_list
[0][0][0],
397 "Incorrect first DB lookup",
400 self
.db
.get_one
.call_args_list
[1][0][0],
402 "Incorrect second DB lookup",
405 def test_get_vnfd_from_vnf_member_no_revision(self
):
406 test_vnfr
= yaml
.safe_load(db_vnfrs_text
)[0]
407 test_vnfr
["revision"] = 3
408 test_vnfd
= yaml
.safe_load(db_vnfds_text
)
409 self
.db
.get_one
.side_effect
= [test_vnfr
, test_vnfd
]
410 _
= self
.nslcmop_topic
._get
_vnfd
_from
_vnf
_member
_index
("1", test_vnfr
["_id"])
412 self
.db
.get_one
.call_args_list
[0][0][0],
414 "Incorrect first DB lookup",
417 self
.db
.get_one
.call_args_list
[1][0][0],
419 "Incorrect second DB lookup",
423 def assertNotRaises(self
, exception_type
):
426 except exception_type
:
427 raise self
.failureException("{} raised".format(exception_type
.__name
__))
429 def test_check_ns_update_operation(self
):
431 self
.nslcmop_topic
= NsLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
434 with self
.subTest(i
=1, t
="VNF instance does not belong to NS"):
435 test_vnfr
= yaml
.safe_load(db_vnfrs_text
)
436 test_vnfr
[0]["revision"] = 2
437 test_nsr
= yaml
.safe_load(db_nsrs_text
)
438 test_nsr
[0]["constituent-vnfr-ref"][
440 ] = "99d90b0c-faff-4b9f-bccd-017f33985984"
441 self
.db
.create_list("vnfrs", test_vnfr
)
442 self
.db
.create_list("nsrs", test_nsr
)
443 nsrs
= self
.db
.get_list("nsrs")[0]
445 "updateType": "CHANGE_VNFPKG",
446 "changeVnfPackageData": {
447 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
448 "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
450 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
452 with self
.assertRaises(EngineException
) as expected_exception
:
453 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "update", indata
)
455 str(expected_exception
.exception
),
456 "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984"
457 " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85",
460 with self
.subTest(i
=2, t
="Ns update request validated with no exception"):
461 test_vnfr
= yaml
.safe_load(db_vnfrs_text
)
462 test_vnfr
[0]["revision"] = 2
463 test_nsr
= yaml
.safe_load(db_nsrs_text
)
464 self
.db
.create_list("vnfrs", test_vnfr
)
465 self
.db
.create_list("nsrs", test_nsr
)
466 nsrs
= self
.db
.get_list("nsrs")[1]
468 "updateType": "CHANGE_VNFPKG",
469 "changeVnfPackageData": {
470 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
471 "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
473 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
475 with self
.assertNotRaises(EngineException
):
476 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "update", indata
)
479 i
=3, t
="Ns update request rejected because of too small timeout"
482 "updateType": "CHANGE_VNFPKG",
483 "changeVnfPackageData": {
484 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
485 "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
487 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
488 "timeout_ns_update": 50,
490 with self
.assertRaises(EngineException
) as expected_exception
:
491 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "update", indata
)
493 str(expected_exception
.exception
),
494 "Error in validating ns-update request: 50 second is not enough "
495 "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
498 with self
.subTest(i
=4, t
="wrong vnfdid is given as an update parameter"):
499 test_vnfr
= yaml
.safe_load(db_vnfrs_text
)
500 test_vnfr
[0]["revision"] = 2
501 test_nsr
= yaml
.safe_load(db_nsrs_text
)
502 self
.db
.create_list("vnfrs", test_vnfr
)
503 self
.db
.create_list("nsrs", test_nsr
)
504 nsrs
= self
.db
.get_list("nsrs")[2]
506 "updateType": "CHANGE_VNFPKG",
507 "changeVnfPackageData": {
508 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
509 "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
511 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
513 with self
.assertRaises(EngineException
) as expected_exception
:
514 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "update", indata
)
516 str(expected_exception
.exception
),
517 "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not "
518 "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of "
519 "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
523 i
=5, t
="Ns update REMOVE_VNF request validated with no exception"
525 test_vnfr
= yaml
.safe_load(db_vnfrs_text
)
526 test_vnfr
[0]["revision"] = 2
527 test_nsr
= yaml
.safe_load(db_nsrs_text
)
528 self
.db
.create_list("vnfrs", test_vnfr
)
529 self
.db
.create_list("nsrs", test_nsr
)
530 nsrs
= self
.db
.get_list("nsrs")[1]
532 "updateType": "REMOVE_VNF",
533 "removeVnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
534 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
536 with self
.assertNotRaises(EngineException
):
537 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "update", indata
)
540 class TestNsrTopic(unittest
.TestCase
):
543 self
.fs
= Mock(FsBase())
544 self
.fs
.get_params
.return_value
= {"./fake/folder"}
545 self
.fs
.file_open
= mock_open()
546 self
.msg
= Mock(MsgBase())
548 self
.nsr_topic
= NsrTopic(self
.db
, self
.fs
, self
.msg
, None)
549 self
.nsr_topic
.check_quota
= Mock(return_value
=None) # skip quota
551 self
.db
.create_list("vim_accounts", yaml
.safe_load(db_vim_accounts_text
))
552 self
.db
.create_list("nsds", yaml
.safe_load(db_nsds_text
))
553 self
.db
.create_list("vnfds", yaml
.safe_load(db_vnfds_text
))
554 self
.db
.create
= Mock(return_value
="created_id")
555 self
.nsd
= self
.db
.get_list("nsds")[0]
556 self
.nsd_id
= self
.nsd
["_id"]
557 self
.nsd_project
= self
.nsd
["_admin"]["projects_read"][0]
559 self
.vim
= self
.db
.get_list("vim_accounts")[0]
560 self
.vim_id
= self
.vim
["_id"]
562 def test_create(self
):
567 "project_id": [self
.nsd_project
],
571 "nsdId": self
.nsd_id
,
573 "vimAccountId": self
.vim_id
,
574 "additionalParamsForVnf": [
576 "member-vnf-index": "hackfest_vnf1",
577 "additionalParams": {"touch_filename": "file"},
580 "member-vnf-index": "hackfest_vnf2",
581 "additionalParams": {"touch_filename": "file"},
589 rollback
, session
, indata
=indata
, kwargs
=None, headers
=headers
592 # check vnfrs and nsrs created in whatever order
596 for _call
in self
.db
.create
.call_args_list
:
597 assert len(_call
[0]) >= 2, "called db.create with few parameters"
598 created_item
= _call
[0][1]
599 if _call
[0][0] == "vnfrs":
600 created_vnfrs
.append(created_item
)
602 "member-vnf-index-ref",
604 "Created item must contain member-vnf-index-ref section",
609 created_item
["nsr-id-ref"],
610 "bad reference id from vnfr to nsr",
613 nsr_id
= created_item
["nsr-id-ref"]
615 "nsState" not in created_item
["_admin"],
616 "Database VNF record must not contain '_admin.nsState'",
619 elif _call
[0][0] == "nsrs":
620 created_nsrs
.append(created_item
)
623 nsr_id
, created_item
["_id"], "bad reference id from vnfr to nsr"
626 nsr_id
= created_item
["_id"]
629 created_item
["_admin"]["nsState"] == "NOT_INSTANTIATED",
630 "Database record must contain '_admin.nsState=NOT INSTANTIATE'",
633 assert True, "created an unknown record {} at database".format(
638 created_item
["_admin"].get("projects_read"),
639 "Database record must contain '_amdin.projects_read'",
643 created_item
["_admin"],
644 "Database record must contain '_admin.created'",
648 len(created_vnfrs
), 2, "created a mismatch number of vnfr at database"
652 created_vnfrs
[0]["vdur"][0]["interfaces"][0]["position"],
654 "vdur first interface position does not match",
658 created_vnfrs
[0]["vdur"][0]["interfaces"][1]["position"],
660 "vdur second interface position does not match",
664 len(created_nsrs
), 1, "Only one nsrs must be created at database"
668 len(created_vnfrs
) + 1,
669 "rollback mismatch with created items at database",
672 # test parameters with error
673 bad_id
= "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
680 HTTPStatus
.NOT_FOUND
,
681 ("not found", bad_id
),
683 # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)), # TODO add "vim"
685 "additional params not supply",
686 {"additionalParamsForVnf.0.member-vnf-index": "k"},
688 HTTPStatus
.BAD_REQUEST
,
692 for message
, kwargs_
, expect_exc
, expect_code
, expect_text_list
in test_set
:
693 with self
.assertRaises(expect_exc
, msg
=message
) as e
:
697 indata
=deepcopy(indata
),
702 self
.assertTrue(e
.exception
.http_code
== expect_code
)
704 for expect_text
in expect_text_list
:
707 str(e
.exception
).lower(),
708 "Expected '{}' at exception text".format(expect_text
),
711 def test_show_instance(self
):
716 "project_id": [self
.nsd_project
],
720 for refresh_status
in ("true", "false"):
721 self
.db
.create_list("nsrs", yaml
.safe_load(db_nsrs_text
))
722 actual_nsr
= self
.db
.get_list("nsrs")[0]
723 nsr_id
= actual_nsr
["_id"]
724 filter_q
["vcaStatus-refresh"] = refresh_status
725 expected_nsr
= self
.nsr_topic
.show(session
, nsr_id
, filter_q
=filter_q
)
726 self
.nsr_topic
.delete(session
, nsr_id
)
727 actual_nsr
.pop("_admin")
728 expected_nsr
.pop("_admin")
730 expected_nsr
, actual_nsr
, "Database nsr and show() nsr do not match."
733 def test_vca_status_refresh(self
):
738 "project_id": [self
.nsd_project
],
741 filter_q
= {"vcaStatus-refresh": "true"}
743 self
.db
.create_list("nsrs", yaml
.safe_load(db_nsrs_text
))
744 nsr
= self
.db
.get_list("nsrs")[0]
746 # When vcaStatus-refresh is true
747 filter_q
["vcaStatus-refresh"] = "true"
748 self
.nsr_topic
.vca_status_refresh(session
, nsr
, filter_q
)
749 msg_args
= self
.msg
.write
.call_args
[0]
750 self
.assertEqual(msg_args
[1], "vca_status_refresh", "Wrong message action")
751 self
.assertGreater(nsr
["_admin"]["modified"], time() - time_delta
)
753 # When vcaStatus-refresh is false but modified time is within threshold
754 filter_q
["vcaStatus-refresh"] = "false"
756 nsr
["_admin"]["modified"] = time_now
757 self
.nsr_topic
.vca_status_refresh(session
, nsr
, filter_q
)
758 msg_args
= self
.msg
.write
.call_args
[1]
759 self
.assertEqual(msg_args
, {}, "Message should not be sent.")
761 nsr
["_admin"]["modified"], time_now
, "Modified time should not be changed."
764 # When vcaStatus-refresh is false but modified time is less than threshold
765 filter_q
["vcaStatus-refresh"] = "false"
766 nsr
["_admin"]["modified"] = time() - (2 * time_delta
)
767 self
.nsr_topic
.vca_status_refresh(session
, nsr
, filter_q
)
768 msg_args
= self
.msg
.write
.call_args
[0]
769 self
.assertEqual(msg_args
[1], "vca_status_refresh", "Wrong message action")
770 self
.nsr_topic
.delete(session
, nsr
["_id"])
772 nsr
["_admin"]["modified"],
774 "Modified time is not changed.",
777 def test_delete_ns(self
):
778 self
.db
.create_list("nsrs", yaml
.safe_load(db_nsrs_text
))
779 self
.nsr
= self
.db
.get_list("nsrs")[0]
780 self
.nsr_id
= self
.nsr
["_id"]
781 self
.db_set_one
= self
.db
.set_one
782 p_id
= self
.nsd_project
789 "project_id": [p_id
],
796 "project_id": [p_other
],
806 with self
.subTest(i
=1, t
="Normal Deletion"):
807 self
.db
.del_one
= Mock()
808 self
.db
.set_one
= Mock()
809 self
.nsr_topic
.delete(session
, self
.nsr_id
)
811 db_args_ro_nsrs
= self
.db
.del_one
.call_args_list
[1][0]
812 db_args
= self
.db
.del_one
.call_args_list
[0][0]
813 msg_args
= self
.msg
.write
.call_args
[0]
815 msg_args
[0], self
.nsr_topic
.topic_msg
, "Wrong message topic"
817 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
818 self
.assertEqual(msg_args
[2], {"_id": self
.nsr_id
}, "Wrong message content")
819 self
.assertEqual(db_args_ro_nsrs
[0], "ro_nsrs", "Wrong DB topic")
820 self
.assertEqual(db_args
[0], self
.nsr_topic
.topic
, "Wrong DB topic")
821 self
.assertEqual(db_args
[1]["_id"], self
.nsr_id
, "Wrong DB ID")
823 db_args
[1]["_admin.projects_read.cont"], [p_id
], "Wrong DB filter"
825 self
.db
.set_one
.assert_not_called()
826 fs_del_calls
= self
.fs
.file_delete
.call_args_list
827 self
.assertEqual(fs_del_calls
[0][0][0], self
.nsr_id
, "Wrong FS file id")
828 with self
.subTest(i
=2, t
="No delete because referenced by other project"):
831 {"_id": self
.nsr_id
},
834 "_admin.projects_read": p_other
,
835 "_admin.projects_write": p_other
,
838 self
.db
.del_one
.reset_mock()
839 self
.db
.set_one
.reset_mock()
840 self
.msg
.write
.reset_mock()
841 self
.fs
.file_delete
.reset_mock()
843 self
.nsr_topic
.delete(session2
, self
.nsr_id
)
844 self
.db
.del_one
.assert_not_called()
845 self
.msg
.write
.assert_not_called()
846 db_s1_args
= self
.db
.set_one
.call_args
847 self
.assertEqual(db_s1_args
[0][0], self
.nsr_topic
.topic
, "Wrong DB topic")
848 self
.assertEqual(db_s1_args
[0][1]["_id"], self
.nsr_id
, "Wrong DB ID")
850 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
853 db_s1_args
[1]["pull_list"],
854 {"_admin.projects_read": [p_other
], "_admin.projects_write": [p_other
]},
855 "Wrong DB pull_list dictionary",
857 self
.fs
.file_delete
.assert_not_called()
858 with self
.subTest(i
=4, t
="Delete with force and admin"):
859 self
.db
.del_one
.reset_mock()
860 self
.db
.set_one
.reset_mock()
861 self
.msg
.write
.reset_mock()
862 self
.fs
.file_delete
.reset_mock()
863 self
.nsr_topic
.delete(session_force
, self
.nsr_id
)
865 db_args_ro_nsrs
= self
.db
.del_one
.call_args_list
[1][0]
866 db_args
= self
.db
.del_one
.call_args_list
[0][0]
867 msg_args
= self
.msg
.write
.call_args
[0]
869 msg_args
[0], self
.nsr_topic
.topic_msg
, "Wrong message topic"
871 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
872 self
.assertEqual(msg_args
[2], {"_id": self
.nsr_id
}, "Wrong message content")
873 self
.assertEqual(db_args_ro_nsrs
[0], "ro_nsrs", "Wrong DB topic")
874 self
.assertEqual(db_args
[0], self
.nsr_topic
.topic
, "Wrong DB topic")
875 self
.assertEqual(db_args
[1]["_id"], self
.nsr_id
, "Wrong DB ID")
876 self
.db
.set_one
.assert_not_called()
877 fs_del_calls
= self
.fs
.file_delete
.call_args_list
878 self
.assertEqual(fs_del_calls
[0][0][0], self
.nsr_id
, "Wrong FS file id")
879 with self
.subTest(i
=3, t
="Conflict on Delete - NS in INSTANTIATED state"):
882 {"_id": self
.nsr_id
},
883 {"_admin.nsState": "INSTANTIATED"},
885 "_admin.projects_read": p_other
,
886 "_admin.projects_write": p_other
,
889 self
.db
.del_one
.reset_mock()
890 self
.db
.set_one
.reset_mock()
891 self
.msg
.write
.reset_mock()
892 self
.fs
.file_delete
.reset_mock()
894 with self
.assertRaises(
895 EngineException
, msg
="Accepted NSR with nsState INSTANTIATED"
897 self
.nsr_topic
.delete(session
, self
.nsr_id
)
899 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
901 self
.assertIn("INSTANTIATED", str(e
.exception
), "Wrong exception text")
902 # TODOD with self.subTest(i=3, t='Conflict on Delete - NS in use by NSI'):
904 with self
.subTest(i
=4, t
="Non-existent NS"):
905 self
.db
.del_one
.reset_mock()
906 self
.db
.set_one
.reset_mock()
907 self
.msg
.write
.reset_mock()
908 self
.fs
.file_delete
.reset_mock()
909 excp_msg
= "Not found"
910 with self
.assertRaises(
911 DbException
, msg
="Accepted non-existent NSD ID"
913 self
.nsr_topic
.delete(session2
, "other_id")
915 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
917 self
.assertIn(excp_msg
, str(e
.exception
), "Wrong exception text")
918 self
.assertIn("other_id", str(e
.exception
), "Wrong exception text")