LangBot/tools/test_regression_ci.py

"""Exercise the daily workflow's published verdict, including incomplete runs."""

import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest

import yaml


class DailySummaryTest(unittest.TestCase):
    def test_summary_reports_observations_without_hiding_outer_failures(self):
        workflow = Path(__file__).resolve().parents[1] / '.github/workflows/daily-endurance.yml'
        steps = yaml.safe_load(workflow.read_text())['jobs']['endurance']['steps']
        command = next(step['run'] for step in steps if step.get('name') == 'Publish observations')
        suite = {
            'passed': True,
            'errors': [],
            'phases': [
                {
                    'achieved_rps': 1.0,
                    'verified_runs': 540,
                    'runner_requests': 540,
                    'ledger_closeout': {'matched': True},
                    'cases': {'chat': {'requests': 270, 'status': {'ok': 270}, 'p95_seconds': 0.5}},
                }
            ],
        }
        for gate, expected in [
            ({'passed': True}, 'Endurance: PASS'),
            ({'passed': False, 'infrastructure_error': 'teardown failed'}, 'teardown failed'),
            ({'passed': False, 'timed_out': True}, '2100-second execution budget'),
            (None, 'Missing verdict'),
        ]:
            with self.subTest(gate=gate), tempfile.TemporaryDirectory() as directory:
                root = Path(directory)
                if gate is not None:
                    target = root / '.qa/endurance/gate'
                    (target / 'suite').mkdir(parents=True)
                    (target / 'gate.json').write_text(json.dumps(gate))
                    (target / 'suite/summary.json').write_text(json.dumps(suite))
                summary = root / 'summary.md'
                env = {
                    **os.environ,
                    'PATH': str(Path(sys.executable).parent) + os.pathsep + os.environ['PATH'],
                    'GITHUB_STEP_SUMMARY': str(summary),
                    'GITHUB_SHA': 'core-revision',
                    'PLUGIN_REVISION': 'plugin-revision',
                }
                subprocess.run(['bash', '-e', '-c', command], cwd=root, env=env, check=True, capture_output=True)
                text = summary.read_text()
                self.assertIn(expected, text)
                if gate and gate.get('passed'):
                    self.assertIn('Completed runs: 540 / 540', text)
                    self.assertIn('1.000 RPS', text)
                else:
                    self.assertIn('Endurance: FAIL / incomplete', text)
                    self.assertNotIn('Endurance: PASS', text)


if __name__ == '__main__':
    unittest.main()