-
Notifications
You must be signed in to change notification settings - Fork 5
/
manage.py
executable file
·104 lines (87 loc) · 2.85 KB
/
manage.py
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
#!/usr/bin/env python
import argparse
import json
import subprocess
import time
from http.client import HTTPConnection
from threading import Thread
from urllib.error import HTTPError
import pusto
ssh = lambda cmd: subprocess.call(
'ssh yadro.org "%s"' % cmd.replace('"', '\"').replace('$', '\$'),
shell=True
)
def process_args(args=None):
parser = argparse.ArgumentParser()
subs = parser.add_subparsers(title='subcommands')
def sub(name, **kw):
s = subs.add_parser(name, **kw)
s.set_defaults(sub=name)
s.arg = lambda *a, **kw: s.add_argument(*a, **kw) and s
s.exe = lambda f: s.set_defaults(exe=f) and s
return s
sub('test_urls', help='test url responses')\
.arg('-v', '--verbose', action='store_true')\
.arg('--host', help='use host for test')\
.exe(lambda a: check_urls(host=a.host, verbose=a.verbose))
sub('deploy', help='deploy to server')\
.exe(lambda a: ssh(
'cd /home/pusto/src'
'&& git pull'
'&& source $(cat .venv)/bin/activate'
'&& ./pusto.py build -b build-tmp'
'&& rm -rf build'
'&& mv build-tmp build'
'&& systemctl restart nginx.service'
))
args = parser.parse_args(args)
if not hasattr(args, 'sub'):
parser.print_usage()
else:
args.exe(args)
def check_urls(host=None, verbose=False):
log = lambda *a: verbose and print(*a)
if not host:
pusto.process('build')
host = 'localhost:9000'
args = 'run --port=9000 --no-reloader'.split(' ')
server = Thread(target=pusto.process, args=args)
server.daemon = True
server.start()
time.sleep(2)
with open('data/urls.json', 'br') as f:
urls = json.loads(f.read().decode())
def get(url, expected_code=200, indent=''):
comment = ''
try:
conn = HTTPConnection(host)
conn.request('HEAD', url)
res = conn.getresponse()
code = res.status
if code == 200:
res_url = url
else:
res_url = res.info().get('Location', '')
res_url = res_url.replace('http://' + host, '')
except HTTPError as e:
code = e.code
res_url = None
err = []
if code != expected_code:
err = ['%s (%r != %r)' % (url, code, expected_code)]
else:
log('%s%s %s %s # %s' % (indent, url, code, res_url, comment))
return err
err = []
for url in sorted(urls.keys()):
aliases = urls.get(url)
err += get(url)
for alias in aliases or []:
err += get(alias, expected_code=301, indent=' ')
if err:
print('Errors:')
print('\n'.join(err))
else:
print('OK')
if __name__ == '__main__':
process_args()