86 lines
3.8 KiB
Python
Executable File
86 lines
3.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
import argparse, json, os, ssl, time, urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
def post_json(api: str, path: str, token: str | None, payload: dict) -> tuple[int, dict]:
|
|
data = json.dumps(payload, ensure_ascii=False).encode('utf-8')
|
|
req = urllib.request.Request(api.rstrip('/') + path, data=data, method='POST')
|
|
req.add_header('Content-Type', 'application/json')
|
|
if token:
|
|
req.add_header('Authorization', f'Bearer {token}')
|
|
ctx = ssl._create_unverified_context()
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30, context=ctx) as r:
|
|
return r.status, json.loads(r.read().decode('utf-8'))
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode('utf-8', errors='ignore')
|
|
try:
|
|
obj = json.loads(body)
|
|
except Exception:
|
|
obj = {'raw': body[:300]}
|
|
return e.code, obj
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument('--api', default=os.environ.get('AIOS_GITOPS_API', 'https://cnnetmkt.top/api'))
|
|
ap.add_argument('--token-file', default=os.environ.get('AIOS_GITOPS_WRITE_TOKEN_FILE', '/data/aios-gitops/runtime/secrets/gitops-write-bot.token'))
|
|
ap.add_argument('--token-secret-ref', default=os.environ.get('AIOS_GITOPS_WRITE_TOKEN_REF', 'secret_ref://aios-gitops/runtime/gitops-write-bot'))
|
|
ap.add_argument('--branch-prefix', default='patch/gitops-write-auth-smoke')
|
|
args = ap.parse_args()
|
|
token = Path(args.token_file).read_text(encoding='utf-8').strip()
|
|
stamp = time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())
|
|
branch = f'{args.branch_prefix}-{stamp}'
|
|
patch_name = f'gitops-write-auth-smoke-{stamp}'
|
|
|
|
noauth_code, noauth_body = post_json(args.api, '/patch', None, {'patch_name': patch_name + '-noauth'})
|
|
if noauth_code != 401:
|
|
raise SystemExit(f'expected noauth 401, got {noauth_code}: {noauth_body}')
|
|
|
|
patch_code, patch_body = post_json(args.api, '/patch', token, {
|
|
'patch_name': patch_name,
|
|
'branch': branch,
|
|
'base': 'main',
|
|
'risk': 'low',
|
|
'description': 'GitOps write auth smoke using secret_ref token',
|
|
'message': f'test(gitops): create write auth smoke patch {stamp}',
|
|
})
|
|
if patch_code != 200 or patch_body.get('status') != 'patch_branch_created':
|
|
raise SystemExit(f'createPatch failed: {patch_code}: {patch_body}')
|
|
|
|
file_code, file_body = post_json(args.api, '/file', token, {
|
|
'branch': branch,
|
|
'base': 'main',
|
|
'create_branch': False,
|
|
'path': f'docs/gitops/write-auth-smoke-{stamp}.md',
|
|
'content': f'# GitOps Write Auth Smoke\n\nTime: {stamp}\nResult: createPatch and commitFile authenticated via secret_ref token.\n',
|
|
'message': f'test(gitops): commit write auth smoke {stamp}',
|
|
})
|
|
if file_code != 200 or file_body.get('status') != 'committed':
|
|
raise SystemExit(f'commitFile failed: {file_code}: {file_body}')
|
|
|
|
def summarize(name: str, code: int, body: dict) -> dict:
|
|
commit = body.get('commit') or {}
|
|
return {
|
|
'operation': name,
|
|
'http': code,
|
|
'status': body.get('status'),
|
|
'branch': body.get('branch') or (body.get('patch') or {}).get('branch'),
|
|
'rev': commit.get('rev') if isinstance(commit, dict) else None,
|
|
'sync_status': ((commit.get('sync') or {}).get('status') if isinstance(commit, dict) else None),
|
|
}
|
|
|
|
print(json.dumps({
|
|
'noauth': {'operation': 'createPatch without token', 'http': noauth_code, 'status': noauth_body.get('error')},
|
|
'createPatch': summarize('createPatch', patch_code, patch_body),
|
|
'commitFile': summarize('commitFile', file_code, file_body),
|
|
'token_secret_ref': args.token_secret_ref,
|
|
}, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|