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 init(ctx: Test) -> Tinc:
35 """Initialize a node."""
48 def try_dot(src: str) -> None:
49 """Try passing graph source through the dot binary, if it's present."""
51 res = subp.run("dot", input=src, stdout=subp.PIPE, check=True, encoding="utf-8")
52 check.true(res.stdout)
53 except FileNotFoundError:
57 def run_offline_tests(command: str, foo: Tinc) -> None:
58 """Run offline tests."""
60 log.info("dump empty invitations")
61 out, err = foo.cmd(command, "invitations")
63 check.is_in("No outstanding invitations", err)
65 for request in ONLINE_REQUESTS:
66 log.info("dump online type %s", request)
67 _, err = foo.cmd(command, *request, code=1)
68 check.is_in("Could not open pid file", err)
71 def dump_pending_invitation(foo: Tinc, bar: Tinc) -> None:
72 """Test dumping of pending invitations."""
74 log.info("dump %s invitation", bar)
75 out, _ = foo.cmd("dump", "invitations")
77 file, node = out.strip().split(" ")
79 check.equals(node, bar.name)
82 def run_unconnected_tests(foo: Tinc, bar: Tinc) -> None:
83 """Run online tests with unconnected nodes."""
85 log.info("dump invalid type")
86 _, err = foo.cmd("dump", "foobar42", code=1)
87 check.is_in("Unknown dump type", err)
89 log.info("use 'reachable' with wrong command")
90 _, err = foo.cmd("dump", "reachable", "edges", code=1)
91 check.is_in("reachable' only supported for nodes", err)
93 log.info("check for too many arguments")
94 _, err = foo.cmd("dump", "edges", "please", code=1)
95 check.is_in("Invalid number of arguments", err)
97 log.info("dump unconnected edges")
98 out, _ = foo.cmd("dump", "edges")
101 log.info("dump unconnected subnets")
102 out, _ = foo.cmd("dump", "subnets")
103 check.lines(out, SUBNETS_BROADCAST + len(SUBNETS_FOO))
104 for sub in SUBNETS_FOO:
105 check.is_in(sub, out)
107 log.info("dump unconnected connections")
108 out, _ = foo.cmd("dump", "connections")
110 check.is_in("<control>", out)
112 log.info("%s knows about %s", foo, bar)
113 out, _ = foo.cmd("dump", "nodes")
115 check.is_in(f"{foo} id ", out)
116 check.is_in(f"{bar} id ", out)
118 log.info("%s can only reach itself", foo)
119 out, _ = foo.cmd("dump", "reachable", "nodes")
121 check.is_in(f"{foo} id ", out)
124 def run_connected_tests(foo: Tinc, bar: Tinc) -> None:
125 """Run online tests with connected nodes."""
127 log.info("dump connected edges")
128 out, _ = foo.cmd("dump", "edges")
130 check.is_in(f"{foo} to {bar}", out)
131 check.is_in(f"{bar} to {foo}", out)
133 log.info("dump connected connections")
134 out, _ = foo.cmd("dump", "connections")
136 check.is_in("<control> at ", out)
137 check.is_in(f"{bar} at ", out)
139 log.info("dump connected subnets")
140 out, _ = foo.cmd("dump", "subnets")
141 check.lines(out, SUBNETS_BROADCAST + len(SUBNETS_FOO) + len(SUBNETS_BAR))
142 for sub in (*SUBNETS_FOO, *SUBNETS_BAR):
143 check.is_in(sub, out)
145 for kind in "graph", "digraph":
146 log.info("dump %s", kind)
147 out, _ = foo.cmd("dump", kind)
148 check.is_in(f"{kind} {{", out)
151 log.info("dump connected nodes")
152 for arg in (("nodes",), ("reachable", "nodes")):
153 out, _ = foo.cmd("dump", *arg)
155 check.is_in(f"{foo} id ", out)
156 check.is_in(f"{bar} id ", out)
159 def run_tests(ctx: Test) -> None:
165 log.info("set %s subnets", foo)
166 for sub in SUBNETS_FOO:
167 foo.cmd("add", "Subnet", sub)
169 for command in "dump", "list":
170 run_offline_tests(command, foo)
172 log.info("start %s", foo)
175 log.info("invite %s", bar)
176 url, _ = foo.cmd("invite", bar.name)
179 dump_pending_invitation(foo, bar)
181 log.info("join %s and set subnets", bar)
183 bar.cmd("set", "DeviceType", "dummy")
184 bar.cmd("set", "Port", "0")
185 for sub in SUBNETS_BAR:
186 bar.cmd("add", "Subnet", sub)
188 run_unconnected_tests(foo, bar)
190 log.info("start %s", bar)
191 foo.add_script(bar.script_up)
193 foo[bar.script_up].wait()
195 run_connected_tests(foo, bar)
198 with Test("run dump tests") as context: