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

Best Python code snippet using lisa_python
xfstesting.py
Source:xfstesting.py
...239    ) -> None:240        environment = result.environment241        assert environment, "fail to get environment from testresult"242        node = cast(RemoteNode, environment.nodes[0])243        self._check_btrfs_supported(node)244        xfstests = self._install_xfstests(node)245        disk = node.features[Disk]246        data_disks = disk.get_raw_data_disks()247        self._execute_xfstests(248            log_path,249            xfstests,250            result,251            data_disks[0],252            f"{data_disks[0]}1",253            f"{data_disks[0]}2",254            file_system=FileSystem.btrfs,255            test_type=FileSystem.btrfs.name,256            excluded_tests=self.EXCLUDED_TESTS,257        )258    @TestCaseMetadata(259        description="""260        This test case will run generic xfstests testing against261         nvme data disk with xfs type system.262        """,263        timeout=TIME_OUT,264        priority=3,265        requirement=simple_requirement(266            supported_features=[Nvme],267        ),268    )269    def xfstesting_generic_nvme_datadisk_validation(270        self, log_path: Path, result: TestResult271    ) -> None:272        environment = result.environment273        assert environment, "fail to get environment from testresult"274        node = cast(RemoteNode, environment.nodes[0])275        xfstests = self._install_xfstests(node)276        nvme_disk = node.features[Nvme]277        nvme_data_disks = nvme_disk.get_raw_data_disks()278        self._execute_xfstests(279            log_path,280            xfstests,281            result,282            nvme_data_disks[0],283            f"{nvme_data_disks[0]}p1",284            f"{nvme_data_disks[0]}p2",285            excluded_tests=self.EXCLUDED_TESTS,286        )287    @TestCaseMetadata(288        description="""289        This test case will run xfs xfstests testing against290         nvme data disk with xfs type system.291        """,292        timeout=TIME_OUT,293        priority=3,294        requirement=simple_requirement(295            supported_features=[Nvme],296        ),297    )298    def xfstesting_xfs_nvme_datadisk_validation(299        self, log_path: Path, result: TestResult300    ) -> None:301        environment = result.environment302        assert environment, "fail to get environment from testresult"303        node = cast(RemoteNode, environment.nodes[0])304        xfstests = self._install_xfstests(node)305        nvme_disk = node.features[Nvme]306        nvme_data_disks = nvme_disk.get_raw_data_disks()307        self._execute_xfstests(308            log_path,309            xfstests,310            result,311            nvme_data_disks[0],312            f"{nvme_data_disks[0]}p1",313            f"{nvme_data_disks[0]}p2",314            test_type=FileSystem.xfs.name,315            excluded_tests=self.EXCLUDED_TESTS,316        )317    @TestCaseMetadata(318        description="""319        This test case will run ext4 xfstests testing against320         nvme data disk with ext4 type system.321        """,322        timeout=TIME_OUT,323        priority=3,324        requirement=simple_requirement(325            supported_features=[Nvme],326        ),327    )328    def xfstesting_ext4_nvme_datadisk_validation(329        self, log_path: Path, result: TestResult330    ) -> None:331        environment = result.environment332        assert environment, "fail to get environment from testresult"333        node = cast(RemoteNode, environment.nodes[0])334        xfstests = self._install_xfstests(node)335        nvme_disk = node.features[Nvme]336        nvme_data_disks = nvme_disk.get_raw_data_disks()337        self._execute_xfstests(338            log_path,339            xfstests,340            result,341            nvme_data_disks[0],342            f"{nvme_data_disks[0]}p1",343            f"{nvme_data_disks[0]}p2",344            file_system=FileSystem.ext4,345            test_type=FileSystem.ext4.name,346            excluded_tests=self.EXCLUDED_TESTS,347        )348    @TestCaseMetadata(349        description="""350        This test case will run btrfs xfstests testing against351         nvme data disk with btrfs type system.352        """,353        timeout=TIME_OUT,354        priority=3,355        requirement=simple_requirement(356            supported_features=[Nvme],357        ),358    )359    def xfstesting_btrfs_nvme_datadisk_validation(360        self, log_path: Path, result: TestResult361    ) -> None:362        environment = result.environment363        assert environment, "fail to get environment from testresult"364        node = cast(RemoteNode, environment.nodes[0])365        self._check_btrfs_supported(node)366        xfstests = self._install_xfstests(node)367        nvme_disk = node.features[Nvme]368        nvme_data_disks = nvme_disk.get_raw_data_disks()369        self._execute_xfstests(370            log_path,371            xfstests,372            result,373            nvme_data_disks[0],374            f"{nvme_data_disks[0]}p1",375            f"{nvme_data_disks[0]}p2",376            file_system=FileSystem.btrfs,377            test_type=FileSystem.btrfs.name,378            excluded_tests=self.EXCLUDED_TESTS,379        )380    @TestCaseMetadata(381        description="""382        This test case will run cifs xfstests testing against383         azure file share.384        """,385        requirement=simple_requirement(386            min_core_count=16,387            supported_platform_type=[AZURE],388        ),389        timeout=TIME_OUT,390        priority=3,391    )392    def xfstesting_azure_file_share_validation(393        self, log: Logger, log_path: Path, result: TestResult394    ) -> None:395        environment = result.environment396        assert environment, "fail to get environment from testresult"397        assert isinstance(environment.platform, AzurePlatform)398        node = cast(RemoteNode, environment.nodes[0])399        if not node.tools[KernelConfig].is_enabled("CONFIG_CIFS"):400            raise UnsupportedDistroException(401                node.os, "current distro not enable cifs module."402            )403        xfstests = self._install_xfstests(node)404        version = _get_smb_version(node)405        fstab_info = (406            f"nofail,vers={version},credentials=/etc/smbcredentials/lisa.cred"407            ",dir_mode=0777,file_mode=0777,serverino"408        )409        mount_opts = (410            f"-o vers={version},credentials=/etc/smbcredentials/lisa.cred"411            ",dir_mode=0777,file_mode=0777,serverino"412        )413        platform = environment.platform414        information = environment.get_information()415        resource_group_name = information["resource_group_name"]416        location = information["location"]417        random_str = generate_random_chars(string.ascii_lowercase + string.digits, 10)418        storage_account_name = f"lisasc{random_str}"419        file_share_name = f"lisa{random_str}fs"420        scratch_name = f"lisa{random_str}scratch"421        fs_url_dict: Dict[str, str] = {file_share_name: "", scratch_name: ""}422        try:423            check_or_create_storage_account(424                credential=platform.credential,425                subscription_id=platform.subscription_id,426                account_name=storage_account_name,427                resource_group_name=resource_group_name,428                location=location,429                log=log,430            )431            for share_name, _ in fs_url_dict.items():432                fs_url_dict[share_name] = get_or_create_file_share(433                    credential=platform.credential,434                    subscription_id=platform.subscription_id,435                    account_name=storage_account_name,436                    file_share_name=share_name,437                    resource_group_name=resource_group_name,438                    log=log,439                )440            account_credential = get_storage_credential(441                credential=platform.credential,442                subscription_id=platform.subscription_id,443                account_name=storage_account_name,444                resource_group_name=resource_group_name,445            )446            _prepare_azure_file_share(447                node,448                account_credential,449                {450                    _test_folder: fs_url_dict[file_share_name],451                    _scratch_folder: fs_url_dict[scratch_name],452                },453                fstab_info,454            )455            self._execute_xfstests(456                log_path,457                xfstests,458                result,459                test_dev=fs_url_dict[file_share_name],460                scratch_dev=fs_url_dict[scratch_name],461                excluded_tests=self.EXCLUDED_TESTS,462                mount_opts=mount_opts,463            )464        finally:465            # clean up resources after testing.466            for share_name in [file_share_name, scratch_name]:467                delete_file_share(468                    credential=platform.credential,469                    subscription_id=platform.subscription_id,470                    account_name=storage_account_name,471                    file_share_name=share_name,472                    resource_group_name=resource_group_name,473                    log=log,474                )475            delete_storage_account(476                credential=platform.credential,477                subscription_id=platform.subscription_id,478                account_name=storage_account_name,479                resource_group_name=resource_group_name,480                log=log,481            )482            # revert file into original status after testing.483            node.execute("cp -f /etc/fstab_cifs /etc/fstab", sudo=True)484    def after_case(self, log: Logger, **kwargs: Any) -> None:485        node: Node = kwargs.pop("node")486        for path in [487            "/dev/mapper/delay-test",488            "/dev/mapper/huge-test",489            "/dev/mapper/huge-test-zero",490        ]:491            if 0 == node.execute(f"ls -lt {path}", sudo=True).exit_code:492                node.execute(f"dmsetup remove {path}", sudo=True)493    def _execute_xfstests(494        self,495        log_path: Path,496        xfstests: Xfstests,497        result: TestResult,498        data_disk: str = "",499        test_dev: str = "",500        scratch_dev: str = "",501        file_system: FileSystem = FileSystem.xfs,502        test_type: str = "generic",503        excluded_tests: str = "",504        mount_opts: str = "",505    ) -> None:506        environment = result.environment507        assert environment, "fail to get environment from testresult"508        node = cast(RemoteNode, environment.nodes[0])509        # TODO: will include generic/641 once the kernel contains below fix.510        # exclude this case generic/641 temporarily511        # it will trigger oops on RHEL8.3/8.4, VM will reboot512        # lack of commit 5808fecc572391867fcd929662b29c12e6d08d81513        if (514            test_type == "generic"515            and isinstance(node.os, Redhat)516            and node.os.information.version >= "8.3.0"517        ):518            excluded_tests += " generic/641"519        # xfs/081 case will hung for long time520        if isinstance(node.os, CBLMariner):521            excluded_tests += " xfs/081"522        # prepare data disk when xfstesting target is data disk523        if data_disk:524            _prepare_data_disk(525                node,526                data_disk,527                {test_dev: _test_folder, scratch_dev: _scratch_folder},528                file_system=file_system,529            )530        xfstests.set_local_config(531            scratch_dev,532            _scratch_folder,533            test_dev,534            _test_folder,535            test_type,536            mount_opts,537        )538        xfstests.set_excluded_tests(excluded_tests)539        output = xfstests.run_test(test_type, self.TIME_OUT)540        xfstests.check_test_results(output, log_path, test_type, result, data_disk)541    def _install_xfstests(self, node: Node) -> Xfstests:542        try:543            xfstests = node.tools[Xfstests]544            return xfstests545        except UnsupportedDistroException as identifier:546            raise SkippedException(identifier)547    def _check_btrfs_supported(self, node: Node) -> None:548        if not node.tools[KernelConfig].is_enabled("CONFIG_BTRFS_FS"):...
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.