diff --git a/pytest/pyproject.toml b/pytest/pyproject.toml index 4786bb8..d43675d 100644 --- a/pytest/pyproject.toml +++ b/pytest/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "testsolar-pytestx" -version = "0.1.69" +version = "0.1.72" description = "Pytest tool for TestSolar" authors = [ {name = "asiazhang2002", email = "asiazhang2002@gmail.com"}, diff --git a/pytest/src/testsolar_pytestx/collector.py b/pytest/src/testsolar_pytestx/collector.py index b499094..5214403 100644 --- a/pytest/src/testsolar_pytestx/collector.py +++ b/pytest/src/testsolar_pytestx/collector.py @@ -3,8 +3,8 @@ import traceback from collections import defaultdict from pathlib import Path -from typing import BinaryIO, Sequence, Optional, List, Dict, Union, Callable - +from typing import BinaryIO, Sequence, Optional, List, Dict, Union, Callable, Set +from loguru import logger from pytest import Item, Collector try: @@ -84,6 +84,22 @@ def collect_testcases( LoadErrors=[], ) + # 检查是否启用 RUN_ALL_CASES 模式 + run_all_cases = os.getenv("TESTSOLAR_TTP_RUNALLCASES", "").lower() in ["1", "true"] + if run_all_cases: + logger.info("[Load] 执行一条命令运行当前项目的所有用例") + load_result.Tests.extend([TestCase(Name=selector, Attributes={}) for selector in entry_param.TestSelectors]) + reporter = FileReporter(Path(entry_param.FileReportPath)) + reporter.report_load_result(load_result) + return + + # 检查是否启用 FILEMODE 模式 + file_mode = os.getenv("TESTSOLAR_TTP_FILEEXECUTEMODE", "").lower() in ["1", "true"] + if file_mode: + logger.info("[Load] 以文件模式运行当前用例") + collect_testcases_file_mode(entry_param, load_result) + return + valid_selectors, load_errors = filter_invalid_selector_path( workspace=entry_param.ProjectPath, selectors=entry_param.TestSelectors, @@ -116,11 +132,11 @@ def collect_testcases( append_extra_args(args) args.extend(testcase_list) - print(f"[Load] try to collect testcases: {args}") + logger.info(f"[Load] try to collect testcases: {args}") _, captured_stderr, exit_code = pytest_main_with_output(args=args, plugin=my_plugin) if exit_code != 0: # 若加载用例失败，则将本批次的用例结果统一作为loaderror上报，并将标准错误流作为用例错误日志上报 - print(f"[Warn][Load] collect testcases exit_code: {exit_code}") + logger.warning(f"[Warn][Load] collect testcases exit_code: {exit_code}") if len(my_plugin.collected) == 0 and len(my_plugin.errors.items()) == 0: for selector in valid_selectors: load_result.LoadErrors.append( @@ -147,13 +163,199 @@ def collect_testcases( ) ) - print(f"[Load] collect testcase count: {len(load_result.Tests)}") - print(f"[Load] collect load error count: {len(load_result.LoadErrors)}") + logger.info(f"[Load] collect testcase count: {len(load_result.Tests)}") + logger.info(f"[Load] collect load error count: {len(load_result.LoadErrors)}") reporter = FileReporter(Path(entry_param.FileReportPath)) reporter.report_load_result(load_result) +def collect_testcases_file_mode(entry_param: EntryParam, load_result: LoadResult) -> None: + """ + 文件模式：只解析到文件层级，不解析里面的类和方法 + 扫描传进来的testselector对应的文件： + - 如果是tests，则扫描tests目录下所有pytest相关的测试文件 + - 如果是.，则扫描整个项目目录下符合pytest规范的文件 + - 排除隐藏目录、文件和缓存文件 + """ + valid_selectors, load_errors = filter_invalid_selector_path( + workspace=entry_param.ProjectPath, + selectors=entry_param.TestSelectors, + ) + + load_result.LoadErrors.extend(load_errors) + + # 收集所有测试文件 + test_files = set() + + for selector in valid_selectors: + # 移除数据驱动部分 + if CASE_DRIVE_SEPARATOR in selector: + selector = selector.split(CASE_DRIVE_SEPARATOR)[0] + + # 提取路径部分（去掉查询参数） + path_part, _, _ = selector.partition("?") + + # 构建完整路径 + project_path_obj = Path(entry_param.ProjectPath) + full_path = project_path_obj / path_part + + if full_path.is_file(): + # 如果是文件，直接添加 + if _is_pytest_test_file(str(full_path)): + test_files.add(path_part) + elif full_path.is_dir(): + # 如果是目录，扫描目录下的所有测试文件 + discovered_files = _scan_pytest_files(str(full_path), entry_param.ProjectPath) + test_files.update(discovered_files) + elif path_part == ".": + # 如果是根目录，扫描整个项目 + discovered_files = _scan_pytest_files(entry_param.ProjectPath, entry_param.ProjectPath) + test_files.update(discovered_files) + else: + # 尝试转换为pytest路径处理（兼容原有逻辑） + pytest_path = selector_to_pytest(test_selector=selector) + if pytest_path: + # 提取文件路径部分（去掉类名和方法名） + file_path = pytest_path.split("::")[0] + full_file_path = project_path_obj / file_path + if full_file_path.exists() and _is_pytest_test_file(str(full_file_path)): + test_files.add(file_path) + + # 为每个测试文件创建一个测试用例 + project_path_obj = Path(entry_param.ProjectPath) + for file_path in sorted(test_files): + # 检查文件是否存在 + full_file_path = project_path_obj / file_path + if full_file_path.exists(): + # 将文件路径转换为选择器格式 + load_result.Tests.append(TestCase(Name=file_path, Attributes={})) + else: + load_result.LoadErrors.append( + LoadError( + name=file_path, + message=f"Test file not found: {full_file_path}", + ) + ) + + print(f"[Load] File mode - collected {len(load_result.Tests)} test files") + print(f"[Load] File mode - load error count: {len(load_result.LoadErrors)}") + + reporter = FileReporter(Path(entry_param.FileReportPath)) + reporter.report_load_result(load_result) + + +def _is_pytest_test_file(file_path: str) -> bool: + """ + 判断文件是否是pytest测试文件 + pytest默认的测试文件命名规则： + 1. test_*.py - 以test_开头的.py文件 + 2. *_test.py - 以_test.py结尾的.py文件（但不能只是_test.py） + """ + file_path_obj = Path(file_path) + if file_path_obj.suffix != ".py": + return False + + filename = file_path_obj.name + + # 规则1: 以test_开头 + if filename.startswith("test_"): + return True + + # 规则2: 以_test.py结尾，但不能只是_test.py + if filename.endswith("_test.py") and filename != "_test.py": + return True + + return False + + +def _scan_pytest_files(scan_path: str, project_path: str) -> Set[str]: + """ + 扫描指定路径下的所有pytest测试文件 + 排除隐藏目录、文件和缓存文件 + """ + test_files = set() + + # 需要排除的目录和文件模式 + exclude_dirs = { + "__pycache__", + ".pytest_cache", + ".git", + ".svn", + ".hg", + "node_modules", + ".venv", + "venv", + ".env", + "env", + ".tox", + "build", + "dist", + ".coverage", + "htmlcov", + ".mypy_cache", + ".idea", + ".vscode", + } + + exclude_patterns = { + ".pyc", + ".pyo", + ".pyd", + ".so", + ".egg-info", + ".coverage", + ".DS_Store", + "Thumbs.db", + } + + try: + scan_path_obj = Path(scan_path) + project_path_obj = Path(project_path) + + def should_exclude_dir(dir_path: Path) -> bool: + """判断是否应该排除目录""" + dir_name = dir_path.name + return ( + dir_name.startswith(".") # 隐藏目录 + or dir_name in exclude_dirs # 特定排除目录 + or dir_name.startswith("__pycache__") # Python缓存目录 + ) + + def scan_directory(current_path: Path) -> None: + """递归扫描目录""" + try: + for item in current_path.iterdir(): + if item.is_dir(): + # 检查是否应该排除此目录 + if not should_exclude_dir(item): + scan_directory(item) + elif item.is_file(): + # 跳过隐藏文件和特定扩展名的文件 + if item.name.startswith("."): + continue + + if any(item.name.endswith(pattern) for pattern in exclude_patterns): + continue + + # 检查是否是pytest测试文件 + if _is_pytest_test_file(str(item)): + # 计算相对于项目根目录的路径 + rel_path = item.relative_to(project_path_obj) + # 统一使用正斜杠 + rel_path_str = str(rel_path).replace("\\", "/") + test_files.add(rel_path_str) + except (OSError, PermissionError) as e: + print(f"[Warning] Failed to access directory {current_path}: {e}") + + scan_directory(scan_path_obj) + + except (OSError, PermissionError) as e: + print(f"[Warning] Failed to scan directory {scan_path}: {e}") + + return test_files + + def show_workspace_files(workdir: str) -> None: print() print(f"Workspace [{workdir}] files:") diff --git a/pytest/src/testsolar_pytestx/executor.py b/pytest/src/testsolar_pytestx/executor.py index 8419223..78deaed 100644 --- a/pytest/src/testsolar_pytestx/executor.py +++ b/pytest/src/testsolar_pytestx/executor.py @@ -285,7 +285,9 @@ def run_testcases( enable_allure = check_allure_enable() if enable_allure: print("Start allure test") - allure_dir = os.path.join(entry.ProjectPath, "allure_results") + # 通过给allure_results增加当前进程pid来生成唯一的路径，避免并发冲突 + process_id = os.getpid() + allure_dir = os.path.join(entry.ProjectPath, f"allure_results_{process_id}") args.append("--alluredir={}".format(allure_dir)) initialization_allure_dir(allure_dir) @@ -319,7 +321,10 @@ def run_testcases( # 注意：传递给pytest中的用例必须在执行时能找到，否则pytest会报错 # TODO: pytest执行出错时，将用例都设置为IGNORED，并设置错误原因 args.extend( - [os.path.join(entry.ProjectPath, selector_to_pytest(it)) for it in valid_selectors] + [ + os.path.join(entry.ProjectPath, selector_to_pytest(it)) if it != "." else "." + for it in valid_selectors + ] ) logger.info(f"Pytest run args: {args}") my_plugin = PytestExecutor(reporter=reporter, comment_fields=case_comment_fields) diff --git a/pytest/src/testsolar_pytestx/extend/allure_extend.py b/pytest/src/testsolar_pytestx/extend/allure_extend.py index 493a9c4..4c19dff 100644 --- a/pytest/src/testsolar_pytestx/extend/allure_extend.py +++ b/pytest/src/testsolar_pytestx/extend/allure_extend.py @@ -80,9 +80,16 @@ def initialization_allure_dir(allure_dir: str) -> None: logger.info(f"Initializing Allure directory: {allure_dir}") if Path(allure_dir).exists(): logger.info(f"Directory {allure_dir} exists. Removing it.") - shutil.rmtree(allure_dir) - os.makedirs(allure_dir, exist_ok=True) - logger.info(f"Directory {allure_dir} created.") + try: + shutil.rmtree(allure_dir) + except (OSError, FileNotFoundError) as e: + logger.warning(f"Failed to remove existing directory {allure_dir}: {e}") + + try: + os.makedirs(allure_dir, exist_ok=True) + logger.info(f"Directory {allure_dir} created.") + except OSError as e: + logger.error(f"Failed to create directory {allure_dir}: {e}") def generate_allure_results( @@ -110,8 +117,19 @@ def generate_allure_results( ) logger.debug(f"Formatted test case name: {testcase_format_name}") + # 处理参数化测试用例：提取基础名称进行匹配 + testcase_base_name = ( + testcase_format_name.split("[")[0] + if "[" in testcase_format_name + else testcase_format_name + ) + if full_name != testcase_format_name: - if full_name in testcase_format_name and testcase_format_name.endswith( + if full_name == testcase_base_name: + logger.info( + f"Test case {testcase_format_name} is a parameterized case, matched base name: {full_name}" + ) + elif full_name in testcase_format_name and testcase_format_name.endswith( allure_data.name ): logger.info( diff --git a/pytest/tests/test_collector.py b/pytest/tests/test_collector.py index e4df2f2..9624eb9 100644 --- a/pytest/tests/test_collector.py +++ b/pytest/tests/test_collector.py @@ -7,7 +7,12 @@ from testsolar_testtool_sdk.model.load import LoadResult from testsolar_testtool_sdk.file_reader import read_file_load_result -from src.testsolar_pytestx.collector import collect_testcases +from src.testsolar_pytestx.collector import ( + collect_testcases, + collect_testcases_file_mode, + _is_pytest_test_file, + _scan_pytest_files, +) class CollectorTest(unittest.TestCase): @@ -251,3 +256,165 @@ def test_collect_testcases_with_coding_testcase_id(self): "test_coding_id.py?test_eval/[2+4-6]", ) self.assertEqual(re.Tests[1].Attributes["coding_testcase_id"], "789") + + def test_collect_testcases_file_mode_with_directory(self): + """测试文件模式：扫描目录""" + with tempfile.TemporaryDirectory() as tmpdir: + report_file = Path(tmpdir) / "result.json" + entry = EntryParam( + TaskId="aa", + ProjectPath=self.testdata_dir, + TestSelectors=["aa/bb"], # 扫描子目录 + FileReportPath=str(report_file), + ) + + load_result = LoadResult(Tests=[], LoadErrors=[]) + collect_testcases_file_mode(entry, load_result) + + # 验证结果 + self.assertEqual(len(load_result.Tests), 1) + self.assertEqual(len(load_result.LoadErrors), 0) + self.assertEqual(load_result.Tests[0].Name, "aa/bb/cc/test_in_sub_class.py") + + def test_collect_testcases_file_mode_with_root_directory(self): + """测试文件模式：扫描根目录""" + with tempfile.TemporaryDirectory() as tmpdir: + report_file = Path(tmpdir) / "result.json" + entry = EntryParam( + TaskId="aa", + ProjectPath=self.testdata_dir, + TestSelectors=["."], # 扫描整个项目 + FileReportPath=str(report_file), + ) + + load_result = LoadResult(Tests=[], LoadErrors=[]) + collect_testcases_file_mode(entry, load_result) + + # 验证结果 - 应该找到多个测试文件 + self.assertGreater(len(load_result.Tests), 5) + + # 验证包含预期的测试文件 + test_names = [test.Name for test in load_result.Tests] + self.assertIn("test_normal_case.py", test_names) + self.assertIn("test_data_drive.py", test_names) + + def test_collect_testcases_file_mode_with_specific_file(self): + """测试文件模式：指定具体文件""" + with tempfile.TemporaryDirectory() as tmpdir: + report_file = Path(tmpdir) / "result.json" + entry = EntryParam( + TaskId="aa", + ProjectPath=self.testdata_dir, + TestSelectors=["test_normal_case.py"], # 指定具体文件 + FileReportPath=str(report_file), + ) + + load_result = LoadResult(Tests=[], LoadErrors=[]) + collect_testcases_file_mode(entry, load_result) + + # 验证结果 + self.assertEqual(len(load_result.Tests), 1) + self.assertEqual(len(load_result.LoadErrors), 0) + self.assertEqual(load_result.Tests[0].Name, "test_normal_case.py") + + def test_collect_testcases_file_mode_with_nonexistent_file(self): + """测试文件模式：不存在的文件""" + with tempfile.TemporaryDirectory() as tmpdir: + report_file = Path(tmpdir) / "result.json" + entry = EntryParam( + TaskId="aa", + ProjectPath=self.testdata_dir, + TestSelectors=["nonexistent_test.py"], + FileReportPath=str(report_file), + ) + + load_result = LoadResult(Tests=[], LoadErrors=[]) + collect_testcases_file_mode(entry, load_result) + + # 验证结果 - 应该有加载错误 + self.assertEqual(len(load_result.Tests), 0) + self.assertEqual(len(load_result.LoadErrors), 1) + + def test_is_pytest_test_file(self): + """测试pytest测试文件识别函数""" + # 应该识别为测试文件的情况 + test_files = [ + "test_example.py", + "test_normal_case.py", + "utils_test.py", + "my_module_test.py", + "aa/bb/test_in_sub_class.py", + ] + + for file_path in test_files: + with self.subTest(file_path=file_path): + self.assertTrue(_is_pytest_test_file(file_path)) + + # 不应该识别为测试文件的情况 + non_test_files = [ + "example.py", + "utils.py", + "config.json", + "README.md", + "_test.py", # 只有下划线开头不算 + "test.txt", # 不是.py文件 + "testfile.py", # 不符合test_*.py格式 + "filetest.py", # 不符合*_test.py格式 + ] + + for file_path in non_test_files: + with self.subTest(file_path=file_path): + self.assertFalse(_is_pytest_test_file(file_path)) + + def test_scan_pytest_files(self): + """测试扫描pytest文件函数""" + # 扫描测试数据目录 + test_files = _scan_pytest_files(self.testdata_dir, self.testdata_dir) + + # 验证结果 + self.assertIsInstance(test_files, set) + self.assertGreater(len(test_files), 0) + + # 验证包含预期的测试文件 + expected_files = { + "test_normal_case.py", + "test_data_drive.py", + "test_coding_id.py", + "aa/bb/cc/test_in_sub_class.py", + } + + for expected_file in expected_files: + self.assertIn(expected_file, test_files) + + # 验证不包含非测试文件 + for file_path in test_files: + self.assertTrue(_is_pytest_test_file(file_path)) + + def test_scan_pytest_files_excludes_hidden_and_cache_dirs(self): + """测试扫描pytest文件时排除隐藏目录和缓存目录""" + # 创建临时目录结构进行测试 + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + + # 创建测试文件结构 + (tmpdir_path / "test_valid.py").write_text("# valid test file") + + # 创建隐藏目录和缓存目录 + hidden_dir = tmpdir_path / ".hidden" + hidden_dir.mkdir(parents=True) + (hidden_dir / "test_hidden.py").write_text("# hidden test file") + + cache_dir = tmpdir_path / "__pycache__" + cache_dir.mkdir(parents=True) + (cache_dir / "test_cache.py").write_text("# cache test file") + + pytest_cache_dir = tmpdir_path / ".pytest_cache" + pytest_cache_dir.mkdir(parents=True) + (pytest_cache_dir / "test_pytest_cache.py").write_text("# pytest cache test file") + + # 扫描文件 + test_files = _scan_pytest_files(str(tmpdir_path), str(tmpdir_path)) + + # 验证只包含有效的测试文件 + self.assertEqual(len(test_files), 1) + self.assertIn("test_valid.py", test_files) diff --git a/pytest/testtool.yaml b/pytest/testtool.yaml index 73469a0..ab48e63 100644 --- a/pytest/testtool.yaml +++ b/pytest/testtool.yaml @@ -2,7 +2,7 @@ schemaVersion: 1.0 name: pytest nameZh: pytest自动化测试 lang: python -version: '0.1.69' +version: '0.1.72' langType: INTERPRETED description: |- pytest是一个成熟的全功能Python测试工具，可以帮助您编写更好的程序。此测试工具允许您在TestSolar上运行pytest。 @@ -68,6 +68,16 @@ parameterDefs: # 用户 use 这个测试工具时可以配置的 with 参数 通过开启该选项可以避免对反斜杠进行编码 default: 'false' inputWidget: switch + - name: runAllCases + value: 单命令运行全部用例 + desc: 默认情况下插件会基于解析所有方法用例来执行，通过开启该选项可以单命令运行全部用例 + default: 'false' + inputWidget: switch + - name: fileExecuteMode + value: 文件执行模式 + desc: 默认情况下插件会解析全部方法用例，通过开启该选项可以只执行文件级别的用例 + default: 'false' + inputWidget: switch entry: load: "python3 /testtools/pytest/src/load.py $1" run: "python3 /testtools/pytest/src/run.py $1" diff --git a/pytest/uv.lock b/pytest/uv.lock index c274e10..aa65191 100644 --- a/pytest/uv.lock +++ b/pytest/uv.lock @@ -1,5 +1,4 @@ version = 1 -revision = 1 requires-python = ">=3.7" [[package]] @@ -362,7 +361,7 @@ name = "portalocker" version = "2.7.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "pywin32", marker = "sys_platform == 'win32'" }, + { name = "pywin32", marker = "platform_system == 'Windows'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/1f/f8/969e6f280201b40b31bcb62843c619f343dcc351dff83a5891530c9dd60e/portalocker-2.7.0.tar.gz", hash = "sha256:032e81d534a88ec1736d03f780ba073f047a06c478b06e2937486f334e955c51", size = 20183 } wheels = [ @@ -463,7 +462,7 @@ wheels = [ [[package]] name = "testsolar-pytestx" -version = "0.1.69" +version = "0.1.72" source = { virtual = "." } dependencies = [ { name = "coverage" },