mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-01-27 10:20:10 +00:00
a4bd6a9b4b
* Convert postgresql.py into a package * Factor out cancellable process into a separate class * Factor out connection handler into a separate class * Move postmaster into postgresql package * Factor out pg_rewind into a separate class * Factor out bootstrap into a separate class * Factor out slots handler into a separate class * Factor out postgresql config handler into a separate class * Move callback_executor into postgresql package This is just a careful refactoring, without code changes.
28 lines
793 B
Python
28 lines
793 B
Python
import unittest
|
|
|
|
from mock import Mock, patch
|
|
from patroni.postgresql.callback_executor import CallbackExecutor
|
|
|
|
|
|
class TestCallbackExecutor(unittest.TestCase):
|
|
|
|
@patch('subprocess.Popen')
|
|
def test_callback_executor(self, mock_popen):
|
|
mock_popen.return_value.wait.side_effect = Exception
|
|
mock_popen.return_value.poll.return_value = None
|
|
|
|
ce = CallbackExecutor()
|
|
self.assertIsNone(ce.call([]))
|
|
ce.join()
|
|
|
|
self.assertIsNone(ce.call([]))
|
|
|
|
mock_popen.return_value.kill.side_effect = OSError
|
|
self.assertIsNone(ce.call([]))
|
|
|
|
mock_popen.side_effect = Exception
|
|
ce = CallbackExecutor()
|
|
ce._callback_event.wait = Mock(side_effect=[None, Exception])
|
|
self.assertIsNone(ce.call([]))
|
|
ce.join()
|