Codebase list logbook / 71f7ae4 tests / test_syslog_handler.py
71f7ae4

Tree @71f7ae4 (Download .tar.gz)

test_syslog_handler.py @71f7ae4 — raw · history · blame 

 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
import os
import re
import socket
from contextlib import closing

import logbook
from logbook.helpers import u

import pytest

unix_socket = "/tmp/__unixsock_logbook.test"

to_test = [
    (socket.AF_INET, socket.SOCK_DGRAM, ('127.0.0.1', 0)),
    (socket.AF_INET, socket.SOCK_STREAM, ('127.0.0.1', 0)),
]
if hasattr(socket, 'AF_UNIX'):
    to_test.append((socket.AF_UNIX, socket.SOCK_DGRAM, unix_socket))

@pytest.mark.usefixtures("unix_sock_path")
@pytest.mark.parametrize("sock_family,socktype,address", to_test)
def test_syslog_handler(logger, activation_strategy,
                        sock_family, socktype, address):
    delimiter = {socket.AF_UNIX: '\x00',
                 socket.AF_INET: '\n'}[sock_family]
    with closing(socket.socket(sock_family, socktype)) as inc:
        inc.bind(address)
        if socktype == socket.SOCK_STREAM:
            inc.listen(0)
        inc.settimeout(1)
        for app_name in [None, 'Testing']:
            if sock_family == socket.AF_UNIX:
                expected = (r'^<12>%stestlogger: Syslog is weird%s$' %
                            (app_name + ':' if app_name else '',
                             delimiter))
            else:
                expected = (r'^<12>1 \d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(\.\d+)?Z %s %s %d '
                            '- - %sSyslog is weird%s$' %
                            (socket.gethostname(),
                             app_name if app_name else 'testlogger',
                             os.getpid(), 'testlogger: ' if app_name else '',
                             delimiter))

            handler = logbook.SyslogHandler(app_name, inc.getsockname(),
                                            socktype=socktype)
            with activation_strategy(handler):
                logger.warn('Syslog is weird')
            try:
                if socktype == socket.SOCK_STREAM:
                    with closing(inc.accept()[0]) as inc2:
                        rv = inc2.recv(1024)
                else:
                    rv = inc.recvfrom(1024)[0]
            except socket.error:
                assert False, 'got timeout on socket'
            rv = rv.decode('utf-8')
            assert re.match(expected, rv), \
                'expected {}, got {}'.format(expected, rv)


@pytest.fixture
def unix_sock_path(request):
    returned = unix_socket

    @request.addfinalizer
    def cleanup():
        if os.path.exists(returned):
            os.unlink(returned)
    return returned