"""Record actual requalification outcomes; reject incomplete or overlapping browser proof.""" import hashlib import json import sys from pathlib import Path import xml.etree.ElementTree as ET from collections import Counter HERE = Path(__file__).resolve().parent ROOT = HERE.parents[3] RAW = ROOT / '.artifacts/p9-requal-01' def load(path): data = path.read_bytes() return json.loads(data.decode('utf-16' if data.startswith(b'\xff\xfe') else 'utf-8-sig')) def identity(path): return {'path': path.relative_to(ROOT).as_posix(), 'sha256': hashlib.sha256(path.read_bytes()).hexdigest()} def specs(report): for suite in report.get('suites', []): yield from suite.get('specs', []) yield from specs(suite) def browser_results(report, exit_code, expected_count=None): if exit_code != 0 or report.get('errors'): raise ValueError('Browser process or run-level failure.') cases = [] for spec in specs(report): for test in spec['tests']: results = test.get('results', []) if test.get('expectedStatus') != 'passed' or len(results) != 1: raise ValueError('Missing, repeated or expected-failure execution.') result = results[0] if result.get('status') != 'passed' or result.get('retry', 0) != 0: raise ValueError('Failed, skipped or retried browser execution.') cases.append({'id': spec['id'] + ':' + test['projectName'], 'title': spec['title'], 'outcome': result['status'], 'durationMs': result['duration'] if 'duration' in result else None}) if len({c['id'] for c in cases}) != len(cases): raise ValueError('Duplicate browser case IDs.') if expected_count is not None and len(cases) != expected_count: raise ValueError('Browser membership count mismatch.') return cases def reconcile_ordinary(discovery, shards): expected = [s['id'] + ':' + t['projectName'] for s in specs(discovery) for t in s['tests']] actual = [case for shard in shards for case in shard] ids = [case['id'] for case in actual] if discovery.get('errors') or len(expected) != len(set(expected)) or len(ids) != len(set(ids)) or set(ids) != set(expected): raise ValueError('Ordinary discovery and actual shard membership differ or overlap.') return actual def require_complete(results): incomplete=[name for name,result in results.items() if result.get('status')!='PASS'] if incomplete: raise ValueError('Required automation incomplete or failing: '+', '.join(incomplete)) def sidecar_results(root, expected): cases=list(root.iter('testcase')) identities=[(case.get('classname'),case.get('name')) for case in cases] if len(cases)!=expected or len(set(identities))!=expected or any(not name for _,name in identities): raise ValueError('Missing or duplicate sidecar cases.') if any(case.find(tag) is not None for case in cases for tag in ['failure','error','skipped']): raise ValueError('Failed or skipped sidecar cases.') return [{'classname':group,'name':name,'outcome':'passed'} for group,name in identities] def jest_results(report, baseline, exit_code, expected): def rows(value): return [(suite['name'].replace('\\','/'),case) for suite in value['testResults'] for case in suite['assertionResults']] actual, reference=rows(report), rows(baseline) # test.each can legitimately generate identical labels; preserve their exact multiplicity. membership=lambda items: Counter((file,case['fullName']) for file,case in items) if exit_code!=0 or len(actual)!=expected or membership(actual)!=membership(reference): raise ValueError('Frontend native case membership differs from unchanged-source reference.') if any(case['status']!='passed' or case.get('invocations',1)!=1 or case.get('retryReasons') for _,case in actual): raise ValueError('Failed, skipped or retried frontend case.') return [{'file':file,'nativeOrdinal':index,'name':case['fullName'],'outcome':case['status']} for index,(file,case) in enumerate(actual)] def security_results(commands, npm, nuget): expected={'supply-chain','evaluator','browser-runtime','ingress','secrets','sbom','node-guards','extension', 'database-runner','browser-database-boundary','nuget-advisories','npm-advisories'} if len(commands)!=len(expected) or {command['name'] for command in commands}!=expected: raise ValueError('Missing or duplicate security command evidence.') projects=nuget.get('projects',[]) if len(projects)!=1 or not projects[0].get('path','').replace('\\','/').endswith('/JobTrackerApi/JobTrackerApi.csproj'): raise ValueError('Missing or unexpected NuGet project evidence.') if nuget.get('problems') or nuget.get('errors') or any(project.get('errors') for project in projects): raise ValueError('NuGet audit errors.') vulnerabilities=[package for project in projects for framework in project.get('frameworks',[]) for key in ['topLevelPackages','transitivePackages'] for package in framework.get(key,[]) if package.get('vulnerabilities')] passed=all(command['exitCode']==0 for command in commands) and npm['metadata']['vulnerabilities']['total']==0 and not vulnerabilities return passed,vulnerabilities def python_security_status(record): if record['advisoryGroups']: return 'FINDINGS' return 'INCOMPLETE' if record['errors'] else 'PASS' def collect_browser(name, count=None): report_path, process_path = RAW / (name + '.json'), RAW / (name + '-exit.json') result = {'lane': name} try: report, process = load(report_path), load(process_path) result.update(report=identity(report_path), processEvidence=identity(process_path), process=process) result['cases'] = browser_results(report, process['exitCode'], count) result.update(status='PASS', passed=len(result['cases']), stats=report.get('stats')) except FileNotFoundError as error: result.update(status='NOT_RUN', reason=str(error)) except (ValueError, KeyError) as error: result.update(status='FAIL', reason=str(error)) return result def collect_trx(name, file, expected): result = {'lane': name} try: path = RAW / file root = ET.parse(path).getroot() counters = next(e.attrib for e in root.iter() if e.tag.endswith('Counters')) cases = [e.attrib for e in root.iter() if e.tag.endswith('UnitTestResult')] process = load(RAW / (name + '-exit.json')) passed = process['exitCode'] == 0 and int(counters['passed']) == expected and int(counters['total']) == expected and len(cases) == expected passed = passed and len({(c['testId'],c['testName']) for c in cases}) == expected and len({c['executionId'] for c in cases}) == expected passed = passed and all(c['outcome'] == 'Passed' for c in cases) result.update(status='PASS' if passed else 'FAIL', report=identity(path), counters=counters, process=process, cases=[{'testId':c['testId'],'executionId':c['executionId'],'name':c['testName'], 'outcome':c['outcome']} for c in cases]) except FileNotFoundError as error: result.update(status='NOT_RUN', reason=str(error)) return result def main(): results = {'backend':collect_trx('backend','backend/backend.trx',1123), 'mariadb':collect_trx('mariadb','mariadb/mariadb.trx',35)} for index in range(1,5): results[f'ordinary-{index}'] = collect_browser(f'ordinary-{index}') try: if not all(results[f'ordinary-{i}']['status']=='PASS' for i in range(1,5)): raise ValueError('Incomplete ordinary shards.') actual=reconcile_ordinary(load(RAW/'discovery.json'),[results[f'ordinary-{i}']['cases'] for i in range(1,5)]) if len(actual)!=362: raise ValueError('Unexpected ordinary discovery count.') results['ordinary']={'status':'PASS','passed':len(actual),'discovery':identity(RAW/'discovery.json')} except (FileNotFoundError,ValueError) as error: results['ordinary']={'status':'NOT_COMPLETE','reason':str(error)} for name,count in {'ai':2,'auth':2,'rate':9,'ingress':6,'pdf-admission':1,'preferences':3, 'rd21':6,'diagnostics':18,'volume':3,'rd21-volume':3,'mariadb-browser':1,'mariadb-observations':2,'linux-renderer':5,'windows-settings-successor':4}.items(): results[name]=collect_browser(name,count) try: path=RAW/'frontend.json'; data=load(path); process=load(RAW/'frontend-exit.json') cases=jest_results(data,load(RAW/'frontend-before-security.json'),process['exitCode'],934) passed=process['exitCode']==0 and data['numPassedTests']==934 and data['numTotalTests']==934 and data['numPendingTests']==0 and data['numTodoTests']==0 and data['numFailedTests']==0 results['frontend']={'status':'PASS' if passed else 'FAIL','passed':data['numPassedTests'], 'suites':data['numPassedTestSuites'],'report':identity(path),'process':process, 'cases':cases,'membershipReference':identity(RAW/'frontend-before-security.json')} except FileNotFoundError as error: results['frontend']={'status':'NOT_RUN','reason':str(error)} except ValueError as error: results['frontend']={'status':'FAIL','reason':str(error)} for name in ['types','lint','build','sidecar']: try: process=load(RAW/(name+'-exit.json')) results[name]={'status':'PASS' if process['exitCode']==0 else 'FAIL','process':process,'log':identity(RAW/(name+'.log'))} if name=='sidecar': try: results[name]['cases']=sidecar_results(ET.parse(RAW/'sidecar.xml').getroot(),56) except ValueError as error: results[name].update(status='FAIL',reason=str(error)) results[name]['report']=identity(RAW/'sidecar.xml') except FileNotFoundError as error: results[name]={'status':'NOT_RUN','reason':str(error)} for name,file in [('performance','performance/result.json'),('native-pdf','native-pdf.json'),('linux-pdf','linux-pdf.json')]: try: data=load(RAW/file) results[name]={'status':data['status'],'record':data,'evidence':identity(RAW/file)} except FileNotFoundError: results[name]={'status':'NOT_RUN'} try: path=RAW/'security/commands.json'; commands=load(path) npm=load(RAW/'security/frontend-audit.json') nuget=load(RAW/'security/nuget-advisories.log') passed,vulnerabilities=security_results(commands,npm,nuget) results['security']={'status':'PASS' if passed else 'FAIL','commands':commands,'evidence':identity(path),'npmAudit':identity(RAW/'security/frontend-audit.json'),'nugetAudit':identity(RAW/'security/nuget-advisories.log'),'nugetVulnerabilities':vulnerabilities,'npmVulnerabilities':npm['metadata']['vulnerabilities']} except FileNotFoundError: results['security']={'status':'NOT_RUN'} except ValueError as error: results['security']={'status':'FAIL','reason':str(error)} try: path=HERE/'python-advisories.json'; data=load(path) if data['lockSha256']!=identity(ROOT/data['lockPath'])['sha256']: raise ValueError('Python lock changed after advisory queries.') results['python-advisories']={'status':python_security_status(data),'coverage':data['coverage'], 'counts':data['counts'],'errors':data['errors'],'evidence':identity(path)} except FileNotFoundError: results['python-advisories']={'status':'NOT_RUN'} except ValueError as error: results['python-advisories']={'status':'FAIL','reason':str(error)} manifest=RAW/'source-freeze.json' source=load(manifest) drift=[name for name,expected in source['inputs'].items() if not (ROOT/name).is_file() or hashlib.sha256((ROOT/name).read_bytes()).hexdigest()!=expected] record={'schemaVersion':1,'baselineCommit':'549b2cba5a791862ba3f854312ea819a86dc42a0', 'branch':'fix/p9-ci-and-requalification','countPolicy':'Separate lanes; discovery-only skips and overlapping reruns never enter executed totals.', 'sourceIdentity':identity(manifest),'frozenInputCount':len(source['inputs']),'runtimeDrift':drift,'results':results} record['automationComplete']=not drift and all(value.get('status')=='PASS' for value in results.values()) (HERE/'validation.json').write_text(json.dumps(record,indent=2)+'\n',encoding='utf-8') print(json.dumps({name:data.get('status','RECORDED') for name,data in results.items()})) if drift: raise SystemExit('Runtime source changed: '+str(drift)) if '--collect-only' not in sys.argv: try: require_complete(results) except ValueError as error: raise SystemExit(str(error)) if __name__=='__main__': main()