3 """Test basic SPTPS features."""
6 import subprocess as subp
9 from testlib import path, util, check
10 from testlib.log import log
12 port_re = re.compile(r"Listening on (\d+)\.\.\.")
16 """Create public/private keypair using sptps_keypair."""
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)
27 log.info("generate keys")
28 server_key = Keypair("server")
29 client_key = Keypair("client")
31 log.info("transfer random data")
32 DATA = util.random_string(256).encode("utf-8")
35 def run_client(port: int, key_priv: str, key_pub: str, *flags: str) -> None:
36 """Start client version of sptps_test."""
47 log.info('start client with "%s"', " ".join(client_cmd))
48 subp.run(client_cmd, input=DATA, check=True)
51 def get_port(server: subp.Popen) -> int:
52 """Get port that sptps_test server is listening on."""
55 line = server.stderr.readline().decode("utf-8")
56 match = port_re.match(line)
59 log.debug("waiting for server to start accepting connections")
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))
67 with subp.Popen(server_cmd, stdout=subp.PIPE, stderr=subp.PIPE) as server:
70 port = get_port(server)
71 run_client(port, key1.private, key0.public, *flags)
74 while len(received) < len(DATA):
75 received += server.stdout.read()
77 if server.returncode is None:
80 check.equals(DATA, received)
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)
88 log.info("running tests with (server, client) keypair and flags %s", flags)
89 test(client_key, server_key)
92 log.info("running tests in stream mode")
95 log.info("running tests in datagram mode")
96 run_keypair_tests("-dq")