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
|
#############################################################################
# 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
# 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)
|
