#!/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())