1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
#############################################################################
# Some helpers to make step functions simpler.
import json
import logging
import os
import random
import re
import shutil
import signal
import socket
import subprocess
import time
import urllib.parse
import yaml
# Run a subprocess, capture its output and exit code in context.
def _run(ctx, argv):
p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = p.communicate("")
ctx["argv"] = argv
ctx["stdout"] = stdout.decode("utf-8")
ctx["stderr"] = stderr.decode("utf-8")
ctx["exit"] = p.returncode
# Check that latest call of _run ended with an specific exit code.
def _run_exit(ctx, expected):
if ctx["exit"] != expected:
print("ctx:", ctx.as_dict())
assert_eq(ctx["exit"], expected)
# Name of Rust binary, debug-build.
def _binary(name):
return os.path.abspath(os.path.join(srcdir, "target", "debug", name))
# Write a file with given content.
def _write(filename, content):
open(filename, "w").write(content)
# Construct a URL that points to server running on localhost by
# replacing the actual scheme and host with ones that work for test.
def _url(ctx, url):
port = ctx["config"]["port"]
c = urllib.parse.urlparse(url)
host = c[1]
c = (c[0], "localhost:{}".format(port)) + c[2:]
return urllib.parse.urlunparse(c), host
#############################################################################
# The actual step functions.
# Fail: use this for unimplemented steps.
def fixme(*args, **kwargs):
assert 0
# Create a file.
def create_file(ctx, filename=None, content=None):
logging.debug(f"Creating file {filename} with {content}")
dirname = os.path.dirname(filename)
os.makedirs(dirname)
_write(filename, content)
# Copy test certificate from source tree, where it's been created previously by
# ./check.
def copy_test_certificate(ctx, cert=None, key=None):
logging.debug(f"Copying test.pem, test.key from srcdir to {cert} and {key}")
shutil.copy(os.path.join(srcdir, "test.pem"), cert)
shutil.copy(os.path.join(srcdir, "test.key"), key)
# Start server using named configuration file.
def start_server(ctx, filename=None):
logging.debug(f"Starting ewww with config file {filename}")
config = get_file(filename).decode("UTF-8")
config = yaml.safe_load(config)
port = config["port"] = random.randint(2000, 30000)
logging.debug(f"Picked randomly port for ewww: {config['port']}")
ctx["config"] = config
config = yaml.safe_dump(config)
_write(filename, config)
start_daemon(ctx, "ewww", [_binary("ewww"), filename])
if not port_open("localhost", port, 5.0):
stderr = open(ctx["daemon"]["ewww"]["stderr"]).read()
logging.debug(f"Stderr from daemon: {stderr!r}")
# Wait for a port to be open
def port_open(host, port, timeout):
logging.debug(f"Waiting for port localhost:{port} to be available")
started = time.time()
while time.time() < started + timeout:
try:
socket.create_connection((host, port), timeout=timeout)
return True
except socket.error:
pass
logging.error(f"Port localhost:{port} is not open")
return False
# Stop previously started server.
def stop_server(ctx):
logging.debug("Stopping ewww")
stop_daemon(ctx, "ewww")
# Make an HTTP request.
def request(ctx, method=None, url=None):
logging.debug(f"Making HTTP request to ewww: {method} {url}")
url, host = _url(ctx, url)
http_request(ctx, host=host, method=method, url=url)
|
