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 try_dot(src: str) -> None:
  35     """Try passing graph source through the dot binary, if it's present."""
  36     try:
  37         res = subp.run("dot", input=src, stdout=subp.PIPE, check=True, encoding="utf-8")
  38         check.true(res.stdout)
  39     except FileNotFoundError:
  40         pass
  41
  42
  43 def run_offline_tests(command: str, foo: Tinc) -> None:
  44     """Run offline tests."""
  45
  46     log.info("dump empty invitations")
  47     out, err = foo.cmd(command, "invitations")
  48     check.false(out)
  49     check.is_in("No outstanding invitations", err)
  50
  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)
  55
  56
  57 def dump_pending_invitation(foo: Tinc, bar: Tinc) -> None:
  58     """Test dumping of pending invitations."""
  59
  60     log.info("dump %s invitation", bar)
  61     out, _ = foo.cmd("dump", "invitations")
  62     check.lines(out, 1)
  63     file, node = out.strip().split(" ")
  64     check.true(file)
  65     check.equals(node, bar.name)
  66
  67
  68 def run_unconnected_tests(foo: Tinc, bar: Tinc) -> None:
  69     """Run online tests with unconnected nodes."""
  70
  71     log.info("dump invalid type")
  72     _, err = foo.cmd("dump", "foobar42", code=1)
  73     check.is_in("Unknown dump type", err)
  74
  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)
  78
  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)
  82
  83     log.info("dump unconnected edges")
  84     out, _ = foo.cmd("dump", "edges")
  85     check.lines(out, 0)
  86
  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:
  91         check.is_in(sub, out)
  92
  93     log.info("dump unconnected connections")
  94     out, _ = foo.cmd("dump", "connections")
  95     check.lines(out, 1)
  96     check.is_in("<control>", out)
  97
  98     log.info("%s knows about %s", foo, bar)
  99     out, _ = foo.cmd("dump", "nodes")
 100     check.lines(out, 2)
 101     check.is_in(f"{foo} id ", out)
 102     check.is_in(f"{bar} id ", out)
 103
 104     log.info("%s can only reach itself", foo)
 105     out, _ = foo.cmd("dump", "reachable", "nodes")
 106     check.lines(out, 1)
 107     check.is_in(f"{foo} id ", out)
 108
 109
 110 def run_connected_tests(foo: Tinc, bar: Tinc) -> None:
 111     """Run online tests with connected nodes."""
 112
 113     log.info("dump connected edges")
 114     out, _ = foo.cmd("dump", "edges")
 115     check.lines(out, 2)
 116     check.is_in(f"{foo} to {bar}", out)
 117     check.is_in(f"{bar} to {foo}", out)
 118
 119     log.info("dump connected connections")
 120     out, _ = foo.cmd("dump", "connections")
 121     check.lines(out, 2)
 122     check.is_in("<control> at ", out)
 123     check.is_in(f"{bar} at ", out)
 124
 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)
 130
 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)
 135         try_dot(out)
 136
 137     log.info("dump connected nodes")
 138     for arg in (("nodes",), ("reachable", "nodes")):
 139         out, _ = foo.cmd("dump", *arg)
 140         check.lines(out, 2)
 141         check.is_in(f"{foo} id ", out)
 142         check.is_in(f"{bar} id ", out)
 143
 144
 145 def run_tests(ctx: Test) -> None:
 146     """Run all tests."""
 147
 148     foo, bar = ctx.node(init=True), ctx.node()
 149
 150     log.info("set %s subnets", foo)
 151     for sub in SUBNETS_FOO:
 152         foo.cmd("add", "Subnet", sub)
 153
 154     for command in "dump", "list":
 155         run_offline_tests(command, foo)
 156
 157     log.info("start %s", foo)
 158     foo.start()
 159
 160     log.info("invite %s", bar)
 161     url, _ = foo.cmd("invite", bar.name)
 162     url = url.strip()
 163
 164     dump_pending_invitation(foo, bar)
 165
 166     log.info("join %s and set subnets", bar)
 167     bar.cmd("join", url)
 168     bar.cmd("set", "DeviceType", "dummy")
 169     bar.cmd("set", "Port", "0")
 170     for sub in SUBNETS_BAR:
 171         bar.cmd("add", "Subnet", sub)
 172
 173     run_unconnected_tests(foo, bar)
 174
 175     log.info("start %s", bar)
 176     foo.add_script(bar.script_up)
 177     bar.cmd("start")
 178     foo[bar.script_up].wait()
 179
 180     run_connected_tests(foo, bar)
 181
 182
 183 with Test("run dump tests") as context:
 184     run_tests(context)