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

   1 #!/usr/bin/env python3
   2
   3 """Test basic SPTPS features."""
   4
   5 import os
   6 import subprocess as subp
   7 import re
   8
   9 from testlib import path, util, check
  10 from testlib.log import log
  11
  12 port_re = re.compile(r"Listening on (\d+)\.\.\.")
  13
  14
  15 class Keypair:
  16     """Create public/private keypair using sptps_keypair."""
  17
  18     private: str
  19     public: str
  20
  21     def __init__(self, name: str) -> None:
  22         self.private = os.path.join(path.TEST_WD, f"{name}.priv")
  23         self.public = os.path.join(path.TEST_WD, f"{name}.pub")
  24         subp.run([path.SPTPS_KEYPAIR_PATH, self.private, self.public], check=True)
  25
  26
  27 log.info("generate keys")
  28 server_key = Keypair("server")
  29 client_key = Keypair("client")
  30
  31 log.info("transfer random data")
  32 DATA = util.random_string(256).encode("utf-8")
  33
  34
  35 def run_client(port: int, key_priv: str, key_pub: str, *flags: str) -> None:
  36     """Start client version of sptps_test."""
  37     client_cmd = [
  38         path.SPTPS_TEST_PATH,
  39         "-4",
  40         "-q",
  41         *flags,
  42         key_priv,
  43         key_pub,
  44         "localhost",
  45         str(port),
  46     ]
  47     log.info('start client with "%s"', " ".join(client_cmd))
  48     subp.run(client_cmd, input=DATA, check=True)
  49
  50
  51 def get_port(server: subp.Popen) -> int:
  52     """Get port that sptps_test server is listening on."""
  53     assert server.stderr
  54     while True:
  55         line = server.stderr.readline().decode("utf-8")
  56         match = port_re.match(line)
  57         if match:
  58             return int(match[1])
  59         log.debug("waiting for server to start accepting connections")
  60
  61
  62 def test(key0: Keypair, key1: Keypair, *flags: str) -> None:
  63     """Run tests using the supplied keypair."""
  64     server_cmd = [path.SPTPS_TEST_PATH, "-4", *flags, key0.private, key1.public, "0"]
  65     log.info('start server with "%s"', " ".join(server_cmd))
  66
  67     with subp.Popen(server_cmd, stdout=subp.PIPE, stderr=subp.PIPE) as server:
  68         assert server.stdout
  69
  70         port = get_port(server)
  71         run_client(port, key1.private, key0.public, *flags)
  72
  73         received = b""
  74         while len(received) < len(DATA):
  75             received += server.stdout.read()
  76
  77         if server.returncode is None:
  78             server.kill()
  79
  80     check.equals(DATA, received)
  81
  82
  83 def run_keypair_tests(*flags: str) -> None:
  84     """Run tests on all generated keypairs."""
  85     log.info("running tests with (client, server) keypair and flags %s", flags)
  86     test(server_key, client_key)
  87
  88     log.info("running tests with (server, client) keypair and flags %s", flags)
  89     test(client_key, server_key)
  90
  91
  92 log.info("running tests in stream mode")
  93 run_keypair_tests()
  94
  95 log.info("running tests in datagram mode")
  96 run_keypair_tests("-dq")