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
53 "vim_accounts", yaml
.load(db_vim_accounts_text
, Loader
=yaml
.Loader
)
55 self
.db
.create_list("nsds", yaml
.load(db_nsds_text
, Loader
=yaml
.Loader
))
56 self
.db
.create_list("vnfds", yaml
.load(db_vnfds_text
, Loader
=yaml
.Loader
))
57 self
.db
.create_list("vnfrs", yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
))
58 self
.db
.create_list("nsrs", yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
))
59 self
.db
.create
= Mock(return_value
="created_id")
60 self
.nsd
= self
.db
.get_list("nsds")[0]
61 self
.nsd_id
= self
.nsd
["_id"]
62 self
.nsr
= self
.db
.get_list("nsrs")[0]
63 self
.nsr_id
= self
.nsr
["_id"]
64 self
.nsr_project
= self
.nsr
["_admin"]["projects_read"][0]
66 self
.vim
= self
.db
.get_list("vim_accounts")[0]
67 self
.vim_id
= self
.vim
["_id"]
69 def test_create_instantiate(self
):
70 self
.db
.set_one
= Mock(return_value
={"updated": 1})
75 "project_id": [self
.nsr_project
],
80 "nsInstanceId": self
.nsr_id
,
82 "vimAccountId": self
.vim_id
,
83 "additionalParamsForVnf": [
85 "member-vnf-index": "1",
86 "additionalParams": {"touch_filename": "file"},
89 "member-vnf-index": "2",
90 "additionalParams": {"touch_filename": "file"},
95 "member-vnf-index": "1",
101 "name": "dataVM-eth0",
102 "ip-address": "10.11.12.13",
103 "floating-ip-required": True,
109 {"name": "internal", "vim-network-id": "vim-net-id"}
113 "lcmOperationType": "instantiate",
118 nslcmop_id
, _
= self
.nslcmop_topic
.new(
119 rollback
, session
, indata
=deepcopy(indata
), kwargs
=None, headers
=headers
122 # check nslcmop is created at database
124 self
.db
.create
.call_count
,
126 "database create not called, or called more than once",
128 _call
= self
.db
.create
.call_args_list
[0]
130 _call
[0][0], "nslcmops", "must be create a nslcmops entry at database"
133 created_nslcmop
= _call
[0][1]
136 created_nslcmop
["_id"],
137 "mismatch between return id and database '_id'",
141 created_nslcmop
["nsInstanceId"],
142 "bad reference id from nslcmop to nsr",
145 created_nslcmop
["_admin"].get("projects_read"),
146 "Database record must contain '_amdin.projects_read'",
150 created_nslcmop
["_admin"],
151 "Database record must contain '_admin.created'",
154 created_nslcmop
["lcmOperationType"] == "instantiate",
155 "Database record must contain 'lcmOperationType=instantiate'",
160 len(self
.db
.set_one
.call_args_list
) + 1,
161 "rollback mismatch with created/set items at database",
164 # test parameters with error
165 bad_id
= "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
169 {"nsInstanceId": bad_id
},
171 HTTPStatus
.NOT_FOUND
,
172 ("not found", bad_id
),
175 # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)), # TODO add "vim"
177 "bad member-vnf-index",
178 {"vnf.0.member-vnf-index": "k"},
180 HTTPStatus
.BAD_REQUEST
,
184 for message
, kwargs_
, expect_exc
, expect_code
, expect_text_list
in test_set
:
185 with self
.assertRaises(expect_exc
, msg
=message
) as e
:
186 self
.nslcmop_topic
.new(
189 indata
=deepcopy(indata
),
194 self
.assertTrue(e
.exception
.http_code
== expect_code
)
196 for expect_text
in expect_text_list
:
199 str(e
.exception
).lower(),
200 "Expected '{}' at exception text".format(expect_text
),
203 def test_check_ns_operation_action(self
):
204 nsrs
= self
.db
.get_list("nsrs")[0]
208 "member_vnf_index": "1",
210 "primitive": "touch",
211 "primitive_params": {"filename": "file"},
214 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "action", indata
)
216 indata_copy
= indata
.copy()
217 if k
== "primitive_params":
219 indata_copy
[k
] = "non_existing"
220 with self
.assertRaises(EngineException
) as exc_manager
:
221 self
.nslcmop_topic
._check
_ns
_operation
(
222 session
, nsrs
, "action", indata_copy
224 exc
= exc_manager
.exception
227 HTTPStatus
.BAD_REQUEST
,
228 "Engine exception bad http_code with {}".format(indata_copy
),
231 def test_update_remove_vnf(self
):
232 vnfr_id
= self
.db
.get_list("vnfrs")[0]["_id"]
236 {"_id": self
.nsr_id
},
237 {"_admin.nsState": "INSTANTIATED"},
240 "lcmOperationType": "update",
241 "updateType": "REMOVE_VNF",
242 "nsInstanceId": self
.nsr_id
,
243 "removeVnfInstanceId": vnfr_id
250 "project_id": [self
.nsr_project
],
256 nslcmop_id
, _
= self
.nslcmop_topic
.new(
257 rollback
, session
, indata
, kwargs
=None, headers
=headers
261 self
.db
.create
.call_count
,
263 "database create not called, or called more than once",
265 _call
= self
.db
.create
.call_args_list
[0]
267 _call
[0][0], "nslcmops", "nslcmops entry must be created at database"
269 created_nslcmop
= _call
[0][1]
272 created_nslcmop
["nsInstanceId"],
273 "mismatch between nsId '_id' in created nslcmop and database nsr",
276 created_nslcmop
["lcmOperationType"] == "update",
277 "Database record must contain 'lcmOperationType=update'",
280 created_nslcmop
["operationParams"]["updateType"] == "REMOVE_VNF",
281 "Database record must contain 'updateType=REMOVE_VNF'",
284 def test_migrate(self
):
285 vnfr_id
= self
.db
.get_list("vnfrs")[0]["_id"]
289 {"_id": self
.nsr_id
},
290 {"_admin.nsState": "INSTANTIATED"},
296 "project_id": [self
.nsr_project
],
302 with self
.subTest(i
=1, t
="Migration for Specific VM"):
304 "lcmOperationType": "migrate",
305 "nsInstanceId": self
.nsr_id
,
306 "migrateToHost":"sample02",
311 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f"
313 nslcmop_id
, _
= self
.nslcmop_topic
.new(
314 rollback
, session
, indata
, kwargs
=None, headers
=headers
318 self
.db
.create
.call_count
,
320 "database create not called, or called more than once",
322 _call
= self
.db
.create
.call_args_list
[0]
324 _call
[0][0], "nslcmops", "nslcmops entry must be created at database"
326 created_nslcmop
= _call
[0][1]
329 created_nslcmop
["nsInstanceId"],
330 "mismatch between nsId '_id' in created nslcmop and database nsr",
333 created_nslcmop
["lcmOperationType"] == "migrate",
334 "Database record must contain 'lcmOperationType=migrate'",
336 with self
.subTest(i
=2, t
="Migration of all VDUs in a VNF"):
338 "lcmOperationType": "migrate",
339 "nsInstanceId": self
.nsr_id
,
340 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f"
342 nslcmop_id
, _
= self
.nslcmop_topic
.new(
343 rollback
, session
, indata
, kwargs
=None, headers
=headers
347 self
.db
.create
.call_count
,
349 "database create not called, or called more than once",
351 _call
= self
.db
.create
.call_args_list
[0]
353 _call
[0][0], "nslcmops", "nslcmops entry must be created at database"
355 created_nslcmop
= _call
[0][1]
358 created_nslcmop
["nsInstanceId"],
359 "mismatch between nsId '_id' in created nslcmop and database nsr",
362 created_nslcmop
["lcmOperationType"] == "migrate",
363 "Database record must contain 'lcmOperationType=migrate'",
365 with self
.subTest(i
=3, t
="Migration failure - vduId not provided in vdu "):
367 "lcmOperationType": "migrate",
368 "nsInstanceId": self
.nsr_id
,
369 "migrateToHost":"sample02",
373 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f"
376 with self
.assertRaises(Exception) as e
:
377 nslcmop_id
, _
= self
.nslcmop_topic
.new(
378 rollback
, session
, indata
, kwargs
=None, headers
=headers
380 self
.assertTrue("Format error at 'vdu' ''vduId' is a required property'" in str(e
.exception
))
383 class TestNsLcmOpTopicWithMock(unittest
.TestCase
):
385 self
.db
= Mock(dbbase
.DbBase())
386 self
.fs
= Mock(FsBase())
387 self
.fs
.get_params
.return_value
= {"./fake/folder"}
388 self
.fs
.file_open
= mock_open()
389 self
.msg
= Mock(MsgBase())
391 self
.nslcmop_topic
= NsLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
393 def test_get_vnfd_from_vnf_member_revision(self
):
394 test_vnfr
= yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
)[0]
395 test_vnfd
= yaml
.load(db_vnfds_text
, Loader
=yaml
.Loader
)
396 self
.db
.get_one
.side_effect
= [test_vnfr
, test_vnfd
]
397 vnfr
= self
.nslcmop_topic
._get
_vnfd
_from
_vnf
_member
_index
("1", test_vnfr
['_id'])
398 self
.assertEqual(self
.db
.get_one
.call_args_list
[0][0][0], 'vnfrs', "Incorrect first DB lookup")
399 self
.assertEqual(self
.db
.get_one
.call_args_list
[1][0][0], 'vnfds', "Incorrect second DB lookup")
401 def test_get_vnfd_from_vnf_member_no_revision(self
):
402 test_vnfr
= yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
)[0]
403 test_vnfr
['revision'] = 3
404 test_vnfd
= yaml
.load(db_vnfds_text
, Loader
=yaml
.Loader
)
405 self
.db
.get_one
.side_effect
= [test_vnfr
, test_vnfd
]
406 vnfr
= self
.nslcmop_topic
._get
_vnfd
_from
_vnf
_member
_index
("1", test_vnfr
['_id'])
407 self
.assertEqual(self
.db
.get_one
.call_args_list
[0][0][0], 'vnfrs', "Incorrect first DB lookup")
408 self
.assertEqual(self
.db
.get_one
.call_args_list
[1][0][0], 'vnfds_revisions', "Incorrect second DB lookup")
411 def assertNotRaises(self
, exception_type
):
414 except exception_type
:
415 raise self
.failureException("{} raised".format(exception_type
.__name
__))
417 def test_check_ns_update_operation(self
):
419 self
.nslcmop_topic
= NsLcmOpTopic(self
.db
, self
.fs
, self
.msg
, None)
422 with self
.subTest(i
=1, t
="VNF instance does not belong to NS"):
423 test_vnfr
= yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
)
424 test_vnfr
[0]["revision"] = 2
425 test_nsr
= yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
)
426 test_nsr
[0]["constituent-vnfr-ref"][
428 ] = "99d90b0c-faff-4b9f-bccd-017f33985984"
429 self
.db
.create_list("vnfrs", test_vnfr
)
430 self
.db
.create_list("nsrs", test_nsr
)
431 nsrs
= self
.db
.get_list("nsrs")[0]
433 "updateType": "CHANGE_VNFPKG",
434 "changeVnfPackageData": {
435 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
436 "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
438 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
440 with self
.assertRaises(EngineException
) as expected_exception
:
441 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "update", indata
)
443 str(expected_exception
.exception
),
444 "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984"
445 " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85",
448 with self
.subTest(i
=2, t
="Ns update request validated with no exception"):
449 test_vnfr
= yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
)
450 test_vnfr
[0]["revision"] = 2
451 test_nsr
= yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
)
452 self
.db
.create_list("vnfrs", test_vnfr
)
453 self
.db
.create_list("nsrs", test_nsr
)
454 nsrs
= self
.db
.get_list("nsrs")[1]
456 "updateType": "CHANGE_VNFPKG",
457 "changeVnfPackageData": {
458 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
459 "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
461 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
463 with self
.assertNotRaises(EngineException
):
464 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "update", indata
)
467 i
=3, t
="Ns update request rejected because of too small timeout"
470 "updateType": "CHANGE_VNFPKG",
471 "changeVnfPackageData": {
472 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
473 "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
475 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
476 "timeout_ns_update": 50,
478 with self
.assertRaises(EngineException
) as expected_exception
:
479 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "update", indata
)
481 str(expected_exception
.exception
),
482 "Error in validating ns-update request: 50 second is not enough "
483 "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
486 with self
.subTest(i
=4, t
="wrong vnfdid is given as an update parameter"):
487 test_vnfr
= yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
)
488 test_vnfr
[0]["revision"] = 2
489 test_nsr
= yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
)
490 self
.db
.create_list("vnfrs", test_vnfr
)
491 self
.db
.create_list("nsrs", test_nsr
)
492 nsrs
= self
.db
.get_list("nsrs")[2]
494 "updateType": "CHANGE_VNFPKG",
495 "changeVnfPackageData": {
496 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
497 "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77",
499 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
501 with self
.assertRaises(EngineException
) as expected_exception
:
502 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "update", indata
)
504 str(expected_exception
.exception
),
505 "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not "
506 "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of "
507 "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984",
510 with self
.subTest(i
=5, t
="Ns update REMOVE_VNF request validated with no exception"):
511 test_vnfr
= yaml
.load(db_vnfrs_text
, Loader
=yaml
.Loader
)
512 test_vnfr
[0]["revision"] = 2
513 test_nsr
= yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
)
514 self
.db
.create_list("vnfrs", test_vnfr
)
515 self
.db
.create_list("nsrs", test_nsr
)
516 nsrs
= self
.db
.get_list("nsrs")[1]
518 "updateType": "REMOVE_VNF",
519 "removeVnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984",
520 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85",
522 with self
.assertNotRaises(EngineException
):
523 self
.nslcmop_topic
._check
_ns
_operation
(session
, nsrs
, "update", indata
)
525 class TestNsrTopic(unittest
.TestCase
):
528 self
.fs
= Mock(FsBase())
529 self
.fs
.get_params
.return_value
= {"./fake/folder"}
530 self
.fs
.file_open
= mock_open()
531 self
.msg
= Mock(MsgBase())
533 self
.nsr_topic
= NsrTopic(self
.db
, self
.fs
, self
.msg
, None)
534 self
.nsr_topic
.check_quota
= Mock(return_value
=None) # skip quota
537 "vim_accounts", yaml
.load(db_vim_accounts_text
, Loader
=yaml
.Loader
)
539 self
.db
.create_list("nsds", yaml
.load(db_nsds_text
, Loader
=yaml
.Loader
))
540 self
.db
.create_list("vnfds", yaml
.load(db_vnfds_text
, Loader
=yaml
.Loader
))
541 self
.db
.create
= Mock(return_value
="created_id")
542 self
.nsd
= self
.db
.get_list("nsds")[0]
543 self
.nsd_id
= self
.nsd
["_id"]
544 self
.nsd_project
= self
.nsd
["_admin"]["projects_read"][0]
546 self
.vim
= self
.db
.get_list("vim_accounts")[0]
547 self
.vim_id
= self
.vim
["_id"]
549 def test_create(self
):
554 "project_id": [self
.nsd_project
],
558 "nsdId": self
.nsd_id
,
560 "vimAccountId": self
.vim_id
,
561 "additionalParamsForVnf": [
563 "member-vnf-index": "hackfest_vnf1",
564 "additionalParams": {"touch_filename": "file"},
567 "member-vnf-index": "hackfest_vnf2",
568 "additionalParams": {"touch_filename": "file"},
576 rollback
, session
, indata
=indata
, kwargs
=None, headers
=headers
579 # check vnfrs and nsrs created in whatever order
583 for _call
in self
.db
.create
.call_args_list
:
584 assert len(_call
[0]) >= 2, "called db.create with few parameters"
585 created_item
= _call
[0][1]
586 if _call
[0][0] == "vnfrs":
587 created_vnfrs
.append(created_item
)
589 "member-vnf-index-ref",
591 "Created item must contain member-vnf-index-ref section",
596 created_item
["nsr-id-ref"],
597 "bad reference id from vnfr to nsr",
600 nsr_id
= created_item
["nsr-id-ref"]
602 elif _call
[0][0] == "nsrs":
603 created_nsrs
.append(created_item
)
606 nsr_id
, created_item
["_id"], "bad reference id from vnfr to nsr"
609 nsr_id
= created_item
["_id"]
611 assert True, "created an unknown record {} at database".format(
616 created_item
["_admin"].get("projects_read"),
617 "Database record must contain '_amdin.projects_read'",
621 created_item
["_admin"],
622 "Database record must contain '_admin.created'",
625 created_item
["_admin"]["nsState"] == "NOT_INSTANTIATED",
626 "Database record must contain '_admin.nstate=NOT INSTANTIATE'",
630 len(created_vnfrs
), 2, "created a mismatch number of vnfr at database"
634 created_vnfrs
[0]["vdur"][0]["interfaces"][0]["position"],
636 "vdur first interface position does not match",
640 created_vnfrs
[0]["vdur"][0]["interfaces"][1]["position"],
642 "vdur second interface position does not match",
646 len(created_nsrs
), 1, "Only one nsrs must be created at database"
650 len(created_vnfrs
) + 1,
651 "rollback mismatch with created items at database",
654 # test parameters with error
655 bad_id
= "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
662 HTTPStatus
.NOT_FOUND
,
663 ("not found", bad_id
),
665 # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)), # TODO add "vim"
667 "additional params not supply",
668 {"additionalParamsForVnf.0.member-vnf-index": "k"},
670 HTTPStatus
.BAD_REQUEST
,
674 for message
, kwargs_
, expect_exc
, expect_code
, expect_text_list
in test_set
:
675 with self
.assertRaises(expect_exc
, msg
=message
) as e
:
679 indata
=deepcopy(indata
),
684 self
.assertTrue(e
.exception
.http_code
== expect_code
)
686 for expect_text
in expect_text_list
:
687 self
.assertIn(expect_text
, str(e
.exception
).lower(),
688 "Expected '{}' at exception text".format(expect_text
))
690 def test_show_instance(self
):
691 session
= {"force": False, "admin": False, "public": False, "project_id": [self
.nsd_project
], "method": "write"}
693 for refresh_status
in ("true", "false"):
694 self
.db
.create_list("nsrs", yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
))
695 actual_nsr
= self
.db
.get_list("nsrs")[0]
696 nsr_id
= actual_nsr
["_id"]
697 filter_q
['vcaStatus-refresh'] = refresh_status
698 expected_nsr
= self
.nsr_topic
.show(session
, nsr_id
, filter_q
=filter_q
)
699 self
.nsr_topic
.delete(session
, nsr_id
)
700 actual_nsr
.pop("_admin")
701 expected_nsr
.pop("_admin")
702 self
.assertEqual(expected_nsr
, actual_nsr
, "Database nsr and show() nsr do not match.")
704 def test_vca_status_refresh(self
):
705 session
= {"force": False, "admin": False, "public": False, "project_id": [self
.nsd_project
], "method": "write"}
706 filter_q
= {'vcaStatus-refresh': 'true'}
708 self
.db
.create_list("nsrs", yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
))
709 nsr
= self
.db
.get_list("nsrs")[0]
711 # When vcaStatus-refresh is true
712 filter_q
['vcaStatus-refresh'] = "true"
713 self
.nsr_topic
.vca_status_refresh(session
, nsr
, filter_q
)
714 msg_args
= self
.msg
.write
.call_args
[0]
715 self
.assertEqual(msg_args
[1], "vca_status_refresh", "Wrong message action")
716 self
.assertGreater(nsr
["_admin"]["modified"], time() - time_delta
)
718 # When vcaStatus-refresh is false but modified time is within threshold
719 filter_q
['vcaStatus-refresh'] = "false"
721 nsr
["_admin"]["modified"] = time_now
722 self
.nsr_topic
.vca_status_refresh(session
, nsr
, filter_q
)
723 msg_args
= self
.msg
.write
.call_args
[1]
724 self
.assertEqual(msg_args
, {}, "Message should not be sent.")
725 self
.assertEqual(nsr
["_admin"]["modified"], time_now
, "Modified time should not be changed.")
727 # When vcaStatus-refresh is false but modified time is less than threshold
728 filter_q
['vcaStatus-refresh'] = "false"
729 nsr
["_admin"]["modified"] = time() - (2*time_delta
)
730 self
.nsr_topic
.vca_status_refresh(session
, nsr
, filter_q
)
731 msg_args
= self
.msg
.write
.call_args
[0]
732 self
.assertEqual(msg_args
[1], "vca_status_refresh", "Wrong message action")
733 self
.nsr_topic
.delete(session
, nsr
["_id"])
734 self
.assertGreater(nsr
["_admin"]["modified"], time() - time_delta
, "Modified time is not changed.")
736 def test_delete_ns(self
):
737 self
.db
.create_list("nsrs", yaml
.load(db_nsrs_text
, Loader
=yaml
.Loader
))
738 self
.nsr
= self
.db
.get_list("nsrs")[0]
739 self
.nsr_id
= self
.nsr
["_id"]
740 self
.db_set_one
= self
.db
.set_one
741 p_id
= self
.nsd_project
748 "project_id": [p_id
],
755 "project_id": [p_other
],
765 with self
.subTest(i
=1, t
="Normal Deletion"):
766 self
.db
.del_one
= Mock()
767 self
.db
.set_one
= Mock()
768 self
.nsr_topic
.delete(session
, self
.nsr_id
)
770 db_args_ro_nsrs
= self
.db
.del_one
.call_args_list
[1][0]
771 db_args
= self
.db
.del_one
.call_args_list
[0][0]
772 msg_args
= self
.msg
.write
.call_args
[0]
774 msg_args
[0], self
.nsr_topic
.topic_msg
, "Wrong message topic"
776 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
777 self
.assertEqual(msg_args
[2], {"_id": self
.nsr_id
}, "Wrong message content")
778 self
.assertEqual(db_args_ro_nsrs
[0], "ro_nsrs", "Wrong DB topic")
779 self
.assertEqual(db_args
[0], self
.nsr_topic
.topic
, "Wrong DB topic")
780 self
.assertEqual(db_args
[1]["_id"], self
.nsr_id
, "Wrong DB ID")
782 db_args
[1]["_admin.projects_read.cont"], [p_id
], "Wrong DB filter"
784 self
.db
.set_one
.assert_not_called()
785 fs_del_calls
= self
.fs
.file_delete
.call_args_list
786 self
.assertEqual(fs_del_calls
[0][0][0], self
.nsr_id
, "Wrong FS file id")
787 with self
.subTest(i
=2, t
="No delete because referenced by other project"):
790 {"_id": self
.nsr_id
},
793 "_admin.projects_read": p_other
,
794 "_admin.projects_write": p_other
,
797 self
.db
.del_one
.reset_mock()
798 self
.db
.set_one
.reset_mock()
799 self
.msg
.write
.reset_mock()
800 self
.fs
.file_delete
.reset_mock()
802 self
.nsr_topic
.delete(session2
, self
.nsr_id
)
803 self
.db
.del_one
.assert_not_called()
804 self
.msg
.write
.assert_not_called()
805 db_s1_args
= self
.db
.set_one
.call_args
806 self
.assertEqual(db_s1_args
[0][0], self
.nsr_topic
.topic
, "Wrong DB topic")
807 self
.assertEqual(db_s1_args
[0][1]["_id"], self
.nsr_id
, "Wrong DB ID")
809 db_s1_args
[1]["update_dict"], "Wrong DB update dictionary"
812 db_s1_args
[1]["pull_list"],
813 {"_admin.projects_read": [p_other
], "_admin.projects_write": [p_other
]},
814 "Wrong DB pull_list dictionary",
816 self
.fs
.file_delete
.assert_not_called()
817 with self
.subTest(i
=4, t
="Delete with force and admin"):
818 self
.db
.del_one
.reset_mock()
819 self
.db
.set_one
.reset_mock()
820 self
.msg
.write
.reset_mock()
821 self
.fs
.file_delete
.reset_mock()
822 self
.nsr_topic
.delete(session_force
, self
.nsr_id
)
824 db_args_ro_nsrs
= self
.db
.del_one
.call_args_list
[1][0]
825 db_args
= self
.db
.del_one
.call_args_list
[0][0]
826 msg_args
= self
.msg
.write
.call_args
[0]
828 msg_args
[0], self
.nsr_topic
.topic_msg
, "Wrong message topic"
830 self
.assertEqual(msg_args
[1], "deleted", "Wrong message action")
831 self
.assertEqual(msg_args
[2], {"_id": self
.nsr_id
}, "Wrong message content")
832 self
.assertEqual(db_args_ro_nsrs
[0], "ro_nsrs", "Wrong DB topic")
833 self
.assertEqual(db_args
[0], self
.nsr_topic
.topic
, "Wrong DB topic")
834 self
.assertEqual(db_args
[1]["_id"], self
.nsr_id
, "Wrong DB ID")
835 self
.db
.set_one
.assert_not_called()
836 fs_del_calls
= self
.fs
.file_delete
.call_args_list
837 self
.assertEqual(fs_del_calls
[0][0][0], self
.nsr_id
, "Wrong FS file id")
838 with self
.subTest(i
=3, t
="Conflict on Delete - NS in INSTANTIATED state"):
841 {"_id": self
.nsr_id
},
842 {"_admin.nsState": "INSTANTIATED"},
844 "_admin.projects_read": p_other
,
845 "_admin.projects_write": p_other
,
848 self
.db
.del_one
.reset_mock()
849 self
.db
.set_one
.reset_mock()
850 self
.msg
.write
.reset_mock()
851 self
.fs
.file_delete
.reset_mock()
853 with self
.assertRaises(
854 EngineException
, msg
="Accepted NSR with nsState INSTANTIATED"
856 self
.nsr_topic
.delete(session
, self
.nsr_id
)
858 e
.exception
.http_code
, HTTPStatus
.CONFLICT
, "Wrong HTTP status code"
860 self
.assertIn("INSTANTIATED", str(e
.exception
), "Wrong exception text")
861 # TODOD with self.subTest(i=3, t='Conflict on Delete - NS in use by NSI'):
863 with self
.subTest(i
=4, t
="Non-existent NS"):
864 self
.db
.del_one
.reset_mock()
865 self
.db
.set_one
.reset_mock()
866 self
.msg
.write
.reset_mock()
867 self
.fs
.file_delete
.reset_mock()
868 excp_msg
= "Not found"
869 with self
.assertRaises(
870 DbException
, msg
="Accepted non-existent NSD ID"
872 self
.nsr_topic
.delete(session2
, "other_id")
874 e
.exception
.http_code
, HTTPStatus
.NOT_FOUND
, "Wrong HTTP status code"
876 self
.assertIn(excp_msg
, str(e
.exception
), "Wrong exception text")
877 self
.assertIn("other_id", str(e
.exception
), "Wrong exception text")