How to use get_xfstests_path method in lisa

Best Python code snippet using lisa_python

xfstests.py

Source:xfstests.py

...141            f"-g {test_type}/quick -E exclude.txt",142            sudo=True,143            shell=True,144            force_run=True,145            cwd=self.get_xfstests_path(),146            timeout=timeout,147        )148        return cmd_result.stdout149    def _initialize(self, *args: Any, **kwargs: Any) -> None:150        super()._initialize(*args, **kwargs)151        self._code_path = self.get_tool_path(use_global=True) / "xfstests-dev"152    def _install_dep(self) -> None:153        posix_os: Posix = cast(Posix, self.node.os)154        # install dependency packages155        package_list = []156        package_list.extend(self.common_dep)157        if isinstance(self.node.os, Redhat):158            package_list.extend(self.fedora_dep)159        elif isinstance(self.node.os, Debian):160            if (161                isinstance(self.node.os, Ubuntu)162                and self.node.os.information.version < "18.4.0"163            ):164                raise UnsupportedDistroException(self.node.os)165            package_list.extend(self.debian_dep)166        elif isinstance(self.node.os, Suse):167            package_list.extend(self.suse_dep)168        elif isinstance(self.node.os, CBLMariner):169            package_list.extend(self.mariner_dep)170        else:171            raise LisaException(172                f"Current distro {self.node.os.name} doesn't support xfstests."173            )174        # if install the packages in one command, the remain available packages can't175        # be installed if one of packages is not available in that distro,176        # so here install it one by one177        for package in list(package_list):178            # to make code simple, put all packages needed by one distro in one list.179            # the package name may be different for the different sku of the180            #  same distro. so, install it when the package exists in the repo.181            if posix_os.is_package_in_repo(package):182                posix_os.install_packages(package)183        # fix compile issue on RHEL/CentOS 7.x184        if (185            isinstance(self.node.os, Redhat)186            and self.node.os.information.version < "8.0.0"187        ):188            posix_os.install_packages(packages="centos-release-scl")189            posix_os.install_packages(190                "http://mirror.centos.org/centos/7/os/x86_64/Packages/"191                "xfsprogs-devel-4.5.0-22.el7.x86_64.rpm"192            )193            posix_os.install_packages(194                packages="devtoolset-7-gcc*", extra_args=["--skip-broken"]195            )196            self.node.execute("rm -f /bin/gcc", sudo=True, shell=True)197            self.node.execute(198                "ln -s /opt/rh/devtoolset-7/root/usr/bin/gcc /bin/gcc",199                sudo=True,200                shell=True,201            )202        # fix compile issue on SLES12SP5203        if (204            isinstance(self.node.os, Suse)205            and self.node.os.information.version < "15.0.0"206        ):207            posix_os.install_packages(packages="gcc5")208            self.node.execute("rm -rf /usr/bin/gcc", sudo=True, shell=True)209            self.node.execute(210                "ln -s /usr/bin/gcc-5 /usr/bin/gcc",211                sudo=True,212                shell=True,213            )214    def _add_test_users(self) -> None:215        # prerequisite for xfstesting216        # these users are used in the test code217        # refer https://github.com/kdave/xfstests218        self.node.execute("useradd -m fsgqa", sudo=True)219        self.node.execute("groupadd fsgqa", sudo=True)220        self.node.execute("useradd 123456-fsgqa", sudo=True)221        self.node.execute("useradd fsgqa2", sudo=True)222    def _install(self) -> bool:223        self._install_dep()224        self._add_test_users()225        tool_path = self.get_tool_path(use_global=True)226        git = self.node.tools[Git]227        git.clone(self.repo, tool_path)228        make = self.node.tools[Make]229        code_path = tool_path.joinpath("xfstests-dev")230        make.make_install(code_path)231        return True232    def get_xfstests_path(self) -> PurePath:233        return self._code_path234    def set_local_config(235        self,236        scratch_dev: str,237        scratch_mnt: str,238        test_dev: str,239        test_folder: str,240        test_type: str,241        mount_opts: str = "",242    ) -> None:243        xfstests_path = self.get_xfstests_path()244        config_path = xfstests_path.joinpath("local.config")245        if self.node.shell.exists(config_path):246            self.node.shell.remove(config_path)247        if "generic" == test_type:248            test_type = "xfs"249        echo = self.node.tools[Echo]250        if mount_opts:251            content = "\n".join(252                [253                    "[cifs]",254                    "FSTYP=cifs",255                    f"TEST_FS_MOUNT_OPTS=''{mount_opts}''",256                    f"MOUNT_OPTIONS=''{mount_opts}''",257                ]258            )259        else:260            content = "\n".join(261                [262                    f"[{test_type}]",263                    f"FSTYP={test_type}",264                ]265            )266        echo.write_to_file(content, config_path, append=True)267        content = "\n".join(268            [269                f"SCRATCH_DEV={scratch_dev}",270                f"SCRATCH_MNT={scratch_mnt}",271                f"TEST_DEV={test_dev}",272                f"TEST_DIR={test_folder}",273            ]274        )275        echo.write_to_file(content, config_path, append=True)276    def set_excluded_tests(self, exclude_tests: str) -> None:277        if exclude_tests:278            xfstests_path = self.get_xfstests_path()279            exclude_file_path = xfstests_path.joinpath("exclude.txt")280            if self.node.shell.exists(exclude_file_path):281                self.node.shell.remove(exclude_file_path)282            echo = self.node.tools[Echo]283            echo.write_to_file(exclude_tests, exclude_file_path)284    def create_send_subtest_msg(285        self,286        test_result: TestResult,287        raw_message: str,288        test_type: str,289        data_disk: str,290    ) -> None:291        environment = test_result.environment292        assert environment, "fail to get environment from testresult"293        all_cases_match = self.__all_cases_pattern.match(raw_message)294        assert all_cases_match, "fail to find run cases from xfstests output"295        all_cases = (all_cases_match.group("all_cases")).split()296        not_run_cases: List[str] = []297        fail_cases: List[str] = []298        not_run_match = self.__not_run_cases_pattern.match(raw_message)299        if not_run_match:300            not_run_cases = (not_run_match.group("not_run_cases")).split()301        fail_match = self.__fail_cases_pattern.match(raw_message)302        if fail_match:303            fail_cases = (fail_match.group("fail_cases")).split()304        pass_cases = [305            x for x in all_cases if x not in not_run_cases and x not in not_run_cases306        ]307        results: List[XfstestsResult] = []308        for case in fail_cases:309            results.append(XfstestsResult(case, TestStatus.FAILED))310        for case in pass_cases:311            results.append(XfstestsResult(case, TestStatus.PASSED))312        for case in not_run_cases:313            results.append(XfstestsResult(case, TestStatus.SKIPPED))314        for result in results:315            # create test result message316            info: Dict[str, Any] = {}317            info["information"] = {}318            info["information"]["test_type"] = test_type319            info["information"]["data_disk"] = data_disk320            subtest_message = create_test_result_message(321                SubTestMessage,322                test_result.id_,323                environment,324                result.name,325                result.status,326                other_fields=info,327            )328            # notify subtest result329            notifier.notify(subtest_message)330    def check_test_results(331        self,332        raw_message: str,333        log_path: Path,334        test_type: str,335        result: TestResult,336        data_disk: str = "",337    ) -> None:338        self.create_send_subtest_msg(result, raw_message, test_type, data_disk)339        xfstests_path = self.get_xfstests_path()340        results_path = xfstests_path / "results/check.log"341        if not self.node.shell.exists(results_path):342            raise LisaException(343                f"Result path {results_path} doesn't exist, please check testing runs"344                " well or not."345            )346        results = self.node.tools[Cat].run(str(results_path), force_run=True, sudo=True)347        results.assert_exit_code()348        pass_match = self.__all_pass_pattern.match(results.stdout)349        if pass_match:350            pass_count = pass_match.group("pass_count")351            self._log.debug(352                f"All pass in xfstests, total pass case count is {pass_count}."353            )354            return355        fail_match = self.__fail_pattern.match(results.stdout)356        assert fail_match357        fail_count = fail_match.group("fail_count")358        total_count = fail_match.group("total_count")359        fail_cases_match = self.__fail_cases_pattern.match(results.stdout)360        assert fail_cases_match361        fail_info = ""362        fail_cases = fail_cases_match.group("fail_cases")363        for fail_case in fail_cases.split():364            fail_info += find_patterns_in_lines(365                raw_message, [re.compile(f".*{fail_case}.*$", re.MULTILINE)]366            )[0][0]367        self.save_xfstests_log(fail_cases.split(), log_path, test_type)368        results_folder = xfstests_path / "results/"369        self.node.execute(f"rm -rf {results_folder}", sudo=True)370        raise LisaException(371            f"Fail {fail_count} cases of total {total_count}, fail cases"372            f" {fail_cases}, details {fail_info}, please investigate."373        )374    def save_xfstests_log(375        self, fail_cases: List[str], log_path: Path, test_type: str376    ) -> None:377        if "generic" == test_type:378            test_type = "xfs"379        xfstests_path = self.get_xfstests_path()380        self.node.shell.copy_back(381            xfstests_path / "results/check.log",382            log_path / "xfstests/check.log",383        )384        for fail_case in fail_cases:385            file_name = f"results/{test_type}/{fail_case}.out.bad"386            result_path = xfstests_path / file_name387            if self.node.shell.exists(result_path):388                self.node.shell.copy_back(result_path, log_path / file_name)389            else:390                self._log.debug(f"{file_name} doesn't exist.")391            file_name = f"results/{test_type}/{fail_case}.full"392            result_path = xfstests_path / file_name393            if self.node.shell.exists(result_path):...

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.