cms-SiteForge/scripts/regression_v821.py

#!/usr/bin/env python3
"""Deterministic, subprocess-isolated pytest regression for V8.21.

All outcomes and raw stdout are preserved. A timed-out or failing batch is a
hard failure, not silently retried into a green result.
"""
from __future__ import annotations
import argparse
import datetime
import html
import json
import os
import re
import signal
import subprocess
import sys
import time
from pathlib import Path

ROOT=Path(__file__).resolve().parents[1]


def launch(modules,seconds):
    """Execute pytest in a disposable process group and bound inherited pipes.

    A test can spawn Worker descendants; killing only the pytest PID may leave
    children holding stdout open. On timeout, terminate the entire group.
    """
    started=time.monotonic()
    kwargs={'cwd':ROOT,'stdout':subprocess.PIPE,'stderr':subprocess.STDOUT,
            'text':True,'encoding':'utf-8','errors':'replace'}
    if os.name=='nt':
        kwargs['creationflags']=subprocess.CREATE_NEW_PROCESS_GROUP
    else:
        kwargs['start_new_session']=True
    child=subprocess.Popen([sys.executable,'-m','pytest','-q',*modules],**kwargs)
    try:
        output,_=child.communicate(timeout=seconds)
        code=child.returncode
        status='PASS' if code==0 else 'FAIL'
    except subprocess.TimeoutExpired:
        if os.name=='nt':
            subprocess.run(['taskkill','/PID',str(child.pid),'/T','/F'],
                           capture_output=True,timeout=10,check=False)
            if child.poll() is None:child.kill()
        else:
            try:os.killpg(child.pid,signal.SIGKILL)
            except ProcessLookupError:pass
        try:output,_=child.communicate(timeout=10)
        except subprocess.TimeoutExpired:
            child.kill();output,_=child.communicate(timeout=5)
        output+='\nTIMEOUT: pytest process group terminated after deadline.\n'
        code=None;status='TIMEOUT'
    m=re.search(r'(\d+) passed(?:\s|,)',output)
    return {'status':status,'passed':int(m.group(1)) if status=='PASS' and m else 0,
            'duration_seconds':round(time.monotonic()-started,2),'exit_code':code,'output':output}


def main(argv=None):
    p=argparse.ArgumentParser(description='SiteForge V8.21 independent pytest batch gate')
    p.add_argument('--batch-size',type=int,default=4)
    p.add_argument('--repeat',type=int,default=1)
    p.add_argument('--timeout',type=int,default=80)
    p.add_argument('--report',type=Path,default=Path('reports/v821-regression'))
    args=p.parse_args(argv)
    if not 1<=args.batch_size<=10 or not 1<=args.repeat<=5 or not 10<=args.timeout<=600:p.error('invalid bounds')
    modules=sorted('tests/'+f.name for f in (ROOT/'tests').glob('test_*.py'))
    if not modules:p.error('no pytest tests found')
    report=args.report if args.report.is_absolute() else ROOT/args.report
    logs=report.parent/(report.name+'_logs')
    logs.mkdir(parents=True,exist_ok=True)
    results=[]
    for round_num in range(1,args.repeat+1):
        for index in range(0,len(modules),args.batch_size):
            group=modules[index:index+args.batch_size]
            name=f'round{round_num}-batch{index//args.batch_size+1}'
            print('RUN',name,flush=True)
            result=launch(group,args.timeout)
            log=logs/(name+'.log')
            log.write_text('$ pytest -q '+' '.join(group)+'\n'+result.pop('output'),encoding='utf-8')
            result.update({'name':name,'modules':group,'log':str(log.relative_to(ROOT))})
            results.append(result)
            print(result['status'],result['passed'],'passed',result['duration_seconds'],'sec',flush=True)
            if result['status']!='PASS':break
        if results[-1]['status']!='PASS':break
    success=len(results)==args.repeat*((len(modules)+args.batch_size-1)//args.batch_size) and all(x['status']=='PASS' for x in results)
    info={'version':'8.25.0-dev.1','generated_at':datetime.datetime.now(datetime.timezone.utc).isoformat(),
          'status':'PASS' if success else 'FAIL','test_modules':len(modules),'repeats':args.repeat,
          'tests_per_round':sum(x['passed'] for x in results if x['name'].startswith('round1-')),
          'total_passes':sum(x['passed'] for x in results),'batches':results,
          'environment_note':'pytest local fixtures only; Docker/PG/Redis/browser/real WordPress not inferred as passed.'}
    report.parent.mkdir(parents=True,exist_ok=True)
    report.with_suffix('.json').write_text(json.dumps(info,ensure_ascii=False,indent=2),encoding='utf-8')
    header='\n'.join(['# SiteForge V8.21 完整自动化回归','',f"**{info['status']}** — {info['test_modules']} 模块 / {info['tests_per_round']} 项每轮 / {args.repeat} 轮",'',
                       '| 批次 | 结果 | 通过用例 | 耗时 | 日志 |','|---|---|---:|---:|---|'])
    table='\n'.join(f"| {r['name']} | {r['status']} | {r['passed']} | {r['duration_seconds']} 秒 | {r['log']} |" for r in results)
    report.with_suffix('.md').write_text(header+'\n'+table+'\n\n真实部署与外部服务另行验收。\n',encoding='utf-8')
    trs=''.join(f'<tr><td>{html.escape(r["name"])}</td><td>{html.escape(r["status"])}</td><td>{r["passed"]}</td><td>{r["duration_seconds"]}</td></tr>' for r in results)
    report.with_suffix('.html').write_text('''<!doctype html><html lang="zh-CN"><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>V8.21 回归报告</title><style>body{font:15px/1.7 system-ui,sans-serif;background:#f2f5f8;padding:25px}main{max-width:950px;margin:auto;background:white;padding:28px;border-radius:14px}table{width:100%;border-collapse:collapse}th,td{padding:10px;border-bottom:1px solid #dde2e7;text-align:left}</style><main><h1>SiteForge V8.21 回归验收</h1><h2>'''+info['status']+'</h2><p>每轮 '+str(info['tests_per_round'])+' 项自动化测试 · '+str(args.repeat)+' 轮 · 无外部环境实测保证</p><table><tr><th>批次</th><th>结果</th><th>通过</th><th>秒数</th></tr>'+trs+'</table><p>请参阅 JSON、Markdown 和原始测试日志。真实客户站点迁移仍需专项验收。</p></main></html>',encoding='utf-8')
    return 0 if success else 1

if __name__=='__main__':raise SystemExit(main())