summaryrefslogtreecommitdiff
path: root/ewww.py
blob: a7ff9303e2f8694821b1ec1a00f669b10e81e680 (plain)
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
#############################################################################
# Some helpers to make step functions simpler.

import json
import os
import re
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):
    assert_eq(ctx['exit'], expected)


# 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):
    c = urllib.parse.urlparse(url)
    host = c[1]
    c = ('http', 'localhost:{}'.format(ctx['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

# Start server using named configuration file.
def start_server(ctx, filename=None):
    config = get_file(filename).decode('UTF-8')
    ctx['config'] = yaml.safe_load(config)
    _write(filename, config)
    pid = os.path.abspath('ewww.pid')
    bin = os.path.join(srcdir, 'target', 'debug', 'ewww')
    _run(ctx, ['/usr/sbin/daemonize', '-p', pid, bin, filename])
    ctx['pid'] = open(pid).read()
    ctx['port'] = 3030
    _run_exit(ctx, 0)

def request(ctx, method=None, url=None):
    url, host = _url(ctx, url)
    _run(ctx, ['curl', '-sv', '-X', method, '-HHost: {}'.format(host), url])
    _run_exit(ctx, 0)

def status_code_is(ctx, code=None):
    pattern = '\n< HTTP/1.1 {} '.format(code)
    assert_eq(pattern in ctx['stderr'], True)