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