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
|
import logging
import sys
import unittest
try:
from unittest import mock
except ImportError:
import mock
import testlib
import mitogen.core
import mitogen.master
import mitogen.parent
import mitogen.utils
from mitogen.core import b
PY2 = sys.version_info[0] == 2
if PY2:
def logging_getLogRecordFactory():
return logging.LogRecord
def logging_setLogRecordFactory(factory):
logging.LogRecord = factory
else:
logging_getLogRecordFactory = logging.getLogRecordFactory
logging_setLogRecordFactory = logging.setLogRecordFactory
def ping():
pass
def log_test():
logging.getLogger(__name__).info("This is a test")
class BufferingTest(testlib.TestCase):
klass = mitogen.core.LogHandler
def record(self):
return logging.LogRecord(
name='name',
level=99,
pathname='pathname',
lineno=123,
msg='msg',
args=(),
exc_info=None,
)
def build(self):
context = mock.Mock()
return context, self.klass(context)
def test_initially_buffered(self):
context, handler = self.build()
rec = self.record()
handler.emit(rec)
self.assertEqual(0, context.send.call_count)
self.assertEqual(1, len(handler._buffer))
def test_uncork(self):
context, handler = self.build()
rec = self.record()
handler.emit(rec)
handler.uncork()
self.assertEqual(1, context.send.call_count)
self.assertEqual(None, handler._buffer)
_, args, _ = context.send.mock_calls[0]
msg, = args
self.assertEqual(mitogen.core.FORWARD_LOG, msg.handle)
self.assertEqual(b('name\x0099\x00msg'), msg.data)
class StartupTest(testlib.RouterMixin, testlib.TestCase):
def test_earliest_messages_logged(self):
log = testlib.LogCapturer()
log.start()
c1 = self.router.local()
c1.shutdown(wait=True)
logs = log.stop()
self.assertIn('Python version is', logs)
self.assertIn('Parent is context 0 (master)', logs)
def test_earliest_messages_logged_via(self):
c1 = self.router.local(name='c1')
# ensure any c1-related msgs are processed before beginning capture
c1.call(ping)
log = testlib.LogCapturer()
log.start()
c2 = self.router.local(via=c1, name='c2', debug=True)
c2.shutdown(wait=True)
logs = log.stop()
self.assertIn('Python version is', logs)
expect = 'Parent is context %s (%s)' % (c1.context_id, 'parent')
self.assertIn(expect, logs)
class LogRecordFactoryTest(testlib.RouterMixin, testlib.TestCase):
def setUp(self):
super(LogRecordFactoryTest, self).setUp()
self.original_factory = logging_getLogRecordFactory()
def tearDown(self):
logging_setLogRecordFactory(self.original_factory)
super(LogRecordFactoryTest, self).tearDown()
def test_logrecordfactory(self):
# Change logging factory and add a custom attribute
old_factory = logging_getLogRecordFactory()
def record_factory(*args, **kwargs):
record = old_factory(*args, **kwargs)
record.custom_attribute = 0xDEADBEEF
return record
logging_setLogRecordFactory(record_factory)
c1 = self.router.local(name="c1")
log = testlib.LogCapturer(
__name__, formatter=logging.Formatter("%(custom_attribute)x - %(message)s")
)
log.start()
c1.call(log_test)
logs = log.stop()
self.assertIn("deadbeef - This is a test", logs)
StartupTest = unittest.skipIf(
condition=sys.version_info < (2, 7) or sys.version_info >= (3, 6),
reason="Message log flaky on Python < 2.7 or >= 3.6"
)(StartupTest)
|
