XFESystem

"""Positive and negative controls for the production native MP verifier."""
import unittest
from MultiprocessorEvidence import validate_multiprocessors


def fixture(cpus):
    text = 'XFE_CPU_OK bsp_apic=0 cr3=0x1000\n'
    for cpu in range(1, cpus):
        text += f'XFE_AP_ONLINE|Index|{cpu}|ApicId|{cpu}|CoreKey|{cpu // 2}|PrivateDescriptors|1\n'
    return text + (f'XFE_MULTIPROCESSOR|Reported|{cpus}|Online|{cpus}|Backend|LimineApKernelJobs|ManagedSchedulerCpus|1\n'
        f'XFE_AP_JOBS_PROOF|Online|{cpus}|Rendezvous|{cpus-1}|CopyCases|32|Passed|1|ManagedSmp|0|Physical240Hz|0\n'
        'XFE_NATIVE_SELFTEST_OK independent_cpu_memory_interrupt_framebuffer\n')


class EvidenceChecks(unittest.TestCase):
    def test_real_counts_remain_separate_from_managed_smp(self):
        for count in (1, 2, 8, 32):
            result = validate_multiprocessors(fixture(count), count)
            self.assertEqual(result['ExecutingCpus'], count)
            self.assertEqual(result['ManagedSchedulerCpus'], 1)
            self.assertFalse(result['ManagedSmpVerified'])
            self.assertFalse(result['Physical240HzVerified'])

    def test_advertised_cpu_count_cannot_substitute_for_execution(self):
        with self.assertRaises(AssertionError): validate_multiprocessors(fixture(1), 8)

    def test_missing_ap(self):
        with self.assertRaises(AssertionError): validate_multiprocessors(fixture(8).replace('XFE_AP_ONLINE|Index|7', 'MISSING|Index|7'), 8)

    def test_duplicate_apic(self):
        with self.assertRaises(AssertionError): validate_multiprocessors(fixture(8).replace('|ApicId|7|', '|ApicId|6|'), 8)

    def test_bsp_repeated_as_ap(self):
        with self.assertRaises(AssertionError): validate_multiprocessors(fixture(8).replace('|ApicId|7|', '|ApicId|0|'), 8)

    def test_duplicate_summary(self):
        text = fixture(8)
        with self.assertRaises(AssertionError): validate_multiprocessors(text + text, 8)

    def test_rendezvous_required(self):
        with self.assertRaises(AssertionError): validate_multiprocessors(fixture(8).replace('|Rendezvous|7|', '|Rendezvous|0|'), 8)

    def test_pixel_failure_rejected(self):
        with self.assertRaises(AssertionError): validate_multiprocessors(fixture(8).replace('|Passed|1|', '|Passed|0|'), 8)

    def test_truncated_records_rejected(self):
        with self.assertRaises(AssertionError): validate_multiprocessors(fixture(8).replace('|PrivateDescriptors|1', ''), 8)

    def test_failure_after_success_rejected(self):
        with self.assertRaises(AssertionError): validate_multiprocessors(fixture(8) + 'XFE_NATIVE_FAIL AP_JOB_JOIN_TIMEOUT', 8)

    def test_false_managed_or_scanout_claim_rejected(self):
        for changed in ('|ManagedSmp|1|Physical240Hz|0', '|ManagedSmp|0|Physical240Hz|1'):
            with self.assertRaises(AssertionError): validate_multiprocessors(fixture(8).replace('|ManagedSmp|0|Physical240Hz|0', changed), 8)

    def test_readiness_before_proof_rejected(self):
        with self.assertRaises(AssertionError): validate_multiprocessors('XFE_NATIVE_SELFTEST_OK\n' + fixture(8), 8)


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