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
|
# Copyright The OpenTelemetry Authors
#
# 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.
from logging import ERROR
from unittest import TestCase
from unittest.mock import Mock, patch
from opentelemetry.sdk.error_handler import (
ErrorHandler,
GlobalErrorHandler,
logger,
)
class TestErrorHandler(TestCase):
@patch("opentelemetry.sdk.error_handler.entry_points")
def test_default_error_handler(self, mock_entry_points):
with self.assertLogs(logger, ERROR):
with GlobalErrorHandler():
# pylint: disable=broad-exception-raised
raise Exception("some exception")
# pylint: disable=no-self-use
@patch("opentelemetry.sdk.error_handler.entry_points")
def test_plugin_error_handler(self, mock_entry_points):
class ZeroDivisionErrorHandler(ErrorHandler, ZeroDivisionError):
# pylint: disable=arguments-differ
_handle = Mock()
class AssertionErrorHandler(ErrorHandler, AssertionError):
# pylint: disable=arguments-differ
_handle = Mock()
mock_entry_point_zero_division_error_handler = Mock()
mock_entry_point_zero_division_error_handler.configure_mock(
**{"load.return_value": ZeroDivisionErrorHandler}
)
mock_entry_point_assertion_error_handler = Mock()
mock_entry_point_assertion_error_handler.configure_mock(
**{"load.return_value": AssertionErrorHandler}
)
mock_entry_points.configure_mock(
**{
"return_value": [
mock_entry_point_zero_division_error_handler,
mock_entry_point_assertion_error_handler,
]
}
)
error = ZeroDivisionError()
with GlobalErrorHandler():
raise error
# pylint: disable=protected-access
ZeroDivisionErrorHandler._handle.assert_called_with(error)
error = AssertionError()
with GlobalErrorHandler():
raise error
AssertionErrorHandler._handle.assert_called_with(error)
@patch("opentelemetry.sdk.error_handler.entry_points")
def test_error_in_handler(self, mock_entry_points):
class ErrorErrorHandler(ErrorHandler, ZeroDivisionError):
# pylint: disable=arguments-differ
def _handle(self, error: Exception):
assert False
mock_entry_point_error_error_handler = Mock()
mock_entry_point_error_error_handler.configure_mock(
**{"load.return_value": ErrorErrorHandler}
)
mock_entry_points.configure_mock(
**{"return_value": [mock_entry_point_error_error_handler]}
)
error = ZeroDivisionError()
with self.assertLogs(logger, ERROR):
with GlobalErrorHandler():
raise error
# pylint: disable=no-self-use
@patch("opentelemetry.sdk.error_handler.entry_points")
def test_plugin_error_handler_context_manager(self, mock_entry_points):
mock_error_handler_instance = Mock()
class MockErrorHandlerClass(IndexError):
def __new__(cls):
return mock_error_handler_instance
mock_entry_point_error_handler = Mock()
mock_entry_point_error_handler.configure_mock(
**{"load.return_value": MockErrorHandlerClass}
)
mock_entry_points.configure_mock(
**{"return_value": [mock_entry_point_error_handler]}
)
error = IndexError()
with GlobalErrorHandler():
raise error
with GlobalErrorHandler():
pass
# pylint: disable=protected-access
mock_error_handler_instance._handle.assert_called_once_with(error)
|
