1 # ============LICENSE_START===================================================
2 # Copyright (C) 2020-2021 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=====================================================
18 from unittest.mock import patch, Mock, MagicMock
20 from mod.api.db_models import NetworkFunctionModel
21 from mod.network_function import NetworkFunction
22 from mod.subscription import AdministrativeState
23 from mod.subscription_handler import SubscriptionHandler
24 from tests.base_setup import BaseClassSetup, get_pmsh_config
27 class SubscriptionHandlerTest(BaseClassSetup):
29 NetworkFunction(nf_name='pnf_1', model_invariant_id='some-id', model_version_id='some-id'),
30 NetworkFunction(nf_name='pnf_2', model_invariant_id='some-id', model_version_id='some-id')]
36 @patch('mod.pmsh_utils._MrSub')
37 @patch('mod.pmsh_utils._MrPub')
38 def setUp(self, mock_mr_pub, mock_mr_sub):
40 self.mock_mr_pub = mock_mr_pub
41 self.mock_mr_sub = mock_mr_sub
42 self.mock_policy_event_thread = Mock()
48 def tearDownClass(cls):
49 super().tearDownClass()
51 @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
52 @patch('mod.subscription.Subscription.get_local_sub_admin_state')
53 @patch('mod.logger.info')
54 @patch('mod.aai_client.get_pmsh_nfs_from_aai')
55 def test_execute_no_change_of_state(self, mock_get_aai, mock_logger, mock_get_sub_status):
56 mock_get_sub_status.return_value = AdministrativeState.UNLOCKED.value
57 mock_get_aai.return_value = self.nfs
58 sub_handler = SubscriptionHandler(self.mock_mr_pub,
59 self.mock_mr_sub, self.app, self.app_conf)
61 mock_logger.assert_called_with('Administrative State did not change '
62 'in the app config: UNLOCKED')
64 @patch('mod.subscription_handler.SubscriptionHandler._start_aai_event_thread',
66 @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
67 @patch('mod.subscription.Subscription.get_local_sub_admin_state')
68 @patch('mod.subscription.Subscription.create_subscription_on_nfs')
69 @patch('mod.aai_client.get_pmsh_nfs_from_aai')
70 def test_execute_change_of_state_to_unlocked(self, mock_get_aai, mock_activate_sub,
72 mock_get_aai.return_value = self.nfs
73 mock_get_sub_status.return_value = AdministrativeState.LOCKED.value
74 sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
77 self.assertEqual(AdministrativeState.UNLOCKED.value,
78 self.app_conf.subscription.administrativeState)
79 mock_activate_sub.assert_called_with(self.nfs, self.mock_mr_pub)
81 @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs))
82 @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
83 @patch('mod.subscription.Subscription.get_local_sub_admin_state')
84 @patch('mod.subscription.Subscription.delete_subscription_from_nfs')
85 def test_execute_change_of_state_to_locked(self, mock_deactivate_sub, mock_get_sub_status):
86 mock_get_sub_status.return_value = AdministrativeState.UNLOCKED.value
87 self.app_conf.subscription.administrativeState = AdministrativeState.LOCKED.value
88 self.app_conf.subscription.update_subscription_status()
89 sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
92 mock_deactivate_sub.assert_called_with(self.nfs, self.mock_mr_pub)
94 @patch('mod.subscription_handler.SubscriptionHandler._start_aai_event_thread', MagicMock())
95 @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
96 @patch('mod.subscription.Subscription.create_subscription_on_nfs')
97 @patch('mod.logger.error')
98 @patch('mod.aai_client.get_pmsh_nfs_from_aai')
99 def test_execute_exception(self, mock_get_aai, mock_logger, mock_activate_sub):
100 mock_get_aai.return_value = self.nfs
101 mock_activate_sub.side_effect = Exception
102 sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
104 sub_handler.execute()
105 mock_logger.assert_called_with('Error occurred during the activation/deactivation process ',
108 @patch('mod.network_function.NetworkFunction.get',
109 MagicMock(return_value=NetworkFunctionModel(nf_name='pnf_1',
110 model_invariant_id='some-id',
111 model_version_id='some-id',
112 ipv4_address='ip_address4',
113 ipv6_address='ip_address6',
114 model_name='model_name',
115 sdnc_model_name='sdnc_model_name',
116 sdnc_model_version='sdnc_model_version')))
117 @patch('mod.subscription.Subscription.get_delete_failed_nfs', MagicMock(return_value=nfs))
118 @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs))
119 @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
120 @patch('mod.subscription.Subscription.get_local_sub_admin_state')
121 @patch('mod.subscription.Subscription.delete_subscription_from_nfs')
122 @patch('mod.network_function.NetworkFunction.increment_retry_count')
123 def test_execute_change_of_state_to_locking_retry_delete(self, mock_retry_inc, mock_delete_sub,
124 mock_get_sub_status):
125 mock_get_sub_status.return_value = AdministrativeState.LOCKING.value
126 sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
128 sub_handler.execute()
129 self.assertEqual(mock_delete_sub.call_count, 2)
130 self.assertEqual(mock_retry_inc.call_count, 2)
132 @patch('mod.subscription.Subscription.get_delete_failed_nfs', MagicMock(return_value=[]))
133 @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs))
134 @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
135 @patch('mod.subscription.Subscription.get_local_sub_admin_state')
136 @patch('mod.subscription.Subscription.update_subscription_status')
137 def test_execute_change_of_state_to_locking_success(self, mock_update_sub,
138 mock_get_sub_status):
139 mock_get_sub_status.return_value = AdministrativeState.LOCKING.value
140 sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
142 sub_handler.execute()
143 mock_update_sub.assert_called_once()
145 @patch('mod.network_function.NetworkFunction.get',
146 MagicMock(return_value=NetworkFunctionModel(nf_name='pnf_1',
147 model_invariant_id='some-id',
148 model_version_id='some-id',
149 ipv4_address='ip_address4',
150 ipv6_address='ip_address6',
151 model_name='model_name',
152 sdnc_model_name='sdnc_model_name',
153 sdnc_model_version='sdnc_model_version',
155 @patch('mod.subscription.Subscription.get_delete_failed_nfs', MagicMock(return_value=[nfs[0]]))
156 @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs[0]))
157 @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config()))
158 @patch('mod.subscription.Subscription.get_local_sub_admin_state')
159 @patch('mod.network_function.NetworkFunction.delete')
160 def test_execute_change_of_state_to_locking_retry_failed(self, mock_nf_del,
161 mock_get_sub_status):
162 mock_get_sub_status.return_value = AdministrativeState.LOCKING.value
163 sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
165 sub_handler.execute()
166 mock_nf_del.assert_called_once()
168 @patch('mod.pmsh_utils.AppConfig._get_pmsh_config',
169 MagicMock(return_value=get_pmsh_config('data/cbs_invalid_data.json')))
170 @patch('mod.subscription_handler.SubscriptionHandler._check_state_change')
171 def test_execute_invalid_schema(self, mock_change_state_check):
172 sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
174 sub_handler.execute()
175 mock_change_state_check.assert_not_called()
177 @patch('mod.pmsh_utils.AppConfig._get_pmsh_config',
178 MagicMock(return_value=get_pmsh_config()))
179 @patch('mod.subscription_handler.SubscriptionHandler._check_state_change')
180 def test_execute_valid_schema(self, mock_change_state_check):
181 sub_handler = SubscriptionHandler(self.mock_mr_pub, self.mock_mr_sub, self.app,
183 sub_handler.execute()
184 mock_change_state_check.assert_called_once()