the form of a ``(host, port)`` tuple. If *address* is not specified,
``('localhost', 514)`` is used. The address is used to open a socket. An
alternative to providing a ``(host, port)`` tuple is providing an address as a
- string, for example '/dev/log'. In this case, a Unix domain socket is used to
+ string or a :class:`bytes` object, for example '/dev/log'.
+ In this case, a Unix domain socket is used to
send the message to the syslog. If *facility* is not specified,
:const:`LOG_USER` is used. The type of socket opened depends on the
*socktype* argument, which defaults to :const:`socket.SOCK_DGRAM` and thus
.. versionchanged:: 3.14
*timeout* was added.
+ .. versionchanged:: next
+ *address* can now be a :class:`bytes` object.
+
.. method:: close()
Closes the socket to the remote host.
"""
Initialize a handler.
- If address is specified as a string, a UNIX socket is used. To log to a
- local syslogd, "SysLogHandler(address="/dev/log")" can be used.
+ If address is specified as a string or bytes, a UNIX socket is used.
+ To log to a local syslogd, "SysLogHandler(address="/dev/log")" can be
+ used.
If facility is not specified, LOG_USER is used. If socktype is
specified as socket.SOCK_DGRAM or socket.SOCK_STREAM, that specific
socket type will be used. For Unix sockets, you can also specify a
address = self.address
socktype = self.socktype
- if isinstance(address, str):
+ if not isinstance(address, (list, tuple)):
self.unixsocket = True
# Syslog server may be unavailable during handler initialisation.
# C's openlog() function also ignores connection errors.
self.addCleanup(os_helper.unlink, self.address)
SysLogHandlerTest.setUp(self)
+ def test_bytes_address(self):
+ # The Unix socket address can also be specified as bytes.
+ if self.server_exception:
+ self.skipTest(self.server_exception)
+ hdlr = logging.handlers.SysLogHandler(os.fsencode(self.address))
+ self.addCleanup(hdlr.close)
+ self.assertTrue(hdlr.unixsocket)
+ logger = logging.getLogger("slh-bytes")
+ logger.addHandler(hdlr)
+ self.addCleanup(logger.removeHandler, hdlr)
+ logger.error("sp\xe4m")
+ self.handled.wait(support.LONG_TIMEOUT)
+ self.assertEqual(self.log_output, b'<11>sp\xc3\xa4m\x00')
+
+@unittest.skipUnless(sys.platform in ('linux', 'android'),
+ 'Linux specific test')
+class AbstractNamespaceSysLogHandlerTest(BaseTest):
+
+ """Test for SysLogHandler with a socket in the abstract namespace."""
+
+ def check(self, address):
+ sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+ self.addCleanup(sock.close)
+ sock.bind(address)
+ sock.settimeout(support.LONG_TIMEOUT)
+ hdlr = logging.handlers.SysLogHandler(address)
+ self.addCleanup(hdlr.close)
+ self.assertTrue(hdlr.unixsocket)
+ hdlr.emit(logging.makeLogRecord({'msg': 'sp\xe4m'}))
+ self.assertEqual(sock.recv(1024), b'<12>sp\xc3\xa4m\x00')
+
+ def test_str_address(self):
+ # A str address is encoded with the filesystem encoding.
+ self.check('\0' + os_helper.TESTFN)
+
+ def test_bytes_address(self):
+ # The name is an arbitrary byte sequence, it need not be decodable.
+ self.check(b'\0test_logging_%d_\xff\xfe' % os.getpid())
+
@unittest.skipUnless(socket_helper.IPV6_ENABLED,
'IPv6 support required for this test.')
class IPv6SysLogHandlerTest(SysLogHandlerTest):