Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use _wait_file_changed method in lisa

Best Python code snippet using lisa_python
timesync.py
Source:timesync.py
...17from lisa.operating_system import CpuArchitecture, Redhat18from lisa.tools import Cat, Chrony, Dmesg, Hwclock, Lscpu, Ntp, Ntpstat, Service19from lisa.tools.lscpu import CpuType20from lisa.util.shell import wait_tcp_port_ready21def _wait_file_changed(22    node: Node,23    path: str,24    expected_value: Optional[Union[str, List[str]]],25) -> bool:26    timeout = 6027    timout_timer = create_timer()28    while timout_timer.elapsed(False) < timeout:29        cat = node.tools[Cat]30        result = cat.run(path, force_run=True)31        if isinstance(expected_value, list):32            if result.stdout in expected_value:33                return True34        else:35            if result.stdout == expected_value:36                return True37        sleep(0.5)38    return False39@TestSuiteMetadata(40    area="time",41    category="functional",42    description="""43    This test suite is related with time sync.44    """,45)46class TimeSync(TestSuite):47    ptp_registered_msg = "PTP clock support registered"48    hyperv_ptp_udev_rule = "ptp_hyperv"49    chrony_path = [50        "/etc/chrony.conf",51        "/etc/chrony/chrony.conf",52        "/etc/chrony.d/azure.conf",53    ]54    current_clocksource = (55        "/sys/devices/system/clocksource/clocksource0/current_clocksource"56    )57    available_clocksource = (58        "/sys/devices/system/clocksource/clocksource0/available_clocksource"59    )60    unbind_clocksource = (61        "/sys/devices/system/clocksource/clocksource0/unbind_clocksource"62    )63    current_clockevent = "/sys/devices/system/clockevents/clockevent0/current_device"64    unbind_clockevent = "/sys/devices/system/clockevents/clockevent0/unbind_device"65    @TestCaseMetadata(66        description="""67        https://docs.microsoft.com/en-us/azure/virtual-machines/linux/time-sync#check-for-ptp-clock-source # noqa: E50168        This test is to check -69            1. PTP time source is available on Azure guests (newer versions of Linux).70            2. PTP device name is hyperv.71            3. When accelerated network is enabled, multiple PTP devices will72             be available, the names of ptp are changeable, create the symlink73             /dev/ptp_hyperv to whichever /dev/ptp entry corresponds to the Azure host.74            4. Chrony should be configured to use the symlink /dev/ptp_hyperv75             instead of /dev/ptp0 or /dev/ptp1.76        """,77        priority=2,78    )79    def timesync_validate_ptp(self, node: Node) -> None:80        # 1. PTP time source is available on Azure guests (newer versions of Linux).81        dmesg = node.tools[Dmesg]82        assert_that(dmesg.get_output()).contains(self.ptp_registered_msg)83        # 2. PTP device name is hyperv.84        cat = node.tools[Cat]85        clock_name_result = cat.run("/sys/class/ptp/ptp0/clock_name")86        assert_that(clock_name_result.stdout).described_as(87            f"ptp clock name should be 'hyperv', meaning the Azure host, "88            f"but it is {clock_name_result.stdout}, more info please refer "89            f"https://docs.microsoft.com/en-us/azure/virtual-machines/linux/time-sync#check-for-ptp-clock-source"  # noqa: E50190        ).is_equal_to("hyperv")91        # 3. When accelerated network is enabled, multiple PTP devices will92        #  be available, the names of ptp are changeable, create the symlink93        #  /dev/ptp_hyperv to whichever /dev/ptp entry corresponds to the Azure host.94        assert_that(node.shell.exists(PurePosixPath("/dev/ptp_hyperv"))).described_as(95            "/dev/ptp_hyperv doesn't exist, make sure there is a udev rule to create "96            "symlink /dev/ptp_hyperv to /dev/ptp entry corresponds to the Azure host. "97            "More info please refer "98            "https://docs.microsoft.com/en-us/azure/virtual-machines/linux/time-sync#check-for-ptp-clock-source"  # noqa: E50199        ).is_true()100        # 4. Chrony should be configured to use the symlink /dev/ptp_hyperv101        #  instead of /dev/ptp0 or /dev/ptp1.102        for chrony_config in self.chrony_path:103            if node.shell.exists(PurePosixPath(chrony_config)):104                chrony_results = cat.run(f"{chrony_config}")105                assert_that(chrony_results.stdout).described_as(106                    "Chrony config file should use the symlink /dev/ptp_hyperv."107                ).contains(self.hyperv_ptp_udev_rule)108    @TestCaseMetadata(109        description="""110        This test is to check -111            1. Check clock source name is one of hyperv_clocksource_tsc_page,112             lis_hv_clocksource_tsc_page, hyperv_clocksource, tsc,113             arch_sys_counter(arm64).114             (thereâs a new feature in the AH2021 host that allows Linux guests so use115              the plain "tsc" instead of the "hyperv_clocksource_tsc_page",116              which produces a modest performance benefit when reading the clock.)117            2. Check CPU flag contains constant_tsc from /proc/cpuinfo.118            3. Check clocksource name shown up in dmesg.119            4. Unbind current clock source if there are 2+ clock sources, check current120             clock source can be switched to a different one.121        """,122        priority=2,123    )124    def timesync_check_unbind_clocksource(self, node: Node, log: Logger) -> None:125        unbind = node.shell.exists(PurePosixPath(self.unbind_clocksource))126        try:127            # 1. Check clock source name is one of hyperv_clocksource_tsc_page,128            #  lis_hv_clocksource_tsc_page, hyperv_clocksource.129            clocksource_map = {130                CpuArchitecture.X64: [131                    "hyperv_clocksource_tsc_page",132                    "lis_hyperv_clocksource_tsc_page",133                    "hyperv_clocksource",134                    "tsc",135                ],136                CpuArchitecture.ARM64: [137                    "arch_sys_counter",138                ],139            }140            lscpu = node.tools[Lscpu]141            arch = lscpu.get_architecture()142            clocksource = clocksource_map.get(CpuArchitecture(arch), None)143            if not clocksource:144                raise UnsupportedCpuArchitectureException(arch)145            cat = node.tools[Cat]146            clock_source_result = cat.run(self.current_clocksource)147            assert_that([clock_source_result.stdout]).described_as(148                f"Expected clocksource name is one of {clocksource},"149                f" but actual it is {clock_source_result.stdout}."150            ).is_subset_of(clocksource)151            # 2. Check CPU flag contains constant_tsc from /proc/cpuinfo.152            if CpuArchitecture.X64 == arch:153                cpu_info_result = cat.run("/proc/cpuinfo")154                if CpuType.Intel == lscpu.get_cpu_type():155                    expected_tsc_str = " constant_tsc "156                elif CpuType.AMD == lscpu.get_cpu_type():157                    expected_tsc_str = " tsc "158                shown_up_times = cpu_info_result.stdout.count(expected_tsc_str)159                assert_that(shown_up_times).described_as(160                    f"Expected {expected_tsc_str} shown up times in cpu flags is"161                    " equal to cpu count."162                ).is_equal_to(lscpu.get_core_count())163            # 3. Check clocksource name shown up in dmesg.164            dmesg = node.tools[Dmesg]165            assert_that(dmesg.get_output()).described_as(166                f"Expected clocksource {clock_source_result.stdout} shown up in dmesg."167            ).contains(f"clocksource {clock_source_result.stdout}")168            # 4. Unbind current clock source if there are 2+ clock sources,169            # check current clock source can be switched to a different one.170            if unbind:171                available_clocksources = cat.run(self.available_clocksource)172                available_clocksources_array = available_clocksources.stdout.split(" ")173                # We can not unbind clock source if there is only one existed.174                if len(available_clocksources_array) > 1:175                    available_clocksources_array.remove(clock_source_result.stdout)176                    cmd_result = node.execute(177                        f"echo {clock_source_result.stdout} >"178                        f" {self.unbind_clocksource}",179                        sudo=True,180                        shell=True,181                    )182                    cmd_result.assert_exit_code()183                    clock_source_result_expected = _wait_file_changed(184                        node, self.current_clocksource, available_clocksources_array185                    )186                    assert_that(clock_source_result_expected).described_as(187                        f"After unbind {clock_source_result.stdout}, current clock"188                        f" source doesn't switch properly."189                    ).is_true()190        finally:191            # after unbind, need reboot vm to bring the previous time clock source back192            if unbind:193                node.reboot()194                remote_node = cast(RemoteNode, node)195                is_ready, _ = wait_tcp_port_ready(196                    remote_node.public_address,197                    remote_node.public_port,198                    log=log,199                    timeout=300,200                )201                if not is_ready:202                    raise LisaException("vm is inaccessible after reboot.")203    @TestCaseMetadata(204        description="""205        This test is to check -206            1. Current clock event name is 'Hyper-V clockevent' for x86,207            'arch_sys_timer' for arm64.208            2. 'Hyper-V clockevent' or 'arch_sys_timer' and 'hrtimer_interrupt'209             show up times in /proc/timer_list should equal to cpu count.210            3. when cpu count is 1 and cpu type is Intel type, unbind current time211             clock event, check current time clock event switch to 'lapic'.212        """,213        priority=2,214    )215    def timesync_check_unbind_clockevent(self, node: Node) -> None:216        if node.shell.exists(PurePosixPath(self.current_clockevent)):217            # 1. Current clock event name is 'Hyper-V clockevent'.218            clockevent_map = {219                CpuArchitecture.X64: "Hyper-V clockevent",220                CpuArchitecture.ARM64: "arch_sys_timer",221            }222            lscpu = node.tools[Lscpu]223            arch = lscpu.get_architecture()224            clock_event_name = clockevent_map.get(CpuArchitecture(arch), None)225            if not clock_event_name:226                raise UnsupportedCpuArchitectureException(arch)227            cat = node.tools[Cat]228            clock_event_result = cat.run(self.current_clockevent)229            assert_that(clock_event_result.stdout).described_as(230                f"Expected clockevent name is {clock_event_name}, "231                f"but actual it is {clock_event_result.stdout}."232            ).is_equal_to(clock_event_name)233            # 2. 'Hyper-V clockevent' and 'hrtimer_interrupt' show up times in234            #  /proc/timer_list should equal to cpu count.235            event_handler_name = "hrtimer_interrupt"236            timer_list_result = cat.run("/proc/timer_list", sudo=True)237            lscpu = node.tools[Lscpu]238            core_count = lscpu.get_core_count()239            event_handler_times = timer_list_result.stdout.count(240                f"{event_handler_name}"241            )242            assert_that(event_handler_times).described_as(243                f"Expected {event_handler_name} shown up {core_count} times in output "244                f"of /proc/timer_list, but actual it shows up "245                f"{event_handler_times} times."246            ).is_equal_to(core_count)247            clock_event_times = timer_list_result.stdout.count(f"{clock_event_name}")248            assert_that(clock_event_times).described_as(249                f"Expected {clock_event_name} shown up {core_count} times in output "250                f"of /proc/timer_list, but actual it shows up "251                f"{clock_event_times} times."252            ).is_equal_to(core_count)253            # 3. when cpu count is 1 and cpu type is Intel type, unbind current time254            #  clock event, check current time clock event switch to 'lapic'.255            if CpuType.Intel == lscpu.get_cpu_type() and 1 == core_count:256                cmd_result = node.execute(257                    f"echo {clock_event_name} > {self.unbind_clockevent}",258                    sudo=True,259                    shell=True,260                )261                cmd_result.assert_exit_code()262                clock_event_result_expected = _wait_file_changed(263                    node, self.current_clockevent, "lapic"264                )265                assert_that(clock_event_result_expected).described_as(266                    f"After unbind {clock_event_name}, current clock event should "267                    f"equal to [lapic]."268                ).is_true()269    @TestCaseMetadata(270        description="""271        This test is to check, ntp works properly.272            1. Stop systemd-timesyncd if this service exists.273            2. Set rtc clock to system time.274            3. Restart Ntp service.275            4. Check and set server setting in config file.276            5. Restart Ntp service to reload with new config....
Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.