jobtrackingapp/docs/redesign/phase-9/evidence.py

"""Summarize completed raw QA artifacts; missing lanes stay missing."""
import hashlib
import base64
import json
from pathlib import Path
import xml.etree.ElementTree as ET
from datetime import datetime, timezone

ROOT = Path(__file__).resolve().parents[3]
HERE = Path(__file__).resolve().parent
RAW = ROOT / '.artifacts/phase9'


def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def browser_cases(suites):
    rows = []
    for suite in suites:
        rows.extend(browser_cases(suite.get('suites', [])))
        for spec in suite.get('specs', []):
            for test in spec.get('tests', []):
                rows.append(dict(file=spec.get('file'), title=spec['title'], status=test['status'],
                                 results=[dict(status=r['status'], retry=r.get('retry', 0), durationMs=r.get('duration', 0),
                                               errors=r.get('errors', []), attachments=r.get('attachments', [])) for r in test.get('results', [])]))
    return rows


def attached_json(attachment):
    body = base64.b64decode(attachment['body']) if 'body' in attachment else Path(attachment['path']).read_bytes()
    return json.loads(body)


def write_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')


assert browser_cases([{'specs': [{'title': 'self-check', 'tests': [{'status': 'expected', 'results': [{'status': 'passed'}]}]}]}])[0]['results'][0]['status'] == 'passed'


if __name__ == '__main__':
    report = dict(observedUtc=datetime.now(timezone.utc).isoformat(), lanes={}, rawFiles={})
    report['documentOracle'] = dict(path='scripts/verify-cv-fidelity.py', sha256=digest(ROOT / 'scripts/verify-cv-fidelity.py'), qualification='Original unchanged oracle; native rendering and admission counts are separate.')
    sources = dict(ordinary='ordinary-accepted', ai='ai', auth='auth', rate='rate', ingress='ingress', pdf='pdf', preferences='preferences', phase9='phase9-frozen')
    for lane, source in sources.items():
        file = RAW / f'{source}.json'
        if not file.exists():
            continue
        data = json.loads(file.read_text(encoding='utf-8-sig'))
        cases = browser_cases(data['suites'])
        report['lanes'][lane] = dict(stats=data['stats'], cases=len(cases),
                                     retries=sum(max(len(c['results']) - 1, 0) for c in cases),
                                     statuses={status: sum(c['status'] == status for c in cases) for status in ['expected', 'unexpected', 'flaky', 'skipped']},
                                     fileFailures=data.get('errors', []))
        folder = HERE / ('p9-01' if lane == 'ordinary' else 'p9-05' if lane == 'phase9' else 'p9-06')
        folder.mkdir(parents=True, exist_ok=True)
        if lane == 'phase9':
            matrix, axe, screenshots = [], [], []
            for case in cases:
                for attachment in case['results'][0]['attachments']:
                    name = attachment['name']
                    if name == 'phase9-route-matrix':
                        sample = attached_json(attachment)
                        for row in sample['samples']:
                            row.pop('bodyText', None)
                            matrix.append(row)
                    elif name == 'phase9-settled-axe':
                        axe.append(attached_json(attachment))
                    elif name == 'phase9-rd21-characterization':
                        write_json(HERE / 'p9-04/rd21.json', attached_json(attachment))
                    elif name == 'phase9-sustained-performance':
                        write_json(HERE / 'p9-05/performance.json', attached_json(attachment))
                    if attachment['contentType'] == 'image/png':
                        path = Path(attachment['path'])
                        screenshots.append(dict(case=case['title'], name=name, path=str(path.relative_to(ROOT)), sha256=digest(path)))
            assert len(matrix) == 396
            write_json(HERE / 'p9-02/route-results.json', matrix)
            write_json(HERE / 'p9-02/screenshots.json', screenshots)
            write_json(HERE / 'p9-03/settled-axe.json', axe)
        if lane == 'ordinary':
            for case in cases:
                for attachment in case['results'][0]['attachments']:
                    if attachment['name'].startswith('route-census-') and attachment['contentType'] == 'application/json':
                        write_json(HERE / 'p9-01' / (attachment['name'] + '.json'), attached_json(attachment))
        for case in cases:
            for result in case['results']:
                for attachment in result['attachments']:
                    if 'body' in attachment:
                        body = base64.b64decode(attachment.pop('body'))
                        attachment.update(bytes=len(body), sha256=hashlib.sha256(body).hexdigest())
        (folder / f'{lane}-cases.json').write_text(json.dumps(cases, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    frontend = RAW / 'frontend.json'
    if frontend.exists():
        data = json.loads(frontend.read_text(encoding='utf-8-sig'))
        report['lanes']['frontend'] = {k: data.get(k) for k in ['success', 'numPassedTestSuites', 'numFailedTestSuites', 'numPassedTests', 'numFailedTests', 'numPendingTests', 'numTodoTests', 'startTime']}
    backend = RAW / 'backend/phase9.trx'
    if backend.exists():
        doc = ET.parse(backend)
        report['lanes']['backend'] = doc.find('.//{*}Counters').attrib
        cases = [dict(name=e.get('testName'), outcome=e.get('outcome'), duration=e.get('duration')) for e in doc.findall('.//{*}UnitTestResult')]
        (HERE / 'p9-00/backend-cases.json').write_text(json.dumps(cases, indent=2) + '\n', encoding='utf-8')
    sidecar = RAW / 'sidecar.xml'
    if sidecar.exists():
        report['lanes']['sidecar'] = ET.parse(sidecar).find('.//testsuite').attrib
    report['failedOrSupersededAttempts'] = {}
    for lane in ['phase9', 'phase9-successor', 'phase9-final', 'phase9-qualified', 'phase9-accepted', 'ordinary', 'ordinary-final', 'landing-red', 'landing-green', 'landing-qualified']:
        file = RAW / f'{lane}.json'
        if file.exists():
            data = json.loads(file.read_text(encoding='utf-8-sig'))
            report['failedOrSupersededAttempts'][lane] = data.get('stats')
    for file in sorted(RAW.glob('*-exit.json')):
        report['lanes'][file.stem.replace('-exit', '') + 'Process'] = json.loads(file.read_text(encoding='utf-8-sig'))
    for file in sorted(RAW.rglob('*')):
        if file.is_file() and file.suffix in ['.json', '.xml', '.trx', '.log'] and 'results' not in file.parts:
            report['rawFiles'][str(file.relative_to(ROOT))] = dict(bytes=file.stat().st_size, sha256=digest(file))
    (HERE / 'validation.json').write_text(json.dumps(report, indent=2, ensure_ascii=False) + '\n', encoding='utf-8')
    print(json.dumps(report['lanes'], indent=2))