From 5c0ba37eda80d39e6ceca59bb1d5f4942f858995 Mon Sep 17 00:00:00 2001 From: "Somhairle H. Marisol" Date: Thu, 17 Sep 2026 14:32:37 +0800 Subject: chore: establish Strategy Lab source baseline (development, not release) --- scripts/qa_api.py | 159 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 scripts/qa_api.py (limited to 'scripts') diff --git a/scripts/qa_api.py b/scripts/qa_api.py new file mode 100644 index 0000000..a97a1ce --- /dev/null +++ b/scripts/qa_api.py @@ -0,0 +1,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) -- cgit v1.2.3