www.tinc-vpn.org Git - tinc/blob - test/integration/testlib/cmd.py

   1 """Wrappers for more complicated tinc/tincd commands."""
   2
   3 import typing as T
   4
   5 from . import check
   6 from .log import log
   7 from .proc import Tinc
   8
   9 ExchangeIO = T.Tuple[
  10     T.Tuple[str, str],
  11     T.Tuple[str, str],
  12     T.Tuple[str, str],
  13 ]
  14
  15
  16 def connect(node0: Tinc, node1: Tinc) -> ExchangeIO:
  17     """Exchange configuration between nodes and start
  18     them in such an order that `Port 0` works on both sides.
  19     """
  20     node0.add_script(node1.script_up)
  21     node0.start()
  22     result = exchange(node0, node1)
  23     node1.add_script(node0.script_up)
  24     node1.cmd("add", "ConnectTo", node0.name)
  25     node1.start()
  26     node0[node1.script_up].wait()
  27     node1[node0.script_up].wait()
  28     return result
  29
  30
  31 def exchange(node0: Tinc, node1: Tinc, export_all: bool = False) -> ExchangeIO:
  32     """Run `export(-all) | exchange | import` between the passed nodes.
  33     `export-all` is used if export_all is set to True.
  34     """
  35     export_cmd = "export-all" if export_all else "export"
  36     log.debug("%s between %s and %s", export_cmd, node0.name, node1.name)
  37
  38     exp_out, exp_err = node0.cmd(export_cmd)
  39     log.debug(
  40         'exchange: %s %s returned ("%s", "%s")', export_cmd, node0, exp_out, exp_err
  41     )
  42     check.is_in("Name =", exp_out)
  43
  44     xch_out, xch_err = node1.cmd("exchange", stdin=exp_out)
  45     log.debug('exchange: exchange %s returned ("%s", "%s")', node1, xch_out, xch_err)
  46     check.is_in("Name =", xch_out)
  47     check.is_in("Imported ", xch_err)
  48
  49     imp_out, imp_err = node0.cmd("import", stdin=xch_out)
  50     log.debug('exchange: import %s returned ("%s", "%s")', node0, imp_out, imp_err)
  51     check.is_in("Imported ", imp_err)
  52
  53     return (
  54         (exp_out, exp_err),
  55         (xch_out, xch_err),
  56         (imp_out, imp_err),
  57     )
  58
  59
  60 def get(tinc: Tinc, var: str) -> str:
  61     """Get the value of the variable, stripped of whitespace."""
  62     assert var
  63     stdout, _ = tinc.cmd("get", var)
  64     return stdout.strip()