3 """Test multicast device."""
10 from testlib import check
11 from testlib.log import log
12 from testlib.proc import Tinc, Script
13 from testlib.test import Test
15 MCAST_ADDR = "224.15.98.12"
19 def multicast_works() -> bool:
20 """Check if multicast is supported and works."""
25 with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server:
26 server.bind((MCAST_ADDR, PORT))
28 req = struct.pack("=4sl", socket.inet_aton(MCAST_ADDR), socket.INADDR_ANY)
29 server.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, req)
31 with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
32 client.sendto(msg, (MCAST_ADDR, PORT))
34 return msg == server.recv(16)
39 def test_no_mcast_support(foo: Tinc) -> None:
40 """Check that startup fails on systems without multicast support."""
42 code = foo.tincd("-D").wait()
44 check.in_file(foo.sub("log"), f"Can't bind to {MCAST_ADDR}")
47 def test_rx_tx(foo: Tinc) -> None:
48 """Test sending real data to a multicast device."""
51 packet = os.urandom(137)
53 with socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) as sock:
55 sent = sock.sendto(packet, (MCAST_ADDR, PORT))
56 log.info("sent broken packet (%d)", sent)
59 foo.add_script(Script.TINC_DOWN)
61 foo[Script.TINC_DOWN].wait()
63 check.in_file(foo.sub("log"), "Read packet of 137 bytes from multicast socket")
66 def test_device_multicast(ctx: Test) -> None:
67 """Test multicast device."""
69 foo = ctx.node(init=True)
70 foo.cmd("set", "DeviceType", "multicast")
72 log.info("check that multicast does not work without Device")
73 _, err = foo.cmd("start", "-D", code=1)
74 check.is_in("Device variable required for multicast socket", err)
76 log.info("check that Device requires a port")
77 foo.cmd("set", "Device", "localhost")
78 _, err = foo.cmd("start", "-D", code=1)
79 check.is_in("Port number required", err)
81 log.info("check that multicast receives data")
82 foo.cmd("set", "Device", f"{MCAST_ADDR} {PORT}")
83 foo.cmd("set", "LogLevel", "10")
88 test_no_mcast_support(foo)
91 with Test("test DeviceType = multicast") as context:
92 test_device_multicast(context)