pulse/internal/api/testdata/test_chart_latency_proof.py

import tempfile
import unittest
from pathlib import Path

import chart_latency_proof as proof


class ChartLatencyProofTests(unittest.TestCase):
    def test_guest_time_is_not_counted_twice(self):
        before = dict(time=0, cpu=[0] * 10, swaps={'pswpin': 0, 'pswpout': 0})
        after = dict(time=1, cpu=[20, 0, 0, 80, 0, 0, 0, 0, 10, 0],
                     swaps=before['swaps'], load=[0.5, 0.5, 0.5], pressure={})
        row = proof.interval(before, after)
        self.assertEqual(row['total_ticks'], 100)
        self.assertEqual(row['busy_pct'], 20)
        self.assertEqual(row['guest_pct'], 10)

    def test_idle_requires_cpu_io_and_swap_quiet(self):
        row = dict(total_ticks=100, load=[1, 1, 1], idle_pct=95, iowait_pct=0,
                   steal_pct=0, guest_pct=0, swap_pages=0)
        self.assertTrue(proof.quiet(row))
        for field, value in [('total_ticks', 0), ('load', [2, 1, 1]), ('idle_pct', 89),
                             ('iowait_pct', 2), ('steal_pct', 1), ('guest_pct', 3), ('swap_pages', 1)]:
            with self.subTest(field=field):
                self.assertFalse(proof.quiet(dict(row, **{field: value})))

    def test_skip_or_hosted_pass_cannot_satisfy_local_slo(self):
        report = 'charts/workloads p50=40ms p95=50ms p99=60ms SLO=90ms\n'
        passed = report + '--- PASS: TestSLO_WorkloadCharts (1.0s)\n'
        proof.validate_output(passed, 0)
        for text, status in [(report + '--- SKIP: TestSLO_WorkloadCharts (1.0s)\n', 0),
                             (passed.replace('90ms', '550ms'), 0), (passed, 1), ('PASS\n', 0),
                             (passed + report, 0), (passed + passed, 0),
                             (passed + '--- SKIP: TestSLO_WorkloadCharts (1.0s)\n', 0)]:
            with self.subTest(text=text, status=status):
                with self.assertRaises(RuntimeError):
                    proof.validate_output(text, status)

    def test_contention_voids_only_the_latency_verdict(self):
        report = '    slo_bench_test.go:637: charts/workloads p50=75ms p95=107ms p99=112ms SLO=90ms\n'
        violation = '    slo_bench_test.go:637: SLO VIOLATION: p95=107ms exceeds target 90ms\n'
        passed = report + '--- PASS: TestSLO_WorkloadCharts (1.0s)\n'
        failed = report + violation + '--- FAIL: TestSLO_WorkloadCharts (1.0s)\nFAIL\n'
        for text, status in [(passed, 0), (failed, 1)]:
            with self.subTest(text=text):
                self.assertEqual(proof.validate_contended_output(text, status),
                                 report.split(': ', 1)[1].strip())
        cleanup = '    testing.go:1369: TempDir RemoveAll cleanup: directory not empty\n'
        for text, status in [(report + '--- SKIP: TestSLO_WorkloadCharts (1.0s)\n', 0),
                             (failed.replace(violation, ''), 1), (failed, 2), (passed, 1),
                             (failed.replace(violation, violation + cleanup), 1),
                             (failed.replace(violation, violation + violation), 1),
                             ('--- FAIL: TestSLO_WorkloadCharts (1.0s)\n', 1),
                             (passed.replace('90ms', '550ms'), 0), (passed + passed, 0)]:
            with self.subTest(text=text, status=status):
                with self.assertRaises(RuntimeError):
                    proof.validate_contended_output(text, status)

    def test_inconclusive_summary_names_competing_processes(self):
        errors = [dict(background_busy_pct=25.2, background_processes=[
                      dict(pid=1, name='provjobd312033', cpu_pct=25.2),
                      dict(pid=2, name='python3', cpu_pct=0.2)]),
                  dict(background_busy_pct=14.4, background_processes=[
                      dict(pid=1, name='provjobd312033', cpu_pct=14.2)])]
        with tempfile.TemporaryDirectory() as directory:
            summary = Path(directory) / 'summary.md'
            proof.report_inconclusive('charts/workloads p50=1ms p95=2ms p99=3ms SLO=90ms',
                                      errors, summary)
            text = summary.read_text()
        self.assertIn('inconclusive', text)
        self.assertIn('2 measurement intervals', text)
        self.assertIn('peak 25.2%', text)
        self.assertIn('provjobd312033', text)
        self.assertNotIn('python3', text)

    def test_measurement_io_stays_in_latency_and_background_cpu_invalidates(self):
        row = dict(background_busy_pct=0, guest_pct=0, steal_pct=0, swap_pages=0,
                   iowait_pct=20)
        self.assertFalse(proof.competing_capacity(row))
        for field, value in [('background_busy_pct', 25), ('guest_pct', 3),
                             ('steal_pct', 1), ('swap_pages', 1)]:
            with self.subTest(field=field):
                self.assertTrue(proof.competing_capacity(dict(row, **{field: value})))

    def test_process_receipt_excludes_measurement_and_handles_new_or_reused_pids(self):
        before = {10: dict(name='test', ticks=20, start='1'),
                  20: dict(name='other', ticks=30, start='2'),
                  30: dict(name='old', ticks=100, start='3')}
        after = {10: dict(name='test', ticks=200, start='1'),
                 20: dict(name='other', ticks=55, start='2'),
                 30: dict(name='new', ticks=10, start='4'),
                 40: dict(name='new process', ticks=5, start='5')}
        self.assertEqual(proof.background_processes(before, after, 10, 100), [
            dict(pid=20, name='other', cpu_pct=25),
            dict(pid=30, name='new', cpu_pct=10),
            dict(pid=40, name='new process', cpu_pct=5)])


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