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
|
#############################################################################
# Some helpers to make step functions simpler.
import logging
import os
import random
import shutil
import socket
import time
import urllib.parse
import yaml
# Name of Rust binary, debug-build.
def _binary(name):
srcdir = globals()["srcdir"]
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):
srcdir = globals()["srcdir"]
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):
get_file = globals()["get_file"]
start_daemon = globals()["start_daemon"]
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):
stop_daemon = globals()["stop_daemon"]
logging.debug("Stopping ewww")
stop_daemon(ctx, "ewww")
# Make an HTTP request.
def request(ctx, method=None, url=None):
http_request = globals()["http_request"]
logging.debug(f"Making HTTP request to ewww: {method} {url}")
url, host = _url(ctx, url)
http_request(ctx, host=host, method=method, url=url)
|
