From 4a3cee1ad33d35bb14b845a4d5cfcebe98c99260 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zbigniew=20J=C4=99drzejewski-Szmek?= Date: Thu, 15 Dec 2016 20:14:19 -0500 Subject: daemon: add wrapper for sd_is_socket_sockaddr and tests Specifying the address without does not work. --- systemd/test/test_daemon.py | 44 +++++++++++++++++++++++++++++++++++++++----- 1 file changed, 39 insertions(+), 5 deletions(-) (limited to 'systemd/test/test_daemon.py') diff --git a/systemd/test/test_daemon.py b/systemd/test/test_daemon.py index e827e1d..272c46f 100644 --- a/systemd/test/test_daemon.py +++ b/systemd/test/test_daemon.py @@ -9,6 +9,7 @@ from systemd.daemon import (booted, is_socket, _is_socket, is_socket_inet, _is_socket_inet, is_socket_unix, _is_socket_unix, + is_socket_sockaddr, _is_socket_sockaddr, is_mq, _is_mq, listen_fds, notify) @@ -122,15 +123,18 @@ def test_no_mismatch(): assert not is_fifo(sock) assert not is_mq_wrapper(sock) assert not is_socket_inet(sock) + assert not is_socket_sockaddr(sock, '127.0.0.1:2000') fd = sock.fileno() assert not is_fifo(fd) assert not is_mq_wrapper(fd) assert not is_socket_inet(fd) + assert not is_socket_sockaddr(fd, '127.0.0.1:2000') assert not _is_fifo(fd) assert not _is_mq_wrapper(fd) assert not _is_socket_inet(fd) + assert not _is_socket_sockaddr(fd, '127.0.0.1:2000') def test_is_socket(): with closing_socketpair(socket.AF_UNIX) as pair: @@ -141,12 +145,42 @@ def test_is_socket(): assert not is_socket(arg, socket.AF_INET) assert is_socket(arg, socket.AF_UNIX, socket.SOCK_STREAM) assert not is_socket(arg, socket.AF_INET, socket.SOCK_DGRAM) + assert not is_socket_sockaddr(arg, '8.8.8.8:2000', socket.SOCK_DGRAM, 0, 0) - assert is_socket(sock) - assert is_socket(arg, socket.AF_UNIX) - assert not is_socket(arg, socket.AF_INET) - assert is_socket(arg, socket.AF_UNIX, socket.SOCK_STREAM) - assert not is_socket(arg, socket.AF_INET, socket.SOCK_DGRAM) + assert _is_socket(arg) + assert _is_socket(arg, socket.AF_UNIX) + assert not _is_socket(arg, socket.AF_INET) + assert _is_socket(arg, socket.AF_UNIX, socket.SOCK_STREAM) + assert not _is_socket(arg, socket.AF_INET, socket.SOCK_DGRAM) + assert not _is_socket_sockaddr(arg, '8.8.8.8:2000', socket.SOCK_DGRAM, 0, 0) + +def test_is_socket_sockaddr(): + with contextlib.closing(socket.socket(socket.AF_INET)) as sock: + sock.bind(('127.0.0.1', 0)) + addr, port = sock.getsockname() + + for listening in (0, 1): + for arg in (sock, sock.fileno()): + # assert is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_STREAM) + assert is_socket_sockaddr(arg, '127.0.0.1:{}'.format(port), socket.SOCK_STREAM) + + assert is_socket_sockaddr(arg, '127.0.0.1:{}'.format(port), listening=listening) + assert is_socket_sockaddr(arg, '127.0.0.1:{}'.format(port), listening=-1) + assert not is_socket_sockaddr(arg, '127.0.0.1:{}'.format(port), listening=not listening) + + with pytest.raises(ValueError): + is_socket_sockaddr(arg, '127.0.0.1', flowinfo=123456) + + assert not is_socket_sockaddr(arg, '129.168.11.11:23'.format(port), socket.SOCK_STREAM) + #assert not is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_DGRAM) + + with pytest.raises(ValueError): + _is_socket_sockaddr(arg, '127.0.0.1', 0, 123456) + + assert not _is_socket_sockaddr(arg, '129.168.11.11:23'.format(port), socket.SOCK_STREAM) + #assert not _is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_DGRAM) + + sock.listen(11) def test__is_socket(): with closing_socketpair(socket.AF_UNIX) as pair: -- cgit v1.2.1 From bc691d8e293a593fbd14ad1d592d06f4f490ed29 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zbigniew=20J=C4=99drzejewski-Szmek?= Date: Thu, 15 Dec 2016 20:47:10 -0500 Subject: Fix handling of addresses without port and add tests --- systemd/test/test_daemon.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) (limited to 'systemd/test/test_daemon.py') diff --git a/systemd/test/test_daemon.py b/systemd/test/test_daemon.py index 272c46f..5e9f5b8 100644 --- a/systemd/test/test_daemon.py +++ b/systemd/test/test_daemon.py @@ -158,27 +158,28 @@ def test_is_socket_sockaddr(): with contextlib.closing(socket.socket(socket.AF_INET)) as sock: sock.bind(('127.0.0.1', 0)) addr, port = sock.getsockname() + port = ':{}'.format(port) for listening in (0, 1): for arg in (sock, sock.fileno()): - # assert is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_STREAM) - assert is_socket_sockaddr(arg, '127.0.0.1:{}'.format(port), socket.SOCK_STREAM) + assert is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_STREAM) + assert is_socket_sockaddr(arg, '127.0.0.1' + port, socket.SOCK_STREAM) - assert is_socket_sockaddr(arg, '127.0.0.1:{}'.format(port), listening=listening) - assert is_socket_sockaddr(arg, '127.0.0.1:{}'.format(port), listening=-1) - assert not is_socket_sockaddr(arg, '127.0.0.1:{}'.format(port), listening=not listening) + assert is_socket_sockaddr(arg, '127.0.0.1' + port, listening=listening) + assert is_socket_sockaddr(arg, '127.0.0.1' + port, listening=-1) + assert not is_socket_sockaddr(arg, '127.0.0.1' + port, listening=not listening) with pytest.raises(ValueError): is_socket_sockaddr(arg, '127.0.0.1', flowinfo=123456) - assert not is_socket_sockaddr(arg, '129.168.11.11:23'.format(port), socket.SOCK_STREAM) - #assert not is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_DGRAM) + assert not is_socket_sockaddr(arg, '129.168.11.11:23', socket.SOCK_STREAM) + assert not is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_DGRAM) with pytest.raises(ValueError): _is_socket_sockaddr(arg, '127.0.0.1', 0, 123456) - assert not _is_socket_sockaddr(arg, '129.168.11.11:23'.format(port), socket.SOCK_STREAM) - #assert not _is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_DGRAM) + assert not _is_socket_sockaddr(arg, '129.168.11.11:23', socket.SOCK_STREAM) + assert not _is_socket_sockaddr(arg, '127.0.0.1', socket.SOCK_DGRAM) sock.listen(11) -- cgit v1.2.1