Rewrite integration test suite in Python
[tinc] / test / integration / ns_ping.py
1 #!/usr/bin/env python3
2
3 """Create two network namespaces and run ping between them."""
4
5 import subprocess as subp
6 import typing as T
7
8 from testlib import external as ext, util, template, cmd
9 from testlib.log import log
10 from testlib.proc import Tinc, Script
11 from testlib.test import Test
12
13 util.require_root()
14 util.require_command("ip", "netns", "list")
15 util.require_path("/dev/net/tun")
16
17 IP_FOO = "192.168.1.1"
18 IP_BAR = "192.168.1.2"
19 MASK = 24
20
21
22 def init(ctx: Test) -> T.Tuple[Tinc, Tinc]:
23     """Initialize new test nodes."""
24     foo, bar = ctx.node(), ctx.node()
25
26     log.info("create network namespaces")
27     assert ext.netns_add(foo.name)
28     assert ext.netns_add(bar.name)
29
30     log.info("initialize two nodes")
31
32     stdin = f"""
33         init {foo}
34         set Port 0
35         set Subnet {IP_FOO}
36         set Interface {foo}
37         set Address localhost
38         set AutoConnect no
39     """
40     foo.cmd(stdin=stdin)
41     foo.add_script(Script.TINC_UP, template.make_netns_config(foo.name, IP_FOO, MASK))
42     foo.start()
43
44     stdin = f"""
45         init {bar}
46         set Port 0
47         set Subnet {IP_BAR}
48         set Interface {bar}
49         set Address localhost
50         set AutoConnect no
51     """
52     bar.cmd(stdin=stdin)
53     bar.add_script(Script.TINC_UP, template.make_netns_config(bar.name, IP_BAR, MASK))
54
55     cmd.exchange(foo, bar)
56
57     return foo, bar
58
59
60 def ping(namespace: str, ip_addr: str) -> int:
61     """Send pings between two network namespaces."""
62     log.info("pinging node from netns %s at %s", namespace, ip_addr)
63     proc = subp.run(
64         ["ip", "netns", "exec", namespace, "ping", "-W1", "-c1", ip_addr], check=False
65     )
66
67     log.info("ping finished with code %d", proc.returncode)
68     return proc.returncode
69
70
71 with Test("ns-ping") as context:
72     foo_node, bar_node = init(context)
73     bar_node.cmd("start")
74
75     log.info("waiting for nodes to come up")
76     bar_node[Script.TINC_UP].wait()
77
78     log.info("ping must not work when there is no connection")
79     assert ping(foo_node.name, IP_BAR)
80
81     log.info("add script foo/host-up")
82     bar_node.add_script(foo_node.script_up)
83
84     log.info("add ConnectTo clause")
85     bar_node.cmd("add", "ConnectTo", foo_node.name)
86
87     log.info("bar waits for foo")
88     bar_node[foo_node.script_up].wait()
89
90     log.info("ping must work after connection is up")
91     assert not ping(foo_node.name, IP_BAR)