131 lines
7.5 KiB
Python
131 lines
7.5 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""V8.21 repeatable integration regression with per-batch deadlines and isolation.
|
||
|
|
|
||
|
|
Runs pytest against only the local repository. Timeout cleanup kills the whole
|
||
|
|
process group on POSIX rather than leaving orphan pytest/Worker children.
|
||
|
|
Nonzero, flaky and timed-out runs always fail the quality gate.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
import argparse
|
||
|
|
import datetime as dt
|
||
|
|
import html
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import re
|
||
|
|
import signal
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
from dataclasses import asdict, dataclass
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
ROOT=Path(__file__).resolve().parent.parent
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class Run:
|
||
|
|
label: str
|
||
|
|
modules: list[str]
|
||
|
|
status: str
|
||
|
|
seconds: float
|
||
|
|
exit_code: int | None
|
||
|
|
passed: int
|
||
|
|
log: str
|
||
|
|
|
||
|
|
|
||
|
|
def run_pytest(label: str, modules: list[str], seconds: int, reports: Path, env: dict[str,str]) -> Run:
|
||
|
|
args=[sys.executable,'-m','pytest','-vv','--tb=short','--disable-warnings','--no-header',*modules]
|
||
|
|
filename=re.sub(r'[^a-zA-Z0-9_-]+','-',label)+'.log'
|
||
|
|
path=reports/filename
|
||
|
|
began=time.monotonic()
|
||
|
|
flags={'start_new_session':True} if os.name!='nt' else {'creationflags':subprocess.CREATE_NEW_PROCESS_GROUP}
|
||
|
|
expired=False
|
||
|
|
with path.open('wb') as out:
|
||
|
|
out.write(('$ '+' '.join(args)+'\n').encode('utf-8'))
|
||
|
|
out.flush()
|
||
|
|
process=subprocess.Popen(args,cwd=ROOT,env=env,stdout=out,stderr=subprocess.STDOUT,**flags)
|
||
|
|
try:
|
||
|
|
process.wait(timeout=seconds)
|
||
|
|
except subprocess.TimeoutExpired:
|
||
|
|
expired=True
|
||
|
|
if os.name!='nt':
|
||
|
|
try: os.killpg(process.pid,signal.SIGTERM)
|
||
|
|
except ProcessLookupError: pass
|
||
|
|
else: process.terminate()
|
||
|
|
try: process.wait(timeout=4)
|
||
|
|
except subprocess.TimeoutExpired:
|
||
|
|
if os.name!='nt':
|
||
|
|
try: os.killpg(process.pid,signal.SIGKILL)
|
||
|
|
except ProcessLookupError: pass
|
||
|
|
else: process.kill()
|
||
|
|
process.wait(timeout=5)
|
||
|
|
out.write(f'\nTIMEOUT: deadline={seconds}s; process group terminated\n'.encode())
|
||
|
|
content=path.read_text(encoding='utf-8',errors='replace')
|
||
|
|
m=re.findall(r'(?m)(\d+) passed(?:,| in |\s*\()',content)
|
||
|
|
passed=int(m[-1]) if m else 0
|
||
|
|
status='TIMEOUT' if expired else ('PASS' if process.returncode==0 and passed>0 else 'FAIL')
|
||
|
|
return Run(label,modules,status,round(time.monotonic()-began,2),process.returncode if not expired else None,passed,str(path))
|
||
|
|
|
||
|
|
|
||
|
|
def write_reports(runs: list[Run],prefix: Path,modules_count: int,repeat: int):
|
||
|
|
statuses={x:sum(r.status==x for r in runs) for x in ('PASS','FAIL','TIMEOUT')}
|
||
|
|
batches=[r for r in runs if re.fullmatch(r'round\d+-batch\d+',r.label)]
|
||
|
|
bad=[r for r in batches if r.status!='PASS']
|
||
|
|
result={'version':'8.21.0-dev.1','date':dt.datetime.now(dt.timezone.utc).isoformat(),
|
||
|
|
'modules':modules_count,'repeats':repeat,'status':'PASS' if not bad else 'FAIL',
|
||
|
|
'tests_last_complete_round':sum(r.passed for r in batches if r.label.startswith(f'round{repeat}-')),
|
||
|
|
'counts':statuses,'failures':[asdict(x) for x in bad],
|
||
|
|
'runs':[asdict(x) for x in runs],
|
||
|
|
'note':'Timed out, failing, or recovered-on-retry tests are not considered passing.'}
|
||
|
|
prefix.parent.mkdir(parents=True,exist_ok=True)
|
||
|
|
prefix.with_suffix('.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf-8')
|
||
|
|
lines=['# SiteForge V8.21 集成回归','',f"状态：**{result['status']}**；模块：{modules_count}；重复轮次：{repeat}",'',
|
||
|
|
'| 批次 | 状态 | 通过用例 | 秒 | 日志 |','|---|---|---:|---:|---|']
|
||
|
|
for r in runs:
|
||
|
|
lines.append(f'| {r.label} | {r.status} | {r.passed} | {r.seconds} | {Path(r.log).name} |')
|
||
|
|
lines+=['','失败、超时或重跑后通过均视为验收未通过；详见日志。','']
|
||
|
|
prefix.with_suffix('.md').write_text('\n'.join(lines),encoding='utf-8')
|
||
|
|
rows=''.join(f'<tr><td>{html.escape(r.label)}</td><td class="{r.status.lower()}">{r.status}</td><td>{r.passed}</td><td>{r.seconds}</td><td>{html.escape(Path(r.log).name)}</td></tr>' for r in runs)
|
||
|
|
doc=f'''<!doctype html><html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>SiteForge V8.21 · 自动化回归</title><style>body{{font:15px/1.5 system-ui,sans-serif;background:#f7f8fa;color:#19273b;margin:0}}header{{background:#182940;color:white;padding:32px 6%}}main{{max-width:1080px;margin:25px auto;background:white;border:1px solid #e0e4eb;border-radius:14px;padding:22px}}table{{width:100%;border-collapse:collapse}}td,th{{padding:11px;text-align:left;border-bottom:1px solid #e6e9ee}}.pass{{color:#14734c}}.fail,.timeout{{color:#a52626}}@media(max-width:700px){{main{{overflow-x:auto;margin:12px}}}}</style></head><body><header><h1>SiteForge · V8.21 集成回归</h1><p>结果：{result['status']} · {modules_count} 模块 · {repeat} 轮</p></header><main><p>每个测试批次有严格超时，发生异常时隔离失败模块并留存原始日志。任何超时均阻止上线判定。</p><table><tr><th>批次</th><th>状态</th><th>通过用例</th><th>耗时（秒）</th><th>日志</th></tr>{rows}</table></main></body></html>'''
|
||
|
|
prefix.with_suffix('.html').write_text(doc,encoding='utf-8')
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv=None):
|
||
|
|
ap=argparse.ArgumentParser(description='Timeout-safe repeatable SiteForge regression')
|
||
|
|
ap.add_argument('--batch-size',type=int,default=4)
|
||
|
|
ap.add_argument('--timeout',type=int,default=90,help='Single pytest batch timeout seconds')
|
||
|
|
ap.add_argument('--module-timeout',type=int,default=50)
|
||
|
|
ap.add_argument('--repeat',type=int,default=1)
|
||
|
|
ap.add_argument('--report',type=Path,default=Path('reports/v820-regression'))
|
||
|
|
args=ap.parse_args(argv)
|
||
|
|
if not 1<=args.batch_size<=50 or not 1<=args.repeat<=10 or args.timeout<5 or args.module_timeout<5:
|
||
|
|
ap.error('Invalid batch-size, repeat, or timeout')
|
||
|
|
modules=sorted('tests/'+p.name for p in (ROOT/'tests').glob('test_*.py'))
|
||
|
|
if not modules:ap.error('No tests/test_*.py modules')
|
||
|
|
logs=args.report.parent/(args.report.name+'_logs');logs.mkdir(parents=True,exist_ok=True)
|
||
|
|
env=os.environ.copy()
|
||
|
|
env.setdefault('PYTHONUNBUFFERED','1')
|
||
|
|
runs=[]
|
||
|
|
for r in range(1,args.repeat+1):
|
||
|
|
for index in range(0,len(modules),args.batch_size):
|
||
|
|
batch=modules[index:index+args.batch_size]
|
||
|
|
label=f'round{r}-batch{index//args.batch_size+1}'
|
||
|
|
print(f'RUN {label}: {len(batch)} modules',flush=True)
|
||
|
|
result=run_pytest(label,batch,args.timeout,logs,env)
|
||
|
|
runs.append(result)
|
||
|
|
print(f' {result.status} / {result.passed} passed / {result.seconds}s',flush=True)
|
||
|
|
if result.status!='PASS':
|
||
|
|
print(' Isolating modules (original failure is retained)',flush=True)
|
||
|
|
for i,mod in enumerate(batch,1):
|
||
|
|
isolated=run_pytest(f'{label}-isolated-{i}',[mod],args.module_timeout,logs,env)
|
||
|
|
runs.append(isolated)
|
||
|
|
print(f' {Path(mod).name}: {isolated.status} ({isolated.seconds}s)',flush=True)
|
||
|
|
# Do not start further rounds after a non-passing batch.
|
||
|
|
break
|
||
|
|
if any(x.status!='PASS' and re.fullmatch(r'round\d+-batch\d+',x.label) for x in runs):break
|
||
|
|
result=write_reports(runs,args.report,len(modules),args.repeat)
|
||
|
|
print(f"RESULT {result['status']}: {len(modules)} modules; report={args.report}.json",flush=True)
|
||
|
|
return 0 if result['status']=='PASS' else 1
|
||
|
|
|
||
|
|
if __name__=='__main__':sys.exit(main())
|