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 #######################################################################################
19 from unittest
.mock
import patch
, Mock
21 from osm_ng_ro
.ns
import Ns
24 __author__
= "Eduardo Sousa"
25 __date__
= "$19-NOV-2021 00:00:00$"
28 class TestNs(unittest
.TestCase
):
32 def test__create_task_without_extra_dict(self
):
34 "target_id": "vim_openstack_1",
35 "action_id": "123456",
37 "task_id": "123456:1",
38 "status": "SCHEDULED",
41 "target_record": "test_target_record",
42 "target_record_id": "test_target_record_id",
45 "action_id": "123456",
50 task
= Ns
._create
_task
(
51 deployment_info
=deployment_info
,
52 target_id
="vim_openstack_1",
55 target_record
="test_target_record",
56 target_record_id
="test_target_record_id",
59 self
.assertEqual(deployment_info
.get("task_index"), 2)
60 self
.assertDictEqual(task
, expected_result
)
62 def test__create_task(self
):
64 "target_id": "vim_openstack_1",
65 "action_id": "123456",
67 "task_id": "123456:1",
68 "status": "SCHEDULED",
71 "target_record": "test_target_record",
72 "target_record_id": "test_target_record_id",
73 # values coming from extra_dict
74 "params": "test_params",
75 "find_params": "test_find_params",
76 "depends_on": "test_depends_on",
79 "action_id": "123456",
84 task
= Ns
._create
_task
(
85 deployment_info
=deployment_info
,
86 target_id
="vim_openstack_1",
89 target_record
="test_target_record",
90 target_record_id
="test_target_record_id",
92 "params": "test_params",
93 "find_params": "test_find_params",
94 "depends_on": "test_depends_on",
98 self
.assertEqual(deployment_info
.get("task_index"), 2)
99 self
.assertDictEqual(task
, expected_result
)
101 @patch("osm_ng_ro.ns.time")
102 def test__create_ro_task(self
, mock_time
: Mock
):
103 now
= 1637324838.994551
104 mock_time
.return_value
= now
106 "target_id": "vim_openstack_1",
107 "action_id": "123456",
109 "task_id": "123456:1",
110 "status": "SCHEDULED",
113 "target_record": "test_target_record",
114 "target_record_id": "test_target_record_id",
115 # values coming from extra_dict
116 "params": "test_params",
117 "find_params": "test_find_params",
118 "depends_on": "test_depends_on",
124 "target_id": "vim_openstack_1",
127 "created_items": None,
140 ro_task
= Ns
._create
_ro
_task
(
141 target_id
="vim_openstack_1",
145 self
.assertDictEqual(ro_task
, expected_result
)
147 def test__process_image_params_with_empty_target_image(self
):
153 result
= Ns
._process
_image
_params
(
154 target_image
=target_image
,
156 target_record_id
=None,
159 self
.assertDictEqual(expected_result
, result
)
161 def test__process_image_params_with_wrong_target_image(self
):
166 "no_image": "to_see_here",
169 result
= Ns
._process
_image
_params
(
170 target_image
=target_image
,
172 target_record_id
=None,
175 self
.assertDictEqual(expected_result
, result
)
177 def test__process_image_params_with_image(self
):
189 result
= Ns
._process
_image
_params
(
190 target_image
=target_image
,
192 target_record_id
=None,
195 self
.assertDictEqual(expected_result
, result
)
197 def test__process_image_params_with_vim_image_id(self
):
206 "vim_image_id": "123456",
209 result
= Ns
._process
_image
_params
(
210 target_image
=target_image
,
212 target_record_id
=None,
215 self
.assertDictEqual(expected_result
, result
)
217 def test__process_image_params_with_image_checksum(self
):
221 "checksum": "e3fc50a88d0a364313df4b21ef20c29e",
226 "image_checksum": "e3fc50a88d0a364313df4b21ef20c29e",
229 result
= Ns
._process
_image
_params
(
230 target_image
=target_image
,
232 target_record_id
=None,
235 self
.assertDictEqual(expected_result
, result
)
237 def test__get_resource_allocation_params_with_empty_target_image(self
):
239 quota_descriptor
= {}
241 result
= Ns
._get
_resource
_allocation
_params
(
242 quota_descriptor
=quota_descriptor
,
245 self
.assertDictEqual(expected_result
, result
)
247 def test__get_resource_allocation_params_with_wrong_target_image(self
):
250 "no_quota": "present_here",
253 result
= Ns
._get
_resource
_allocation
_params
(
254 quota_descriptor
=quota_descriptor
,
257 self
.assertDictEqual(expected_result
, result
)
259 def test__get_resource_allocation_params_with_limit(self
):
267 result
= Ns
._get
_resource
_allocation
_params
(
268 quota_descriptor
=quota_descriptor
,
271 self
.assertDictEqual(expected_result
, result
)
273 def test__get_resource_allocation_params_with_reserve(self
):
281 result
= Ns
._get
_resource
_allocation
_params
(
282 quota_descriptor
=quota_descriptor
,
285 self
.assertDictEqual(expected_result
, result
)
287 def test__get_resource_allocation_params_with_shares(self
):
295 result
= Ns
._get
_resource
_allocation
_params
(
296 quota_descriptor
=quota_descriptor
,
299 self
.assertDictEqual(expected_result
, result
)
301 def test__get_resource_allocation_params(self
):
313 result
= Ns
._get
_resource
_allocation
_params
(
314 quota_descriptor
=quota_descriptor
,
317 self
.assertDictEqual(expected_result
, result
)