209 lines
7.1 KiB
Python
209 lines
7.1 KiB
Python
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import signal
|
|
from unittest import mock
|
|
|
|
from oslo_config import cfg
|
|
from oslotest import base
|
|
|
|
CONF = cfg.CONF
|
|
|
|
NUMBER_OF_METRICS_PROCESSES = 2
|
|
NUMBER_OF_ALARM_HIST_PROCESSES = 3
|
|
NUMBER_OF_EVENTS_PROCESSES = 0
|
|
|
|
|
|
class FakeException(Exception):
|
|
pass
|
|
|
|
|
|
class TestPersister(base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestPersister, self).setUp()
|
|
|
|
from monasca_persister import persister
|
|
self.persister = persister
|
|
|
|
self._set_patchers()
|
|
self._set_mocks()
|
|
|
|
def _set_patchers(self):
|
|
self.sys_exit_patcher = mock.patch.object(self.persister.sys, 'exit')
|
|
self.log_patcher = mock.patch.object(self.persister, 'log')
|
|
self.simport_patcher = mock.patch.object(self.persister, 'simport')
|
|
self.cfg_patcher = mock.patch.object(self.persister, 'cfg')
|
|
self.sleep_patcher = mock.patch.object(self.persister.time, 'sleep')
|
|
self.process_patcher = mock.patch.object(
|
|
self.persister.multiprocessing, 'Process')
|
|
|
|
def _set_mocks(self):
|
|
self.mock_sys_exit = self.sys_exit_patcher.start()
|
|
self.mock_log = self.log_patcher.start()
|
|
self.mock_simport = self.simport_patcher.start()
|
|
self.mock_cfg = self.cfg_patcher.start()
|
|
self.mock_sleep = self.sleep_patcher.start()
|
|
self.mock_process_class = self.process_patcher.start()
|
|
|
|
self.mock_cfg.CONF.kafka_metrics.num_processors = NUMBER_OF_METRICS_PROCESSES
|
|
self.mock_cfg.CONF.kafka_alarm_history.num_processors = NUMBER_OF_ALARM_HIST_PROCESSES
|
|
self.mock_cfg.CONF.kafka_events.num_processors = NUMBER_OF_EVENTS_PROCESSES
|
|
|
|
self.mock_cfg.CONF.zookeeper = 'zookeeper'
|
|
|
|
self.mock_sleep.side_effect = [FakeException, None]
|
|
|
|
self.process_pool = _get_process_list(
|
|
NUMBER_OF_METRICS_PROCESSES + NUMBER_OF_ALARM_HIST_PROCESSES + 1)
|
|
self.mock_process_class.side_effect = self.process_pool
|
|
|
|
def tearDown(self):
|
|
super(TestPersister, self).tearDown()
|
|
|
|
self.mock_sleep.side_effect = None
|
|
|
|
self.mock_sys_exit.side_effect = None
|
|
self.mock_sys_exit.reset_mock()
|
|
|
|
self.sys_exit_patcher.stop()
|
|
self.log_patcher.stop()
|
|
self.simport_patcher.stop()
|
|
self.cfg_patcher.stop()
|
|
self.sleep_patcher.stop()
|
|
self.process_patcher.stop()
|
|
|
|
self.persister.processors = []
|
|
self.persister.exiting = False
|
|
|
|
def _run_persister(self):
|
|
self.persister.main()
|
|
self.assertTrue(self.mock_process_class.called)
|
|
|
|
def _assert_clean_exit_handler_terminates_with_expected_signal(
|
|
self, signum, expected):
|
|
# in order to really exit, an exception is thrown
|
|
|
|
self.mock_sys_exit.side_effect = FakeException
|
|
self.assertRaises(FakeException, self.persister.clean_exit, signum)
|
|
|
|
self.mock_sys_exit.assert_called_once_with(expected)
|
|
|
|
def _assert_correct_number_of_processes_created(self):
|
|
|
|
for p in self.process_pool[:-1]:
|
|
self.assertTrue(p.start.called)
|
|
|
|
self.assertFalse(self.process_pool[-1].called)
|
|
|
|
def _assert_process_alive_status_called(self):
|
|
|
|
for p in self.process_pool[:-1]:
|
|
self.assertTrue(p.is_alive.called)
|
|
|
|
def _assert_process_terminate_called(self):
|
|
|
|
for p in self.process_pool[:-1]:
|
|
self.assertTrue(p.terminate.called)
|
|
|
|
def test_active_children_are_killed_during_exit(self):
|
|
|
|
with mock.patch.object(self.persister.multiprocessing,
|
|
'active_children') as active_children,\
|
|
mock.patch.object(self.persister.os, 'kill') as mock_kill:
|
|
|
|
active_children.return_value = [mock.Mock(name='child-1', pid=1),
|
|
mock.Mock(name='child-2', pid=2)]
|
|
|
|
self.persister.clean_exit(0)
|
|
|
|
mock_kill.assert_has_calls([mock.call(1, signal.SIGKILL),
|
|
mock.call(2, signal.SIGKILL)])
|
|
|
|
def test_active_children_kill_exception_is_ignored(self):
|
|
|
|
with mock.patch.object(self.persister.multiprocessing,
|
|
'active_children') as active_children, \
|
|
mock.patch.object(self.persister.os, 'kill') as mock_kill:
|
|
|
|
active_children.return_value = [mock.Mock()]
|
|
mock_kill.side_effect = FakeException
|
|
|
|
self.persister.clean_exit(0)
|
|
|
|
self.assertTrue(mock_kill.called)
|
|
|
|
def test_clean_exit_does_nothing_when_exiting_is_true(self):
|
|
self.persister.exiting = True
|
|
|
|
self.assertIsNone(self.persister.clean_exit(0))
|
|
|
|
self.assertFalse(self.mock_sys_exit.called)
|
|
|
|
def test_exception_during_process_termination_is_ignored(self):
|
|
dead_process = _get_process('raises_when_terminated')
|
|
dead_process.terminate.side_effect = FakeException
|
|
self.process_pool[0] = dead_process
|
|
|
|
self._run_persister()
|
|
|
|
self.assertTrue(dead_process.terminate.called)
|
|
|
|
def test_if_not_sigterm_then_clean_exit_handler_terminates_with_signal(self):
|
|
|
|
self._assert_clean_exit_handler_terminates_with_expected_signal(
|
|
signal.SIGINT, signal.SIGINT)
|
|
|
|
def test_if_sigterm_then_clean_exit_handler_terminates_with_zero(self):
|
|
|
|
self._assert_clean_exit_handler_terminates_with_expected_signal(
|
|
signal.SIGTERM, 0)
|
|
|
|
def test_non_running_process_not_terminated(self):
|
|
dead_process = _get_process('dead_process', is_alive=False)
|
|
self.process_pool[0] = dead_process
|
|
|
|
self._run_persister()
|
|
|
|
self.assertTrue(dead_process.is_alive.called)
|
|
self.assertFalse(dead_process.terminate.called)
|
|
|
|
def test_running_processes_are_created_and_terminated(self):
|
|
|
|
self._run_persister()
|
|
|
|
self._assert_correct_number_of_processes_created()
|
|
self._assert_process_alive_status_called()
|
|
self._assert_process_terminate_called()
|
|
|
|
def test_start_process_handler_creates_and_runs_persister(self):
|
|
fake_kafka_config = mock.Mock()
|
|
fake_repository = mock.Mock()
|
|
|
|
with mock.patch('monasca_persister.kafka.legacy_kafka_persister.'
|
|
'LegacyKafkaPersister') as mock_persister_class:
|
|
self.persister.start_process(fake_repository, fake_kafka_config)
|
|
|
|
mock_persister_class.assert_called_once_with(
|
|
fake_kafka_config, 'zookeeper', fake_repository)
|
|
|
|
|
|
def _get_process(name, is_alive=True):
|
|
return mock.Mock(name=name, is_alive=mock.Mock(return_value=is_alive))
|
|
|
|
|
|
def _get_process_list(number_of_processes):
|
|
processes = []
|
|
for i in range(number_of_processes):
|
|
processes.append(_get_process(name='process_{}'.format(i)))
|
|
return processes
|