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
|
import logging
import json
import os
import subprocess
import time
import requests
CONTROLLER_URL = 'https://wmf2-controller.vm.liw.fi'
TOKEN = 'eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzUxMiJ9.eyJpc3MiOiJpc3MiLCJzdWIiOiJzdWJqZWN0LXV1aWQiLCJhdWQiOiJhdWQiLCJleHAiOjE1NjQxNDc5MTksInNjb3BlIjoidHJpZ2dlciBzdGF0dXMifQ.WZ7wQgzPSrhIb7ufTbdnnXIYu6sOjwMmhouJrboeJQQMZKHwp-6MRbJ8dccYU7UuPyp17IlZf2w875FkYwvCTcfySwkbocBw7LmR4FqrWhHKhG6eYJBrcSLU_72t1pQaYR_-m1MGeY7aXtmUclLhzl1SEAFxOOhWwPuXb2Pk7_JeXxN0ErUka4vA2uPtqCH7rCICgbyY1mBApaZQmucWORwLs5FmhFy5STWmcgYNSesId2rj0_0Y5jW3Ws5yif0PmxCx0ik5qCebmWbvvAHQ7vDCsfBYNBQbv59MkfeSOFin4CnWy_WgqVE7A20ZymMTTjle_5au014vhn-9g3OzmIGPX1dokxa_ZeSzqcoa8ZxASwH2TLIHUAbZh_PT68dyR9IZNPZLbK2vh5OK_TEXxQ9PM38CGoa0Sl7RIzKu7i9U6vtTz0npCwi7JMQFHltrGKJtkRcqUvI3BAl6ZjRp34y3PA0HyvjVrHtcqhyagEwQqsBggloyOwVibVAmMAqMsp59Jjk4oxc7lRFcw2SADYtRkSDswJsWkjxvDr4_L2PZ7LdklcfKomLXCb4n46VM91il3PWxPIJMzNcd8uQAyQcLrzZNND4-kmlNaVIofccYNqPk_qPlQ35OAnncZn6WXc8lIzp7bMjXMzMZCThb_m-K6MPTeMcepVFV8ZKEwIE'
def _POST(token, baseurl, path, body):
logging.info('POST to %s%s with %r', baseurl, path, body)
url = '{}{}'.format(baseurl, path)
headers = {
'Authorization': 'Bearer {}'.format(TOKEN),
'Content-Type': 'application/json',
}
return requests.post(url, headers=headers, data=json.dumps(body))
def _GET(url, filename):
logging.info('GET %s to %s', url, filename)
r = requests.get(url)
logging.info('Status %s', r.status_code)
assert r.ok
with open(filename, 'wb') as f:
f.write(r.content)
def _get_token():
return TOKEN
def nop(ctx, url=None):
logging.info('Blithely assuming %s exists', url)
def wait(ctx, seconds):
logging.info('Waiting for %r seconds', seconds)
seconds = int(seconds)
time.sleep(seconds)
def trigger_build(ctx, body=None):
logging.info('Triggering build with %r', body)
token = _get_token()
body = json.loads(body)
r = _POST(token, CONTROLLER_URL, '/trigger', body)
ctx['status'] = r.status_code # FIXME
def status_code_is(ctx, code=None):
logging.info('Checking HTTP respons status code is %r', code)
code = int(code)
assert ctx['status'] == code
def get_build_status(ctx):
# FIXME
logging.info('Retrieving build status for all builds')
old = ctx.get('builds')
if old is None:
ctx['builds'] = {'builds': {'hithere': 'building'}}
else:
ctx['builds'] = {'builds': {'hithere': 'success'}}
def build_status_is(ctx, name=None, status=None):
# FIXME
logging.info('Retrieving that %r build status is %r', name, status)
builds = ctx.get('builds', {}).get('builds')
assert builds is not None and builds.get(name) == status
def get_file(ctx, url=None, filename=None):
logging.info('Retrieving %r to %r', url, filename)
_GET(url, filename)
def run_file(ctx, filename=None):
# FIXME
logging.info('Executing %r', filename)
filename = './' + filename
os.chmod(filename, 0o700)
ctx['stdout'] = subprocess.check_output([filename]).decode('utf-8')
def stdout_is(ctx, expected=None):
logging.info('Checking captured stdout is %r', expected)
logging.info('Captured stdout is %r', ctx['stdout'])
assert ctx['stdout'] == (expected + '\n')
|
