tinc-vpn.org Git - tinc/blob - test/integration/address_cache.py

   1 #!/usr/bin/env python3
   2
   3 """Test recent address cache."""
   4
   5 import os
   6 import typing as T
   7 import shutil
   8
   9 from testlib import check
  10 from testlib.log import log
  11 from testlib.proc import Tinc, Script
  12 from testlib.test import Test
  13
  14
  15 def init(ctx: Test) -> T.Tuple[Tinc, Tinc]:
  16     """Create test node."""
  17     bar = ctx.node()
  18     foo = ctx.node(init="set AutoConnect no")
  19     return foo, bar
  20
  21
  22 def connect_nodes(foo: Tinc, bar: Tinc) -> None:
  23     """Start second node and wait for connection."""
  24     log.info("connect nodes")
  25     bar.cmd("start")
  26     bar[foo.script_up].wait()
  27     foo[bar.script_up].wait()
  28
  29
  30 def run_tests(ctx: Test) -> None:
  31     """Run tests."""
  32     foo, bar = init(ctx)
  33
  34     log.info("cache directory must exist after init")
  35     check.dir_exists(foo.sub("cache"))
  36
  37     foo.add_script(Script.TINC_UP)
  38     foo.add_script(Script.INVITATION_ACCEPTED)
  39     foo.start()
  40
  41     log.info("invite %s to %s", bar, foo)
  42     invite, _ = foo.cmd("invite", bar.name)
  43     invite = invite.strip()
  44
  45     log.info("join %s to %s", bar, foo)
  46     bar.cmd("join", invite)
  47
  48     log.info("cache directory must exist after join")
  49     check.dir_exists(bar.sub("cache"))
  50
  51     log.info("invitee address must be cached after invitation is accepted")
  52     foo[Script.INVITATION_ACCEPTED].wait()
  53     check.file_exists(foo.sub(f"cache/{bar}"))
  54     os.remove(foo.sub(f"cache/{bar}"))
  55
  56     log.info("configure %s", bar)
  57     bar.cmd("set", "DeviceType", "dummy")
  58     bar.cmd("set", "Port", "0")
  59
  60     log.info("add host-up scripts")
  61     foo.add_script(bar.script_up)
  62     bar.add_script(foo.script_up)
  63
  64     connect_nodes(foo, bar)
  65
  66     log.info("%s must cache %s's public address", bar, foo)
  67     check.file_exists(bar.sub(f"cache/{foo}"))
  68
  69     log.info("%s must not cache %s's outgoing address", foo, bar)
  70     assert not os.path.exists(foo.sub(f"cache/{bar}"))
  71
  72     log.info("stop node %s", bar)
  73     bar.cmd("stop")
  74
  75     log.info("remove %s cache directory", bar)
  76     shutil.rmtree(bar.sub("cache"))
  77
  78     connect_nodes(foo, bar)
  79
  80     log.info("make sure %s cache was recreated", bar)
  81     check.file_exists(bar.sub(f"cache/{foo}"))
  82
  83     log.info("stop nodes")
  84     bar.cmd("stop")
  85     foo.cmd("stop")
  86
  87     log.info("remove Address from all nodes")
  88     for node in foo, bar:
  89         node.cmd("del", "Address", code=None)
  90         for peer in foo, bar:
  91             node.cmd("del", f"{peer}.Address", code=None)
  92     bar.cmd("add", "ConnectTo", foo.name)
  93
  94     log.info("make sure connection works using just the cached address")
  95     foo.cmd("start")
  96     connect_nodes(foo, bar)
  97
  98
  99 with Test("run address cache tests") as context:
 100     run_tests(context)