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

   1 # ============LICENSE_START===================================================
   2 #  Copyright (C) 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 import json
  19 import os
  20 from signal import SIGTERM, signal
  21 from test.support import EnvironmentVarGuard
  22 from unittest import TestCase
  23 from unittest.mock import patch, Mock
  24
  25 from mod.api.db_models import NetworkFunctionModel
  26 from mod.exit_handler import ExitHandler
  27 from mod.pmsh_utils import AppConfig
  28 from mod.subscription import Subscription
  29
  30
  31 class ExitHandlerTests(TestCase):
  32     @patch('mod.subscription.Subscription.create')
  33     @patch('mod.pmsh_utils.AppConfig._get_pmsh_config')
  34     @patch('mod.pmsh_utils.PeriodicTask')
  35     def setUp(self, mock_periodic_task, mock_get_pmsh_config, mock_sub_create):
  36         self.env = EnvironmentVarGuard()
  37         self.env.set('LOGGER_CONFIG', os.path.join(os.path.dirname(__file__), 'log_config.yaml'))
  38         with open(os.path.join(os.path.dirname(__file__), 'data/cbs_data_1.json'), 'r') as data:
  39             self.cbs_data = json.load(data)
  40         mock_get_pmsh_config.return_value = self.cbs_data
  41         self.mock_aai_event_thread = mock_periodic_task
  42         self.app_conf = AppConfig()
  43         self.sub = self.app_conf.subscription
  44
  45     @patch('mod.logger.debug')
  46     @patch.object(Subscription, 'update_sub_nf_status')
  47     @patch.object(Subscription, 'update_subscription_status')
  48     @patch.object(Subscription, '_get_nf_models',
  49                   return_value=[NetworkFunctionModel('pnf1', 'ACTIVE')])
  50     def test_terminate_signal_successful(self, mock_sub_get_nf_models, mock_upd_sub_status,
  51                                          mock_upd_subnf_status, mock_logger):
  52         handler = ExitHandler(periodic_tasks=[self.mock_aai_event_thread],
  53                               app_conf=self.app_conf,
  54                               subscription_handler=Mock())
  55         signal(SIGTERM, handler)
  56         os.kill(os.getpid(), SIGTERM)
  57         self.assertTrue(ExitHandler.shutdown_signal_received)