components/pm-subscription-handler/tests/test_policy_response_handler.py

   1 # ============LICENSE_START===================================================
   2 #  Copyright (C) 2019-2020 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
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  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.
  15 #
  16 # SPDX-License-Identifier: Apache-2.0
  17 # ============LICENSE_END=====================================================
  18 from unittest.mock import patch
  19
  20 from mod.api.db_models import SubscriptionModel
  21 from mod.network_function import NetworkFunction
  22 from mod.policy_response_handler import PolicyResponseHandler, policy_response_handle_functions
  23 from mod.subscription import AdministrativeState, SubNfState
  24 from tests.base_setup import BaseClassSetup
  25
  26
  27 class PolicyResponseHandlerTest(BaseClassSetup):
  28
  29     @classmethod
  30     def setUpClass(cls):
  31         super().setUpClass()
  32
  33     @patch('mod.create_app')
  34     @patch('mod.pmsh_utils._MrSub')
  35     def setUp(self, mock_mr_sub, mock_app):
  36         super().setUp()
  37         self.mock_policy_mr_sub = mock_mr_sub
  38         self.nf = NetworkFunction(nf_name='nf1')
  39         self.policy_response_handler = PolicyResponseHandler(self.mock_policy_mr_sub,
  40                                                              self.app_conf,
  41                                                              mock_app)
  42
  43     def tearDown(self):
  44         super().tearDown()
  45
  46     @classmethod
  47     def tearDownClass(cls):
  48         super().tearDownClass()
  49
  50     @patch('mod.network_function.NetworkFunction.delete')
  51     def test_handle_response_locked_success(self, mock_delete):
  52         with patch.dict(policy_response_handle_functions,
  53                         {AdministrativeState.LOCKED.value: {'success': mock_delete}}):
  54             self.policy_response_handler._handle_response(
  55                 self.app_conf.subscription.subscriptionName,
  56                 AdministrativeState.LOCKED.value,
  57                 self.nf.nf_name, 'success')
  58
  59             mock_delete.assert_called()
  60
  61     @patch('mod.subscription.Subscription.update_sub_nf_status')
  62     def test_handle_response_locked_failed(self, mock_update_sub_nf):
  63         with patch.dict(policy_response_handle_functions,
  64                         {AdministrativeState.LOCKED.value: {'failed': mock_update_sub_nf}}):
  65             self.policy_response_handler._handle_response(
  66                 self.app_conf.subscription.subscriptionName,
  67                 AdministrativeState.LOCKED.value,
  68                 self.nf.nf_name, 'failed')
  69             mock_update_sub_nf.assert_called_with(
  70                 subscription_name=self.app_conf.subscription.subscriptionName,
  71                 status=SubNfState.DELETE_FAILED.value, nf_name=self.nf.nf_name)
  72
  73     @patch('mod.subscription.Subscription.update_sub_nf_status')
  74     def test_handle_response_unlocked_success(self, mock_update_sub_nf):
  75         with patch.dict(policy_response_handle_functions,
  76                         {AdministrativeState.UNLOCKED.value: {'success': mock_update_sub_nf}}):
  77             self.policy_response_handler._handle_response(
  78                 self.app_conf.subscription.subscriptionName,
  79                 AdministrativeState.UNLOCKED.value,
  80                 self.nf.nf_name, 'success')
  81             mock_update_sub_nf.assert_called_with(
  82                 subscription_name=self.app_conf.subscription.subscriptionName,
  83                 status=SubNfState.CREATED.value, nf_name=self.nf.nf_name)
  84
  85     @patch('mod.subscription.Subscription.update_sub_nf_status')
  86     def test_handle_response_unlocked_failed(self, mock_update_sub_nf):
  87         with patch.dict(policy_response_handle_functions,
  88                         {AdministrativeState.UNLOCKED.value: {'failed': mock_update_sub_nf}}):
  89             self.policy_response_handler._handle_response(
  90                 self.app_conf.subscription.subscriptionName,
  91                 AdministrativeState.UNLOCKED.value,
  92                 self.nf.nf_name, 'failed')
  93             mock_update_sub_nf.assert_called_with(
  94                 subscription_name=self.app_conf.subscription.subscriptionName,
  95                 status=SubNfState.CREATE_FAILED.value, nf_name=self.nf.nf_name)
  96
  97     def test_handle_response_exception(self):
  98         self.assertRaises(Exception, self.policy_response_handler._handle_response, 'sub1',
  99                           'wrong_state', 'nf1', 'wrong_message')
 100
 101     @patch('mod.policy_response_handler.PolicyResponseHandler._handle_response')
 102     @patch('mod.subscription.Subscription.get')
 103     def test_poll_policy_topic_calls_methods_correct_sub(self, mock_get_sub, mock_handle_response):
 104         response_data = ['{"name": "ResponseEvent","status": { "subscriptionName": '
 105                          '"ExtraPM-All-gNB-R2B", "nfName": "pnf300", "message": "success" } }']
 106         self.mock_policy_mr_sub.get_from_topic.return_value = response_data
 107         mock_get_sub.return_value = SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
 108                                                       status=AdministrativeState.UNLOCKED.value)
 109         self.policy_response_handler.poll_policy_topic()
 110         self.mock_policy_mr_sub.get_from_topic.assert_called()
 111         mock_handle_response.assert_called_with(self.app_conf.subscription.subscriptionName,
 112                                                 AdministrativeState.UNLOCKED.value, 'pnf300',
 113                                                 'success')
 114
 115     @patch('mod.policy_response_handler.PolicyResponseHandler._handle_response')
 116     @patch('mod.subscription.Subscription.get')
 117     def test_poll_policy_topic_no_method_calls_incorrect_sub(self, mock_get_sub,
 118                                                              mock_handle_response):
 119         response_data = ['{"name": "ResponseEvent","status": { "subscriptionName": '
 120                          '"Different_Subscription", "nfName": "pnf300", "message": "success" } }']
 121         self.mock_policy_mr_sub.get_from_topic.return_value = response_data
 122         mock_get_sub.return_value = SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
 123                                                       status=AdministrativeState.UNLOCKED.value)
 124         self.policy_response_handler.poll_policy_topic()
 125
 126         self.mock_policy_mr_sub.get_from_topic.assert_called()
 127
 128         mock_handle_response.assert_not_called()
 129
 130     @patch('mod.logger.error')
 131     @patch('mod.subscription.Subscription.get')
 132     def test_poll_policy_topic_exception(self, mock_get_sub, mock_logger):
 133         self.mock_policy_mr_sub.get_from_topic.return_value = 'wrong_return'
 134         mock_get_sub.return_value = SubscriptionModel(subscription_name='ExtraPM-All-gNB-R2B',
 135                                                       status=AdministrativeState.UNLOCKED.value)
 136         self.policy_response_handler.poll_policy_topic()
 137         mock_logger.assert_called()