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 exchange(node0: Tinc, node1: Tinc, export_all: bool = False) -> ExchangeIO:
  17     """Run `export(-all) | exchange | import` between the passed nodes.
  18     `export-all` is used if export_all is set to True.
  19     """
  20     export_cmd = "export-all" if export_all else "export"
  21     log.debug("%s between %s and %s", export_cmd, node0.name, node1.name)
  22
  23     exp_out, exp_err = node0.cmd(export_cmd)
  24     log.debug(
  25         'exchange: %s %s returned ("%s", "%s")', export_cmd, node0, exp_out, exp_err
  26     )
  27     check.is_in("Name =", exp_out)
  28
  29     xch_out, xch_err = node1.cmd("exchange", stdin=exp_out)
  30     log.debug('exchange: exchange %s returned ("%s", "%s")', node1, xch_out, xch_err)
  31     check.is_in("Name =", xch_out)
  32     check.is_in("Imported ", xch_err)
  33
  34     imp_out, imp_err = node0.cmd("import", stdin=xch_out)
  35     log.debug('exchange: import %s returned ("%s", "%s")', node0, imp_out, imp_err)
  36     check.is_in("Imported ", imp_err)
  37
  38     return (
  39         (exp_out, exp_err),
  40         (xch_out, xch_err),
  41         (imp_out, imp_err),
  42     )
  43
  44
  45 def get(tinc: Tinc, var: str) -> str:
  46     """Get the value of the variable, stripped of whitespace."""
  47     assert var
  48     stdout, _ = tinc.cmd("get", var)
  49     return stdout.strip()