forked from fluent/fluent-logger-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_handler.py
More file actions
46 lines (37 loc) · 1.22 KB
/
test_handler.py
File metadata and controls
46 lines (37 loc) · 1.22 KB
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
# -*- coding: utf-8 -*-
import logging
import unittest
import fluent.handler
from tests import mockserver
class TestHandler(unittest.TestCase):
def setUp(self):
super(TestHandler, self).setUp()
for port in range(10000, 20000):
try:
self._server = mockserver.MockRecvServer('localhost', port)
self._port = port
break
except IOError:
pass
def get_data(self):
return self._server.get_recieved()
def test_simple(self):
handler = fluent.handler.FluentHandler('app.follow', port=self._port)
logging.basicConfig(level=logging.INFO)
log = logging.getLogger('fluent.test')
handler.setFormatter(fluent.handler.FluentRecordFormatter())
log.addHandler(handler)
log.info({
'from': 'userA',
'to': 'userB'
})
handler.close()
data = self.get_data()
eq = self.assertEqual
eq(1, len(data))
eq(3, len(data[0]))
eq('app.follow', data[0][0])
eq('userA', data[0][2]['from'])
eq('userB', data[0][2]['to'])
self.assertTrue(data[0][1])
self.assertTrue(isinstance(data[0][1], int))