1 #######################################################################################
2 # Copyright ETSI Contributors and Others.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #######################################################################################
20 from unittest
.mock
import MagicMock
, patch
22 from osm_ng_ro
.ns_thread
import VimInteractionAffinityGroup
25 class TestVimInteractionAffinityGroup(unittest
.TestCase
):
27 module_name
= "osm_ro_plugin"
28 self
.target_vim
= MagicMock(name
=f
"{module_name}.vimconn.VimConnector")
29 self
.task_depends
= None
31 patches
= [patch(f
"{module_name}.vimconn.VimConnector", self
.target_vim
)]
33 # Enabling mocks and add cleanups
36 self
.addCleanup(mock
.stop
)
38 def test__new_affinity_group_ok(self
):
40 create affinity group with attributes set in params
43 logger
= "test_logger"
51 instance
= VimInteractionAffinityGroup(db
, logger
, my_vims
, db_vims
)
52 with patch
.object(instance
, "my_vims", [self
.target_vim
]), patch
.object(
53 instance
, "logger", logging
54 ), patch
.object(instance
, "db_vims", db_vims
):
60 "action_id": "123456",
62 "task_id": "123456:1",
63 "status": "SCHEDULED",
66 "target_record": "test_target_record",
67 "target_record_id": "test_target_record_id",
68 # values coming from extra_dict
70 "affinity_group_data": {
71 "name": "affinity_group_1",
77 "depends_on": "test_depends_on",
82 task_index
= "task_index_1"
83 self
.target_vim
.new_affinity_group
.return_value
= (
84 "sample_affinity_group_id_1"
86 result
= instance
.new(ro_task
, task_index
, self
.task_depends
)
87 self
.assertEqual(result
[0], "DONE")
88 self
.assertEqual(result
[1].get("vim_id"), "sample_affinity_group_id_1")
89 self
.assertEqual(result
[1].get("created"), True)
90 self
.assertEqual(result
[1].get("vim_status"), "DONE")
92 def test__new_affinity_group_failed(self
):
94 create affinity group with no attributes set in params
97 logger
= "test_logger"
105 instance
= VimInteractionAffinityGroup(db
, logger
, my_vims
, db_vims
)
106 with patch
.object(instance
, "my_vims", [self
.target_vim
]), patch
.object(
107 instance
, "logger", logging
108 ), patch
.object(instance
, "db_vims", db_vims
):
114 "action_id": "123456",
116 "task_id": "123456:1",
117 "status": "SCHEDULED",
120 "target_record": "test_target_record",
121 "target_record_id": "test_target_record_id",
122 # values coming from extra_dict
125 "depends_on": "test_depends_on",
130 task_index
= "task_index_2"
131 self
.target_vim
.new_affinity_group
.return_value
= (
132 "sample_affinity_group_id_1"
134 result
= instance
.new(ro_task
, task_index
, self
.task_depends
)
135 self
.assertEqual(result
[0], "DONE")
136 self
.assertEqual(result
[1].get("vim_id"), None)
137 self
.assertEqual(result
[1].get("created"), False)
138 self
.assertEqual(result
[1].get("vim_status"), "DONE")
140 def test__delete_affinity_group_ok(self
):
142 delete affinity group with a proper vim_id
145 logger
= "test_logger"
153 instance
= VimInteractionAffinityGroup(db
, logger
, my_vims
, db_vims
)
154 with patch
.object(instance
, "my_vims", [self
.target_vim
]), patch
.object(
155 instance
, "logger", logging
156 ), patch
.object(instance
, "db_vims", db_vims
):
162 "task_id": "123456:1",
167 "created_items": None,
168 "vim_id": "sample_affinity_group_id_3",
169 "vim_name": "sample_affinity_group_id_3",
171 "vim_details": "some-details",
176 task_index
= "task_index_3"
177 self
.target_vim
.delete_affinity_group
.return_value
= (
178 "sample_affinity_group_id_3"
180 result
= instance
.delete(ro_task
, task_index
)
181 self
.assertEqual(result
[0], "DONE")
182 self
.assertEqual(result
[1].get("vim_details"), "DELETED")
183 self
.assertEqual(result
[1].get("created"), False)
184 self
.assertEqual(result
[1].get("vim_status"), "DELETED")
186 def test__delete_affinity_group_failed(self
):
188 delete affinity group with missing vim_id
191 logger
= "test_logger"
199 instance
= VimInteractionAffinityGroup(db
, logger
, my_vims
, db_vims
)
200 with patch
.object(instance
, "my_vims", [self
.target_vim
]), patch
.object(
201 instance
, "logger", logging
202 ), patch
.object(instance
, "db_vims", db_vims
):
208 "task_id": "123456:1",
213 "created_items": None,
217 "vim_details": "some-details",
222 task_index
= "task_index_4"
223 self
.target_vim
.delete_affinity_group
.return_value
= ""
224 result
= instance
.delete(ro_task
, task_index
)
225 self
.assertEqual(result
[0], "DONE")
226 self
.assertEqual(result
[1].get("vim_details"), "DELETED")
227 self
.assertEqual(result
[1].get("created"), False)
228 self
.assertEqual(result
[1].get("vim_status"), "DELETED")