Add tests for dump commands
[tinc] / test / integration / cmd_dump.py
1 #!/usr/bin/env python3
2
3 """Test dump commands."""
4
5 import subprocess as subp
6
7 from testlib import check
8 from testlib.log import log
9 from testlib.proc import Tinc
10 from testlib.test import Test
11
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(
15     (
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)",
20     )
21 )
22 ONLINE_REQUESTS = (
23     ("connections",),
24     ("digraph",),
25     ("edges",),
26     ("foobar",),
27     ("graph",),
28     ("nodes",),
29     ("reachable", "nodes"),
30     ("subnets",),
31 )
32
33
34 def init(ctx: Test) -> Tinc:
35     """Initialize a node."""
36
37     node = ctx.node()
38     stdin = f"""
39         init {node}
40         set Port 0
41         set Address localhost
42         set DeviceType dummy
43     """
44     node.cmd(stdin=stdin)
45     return node
46
47
48 def try_dot(src: str) -> None:
49     """Try passing graph source through the dot binary, if it's present."""
50     try:
51         res = subp.run("dot", input=src, stdout=subp.PIPE, check=True, encoding="utf-8")
52         check.true(res.stdout)
53     except FileNotFoundError:
54         pass
55
56
57 def run_offline_tests(command: str, foo: Tinc) -> None:
58     """Run offline tests."""
59
60     log.info("dump empty invitations")
61     out, err = foo.cmd(command, "invitations")
62     check.false(out)
63     check.is_in("No outstanding invitations", err)
64
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)
69
70
71 def dump_pending_invitation(foo: Tinc, bar: Tinc) -> None:
72     """Test dumping of pending invitations."""
73
74     log.info("dump %s invitation", bar)
75     out, _ = foo.cmd("dump", "invitations")
76     check.lines(out, 1)
77     file, node = out.strip().split(" ")
78     check.true(file)
79     check.equals(node, bar.name)
80
81
82 def run_unconnected_tests(foo: Tinc) -> None:
83     """Run online tests with unconnected nodes."""
84
85     log.info("dump invalid type")
86     _, err = foo.cmd("dump", "foobar42", code=1)
87     check.is_in("Unknown dump type", err)
88
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)
92
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)
96
97     log.info("dump unconnected edges")
98     out, _ = foo.cmd("dump", "edges")
99     check.lines(out, 0)
100
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)
106
107     log.info("dump unconnected connections")
108     out, _ = foo.cmd("dump", "connections")
109     check.lines(out, 1)
110     check.is_in("<control>", out)
111
112     log.info("dump unconnected nodes")
113     for arg in (("nodes",), ("reachable", "nodes")):
114         out, _ = foo.cmd("dump", *arg)
115         check.lines(out, 1)
116         check.is_in(f"{foo} id ", out)
117
118
119 def run_connected_tests(foo: Tinc, bar: Tinc) -> None:
120     """Run online tests with connected nodes."""
121
122     log.info("dump connected edges")
123     out, _ = foo.cmd("dump", "edges")
124     check.lines(out, 2)
125     check.is_in(f"{foo} to {bar}", out)
126     check.is_in(f"{bar} to {foo}", out)
127
128     log.info("dump connected connections")
129     out, _ = foo.cmd("dump", "connections")
130     check.lines(out, 2)
131     check.is_in("<control> at ", out)
132     check.is_in(f"{bar} at ", out)
133
134     log.info("dump connected subnets")
135     out, _ = foo.cmd("dump", "subnets")
136     check.lines(out, SUBNETS_BROADCAST + len(SUBNETS_FOO) + len(SUBNETS_BAR))
137     for sub in (*SUBNETS_FOO, *SUBNETS_BAR):
138         check.is_in(sub, out)
139
140     for kind in "graph", "digraph":
141         log.info("dump %s", kind)
142         out, _ = foo.cmd("dump", kind)
143         check.is_in(f"{kind} {{", out)
144         try_dot(out)
145
146     log.info("dump connected nodes")
147     for arg in (("nodes",), ("reachable", "nodes")):
148         out, _ = foo.cmd("dump", *arg)
149         check.lines(out, 2)
150         check.is_in(f"{foo} id ", out)
151         check.is_in(f"{bar} id ", out)
152
153
154 def run_tests(ctx: Test) -> None:
155     """Run all tests."""
156
157     foo = init(ctx)
158     bar = ctx.node()
159
160     log.info("set %s subnets", foo)
161     for sub in SUBNETS_FOO:
162         foo.cmd("add", "Subnet", sub)
163
164     for command in "dump", "list":
165         run_offline_tests(command, foo)
166
167     log.info("start %s", foo)
168     foo.start()
169
170     log.info("invite %s", bar)
171     url, _ = foo.cmd("invite", bar.name)
172     url = url.strip()
173
174     dump_pending_invitation(foo, bar)
175
176     log.info("join %s and set subnets", bar)
177     bar.cmd("join", url)
178     bar.cmd("set", "DeviceType", "dummy")
179     bar.cmd("set", "Port", "0")
180     for sub in SUBNETS_BAR:
181         bar.cmd("add", "Subnet", sub)
182
183     run_unconnected_tests(foo)
184
185     log.info("start %s", bar)
186     foo.add_script(bar.script_up)
187     bar.cmd("start")
188     foo[bar.script_up].wait()
189
190     run_connected_tests(foo, bar)
191
192
193 with Test("run dump tests") as context:
194     run_tests(context)