diff options
Diffstat (limited to 'ewww.py')
-rw-r--r-- | ewww.py | 94 |
1 files changed, 58 insertions, 36 deletions
@@ -5,6 +5,7 @@ import json import os import random import re +import shutil import subprocess import time import urllib.parse @@ -16,36 +17,36 @@ import yaml 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 + 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) + 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)) + 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) + 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'] + port = ctx["config"]["port"] c = urllib.parse.urlparse(url) host = c[1] - c = ('http', 'localhost:{}'.format(port)) + c[2:] + c = (c[0], "localhost:{}".format(port)) + c[2:] return urllib.parse.urlunparse(c), host @@ -57,59 +58,80 @@ def _url(ctx, url): def fixme(*args, **kwargs): assert 0 + # Create a file. def create_file(ctx, filename=None, content=None): 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): + 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): - config = get_file(filename).decode('UTF-8') + config = get_file(filename).decode("UTF-8") config = yaml.safe_load(config) - config['port'] = random.randint(2000, 30000) - ctx['config'] = config + config["port"] = random.randint(2000, 30000) + ctx["config"] = config config = yaml.safe_dump(config) _write(filename, config) - _run(ctx, [ - '/usr/sbin/daemonize', - '-c', os.getcwd(), - '-p', 'ewww.pid', - '-o', 'ewww.stdout', - '-e', 'ewww.stderr', - _binary('ewww'), - filename - ]) + _run( + ctx, + [ + "/usr/sbin/daemonize", + "-c", + os.getcwd(), + "-p", + "ewww.pid", + "-o", + "ewww.stdout", + "-e", + "ewww.stderr", + _binary("ewww"), + filename, + ], + ) _run_exit(ctx, 0) - ctx['pid'] = open('ewww.pid').read() + ctx["pid"] = open("ewww.pid").read() + # Make a HTTP request. def request(ctx, method=None, url=None): url, host = _url(ctx, url) - _run(ctx, ['curl', '-sv', '-X', method, '-HHost: {}'.format(host), url]) + print(url) + _run(ctx, ["curl", "-ksv", "-X", method, "-HHost: {}".format(host), url]) _run_exit(ctx, 0) + # Check status code of latest HTTP request. def status_code_is(ctx, code=None): - pattern = '\n< HTTP/1.1 {} '.format(code) - assert_eq(pattern in ctx['stderr'], True) + pattern = "\n< HTTP/2 {} ".format(code) + assert_eq(pattern in ctx["stderr"], True) + # Check a HTTP response header for latest request has a given value. def http_header_is(ctx, header=None, value=None): - s = ctx['stderr'] - pattern = '\n< {}: {}'.format(header, value) + s = ctx["stderr"] + pattern = "\n< {}: {}".format(header, value) if pattern not in s: - print('stderr:', repr(s)) - print('pattern:', repr(pattern)) + print("stderr:", repr(s)) + print("pattern:", repr(pattern)) assert_eq(pattern in s, True) - + + # Check a HTTP body response for latest request has a given value. def http_body_is(ctx, body=None): - s = ctx['stdout'] - body = body.encode('UTF8').decode('unicode-escape') + s = ctx["stdout"] + body = body.encode("UTF8").decode("unicode-escape") if body != s: - print('stdout:', repr(s)) - prin('pattern:', repr(body)) + print("stdout:", repr(s)) + prin("pattern:", repr(body)) assert_eq(body, s) |