def verify_vmbus_interrupts()

in microsoft/testsuites/core/cpu.py [0:0]

• 56 lines of code
• 15 McCabe index (conditional complexity)

    def verify_vmbus_interrupts(self, node: Node, log: Logger) -> None:
        found_hyperv_interrupt = False
        cpu_count = node.tools[Lscpu].get_core_count()
        log.debug(f"{cpu_count} CPU cores detected...")

        self._create_stimer_interrupts(node, cpu_count)
        interrupt_inspector = node.tools[InterruptInspector]
        interrupts = interrupt_inspector.get_interrupt_data()
        for interrupt in interrupts:
            is_hyperv_interrupt = any(
                [(substr in interrupt.metadata) for substr in hyperv_interrupt_substr]
            )
            if not is_hyperv_interrupt:
                continue
            log.debug(f"Processing Hyper-V interrupt : {interrupt}")
            assert_that(
                len(interrupt.cpu_counter),
                "Hyper-v interrupts should have count for each cpu.",
            ).is_equal_to(cpu_count)
            if interrupt.irq_number == "HRE" or "reenlightenment" in interrupt.metadata:
                assert_that(
                    all(
                        [
                            interrupt_count == 0
                            for interrupt_count in interrupt.cpu_counter
                        ]
                    ),
                    "Hyper-V reenlightenment interrupts should be 0 on each vCPU "
                    "unless the VM is doing migration.",
                ).is_greater_than_or_equal_to(True)
            elif interrupt.irq_number == "HYP" or "callback" in interrupt.metadata:
                assert_that(
                    sum(
                        [
                            interrupt_count > 0
                            for interrupt_count in interrupt.cpu_counter
                        ]
                    ),
                    "Hypervisor callback interrupt should be processed by "
                    "atleast min(#vCPU, 4) vCPU's",
                ).is_greater_than_or_equal_to(min(cpu_count, 4))
            elif interrupt.irq_number == "HVS" or "stimer" in interrupt.metadata:
                assert_that(
                    all(
                        [
                            interrupt_count > 0
                            for interrupt_count in interrupt.cpu_counter
                        ]
                    ),
                    "Hypervisor synthetic timer interrupt should be processed by "
                    "all vCPU's",
                ).is_equal_to(True)
            else:
                continue

            found_hyperv_interrupt = True

        # Fail test execution if these hyper-v interrupts are not showing up
        if not found_hyperv_interrupt:
            raise LisaException("Hyper-V interrupts are not recorded.")