daemon: add wrapper for sd_is_socket_sockaddr and tests

Specifying the address without does not work.

2 files changed, 54 insertions, 5 deletions

diff --git a/systemd/daemon.py b/systemd/daemon.py
index 82011ca..5d398f2 100644
--- a/systemd/daemon.py
+++ b/systemd/daemon.py
@@ -5,6 +5,7 @@ from ._daemon import (__version__,
                       _is_fifo,
                       _is_socket,
                       _is_socket_inet,
+                      _is_socket_sockaddr,
                       _is_socket_unix,
                       _is_mq,
                       LISTEN_FDS_START)
@@ -28,6 +29,20 @@ def is_socket_inet(fileobj, family=_AF_UNSPEC, type=0, listening=-1, port=0):
     fd = _convert_fileobj(fileobj)
     return _is_socket_inet(fd, family, type, listening, port)
 
+def is_socket_sockaddr(fileobj, address, type=0, flowinfo=0, listening=-1):
+    """Check socket type, address and/or port, flowinfo, listening state.
+
+    Wraps sd_is_socket_inet_sockaddr(3).
+
+    `address` is a systemd-style numerical IPv4 or IPv6 address as used in
+    ListenStream=. A port may be included after a colon (":").
+    See systemd.socket(5) for details.
+
+    Constants for `family` are defined in the socket module.
+    """
+    fd = _convert_fileobj(fileobj)
+    return _is_socket_sockaddr(fd, address, type, flowinfo, listening)
+
 def is_socket_unix(fileobj, type=0, listening=-1, path=None):
     fd = _convert_fileobj(fileobj)
     return _is_socket_unix(fd, type, listening, path)
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: