1 # ============LICENSE_START===================================================
2 # Copyright (C) 2019-2022 Nordix Foundation.
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
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 implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
16 # SPDX-License-Identifier: Apache-2.0
17 # ============LICENSE_END=====================================================
20 from unittest.mock import patch, MagicMock
21 from http import HTTPStatus
23 from mod import aai_client, db
24 from mod.api.controller import status, post_subscription, get_subscription_by_name, \
25 get_subscriptions, get_meas_group_with_nfs, delete_subscription_by_name, update_admin_state
26 from tests.base_setup import BaseClassSetup
27 from mod.api.custom_exception import InvalidDataException, DataConflictException
28 from mod.api.db_models import SubscriptionModel, NfMeasureGroupRelationalModel
29 from mod.subscription import SubNfState
30 from mod.network_function import NetworkFunctionFilter
31 from tests.base_setup import create_subscription_data, create_multiple_subscription_data, \
32 create_multiple_network_function_data
33 from mod.api.services import measurement_group_service, nf_service, subscription_service
36 class ControllerTestCase(BaseClassSetup):
44 super().setUpAppConf()
45 with open(os.path.join(os.path.dirname(__file__), 'data/aai_xnfs.json'), 'r') as data:
46 self.aai_response_data = data.read()
47 with open(os.path.join(os.path.dirname(__file__), 'data/aai_model_info.json'), 'r') as data:
48 self.good_model_info = data.read()
49 with open(os.path.join(os.path.dirname(__file__),
50 'data/create_subscription_request.json'), 'r') as data:
51 self.subscription_request = data.read()
57 def tearDownClass(cls):
58 super().tearDownClass()
60 def test_status_response_healthy(self):
61 self.assertEqual(status()['status'], 'healthy')
63 def create_test_subs(self, new_sub_name, new_msrmt_grp_name):
64 subscription = self.subscription_request.replace('ExtraPM-All-gNB-R2B', new_sub_name)
65 subscription = subscription.replace('msrmt_grp_name', new_msrmt_grp_name)
68 @patch('mod.api.services.subscription_service.save_nf_filter', MagicMock(return_value=None))
69 @patch('mod.pmsh_config.AppConfig.publish_to_topic', MagicMock(return_value=None))
70 @patch.object(aai_client, '_get_all_aai_nf_data')
71 @patch.object(aai_client, 'get_aai_model_data')
72 @patch.object(NetworkFunctionFilter, 'get_network_function_filter')
73 def test_post_subscription(self, mock_filter_call, mock_model_aai, mock_aai):
74 mock_aai.return_value = json.loads(self.aai_response_data)
75 mock_model_aai.return_value = json.loads(self.good_model_info)
76 subscription = self.create_test_subs('xtraPM-All-gNB-R2B-post', 'msrmt_grp_name-post')
77 subscription = json.loads(subscription)
78 mock_filter_call.return_value = NetworkFunctionFilter(
79 **subscription['subscription']["nfFilter"])
80 sub_name = subscription['subscription']['subscriptionName']
81 mes_grp = subscription['subscription']['measurementGroups'][0]['measurementGroup']
82 mes_grp_name = mes_grp['measurementGroupName']
83 response = post_subscription(subscription)
84 subscription = (SubscriptionModel.query.filter(
85 SubscriptionModel.subscription_name == sub_name).one_or_none())
86 self.assertIsNotNone(subscription)
87 msr_grp_nf_rel = (NfMeasureGroupRelationalModel.query.filter(
88 NfMeasureGroupRelationalModel.measurement_grp_name == mes_grp_name)).all()
89 for published_event in msr_grp_nf_rel:
90 self.assertEqual(published_event.nf_measure_grp_status,
91 SubNfState.PENDING_CREATE.value)
92 self.assertEqual(response[1], 201)
94 def test_post_subscription_duplicate_sub(self):
95 # Posting the same subscription request stored in previous test to get duplicate response
96 response = post_subscription(json.loads(self.subscription_request))
97 self.assertEqual(response[1], 409)
98 self.assertEqual(response[0], 'subscription Name: ExtraPM-All-gNB-R2B already exists.')
100 def test_post_subscription_invalid_filter(self):
101 subscription = self.create_test_subs('xtraPM-All-gNB-R2B-invalid', 'msrmt_grp_name-invalid')
102 subscription = json.loads(subscription)
103 subscription['subscription']['nfFilter']['nfNames'] = []
104 subscription['subscription']['nfFilter']['modelInvariantIDs'] = []
105 subscription['subscription']['nfFilter']['modelVersionIDs'] = []
106 subscription['subscription']['nfFilter']['modelNames'] = []
107 response = post_subscription(subscription)
108 self.assertEqual(response[1], 400)
109 self.assertEqual(response[0], 'At least one filter within nfFilter must not be empty')
111 def test_post_subscription_missing(self):
112 subscription = json.loads(self.subscription_request)
113 subscription['subscription']['subscriptionName'] = ''
114 response = post_subscription(subscription)
115 self.assertEqual(response[1], 400)
116 self.assertEqual(response[0], 'No value provided in subscription name')
118 @patch('mod.api.services.subscription_service.query_subscription_by_name',
119 MagicMock(return_value=create_subscription_data('sub_demo')))
120 def test_get_subscription_by_name_api(self):
121 sub, status_code = get_subscription_by_name('sub_demo')
122 self.assertEqual(status_code, HTTPStatus.OK.value)
123 self.assertEqual(sub['subscription']['subscriptionName'], 'sub_demo')
124 self.assertEqual(sub['subscription']['nfFilter']['nfNames'],
125 ['^pnf.*', '^vnf.*'])
126 self.assertEqual(sub['subscription']['controlLoopName'],
127 'pmsh_control_loop_name')
128 self.assertEqual(len(sub['subscription']['measurementGroups']), 2)
129 self.assertEqual(sub['subscription']['operationalPolicyName'],
130 'pmsh_operational_policy')
132 @patch('mod.api.services.subscription_service.query_subscription_by_name',
133 MagicMock(return_value=None))
134 def test_get_subscription_by_name_api_none(self):
135 sub, status_code = get_subscription_by_name('sub_demo')
136 self.assertEqual(status_code, HTTPStatus.NOT_FOUND.value)
137 self.assertEqual(sub['error'],
138 'Subscription was not defined with the name : sub_demo')
140 @patch('mod.api.services.subscription_service.query_subscription_by_name',
141 MagicMock(side_effect=Exception('something failed')))
142 def test_get_subscription_by_name_api_exception(self):
143 sub, status_code = get_subscription_by_name('sub_demo')
144 self.assertEqual(status_code, HTTPStatus.INTERNAL_SERVER_ERROR.value)
146 @patch('mod.api.services.subscription_service.query_all_subscriptions',
147 MagicMock(return_value=create_multiple_subscription_data(
148 ['sub_demo_one', 'sub_demo_two'])))
149 def test_get_subscriptions_api(self):
150 subs, status_code = get_subscriptions()
151 self.assertEqual(status_code, HTTPStatus.OK.value)
152 self.assertEqual(subs[0]['subscription']['subscriptionName'], 'sub_demo_one')
153 self.assertEqual(subs[1]['subscription']['subscriptionName'], 'sub_demo_two')
154 self.assertEqual(subs[1]['subscription']['measurementGroups'][0]['measurementGroup']
155 ['measurementGroupName'], 'MG1')
156 self.assertEqual(len(subs[1]['subscription']['measurementGroups']), 2)
157 self.assertEqual(subs[0]['subscription']['nfs'][0], 'pnf_101')
158 self.assertEqual(subs[0]['subscription']['nfs'][1], 'pnf_102')
159 self.assertEqual(subs[1]['subscription']['nfs'], [])
160 self.assertEqual(len(subs), 2)
162 @patch('mod.api.services.subscription_service.query_all_subscriptions',
163 MagicMock(return_value=None))
164 def test_get_subscriptions_api_none(self):
165 subs, status_code = get_subscriptions()
166 self.assertEqual(status_code, HTTPStatus.OK.value)
167 self.assertEqual(subs, [])
169 @patch('mod.api.services.subscription_service.query_all_subscriptions',
170 MagicMock(side_effect=Exception('something failed')))
171 def test_get_subscriptions_api_exception(self):
172 subs, status_code = get_subscriptions()
173 self.assertEqual(status_code, HTTPStatus.INTERNAL_SERVER_ERROR.value)
175 def test_get_meas_group_with_nfs_api(self):
176 sub = create_subscription_data('sub1')
177 nf_list = create_multiple_network_function_data(['pnf101', 'pnf102'])
178 measurement_group_service.save_measurement_group(sub.measurement_groups[0].
179 serialize()['measurementGroup'],
180 sub.subscription_name)
182 nf_service.save_nf(nf)
183 measurement_group_service. \
184 apply_nf_status_to_measurement_group(nf.nf_name, sub.measurement_groups[0].
185 measurement_group_name,
186 SubNfState.PENDING_CREATE.value)
188 mg_with_nfs, status_code = get_meas_group_with_nfs('sub1', 'MG1')
189 self.assertEqual(status_code, HTTPStatus.OK.value)
190 self.assertEqual(mg_with_nfs['subscriptionName'], 'sub1')
191 self.assertEqual(mg_with_nfs['measurementGroupName'], 'MG1')
192 self.assertEqual(mg_with_nfs['administrativeState'], 'UNLOCKED')
193 self.assertEqual(len(mg_with_nfs['networkFunctions']), 2)
195 def test_get_meas_group_with_nfs_api_none(self):
196 error, status_code = get_meas_group_with_nfs('sub1', 'MG1')
197 self.assertEqual(error['error'], 'measurement group was not defined with '
198 'the sub name: sub1 and meas group name: MG1')
199 self.assertEqual(status_code, HTTPStatus.NOT_FOUND.value)
201 @patch('mod.api.services.measurement_group_service.query_meas_group_by_name',
202 MagicMock(side_effect=Exception('something failed')))
203 def test_get_meas_group_with_nfs_api_exception(self):
204 error, status_code = get_meas_group_with_nfs('sub1', 'MG1')
205 self.assertEqual(status_code, HTTPStatus.INTERNAL_SERVER_ERROR.value)
207 def test_delete_when_state_unlocked(self):
208 subscription_unlocked_data = create_subscription_data('MG_unlocked')
209 subscription_unlocked_data.measurement_groups[0].measurement_group_name = 'unlock'
210 subscription_unlocked_data.measurement_groups[0].administrative_state = 'UNLOCKED'
211 db.session.add(subscription_unlocked_data)
212 db.session.add(subscription_unlocked_data.measurement_groups[0])
215 message, status_code = delete_subscription_by_name('MG_unlocked')
216 self.assertEqual(status_code, HTTPStatus.CONFLICT.value)
217 self.assertEqual(subscription_service.query_subscription_by_name('MG_unlocked')
218 .subscription_name, 'MG_unlocked')
220 def test_delete_when_state_locked(self):
221 subscription_unlocked_data = create_subscription_data('MG_locked')
222 subscription_unlocked_data.measurement_groups[0].measurement_group_name = 'lock'
223 subscription_unlocked_data.measurement_groups[0].administrative_state = 'LOCKED'
224 db.session.add(subscription_unlocked_data)
225 db.session.add(subscription_unlocked_data.measurement_groups[0])
228 none_type, status_code = delete_subscription_by_name('MG_locked')
229 self.assertEqual(none_type, None)
230 self.assertEqual(status_code, HTTPStatus.NO_CONTENT.value)
231 self.assertEqual(subscription_service.query_subscription_by_name('MG_locked'), None)
233 def test_delete_when_state_locking(self):
234 subscription_locking_data = create_subscription_data('MG_locking')
235 subscription_locking_data.measurement_groups[0].measurement_group_name = 'locking'
236 subscription_locking_data.measurement_groups[0].administrative_state = 'LOCKING'
237 db.session.add(subscription_locking_data)
238 db.session.add(subscription_locking_data.measurement_groups[0])
241 message, status_code = delete_subscription_by_name('MG_locking')
242 self.assertEqual(status_code, HTTPStatus.CONFLICT.value)
243 self.assertEqual(subscription_service.query_subscription_by_name('MG_locking')
244 .subscription_name, 'MG_locking')
246 def test_delete_sub_none(self):
247 message, status_code = delete_subscription_by_name('None')
248 self.assertEqual(message['error'], 'Subscription is not defined with name None')
249 self.assertEqual(status_code, HTTPStatus.NOT_FOUND.value)
251 @patch('mod.api.services.subscription_service.query_to_delete_subscription_by_name',
252 MagicMock(side_effect=Exception('something failed')))
253 def test_delete_sub_exception(self):
254 error, status_code = delete_subscription_by_name('None')
255 self.assertEqual(status_code, HTTPStatus.INTERNAL_SERVER_ERROR.value)
257 @patch('mod.pmsh_config.AppConfig.publish_to_topic', MagicMock(return_value=None))
258 def test_update_admin_state_api_for_locked_update(self):
259 sub = create_subscription_data('sub1')
260 nf_list = create_multiple_network_function_data(['pnf_101', 'pnf_102'])
263 nf_service.save_nf(nf)
264 measurement_group_service. \
265 apply_nf_status_to_measurement_group(nf.nf_name, sub.measurement_groups[0].
266 measurement_group_name,
267 SubNfState.CREATED.value)
269 response = update_admin_state('sub1', 'MG1', {'administrativeState': 'LOCKED'})
270 self.assertEqual(response[1], HTTPStatus.OK.value)
271 self.assertEqual(response[0], 'Successfully updated admin state')
272 mg_with_nfs, status_code = get_meas_group_with_nfs('sub1', 'MG1')
273 self.assertEqual(mg_with_nfs['subscriptionName'], 'sub1')
274 self.assertEqual(mg_with_nfs['measurementGroupName'], 'MG1')
275 self.assertEqual(mg_with_nfs['administrativeState'], 'LOCKING')
276 for nf in mg_with_nfs['networkFunctions']:
277 self.assertEqual(nf['nfMgStatus'], SubNfState.PENDING_DELETE.value)
279 @patch('mod.pmsh_config.AppConfig.publish_to_topic', MagicMock(return_value=None))
280 @patch.object(aai_client, '_get_all_aai_nf_data')
281 @patch.object(aai_client, 'get_aai_model_data')
282 @patch.object(NetworkFunctionFilter, 'get_network_function_filter')
283 def test_update_admin_state_api_for_unlocked_update(self, mock_filter_call,
284 mock_model_aai, mock_aai):
285 mock_aai.return_value = json.loads(self.aai_response_data)
286 mock_model_aai.return_value = json.loads(self.good_model_info)
287 sub = create_subscription_data('sub1')
289 nf_list = create_multiple_network_function_data(['pnf_101', 'pnf_102'])
290 for network_function in nf_list:
291 db.session.add(network_function)
293 mock_filter_call.return_value = NetworkFunctionFilter.get_network_function_filter('sub')
294 response = update_admin_state('sub1', 'MG2', {'administrativeState': 'UNLOCKED'})
295 self.assertEqual(response[1], HTTPStatus.OK.value)
296 self.assertEqual(response[0], 'Successfully updated admin state')
297 mg_with_nfs, status_code = get_meas_group_with_nfs('sub1', 'MG2')
298 self.assertEqual(mg_with_nfs['subscriptionName'], 'sub1')
299 self.assertEqual(mg_with_nfs['measurementGroupName'], 'MG2')
300 self.assertEqual(mg_with_nfs['administrativeState'], 'UNLOCKED')
301 for nf in mg_with_nfs['networkFunctions']:
302 self.assertEqual(nf['nfMgStatus'], SubNfState.PENDING_CREATE.value)
304 @patch('mod.api.services.measurement_group_service.update_admin_status',
305 MagicMock(side_effect=InvalidDataException('Bad request')))
306 def test_update_admin_state_api_invalid_data_exception(self):
307 error, status_code = update_admin_state('sub4', 'MG2',
308 {'administrativeState': 'UNLOCKED'})
309 self.assertEqual(status_code, HTTPStatus.BAD_REQUEST.value)
310 self.assertEqual(error, 'Bad request')
312 @patch('mod.api.services.measurement_group_service.update_admin_status',
313 MagicMock(side_effect=DataConflictException('Data conflict')))
314 def test_update_admin_state_api_data_conflict_exception(self):
315 error, status_code = update_admin_state('sub4', 'MG2',
316 {'administrativeState': 'UNLOCKED'})
317 self.assertEqual(status_code, HTTPStatus.CONFLICT.value)
318 self.assertEqual(error, 'Data conflict')
320 @patch('mod.api.services.measurement_group_service.update_admin_status',
321 MagicMock(side_effect=Exception('Server Error')))
322 def test_update_admin_state_api_exception(self):
323 error, status_code = update_admin_state('sub4', 'MG2',
324 {'administrativeState': 'UNLOCKED'})
325 self.assertEqual(status_code, HTTPStatus.INTERNAL_SERVER_ERROR.value)
326 self.assertEqual(error, 'Update admin status request was not processed for sub name: sub4 '
327 'and meas group name: MG2 due to Exception : Server Error')