diff options
Diffstat (limited to 'components/pm-subscription-handler/tests/test_subscription_handler.py')
| -rw-r--r-- | components/pm-subscription-handler/tests/test_subscription_handler.py | 106 |
1 files changed, 80 insertions, 26 deletions
diff --git a/components/pm-subscription-handler/tests/test_subscription_handler.py b/components/pm-subscription-handler/tests/test_subscription_handler.py index bf72382a..f77dfd10 100644 --- a/components/pm-subscription-handler/tests/test_subscription_handler.py +++ b/components/pm-subscription-handler/tests/test_subscription_handler.py @@ -15,15 +15,19 @@ # # SPDX-License-Identifier: Apache-2.0 # ============LICENSE_END===================================================== -from unittest.mock import patch, Mock +from unittest.mock import patch, Mock, MagicMock +from mod.api.db_models import NetworkFunctionModel from mod.network_function import NetworkFunction from mod.subscription import AdministrativeState from mod.subscription_handler import SubscriptionHandler -from tests.base_setup import BaseClassSetup +from tests.base_setup import BaseClassSetup, get_pmsh_config class SubscriptionHandlerTest(BaseClassSetup): + nfs = [ + NetworkFunction(nf_name='pnf_1', model_invariant_id='some-id', model_version_id='some-id'), + NetworkFunction(nf_name='pnf_2', model_invariant_id='some-id', model_version_id='some-id')] @classmethod def setUpClass(cls): @@ -32,12 +36,6 @@ class SubscriptionHandlerTest(BaseClassSetup): @patch('mod.pmsh_utils._MrPub') def setUp(self, mock_mr_pub): super().setUp() - self.nfs = [NetworkFunction(nf_name='pnf_1', - model_invariant_id='some-id', - model_version_id='some-id'), - NetworkFunction(nf_name='pnf_2', - model_invariant_id='some-id', - model_version_id='some-id')] self.mock_mr_pub = mock_mr_pub self.mock_aai_event_thread = Mock() self.mock_policy_event_thread = Mock() @@ -49,6 +47,7 @@ class SubscriptionHandlerTest(BaseClassSetup): def tearDownClass(cls): super().tearDownClass() + @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config())) @patch('mod.subscription.Subscription.get_local_sub_admin_state') @patch('mod.logger.info') @patch('mod.aai_client.get_pmsh_nfs_from_aai') @@ -57,13 +56,14 @@ class SubscriptionHandlerTest(BaseClassSetup): mock_get_aai.return_value = self.nfs sub_handler = SubscriptionHandler(self.mock_mr_pub, self.app, self.app_conf, - self.mock_aai_event_thread, - self.mock_policy_event_thread) + self.mock_aai_event_thread) sub_handler.execute() - mock_logger.assert_called_with('Administrative State did not change in the Config') + mock_logger.assert_called_with('Administrative State did not change ' + 'in the app config: UNLOCKED') + @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config())) @patch('mod.subscription.Subscription.get_local_sub_admin_state') - @patch('mod.subscription.Subscription.activate_subscription') + @patch('mod.subscription.Subscription.create_subscription_on_nfs') @patch('mod.aai_client.get_pmsh_nfs_from_aai') def test_execute_change_of_state_to_unlocked(self, mock_get_aai, mock_activate_sub, mock_get_sub_status): @@ -72,33 +72,30 @@ class SubscriptionHandlerTest(BaseClassSetup): self.mock_aai_event_thread.return_value.start.return_value = 'start_method' sub_handler = SubscriptionHandler(self.mock_mr_pub, self.app, self.app_conf, - self.mock_aai_event_thread.return_value, - self.mock_policy_event_thread) + self.mock_aai_event_thread.return_value) sub_handler.execute() self.assertEqual(AdministrativeState.UNLOCKED.value, self.app_conf.subscription.administrativeState) mock_activate_sub.assert_called_with(self.nfs, self.mock_mr_pub, self.app_conf) - self.mock_aai_event_thread.return_value.start.assert_called() + @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs)) + @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config())) @patch('mod.subscription.Subscription.get_local_sub_admin_state') - @patch('mod.subscription.Subscription.deactivate_subscription') - @patch('mod.aai_client.get_pmsh_nfs_from_aai') - def test_execute_change_of_state_to_locked(self, mock_get_aai, mock_deactivate_sub, - mock_get_sub_status): + @patch('mod.subscription.Subscription.delete_subscription_from_nfs') + def test_execute_change_of_state_to_locked(self, mock_deactivate_sub, mock_get_sub_status): mock_get_sub_status.return_value = AdministrativeState.UNLOCKED.value self.app_conf.subscription.administrativeState = AdministrativeState.LOCKED.value self.app_conf.subscription.update_subscription_status() - mock_get_aai.return_value = self.nfs self.mock_aai_event_thread.return_value.cancel.return_value = 'cancel_method' sub_handler = SubscriptionHandler(self.mock_mr_pub, self.app, self.app_conf, - self.mock_aai_event_thread.return_value, - self.mock_policy_event_thread) + self.mock_aai_event_thread.return_value) sub_handler.execute() - mock_deactivate_sub.assert_called_with(self.mock_mr_pub, self.app_conf) + mock_deactivate_sub.assert_called_with(self.nfs, self.mock_mr_pub, self.app_conf) self.mock_aai_event_thread.return_value.cancel.assert_called() - @patch('mod.subscription.Subscription.activate_subscription') + @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config())) + @patch('mod.subscription.Subscription.create_subscription_on_nfs') @patch('mod.logger.error') @patch('mod.aai_client.get_pmsh_nfs_from_aai') def test_execute_exception(self, mock_get_aai, mock_logger, mock_activate_sub): @@ -106,8 +103,65 @@ class SubscriptionHandlerTest(BaseClassSetup): mock_activate_sub.side_effect = Exception sub_handler = SubscriptionHandler(self.mock_mr_pub, self.app, self.app_conf, - self.mock_aai_event_thread, - self.mock_policy_event_thread) + self.mock_aai_event_thread) sub_handler.execute() mock_logger.assert_called_with('Error occurred during the activation/deactivation process ', exc_info=True) + + @patch('mod.network_function.NetworkFunction.get', + MagicMock(return_value=NetworkFunctionModel(nf_name='pnf_1', + model_invariant_id='some-id', + model_version_id='some-id', + ip_address='ip_address', + model_name='model_name', + sdnc_model_name='sdnc_model_name', + sdnc_model_version='sdnc_model_version'))) + @patch('mod.subscription.Subscription.get_delete_failed_nfs', MagicMock(return_value=nfs)) + @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs)) + @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config())) + @patch('mod.subscription.Subscription.get_local_sub_admin_state') + @patch('mod.subscription.Subscription.delete_subscription_from_nfs') + @patch('mod.network_function.NetworkFunction.increment_retry_count') + def test_execute_change_of_state_to_locking_retry_delete(self, mock_retry_inc, mock_delete_sub, + mock_get_sub_status): + mock_get_sub_status.return_value = AdministrativeState.LOCKING.value + sub_handler = SubscriptionHandler(self.mock_mr_pub, self.app, self.app_conf, + self.mock_aai_event_thread) + sub_handler.execute() + self.assertEqual(mock_delete_sub.call_count, 2) + self.assertEqual(mock_retry_inc.call_count, 2) + + @patch('mod.subscription.Subscription.get_delete_failed_nfs', MagicMock(return_value=[])) + @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs)) + @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config())) + @patch('mod.subscription.Subscription.get_local_sub_admin_state') + @patch('mod.subscription.Subscription.update_subscription_status') + def test_execute_change_of_state_to_locking_success(self, mock_update_sub, + mock_get_sub_status): + mock_get_sub_status.return_value = AdministrativeState.LOCKING.value + sub_handler = SubscriptionHandler(self.mock_mr_pub, self.app, self.app_conf, + self.mock_aai_event_thread) + sub_handler.execute() + mock_update_sub.assert_called_once() + + @patch('mod.network_function.NetworkFunction.get', + MagicMock(return_value=NetworkFunctionModel(nf_name='pnf_1', + model_invariant_id='some-id', + model_version_id='some-id', + ip_address='ip_address', + model_name='model_name', + sdnc_model_name='sdnc_model_name', + sdnc_model_version='sdnc_model_version', + retry_count=3))) + @patch('mod.subscription.Subscription.get_delete_failed_nfs', MagicMock(return_value=[nfs[0]])) + @patch('mod.subscription.Subscription.get_network_functions', MagicMock(return_value=nfs[0])) + @patch('mod.pmsh_utils.AppConfig.refresh_config', MagicMock(return_value=get_pmsh_config())) + @patch('mod.subscription.Subscription.get_local_sub_admin_state') + @patch('mod.network_function.NetworkFunction.delete') + def test_execute_change_of_state_to_locking_retry_failed(self, mock_nf_del, + mock_get_sub_status): + mock_get_sub_status.return_value = AdministrativeState.LOCKING.value + sub_handler = SubscriptionHandler(self.mock_mr_pub, self.app, self.app_conf, + self.mock_aai_event_thread) + sub_handler.execute() + mock_nf_del.assert_called_once() |