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, bar: 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("%s knows about %s", foo, bar)
 113     out, _ = foo.cmd("dump", "nodes")
 114     check.lines(out, 2)
 115     check.is_in(f"{foo} id ", out)
 116     check.is_in(f"{bar} id ", out)
 117
 118     log.info("%s can only reach itself", foo)
 119     out, _ = foo.cmd("dump", "reachable", "nodes")
 120     check.lines(out, 1)
 121     check.is_in(f"{foo} id ", out)
 122
 123
 124 def run_connected_tests(foo: Tinc, bar: Tinc) -> None:
 125     """Run online tests with connected nodes."""
 126
 127     log.info("dump connected edges")
 128     out, _ = foo.cmd("dump", "edges")
 129     check.lines(out, 2)
 130     check.is_in(f"{foo} to {bar}", out)
 131     check.is_in(f"{bar} to {foo}", out)
 132
 133     log.info("dump connected connections")
 134     out, _ = foo.cmd("dump", "connections")
 135     check.lines(out, 2)
 136     check.is_in("<control> at ", out)
 137     check.is_in(f"{bar} at ", out)
 138
 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)
 144
 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)
 149         try_dot(out)
 150
 151     log.info("dump connected nodes")
 152     for arg in (("nodes",), ("reachable", "nodes")):
 153         out, _ = foo.cmd("dump", *arg)
 154         check.lines(out, 2)
 155         check.is_in(f"{foo} id ", out)
 156         check.is_in(f"{bar} id ", out)
 157
 158
 159 def run_tests(ctx: Test) -> None:
 160     """Run all tests."""
 161
 162     foo = init(ctx)
 163     bar = ctx.node()
 164
 165     log.info("set %s subnets", foo)
 166     for sub in SUBNETS_FOO:
 167         foo.cmd("add", "Subnet", sub)
 168
 169     for command in "dump", "list":
 170         run_offline_tests(command, foo)
 171
 172     log.info("start %s", foo)
 173     foo.start()
 174
 175     log.info("invite %s", bar)
 176     url, _ = foo.cmd("invite", bar.name)
 177     url = url.strip()
 178
 179     dump_pending_invitation(foo, bar)
 180
 181     log.info("join %s and set subnets", bar)
 182     bar.cmd("join", url)
 183     bar.cmd("set", "DeviceType", "dummy")
 184     bar.cmd("set", "Port", "0")
 185     for sub in SUBNETS_BAR:
 186         bar.cmd("add", "Subnet", sub)
 187
 188     run_unconnected_tests(foo, bar)
 189
 190     log.info("start %s", bar)
 191     foo.add_script(bar.script_up)
 192     bar.cmd("start")
 193     foo[bar.script_up].wait()
 194
 195     run_connected_tests(foo, bar)
 196
 197
 198 with Test("run dump tests") as context:
 199     run_tests(context)