Add tests for some device & address variables

[tinc] / test / integration / bind_address.py

diff --git a/test/integration/bind_address.py b/test/integration/bind_address.py
new file mode 100755 (executable)
index 0000000..8db9998
--- /dev/null
+++ b/test/integration/bind_address.py
@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+
+"""Test binding to interfaces and addresses."""
+
+import json
+import socket
+import subprocess as subp
+import sys
+import typing as T
+
+from testlib import check, util
+from testlib.const import EXIT_SKIP
+from testlib.log import log
+from testlib.test import Test
+
+util.require_command("ss", "-nlup")
+util.require_command("ip", "--json", "addr")
+
+
+def connect_tcp(address: str, port: int) -> None:
+    """Check that a TCP connection to (address, port) works."""
+
+    family = socket.AF_INET if "." in address else socket.AF_INET6
+
+    with socket.socket(family, socket.SOCK_STREAM) as sock:
+        sock.connect((address, port))
+
+
+def get_interfaces() -> T.List[T.Tuple[str, T.List[str]]]:
+    """Get a list of network interfaces with assigned addresses."""
+
+    output = subp.run(
+        ["ip", "--json", "addr"], check=True, encoding="utf-8", stdout=subp.PIPE
+    ).stdout
+
+    result: T.List[T.Tuple[str, T.List[str]]] = []
+
+    for line in json.loads(output):
+        if not "UP" in line["flags"]:
+            continue
+        local: T.List[str] = []
+        for addr in line["addr_info"]:
+            if addr["family"] in ("inet", "inet6"):
+                local.append(addr["local"])
+        if local:
+            result.append((line["ifname"], local))
+
+    return result
+
+
+INTERFACES = get_interfaces()
+
+
+def get_udp_listen(pid: int) -> T.List[str]:
+    """Get a list of the currently listening UDP sockets."""
+
+    listen = subp.run(["ss", "-nlup"], check=True, stdout=subp.PIPE, encoding="utf-8")
+    addresses: T.List[str] = []
+
+    for line in listen.stdout.splitlines():
+        if f"pid={pid}," in line:
+            _, _, _, addr, _ = line.split(maxsplit=4)
+            addresses.append(addr)
+
+    return addresses
+
+
+def test_bind_interface(ctx: Test) -> None:
+    """Test BindToInterface."""
+
+    devname, addresses = INTERFACES[0]
+    log.info("using interface %s, addresses (%s)", devname, addresses)
+
+    init = f"""
+        set BindToInterface {devname}
+        set LogLevel 5
+    """
+    foo = ctx.node(init=init)
+    foo.start()
+
+    log.info("check that tincd opened UDP sockets")
+    listen = get_udp_listen(foo.pid)
+    check.is_in(f"%{devname}:{foo.port}", *listen)
+
+    log.info("check TCP sockets")
+    for addr in addresses:
+        connect_tcp(addr, foo.port)
+
+
+def test_bind_address(ctx: Test, kind: str) -> None:
+    """Test BindToAddress or ListenAddress."""
+
+    _, addresses = INTERFACES[0]
+
+    log.info("create and start tincd node")
+    foo = ctx.node(init="set LogLevel 10")
+    for addr in addresses:
+        foo.cmd("add", kind, addr)
+    foo.start()
+
+    log.info("check for correct log message")
+    for addr in addresses:
+        check.in_file(foo.sub("log"), f"Listening on {addr}")
+
+    log.info("test TCP connections")
+    for addr in addresses:
+        connect_tcp(addr, foo.port)
+
+    log.info("check that tincd opened UDP sockets")
+    listen = get_udp_listen(foo.pid)
+    for addr in addresses:
+        check.is_in(addr, *listen)
+        check.is_in(f":{foo.port}", *listen)
+    check.equals(len(addresses), len(listen))
+
+
+if not INTERFACES:
+    log.info("interface list is empty, skipping test")
+    sys.exit(EXIT_SKIP)
+
+with Test("test ListenAddress") as context:
+    test_bind_address(context, "ListenAddress")
+
+with Test("test BindToAddress") as context:
+    test_bind_address(context, "BindToAddress")
+
+with Test("test BindToInterface") as context:
+    test_bind_interface(context)