Welcome to mirror list, hosted at ThFree Co, Russian Federation.

error_handler_test.py « tests « certbot - github.com/certbot/certbot.git - Unnamed repository; edit this file 'description' to name the repository.
summaryrefslogtreecommitdiff
blob: 8508a3df5a31264cc2ba15e752c481e0281a9db1 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
"""Tests for certbot.error_handler."""
import contextlib
import os
import signal
import sys
import unittest

import mock
# pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Callable, Dict, Union
# pylint: enable=unused-import, no-name-in-module

import certbot.tests.util as test_util

def get_signals(signums):
    """Get the handlers for an iterable of signums."""
    return dict((s, signal.getsignal(s)) for s in signums)


def set_signals(sig_handler_dict):
    """Set the signal (keys) with the handler (values) from the input dict."""
    for s, h in sig_handler_dict.items():
        signal.signal(s, h)


@contextlib.contextmanager
def signal_receiver(signums):
    """Context manager to catch signals"""
    signals = []
    prev_handlers = get_signals(signums)  # type: Dict[int, Union[int, None, Callable]]
    set_signals(dict((s, lambda s, _: signals.append(s)) for s in signums))
    yield signals
    set_signals(prev_handlers)


def send_signal(signum):
    """Send the given signal"""
    os.kill(os.getpid(), signum)


class ErrorHandlerTest(unittest.TestCase):
    """Tests for certbot.error_handler.ErrorHandler."""

    def setUp(self):
        from certbot import error_handler

        self.init_func = mock.MagicMock()
        self.init_args = set((42,))
        self.init_kwargs = {'foo': 'bar'}
        self.handler = error_handler.ErrorHandler(self.init_func,
                                                  *self.init_args,
                                                  **self.init_kwargs)

        # pylint: disable=protected-access
        self.signals = error_handler._SIGNALS

    def test_context_manager(self):
        exception_raised = False
        try:
            with self.handler:
                raise ValueError
        except ValueError:
            exception_raised = True

        self.assertTrue(exception_raised)
        self.init_func.assert_called_once_with(*self.init_args,
                                               **self.init_kwargs)

    # On Windows, this test kills pytest itself !
    @test_util.broken_on_windows
    def test_context_manager_with_signal(self):
        init_signals = get_signals(self.signals)
        with signal_receiver(self.signals) as signals_received:
            with self.handler:
                should_be_42 = 42
                send_signal(self.signals[0])
                should_be_42 *= 10

        # check execution stoped when the signal was sent
        self.assertEqual(42, should_be_42)
        # assert signals were caught
        self.assertEqual([self.signals[0]], signals_received)
        # assert the error handling function was just called once
        self.init_func.assert_called_once_with(*self.init_args,
                                               **self.init_kwargs)
        for signum in self.signals:
            self.assertEqual(init_signals[signum], signal.getsignal(signum))

    def test_bad_recovery(self):
        bad_func = mock.MagicMock(side_effect=[ValueError])
        self.handler.register(bad_func)
        try:
            with self.handler:
                raise ValueError
        except ValueError:
            pass
        self.init_func.assert_called_once_with(*self.init_args,
                                               **self.init_kwargs)
        bad_func.assert_called_once_with()

    # On Windows, this test kills pytest itself !
    @test_util.broken_on_windows
    def test_bad_recovery_with_signal(self):
        sig1 = self.signals[0]
        sig2 = self.signals[-1]
        bad_func = mock.MagicMock(side_effect=lambda: send_signal(sig1))
        self.handler.register(bad_func)
        with signal_receiver(self.signals) as signals_received:
            with self.handler:
                send_signal(sig2)
        self.assertEqual([sig2, sig1], signals_received)
        self.init_func.assert_called_once_with(*self.init_args,
                                               **self.init_kwargs)
        bad_func.assert_called_once_with()

    def test_sysexit_ignored(self):
        try:
            with self.handler:
                sys.exit(0)
        except SystemExit:
            pass
        self.assertFalse(self.init_func.called)

    def test_regular_exit(self):
        func = mock.MagicMock()
        self.handler.register(func)
        with self.handler:
            pass
        self.init_func.assert_not_called()
        func.assert_not_called()


class ExitHandlerTest(ErrorHandlerTest):
    """Tests for certbot.error_handler.ExitHandler."""

    def setUp(self):
        from certbot import error_handler
        super(ExitHandlerTest, self).setUp()
        self.handler = error_handler.ExitHandler(self.init_func,
                                                 *self.init_args,
                                                 **self.init_kwargs)

    def test_regular_exit(self):
        func = mock.MagicMock()
        self.handler.register(func)
        with self.handler:
            pass
        self.init_func.assert_called_once_with(*self.init_args,
                                               **self.init_kwargs)
        func.assert_called_once_with()

    # On Windows, this test kills pytest itself !
    @test_util.broken_on_windows
    def test_bad_recovery_with_signal(self):
        super(ExitHandlerTest, self).test_bad_recovery_with_signal()

if __name__ == "__main__":
    unittest.main()  # pragma: no cover