def timesync_check_unbind_clocksource()

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

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

    def timesync_check_unbind_clocksource(self, node: Node) -> None:
        # 1. Check clock source name is one of hyperv_clocksource_tsc_page,
        #  lis_hv_clocksource_tsc_page, hyperv_clocksource.
        clocksource_map = {
            CpuArchitecture.X64: [
                "hyperv_clocksource_tsc_page",
                "lis_hyperv_clocksource_tsc_page",
                "hyperv_clocksource",
                "tsc",
            ],
            CpuArchitecture.ARM64: [
                "arch_sys_counter",
            ],
        }
        lscpu = node.tools[Lscpu]
        arch = lscpu.get_architecture()
        clocksource = clocksource_map.get(CpuArchitecture(arch), None)
        if not clocksource:
            raise UnsupportedCpuArchitectureException(arch)
        cat = node.tools[Cat]
        clock_source_result = cat.run(self.current_clocksource)
        assert_that([clock_source_result.stdout]).described_as(
            f"Expected clocksource name is one of {clocksource},"
            f" but actual it is {clock_source_result.stdout}."
        ).is_subset_of(clocksource)

        # 2. Check CPU flag contains constant_tsc from /proc/cpuinfo.
        if CpuArchitecture.X64 == arch:
            cpu_info_result = cat.run("/proc/cpuinfo")
            if CpuType.Intel == lscpu.get_cpu_type():
                expected_tsc_str = " constant_tsc "
            elif CpuType.AMD == lscpu.get_cpu_type():
                expected_tsc_str = " tsc "
            shown_up_times = cpu_info_result.stdout.count(expected_tsc_str)
            assert_that(shown_up_times).described_as(
                f"Expected {expected_tsc_str} shown up times in cpu flags is"
                " equal to cpu count."
            ).is_equal_to(lscpu.get_core_count())

        # 3. Check clocksource name shown up in dmesg.
        dmesg = node.tools[Dmesg]
        assert_that(dmesg.get_output()).described_as(
            f"Expected clocksource {clock_source_result.stdout} shown up in dmesg."
        ).contains(f"clocksource {clock_source_result.stdout}")

        # 4. Unbind current clock source if there are 2+ clock sources,
        # check current clock source can be switched to a different one.
        if node.shell.exists(PurePosixPath(self.unbind_clocksource)):
            available_clocksources = cat.run(self.available_clocksource)
            available_clocksources_array = available_clocksources.stdout.split(" ")
            # We can not unbind clock source if there is only one existed.
            if len(available_clocksources_array) > 1:
                available_clocksources_array.remove(clock_source_result.stdout)
                cmd_result = node.execute(
                    f"echo {clock_source_result.stdout} > {self.unbind_clocksource}",
                    sudo=True,
                    shell=True,
                )
                cmd_result.assert_exit_code()

                clock_source_result_expected = _wait_file_changed(
                    node, self.current_clocksource, available_clocksources_array
                )
                assert_that(clock_source_result_expected).described_as(
                    f"After unbind {clock_source_result.stdout}, current clock source "
                    f"doesn't switch properly."
                ).is_true()