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
|
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
def ping():
pass
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)
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)
|
