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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
"""Black-box release gates. Uses isolated QA accounts, no production credentials in output."""
import argparse
import json
import os
import pathlib
import secrets
import time
import requests
BASE = os.environ.get('QA_BASE', 'http://127.0.0.1:8787').rstrip('/')
OUT = pathlib.Path(os.environ.get('QA_OUTPUT', '/home/somhairle/.hermes/cache/strategy-lab-qa/results'))
OUT.mkdir(parents=True, exist_ok=True)
checks = []
def check(name, condition, detail=''):
checks.append({'name': name, 'passed': bool(condition), 'detail': detail})
print(('PASS ' if condition else 'FAIL ') + name, flush=True)
def session():
s = requests.Session()
s.headers.update({'Content-Type': 'application/json', 'Origin': BASE})
return s
def call(s, method, path, body=None, expected=(200, 201, 202)):
r = s.request(method, BASE + '/api' + path, json=body, timeout=150)
if r.status_code not in expected:
raise RuntimeError(f'{method} {path}: HTTP {r.status_code}: {r.text[:600]}')
return r.json() if r.content else {}
def denied(s, method, path, body=None, headers=None):
r = s.request(method, BASE + '/api' + path, json=body, headers=headers, timeout=20)
return r.status_code in (401, 403, 404)
def wait(s, path, terminal, seconds=300):
until = time.monotonic() + seconds
while time.monotonic() < until:
obj = call(s, 'GET', path)
if obj.get('status') in terminal:
return obj
time.sleep(2)
raise RuntimeError('timeout waiting for ' + path)
def accounts(admin):
nonce = secrets.token_hex(5)
users = []
for label in ['a', 'b']:
email = f'qa-{label}-{nonce}@strategy-lab.invalid'
pw = secrets.token_urlsafe(22)
invite = call(admin, 'POST', '/admin/invitations', {'email': email, 'role': 'member', 'expires_hours': 1})
s = session()
body = {'invite_token': invite['token'], 'email': email, 'password': pw, 'name': '验收账户' + label, 'role': 'admin'}
reg = call(s, 'POST', '/auth/register', body)
me = call(s, 'GET', '/auth/me')['user']
check('register ignores supplied admin role ' + label, me['role'] == 'member')
check('member cannot enumerate users ' + label, denied(s, 'GET', '/admin/users'))
retry = session().post(BASE + '/api/auth/register', json=body, timeout=20)
check('invitation single-use ' + label, retry.status_code in (400, 401, 403, 409))
users.append({'session': s, 'email': email, 'password': pw, 'user': me})
a, b = users
project = call(a['session'], 'POST', '/projects', {'name': '端到端验收项目', 'description': '自动化验收；真实行情与合成测试严格分离'})
pid = project['id']
check('other member cannot read project', denied(b['session'], 'GET', '/projects/' + pid))
check('admin cannot read private project', denied(admin, 'GET', '/projects/' + pid))
check('other member cannot overwrite project', denied(b['session'], 'PUT', f'/projects/{pid}/draft', {'code': 'leak', 'expected_generation': project['draft_generation']}))
check('csrf foreign origin blocked', denied(a['session'], 'PATCH', '/projects/' + pid, {'name': 'bad'}, {'Origin': 'https://evil.invalid'}))
host = BASE.split('://', 1)[1]
check('csrf substring origin blocked', denied(a['session'], 'PATCH', '/projects/' + pid, {'name': 'bad'}, {'Origin': 'https://' + host + '.evil.invalid'}))
code = 'import backtrader as bt\nclass Strategy(bt.Strategy):\n def next(self):\n if not self.position and len(self) == 2:\n self.buy(size=100)\n'
project = call(a['session'], 'PUT', f'/projects/{pid}/draft', {'code': code, 'expected_generation': project['draft_generation']})
stale = a['session'].put(BASE + f'/api/projects/{pid}/draft', json={'code': '# stale', 'expected_generation': project['draft_generation'] - 1}, timeout=20)
check('stale draft rejected', stale.status_code == 409)
version = call(a['session'], 'POST', f'/projects/{pid}/versions', {'message': '验收：固定源代码'})
edited = call(a['session'], 'PUT', f'/projects/{pid}/draft', {'code': code + '# changed\n', 'expected_generation': project['draft_generation']})
old = call(a['session'], 'GET', f'/projects/{pid}/versions')['items']
check('immutable saved source', any(v['id'] == version['id'] and v['code'] == code for v in old))
restored = call(a['session'], 'POST', f'/projects/{pid}/restore', {'version_id': version['id'], 'expected_generation': edited['draft_generation']})
check('restore creates new draft generation', restored['draft_code'] == code and restored['draft_generation'] > edited['draft_generation'])
check('last admin cannot be disabled', admin.patch(BASE + '/api/admin/users/' + call(admin, 'GET', '/auth/me')['user']['id'], json={'active': False}, timeout=20).status_code in (400, 403, 409))
second = session()
call(second, 'POST', '/auth/login', {'email': b['email'], 'password': b['password']})
reset = call(admin, 'POST', '/admin/users/' + b['user']['id'] + '/reset-password', {})
new_pw = secrets.token_urlsafe(22)
call(session(), 'POST', '/auth/reset-password', {'token': reset['reset_token'], 'new_password': new_pw})
check('reset revokes existing sessions', denied(second, 'GET', '/auth/me') and denied(b['session'], 'GET', '/auth/me'))
check('reset token single-use', session().post(BASE + '/api/auth/reset-password', json={'token': reset['reset_token'], 'new_password': new_pw}, timeout=20).status_code in (400, 401, 403, 409))
b['password'] = new_pw
call(b['session'], 'POST', '/auth/login', {'email': b['email'], 'password': new_pw})
call(admin, 'PATCH', '/admin/users/' + b['user']['id'], {'active': False})
check('disabled account session invalid', denied(b['session'], 'GET', '/auth/me'))
check('disabled account login denied', denied(session(), 'POST', '/auth/login', {'email': b['email'], 'password': new_pw}))
call(admin, 'PATCH', '/admin/users/' + b['user']['id'], {'active': True})
call(b['session'], 'POST', '/auth/login', {'email': b['email'], 'password': new_pw})
a['project'] = restored
return a, b
def market(a, b):
s = a['session']
req = {'name': '真实行情验收：浦发银行', 'instruments': [{'symbol': '600000', 'market': 'cn', 'asset_type': 'stock', 'name': '浦发银行'}], 'start_date': '2024-01-01', 'end_date': '2024-06-30', 'frequency': 'daily', 'adjustment': 'none', 'fields': ['open', 'high', 'low', 'close', 'volume']}
d = call(s, 'POST', '/datasets', req)
d = wait(s, '/datasets/' + d['id'], ['ready', 'failed'], 360)
check('real dataset ready', d['status'] == 'ready', str(d.get('error', ''))[:400])
if d['status'] != 'ready':
return
check('other account cannot read dataset', denied(b['session'], 'GET', '/datasets/' + d['id']))
check('no server paths in manifest', '/home/' not in json.dumps(d) and '"path"' not in json.dumps(d.get('manifest', {})))
preview = call(s, 'GET', '/datasets/' + d['id'] + '/preview')
check('real preview has rows', len(preview.get('rows', [])) > 0)
dupe = call(b['session'], 'POST', '/datasets', req)
dupe = wait(b['session'], '/datasets/' + dupe['id'], ['ready', 'failed'], 360)
check('shared cache reused across users', dupe['status'] == 'ready' and dupe.get('cache_hit') is True)
check('shared immutable data hash', dupe.get('manifest', {}).get('hash') == d.get('manifest', {}).get('hash'))
runbody = {'project_id': a['project']['id'], 'dataset_id': d['id'], 'capital': 100000, 'commission': 0.0003, 'slippage': 0.001, 'benchmark_symbol': '600000', 'parameters': {}, 'acknowledge_warnings': True}
run = call(s, 'POST', '/runs', runbody)
run = wait(s, '/runs/' + run['id'], ['succeeded', 'failed', 'cancelled'], 240)
check('real container backtest succeeded', run['status'] == 'succeeded', str(run.get('error', ''))[:400])
check('other member cannot read run', denied(b['session'], 'GET', '/runs/' + run['id']))
if run['status'] == 'succeeded':
result = run['result']
check('backtest has real equity and fills', len(result.get('equity', [])) > 10 and len(result.get('trades', [])) > 0)
check('result pins data manifest', result.get('data_manifest_hash') == d['manifest']['hash'])
rerun = call(s, 'POST', '/runs/' + run['id'] + '/rerun', {'use_original_data': True})
rerun = wait(s, '/runs/' + rerun['id'], ['succeeded', 'failed', 'cancelled'], 240)
check('original run reproducible', rerun['status'] == 'succeeded' and rerun.get('result', {}).get('equity') == result.get('equity'))
a['dataset_id'] = d['id']
a['run_id'] = run['id']
(OUT / 'real-backtest.json').write_text(json.dumps(run, ensure_ascii=False, indent=2))
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--market', action='store_true')
args = parser.parse_args()
admin = session()
call(admin, 'POST', '/auth/login', {'email': os.environ['QA_ADMIN_EMAIL'], 'password': os.environ['QA_ADMIN_PASSWORD']})
a, b = accounts(admin)
if args.market:
market(a, b)
safe_state = {k: v for k, v in a.items() if k != 'session'}
private = OUT / 'browser-account.private.json'
fd = os.open(private, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, 'w') as f:
json.dump(safe_state, f, ensure_ascii=False)
if __name__ == '__main__':
try:
main()
except Exception as exc:
check('workflow precondition', False, str(exc))
finally:
(OUT / 'api-checks.json').write_text(json.dumps(checks, ensure_ascii=False, indent=2))
raise SystemExit(0 if checks and all(c['passed'] for c in checks) else 1)
|
