www.tinc-vpn.org Git - tinc/blob - 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)