XFESystem

"""Small offline negative controls for native pointer cadence evidence."""
import copy
import unittest
from PointerCadenceEvidence import parse_timing_dump, summarize_timing, cross_check_host_clock


def fixture():
    meta = {'Frequency': 1_000_000, 'Capacity': 4096, 'StartTicks': 1000, 'EndTicks': 26100,
            'FramesBefore': 10, 'FramesAfter': 17, 'InputsBefore': 100, 'InputsAfter': 113,
            'ObservedBefore': 100, 'ObservedAfter': 113, 'ConsumedBefore': 100, 'ConsumedAfter': 113,
            'CapturedFrames': 7, 'Overflow': 0, 'FramesPresentedDelta': 7, 'InvalidInputReports': 0,
            'FirstInputTicks': 1100, 'LastInputTicks': 25100, 'LostBefore': 0, 'LostAfter': 0, 'DroppedFrames': 0}
    frames = [{'Seq': 11, 'BeginTicks': 1000, 'EndTicks': 1050, 'InputSeq': 100, 'InputTicks': 0,
               'DispatchTicks': 0, 'X': 99, 'Y': 50}]
    for index in range(6):
        begin = 1200 + index * 4000
        frames.append({'Seq': 12 + index, 'BeginTicks': begin, 'EndTicks': begin + 100,
                       'InputSeq': 101 + index, 'InputTicks': begin - 100, 'DispatchTicks': begin - 20, 'X': 100 + index, 'Y': 50})
    return {'Meta': meta, 'Frames': frames}


def steady_240_fixture():
    # 240 kHz gives exactly 1,000 ticks per 240 Hz transition; the first
    # actual report takes 2 ms to appear, then every actual position is fresh.
    meta = {'Frequency': 240_000, 'Capacity': 4096, 'StartTicks': 800, 'EndTicks': 8000,
            'FramesBefore': 10, 'FramesAfter': 17, 'InputsBefore': 100, 'InputsAfter': 129,
            'ObservedBefore': 100, 'ObservedAfter': 129, 'ConsumedBefore': 100, 'ConsumedAfter': 129,
            'CapturedFrames': 7, 'Overflow': 0, 'FramesPresentedDelta': 7, 'InvalidInputReports': 0,
            'FirstInputTicks': 1000, 'LastInputTicks': 7720, 'LostBefore': 0, 'LostAfter': 0, 'DroppedFrames': 0}
    frames = []
    for index in range(7):
        end = 1480 + index * 1000
        frames.append({'Seq': 11 + index, 'BeginTicks': end - 20, 'EndTicks': end,
                       'InputSeq': 101 + index, 'InputTicks': end - 480, 'DispatchTicks': end - 30,
                       'X': 100 + index, 'Y': 50})
    return {'Meta': meta, 'Frames': frames}


def encode(value):
    def row(prefix, item): return prefix + ''.join('|' + name + '|' + str(number) for name, number in item.items()) + '\n'
    return 'XFE_POINTER_TIMING_BEGIN|1\n' + row('META', value['Meta']) + ''.join(row('FRAME', frame) for frame in value['Frames']) + 'XFE_POINTER_TIMING_END\n'


class PointerCadenceGuards(unittest.TestCase):
    def test_real_timestamps_and_counter_binding(self):
        value = summarize_timing(parse_timing_dump(encode(fixture())), 240)
        self.assertTrue(value['EvidenceValid'])
        self.assertTrue(value['SoftwareCadenceGoalMet'])
        self.assertIsNone(value['NativeScanoutHzMeasured'])
        self.assertFalse(value['PhysicalMonitorRefreshMatched'])

    def test_incomplete_dump_cannot_pass(self):
        self.assertIsNone(parse_timing_dump(encode(fixture()).removesuffix('XFE_POINTER_TIMING_END\n')))

    def test_native_logger_newline_prefixed_chunks_keep_every_frame(self):
        encoded = encode(fixture()).replace('\nFRAME|Seq|14', '\n\nFRAME|Seq|14').replace('XFE_POINTER_TIMING_END', '\n \t\nXFE_POINTER_TIMING_END')
        parsed = parse_timing_dump(encoded)
        self.assertEqual(7, len(parsed['Frames']))
        self.assertTrue(summarize_timing(parsed)['EvidenceValid'])
        missing = encoded.replace(next(line for line in encoded.splitlines() if line.startswith('FRAME|Seq|14')) + '\n', '')
        with self.assertRaises(ValueError): summarize_timing(parse_timing_dump(missing))

    def test_duplicate_field_rejected(self):
        with self.assertRaises(ValueError): parse_timing_dump(encode(fixture()).replace('META|', 'META|Frequency|1000000|', 1))

    def test_omitted_slow_frame_and_ring_overflow_rejected(self):
        for mutate in (lambda value: value['Frames'].pop(2), lambda value: value['Meta'].update(DroppedFrames=1)):
            value = fixture(); mutate(value)
            with self.assertRaises(ValueError): summarize_timing(value)

    def test_future_input_cannot_create_negative_latency(self):
        value = fixture(); value['Frames'][2]['DispatchTicks'] = value['Frames'][2]['BeginTicks'] + 1
        with self.assertRaises(ValueError): summarize_timing(value)

    def test_input_overflow_and_zero_frequency_rejected(self):
        for changes in ({'LostAfter': 1}, {'Frequency': 0}):
            value = fixture(); value['Meta'].update(changes)
            with self.assertRaises(ValueError): summarize_timing(value)

    def test_observed_pressure_cannot_fake_consumed_report_count(self):
        value = fixture(); value['Meta']['ConsumedAfter'] = 101
        with self.assertRaises(ValueError): summarize_timing(value)

    def test_repainted_animation_is_not_new_pointer_positions(self):
        value = fixture()
        for frame in value['Frames'][2:]: frame['X'] = 101
        summary = summarize_timing(value)
        self.assertTrue(summary['SoftwarePresentCadenceGoalMet'])
        self.assertFalse(summary['SoftwarePointerCadenceGoalMet'])
        self.assertFalse(summary['SoftwareCadenceGoalMet'])
        self.assertEqual(1, summary['FreshPointerPresentations'])
        self.assertGreater(summary['RepeatedPointerPresentationRatio'], .6)

    def test_reused_input_sequence_cannot_claim_new_pointer_frames(self):
        value = fixture()
        for frame in value['Frames'][2:]: frame['InputSeq'] = 101
        summary = summarize_timing(value)
        self.assertTrue(summary['SoftwarePresentCadenceGoalMet'])
        self.assertFalse(summary['SoftwarePointerCadenceGoalMet'])

    def test_first_position_is_baseline_not_invented_movement(self):
        value = fixture()
        value['Frames'].pop(0)
        value['Meta'].update(FramesBefore=11, FramesPresentedDelta=6, CapturedFrames=6)
        summary = summarize_timing(value)
        self.assertEqual(5, summary['FreshPointerPresentations'])
        self.assertEqual(12, summary['FreshPointerBaselineFrameSequence'])
        self.assertTrue(summary['SoftwarePointerCadenceGoalMet'])

    def test_repeated_pointer_frames_cannot_hide_first_transition_stall(self):
        value = fixture()
        value['Frames'][2]['X'] = value['Frames'][1]['X']
        summary = summarize_timing(value)
        self.assertTrue(summary['SoftwarePresentCadenceGoalMet'])
        self.assertEqual(12, summary['FreshPointerBaselineFrameSequence'])
        self.assertGreater(summary['FreshPointerFrameIntervals']['MaximumMs'], 1000 / 240)
        self.assertFalse(summary['SoftwarePointerCadenceGoalMet'])

    def test_exact_240_steady_cadence_with_two_ms_first_response_passes(self):
        value = steady_240_fixture()
        summary = summarize_timing(value)
        self.assertTrue(summary['SoftwareCadenceGoalMet'])
        self.assertTrue(summary['FreshPointerCadenceEvidenceSufficient'])
        self.assertEqual(6, summary['FreshPointerPresentations'])
        self.assertEqual(11, summary['FreshPointerBaselineFrameSequence'])
        self.assertEqual(1480, summary['FreshPointerBaselineEndTicks'])
        self.assertAlmostEqual(2, summary['FreshPointerFirstResponseMs'])
        self.assertAlmostEqual(.025, summary['FreshPointerCadenceWindowSeconds'])
        self.assertEqual(240, summary['FreshPointerPresentHz'])
        self.assertEqual(6, summary['FreshPointerFrameIntervals']['Samples'])
        self.assertAlmostEqual(1000 / 240, summary['FreshPointerFrameIntervals']['MaximumMs'])
        checked = cross_check_host_clock(summary, summary['ContinuousMotionSeconds'])
        self.assertTrue(checked['SoftwareCadenceGoalMet'])

    def test_two_period_first_real_transition_fails(self):
        value = steady_240_fixture()
        for frame in value['Frames'][1:]:
            for field in ('BeginTicks', 'EndTicks', 'InputTicks', 'DispatchTicks'):
                frame[field] += 1000
        value['Meta']['LastInputTicks'] += 1000
        value['Meta']['EndTicks'] += 1000
        summary = summarize_timing(value)
        self.assertAlmostEqual(2000 / 240, summary['FreshPointerFrameIntervals']['MaximumMs'])
        self.assertFalse(summary['SoftwarePointerCadenceGoalMet'])

    def test_slow_first_response_cannot_be_hidden_by_later_exact_240(self):
        value = steady_240_fixture()
        for frame in value['Frames']:
            for field in ('BeginTicks', 'EndTicks', 'DispatchTicks'):
                frame[field] += 1200
        value['Meta']['LastInputTicks'] += 1200
        value['Meta']['EndTicks'] += 1200
        summary = summarize_timing(value)
        self.assertEqual(240, summary['FreshPointerPresentHz'])
        self.assertAlmostEqual(7, summary['FreshPointerFirstResponseMs'])
        self.assertEqual(11, summary['FreshPointerBaselineFrameSequence'])
        self.assertFalse(summary['SoftwarePointerCadenceGoalMet'])

    def test_zero_initialized_position_is_not_a_movement_baseline(self):
        value = fixture()
        value['Frames'][0].update(InputSeq=0, X=0, Y=0)
        summary = summarize_timing(value)
        self.assertEqual(12, summary['FreshPointerBaselineFrameSequence'])
        self.assertEqual(5, summary['FreshPointerPresentations'])
        self.assertTrue(summary['SoftwarePointerCadenceGoalMet'])

    def test_baseline_cannot_move_forward_to_hide_repeated_position_stall(self):
        value = steady_240_fixture()
        for frame in value['Frames'][1:3]:
            frame['X'] = value['Frames'][0]['X']
        summary = summarize_timing(value)
        self.assertEqual(11, summary['FreshPointerBaselineFrameSequence'])
        self.assertAlmostEqual(3000 / 240, summary['FreshPointerFrameIntervals']['MaximumMs'])
        self.assertFalse(summary['SoftwarePointerCadenceGoalMet'])

    def test_only_one_real_transition_is_insufficient_cadence_evidence(self):
        value = steady_240_fixture()
        for frame in value['Frames'][2:]:
            frame['X'] = value['Frames'][1]['X']
        summary = summarize_timing(value)
        self.assertEqual(1, summary['FreshPointerPresentations'])
        self.assertFalse(summary['FreshPointerCadenceEvidenceSufficient'])
        self.assertFalse(summary['SoftwarePointerCadenceGoalMet'])

    def test_no_new_report_in_actual_presents_has_no_pointer_baseline(self):
        value = steady_240_fixture()
        for frame in value['Frames']:
            frame.update(InputSeq=0, InputTicks=0, DispatchTicks=0)
        summary = summarize_timing(value)
        self.assertIsNone(summary['FreshPointerBaselineFrameSequence'])
        self.assertIsNone(summary['FreshPointerFirstResponseMs'])
        self.assertFalse(summary['FreshPointerCadenceEvidenceSufficient'])
        self.assertEqual(0, summary['FreshPointerPresentations'])
        self.assertEqual(0, summary['FreshPointerPresentHz'])
        self.assertFalse(summary['SoftwarePointerCadenceGoalMet'])

    def test_tail_stall_cannot_be_hidden_by_exact_240_subwindow(self):
        value = steady_240_fixture()
        value['Meta'].update(LastInputTicks=10_000, EndTicks=10_100)
        summary = summarize_timing(value)
        self.assertEqual(240, summary['FreshPointerPresentHz'])
        self.assertGreater(summary['FreshPointerTailGapMs'], 1000 / 240)
        self.assertFalse(summary['SoftwarePointerCadenceGoalMet'])

    def test_stall_remains_in_tail_and_fails_exact_target(self):
        value = fixture()
        value['Frames'][-1]['EndTicks'] = 26000
        value['Meta']['LastInputTicks'] = 26000
        summary = summarize_timing(value)
        self.assertGreater(summary['ActiveFrameIntervals']['MaximumMs'], 4.167)
        self.assertGreater(summary['ActiveIntervalsOverTargetBudget'], 0)
        self.assertFalse(summary['SoftwareCadenceGoalMet'])


class CrossClockGuards(unittest.TestCase):
    @staticmethod
    def summary(seconds=5, hz=250):
        return {'EvidenceValid': True, 'SoftwareCadenceGoalMet': True, 'SoftwarePresentCadenceGoalMet': True,
                'SoftwarePointerCadenceGoalMet': True, 'ContinuousMotionSeconds': seconds,
                'SoftwarePresentHzDuringContinuousMotion': hz, 'TargetHz': 240, 'TargetPeriodMs': 1000 / 240,
                'ActiveFrameIntervals': {'MaximumMs': 4}, 'ActiveBoundaryGapsMs': [.3, .3],
                'FreshPointerPresentHz': hz, 'FreshPointerFrameIntervals': {'MaximumMs': 4},
                'FreshPointerBoundaryGapsMs': [.3, .3]}

    def test_matching_clocks_retain_native_evidence(self):
        result = cross_check_host_clock(self.summary(), 5)
        self.assertTrue(result['EvidenceValid'])
        self.assertTrue(result['SoftwareCadenceGoalMet'])
        self.assertEqual(1, result['ClockConsistency']['NativeSecondsPerHostSecond'])

    def test_guest_frequency_error_cannot_inflate_a_pass(self):
        for native in (2.5, 10):
            result = cross_check_host_clock(self.summary(seconds=native), 5)
            self.assertFalse(result['EvidenceValid'])
            self.assertFalse(result['SoftwareCadenceGoalMet'])
            self.assertFalse(result['ClockConsistency']['Passed'])

    def test_small_error_inside_tolerance_cannot_raise_actual_238_to_240(self):
        result = cross_check_host_clock(self.summary(seconds=4.95, hz=1190 / 4.95), 5)
        self.assertTrue(result['ClockConsistency']['Passed'])
        self.assertTrue(result['EvidenceValid'])
        self.assertAlmostEqual(238, result['ConservativeHostAnchoredSoftwarePresentHz'])
        self.assertFalse(result['SoftwareCadenceGoalMet'])

    def test_host_clock_correction_covers_pointer_rate_separately(self):
        summary = self.summary(seconds=4.95, hz=250)
        summary['FreshPointerPresentHz'] = 1190 / 4.95
        result = cross_check_host_clock(summary, 5)
        self.assertTrue(result['EvidenceValid'])
        self.assertTrue(result['SoftwarePresentCadenceGoalMet'])
        self.assertAlmostEqual(238, result['ConservativeHostAnchoredFreshPointerPresentHz'])
        self.assertFalse(result['SoftwarePointerCadenceGoalMet'])
        self.assertFalse(result['SoftwareCadenceGoalMet'])

    def test_full_window_clock_scale_applies_to_baseline_subwindow_and_boundaries(self):
        summary = summarize_timing(steady_240_fixture())
        result = cross_check_host_clock(summary, summary['ContinuousMotionSeconds'] * 1.01)
        self.assertTrue(result['EvidenceValid'])
        self.assertAlmostEqual(240 / 1.01, result['ConservativeHostAnchoredFreshPointerPresentHz'])
        self.assertAlmostEqual(2.02, result['ConservativeHostAnchoredFreshPointerBoundaryGapsMs'][0])
        self.assertGreater(result['ConservativeHostAnchoredFreshPointerMaximumIntervalMs'], 1000 / 240)
        self.assertFalse(result['SoftwarePointerCadenceGoalMet'])

    def test_missing_or_nonfinite_cross_clock_interval_is_not_valid_evidence(self):
        for host in (0, -1, float('nan'), float('inf'), None):
            result = cross_check_host_clock(self.summary(), host)
            self.assertFalse(result['EvidenceValid'])
            self.assertFalse(result['SoftwareCadenceGoalMet'])

    def test_240_edid_cannot_turn_60_software_into_pass(self):
        value = fixture()
        for frame in value['Frames']:
            for key in ('BeginTicks', 'EndTicks', 'InputTicks', 'DispatchTicks'):
                frame[key] *= 4
        for key in ('StartTicks', 'EndTicks', 'FirstInputTicks', 'LastInputTicks'):
            value['Meta'][key] *= 4
        summary = summarize_timing(value, 240)
        self.assertFalse(summary['SoftwareCadenceGoalMet'])
        self.assertFalse(summary['PhysicalMonitorRefreshMatched'])

    def test_burst_after_initial_stall_does_not_hide_boundary_gap(self):
        value = fixture()
        for index, frame in enumerate(value['Frames']):
            frame.update(BeginTicks=10000 + index * 1000, EndTicks=10100 + index * 1000,
                         InputTicks=9900 + index * 1000, DispatchTicks=9950 + index * 1000)
        summary = summarize_timing(value)
        self.assertGreaterEqual(summary['SoftwarePresentHzDuringContinuousMotion'], 240)
        self.assertLess(summary['ActiveFrameIntervals']['MaximumMs'], 4.167)
        self.assertGreater(max(summary['ActiveBoundaryGapsMs']), 4.167)
        self.assertFalse(summary['SoftwareCadenceGoalMet'])


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