3 """Test dump commands."""
5 import subprocess as subp
7 from testlib import check
8 from testlib.log import log
9 from testlib.proc import Tinc
10 from testlib.test import Test
12 SUBNETS_FOO = ("10.0.0.0/16", "10.1.2.0/24")
13 SUBNETS_BAR = ("10.3.2.0/27", "fe80::/64")
14 SUBNETS_BROADCAST = len(
16 "ff:ff:ff:ff:ff:ff owner (broadcast)",
17 "255.255.255.255 owner (broadcast)",
18 "224.0.0.0/4 owner (broadcast)",
19 "ff00::/8 owner (broadcast)",
29 ("reachable", "nodes"),
34 def try_dot(src: str) -> None:
35 """Try passing graph source through the dot binary, if it's present."""
37 res = subp.run("dot", input=src, stdout=subp.PIPE, check=True, encoding="utf-8")
38 check.true(res.stdout)
39 except FileNotFoundError:
43 def run_offline_tests(command: str, foo: Tinc) -> None:
44 """Run offline tests."""
46 log.info("dump empty invitations")
47 out, err = foo.cmd(command, "invitations")
49 check.is_in("No outstanding invitations", err)
51 for request in ONLINE_REQUESTS:
52 log.info("dump online type %s", request)
53 _, err = foo.cmd(command, *request, code=1)
54 check.is_in("Could not open pid file", err)
57 def dump_pending_invitation(foo: Tinc, bar: Tinc) -> None:
58 """Test dumping of pending invitations."""
60 log.info("dump %s invitation", bar)
61 out, _ = foo.cmd("dump", "invitations")
63 file, node = out.strip().split(" ")
65 check.equals(node, bar.name)
68 def run_unconnected_tests(foo: Tinc, bar: Tinc) -> None:
69 """Run online tests with unconnected nodes."""
71 log.info("dump invalid type")
72 _, err = foo.cmd("dump", "foobar42", code=1)
73 check.is_in("Unknown dump type", err)
75 log.info("use 'reachable' with wrong command")
76 _, err = foo.cmd("dump", "reachable", "edges", code=1)
77 check.is_in("reachable' only supported for nodes", err)
79 log.info("check for too many arguments")
80 _, err = foo.cmd("dump", "edges", "please", code=1)
81 check.is_in("Invalid number of arguments", err)
83 log.info("dump unconnected edges")
84 out, _ = foo.cmd("dump", "edges")
87 log.info("dump unconnected subnets")
88 out, _ = foo.cmd("dump", "subnets")
89 check.lines(out, SUBNETS_BROADCAST + len(SUBNETS_FOO))
90 for sub in SUBNETS_FOO:
93 log.info("dump unconnected connections")
94 out, _ = foo.cmd("dump", "connections")
96 check.is_in("<control>", out)
98 log.info("%s knows about %s", foo, bar)
99 out, _ = foo.cmd("dump", "nodes")
101 check.is_in(f"{foo} id ", out)
102 check.is_in(f"{bar} id ", out)
104 log.info("%s can only reach itself", foo)
105 out, _ = foo.cmd("dump", "reachable", "nodes")
107 check.is_in(f"{foo} id ", out)
110 def run_connected_tests(foo: Tinc, bar: Tinc) -> None:
111 """Run online tests with connected nodes."""
113 log.info("dump connected edges")
114 out, _ = foo.cmd("dump", "edges")
116 check.is_in(f"{foo} to {bar}", out)
117 check.is_in(f"{bar} to {foo}", out)
119 log.info("dump connected connections")
120 out, _ = foo.cmd("dump", "connections")
122 check.is_in("<control> at ", out)
123 check.is_in(f"{bar} at ", out)
125 log.info("dump connected subnets")
126 out, _ = foo.cmd("dump", "subnets")
127 check.lines(out, SUBNETS_BROADCAST + len(SUBNETS_FOO) + len(SUBNETS_BAR))
128 for sub in (*SUBNETS_FOO, *SUBNETS_BAR):
129 check.is_in(sub, out)
131 for kind in "graph", "digraph":
132 log.info("dump %s", kind)
133 out, _ = foo.cmd("dump", kind)
134 check.is_in(f"{kind} {{", out)
137 log.info("dump connected nodes")
138 for arg in (("nodes",), ("reachable", "nodes")):
139 out, _ = foo.cmd("dump", *arg)
141 check.is_in(f"{foo} id ", out)
142 check.is_in(f"{bar} id ", out)
145 def run_tests(ctx: Test) -> None:
148 foo, bar = ctx.node(init=True), ctx.node()
150 log.info("set %s subnets", foo)
151 for sub in SUBNETS_FOO:
152 foo.cmd("add", "Subnet", sub)
154 for command in "dump", "list":
155 run_offline_tests(command, foo)
157 log.info("start %s", foo)
160 log.info("invite %s", bar)
161 url, _ = foo.cmd("invite", bar.name)
164 dump_pending_invitation(foo, bar)
166 log.info("join %s and set subnets", bar)
168 bar.cmd("set", "DeviceType", "dummy")
169 bar.cmd("set", "Port", "0")
170 for sub in SUBNETS_BAR:
171 bar.cmd("add", "Subnet", sub)
173 run_unconnected_tests(foo, bar)
175 log.info("start %s", bar)
176 foo.add_script(bar.script_up)
178 foo[bar.script_up].wait()
180 run_connected_tests(foo, bar)
183 with Test("run dump tests") as context: