59 lines
3.7 KiB
Python
Executable File
59 lines
3.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
import json, os, subprocess, sys, time, urllib.request, tempfile, pathlib
|
|
PORT = int(os.getenv('AIOS_TEST_PORT','18990'))
|
|
ROOT = pathlib.Path(tempfile.mkdtemp(prefix='aios-v1r1c0-uat-'))
|
|
WORKSPACE = ROOT / 'workspace'; WORKSPACE.mkdir()
|
|
subprocess.run(['git','init'],cwd=WORKSPACE,check=True,stdout=subprocess.DEVNULL)
|
|
subprocess.run(['git','config','user.name','AIOS UAT'],cwd=WORKSPACE,check=True)
|
|
subprocess.run(['git','config','user.email','uat@cnnetmkt.top'],cwd=WORKSPACE,check=True)
|
|
(WORKSPACE/'README.md').write_text('uat')
|
|
subprocess.run(['git','add','.'],cwd=WORKSPACE,check=True)
|
|
subprocess.run(['git','commit','-m','init'],cwd=WORKSPACE,check=True,stdout=subprocess.DEVNULL)
|
|
subprocess.run(['git','branch','-M','main'],cwd=WORKSPACE,check=True,stdout=subprocess.DEVNULL)
|
|
env=os.environ.copy(); env.update({'AIOS_GITOPS_ROOT':str(ROOT),'AIOS_GITOPS_WORKSPACE':str(WORKSPACE),'AIOS_GITOPS_DB':str(ROOT/'runtime'/'gitops_model.sqlite3'),'AIOS_GITOPS_DEV_ALLOW':'1','AIOS_GITOPS_HOST':'127.0.0.1','AIOS_GITOPS_PORT':str(PORT)})
|
|
server=pathlib.Path(__file__).resolve().parents[1]/'active-runtime'/'app'/'aios_gitops_model_api.py'
|
|
p=subprocess.Popen([sys.executable,str(server)],env=env,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,text=True)
|
|
# Wait until the local API is ready; slow CI or busy servers may need a few seconds.
|
|
def _wait_ready():
|
|
import urllib.error
|
|
for _ in range(40):
|
|
if p.poll() is not None:
|
|
out = p.stdout.read() if p.stdout else ''
|
|
raise RuntimeError('server exited early: '+out)
|
|
try:
|
|
urllib.request.urlopen(f'http://127.0.0.1:{PORT}/health', timeout=1).read()
|
|
return
|
|
except Exception:
|
|
time.sleep(0.25)
|
|
out = p.stdout.read() if p.stdout else ''
|
|
raise RuntimeError('server not ready: '+out)
|
|
_wait_ready()
|
|
def call(path,payload=None,user='alice',role='user'):
|
|
url=f'http://127.0.0.1:{PORT}{path}'; data=None if payload is None else json.dumps(payload).encode(); req=urllib.request.Request(url,data=data,method='POST' if payload is not None else 'GET')
|
|
if data: req.add_header('Content-Type','application/json')
|
|
req.add_header('Authorization','Bearer dev'); req.add_header('X-AIOS-User',user); req.add_header('X-AIOS-Role',role)
|
|
with urllib.request.urlopen(req,timeout=10) as r: return json.loads(r.read().decode())
|
|
ok=[]
|
|
try:
|
|
ok.append(('health',call('/health')['status']=='ok'))
|
|
add=call('/api/model/custom/add',{'provider_name':'local-test','base_url':'http://127.0.0.1:9999/v1','model_name':'test-model','priority':5},role='user')
|
|
ok.append(('custom_model_add',add.get('status')=='ok'))
|
|
pr=call('/api/model/priority/set',{'items':[{'model_id':add['model_id'],'priority':1},'base.deepseekv4-flash']},role='user')
|
|
ok.append(('priority_set',pr.get('status')=='updated'))
|
|
route=call('/api/model/route/resolve',{'task_type':'chat'},role='user')
|
|
ok.append(('route_resolve',route.get('selected',{}).get('id')==add['model_id']))
|
|
rt=call('/api/runtime/select',{'tier':'regular','risk':'normal'},role='user')
|
|
ok.append(('runtime_select_regular',rt.get('runtime_type')=='shared'))
|
|
b=call('/api/branch',{'branch':'patch/uat'},role='admin')
|
|
ok.append(('branch_create',b.get('status')=='ok'))
|
|
f=call('/api/file',{'branch':'patch/uat','path':'docs/uat.md','content':'ok','message':'test: uat file'},role='admin')
|
|
ok.append(('file_commit',f.get('status')=='committed'))
|
|
finally:
|
|
p.terminate()
|
|
try: p.communicate(timeout=5)
|
|
except Exception: pass
|
|
failed=[name for name,passed in ok if not passed]
|
|
print(json.dumps({'status':'PASS' if not failed else 'FAIL','cases':ok,'failed':failed,'root':str(ROOT)},ensure_ascii=False,indent=2))
|
|
raise SystemExit(0 if not failed else 1)
|