44from collections import defaultdict
55from pathlib import Path
66from typing import BinaryIO , Sequence , Optional , List , Dict , Union , Callable
7-
7+ from loguru import logger
88from pytest import Item , Collector
99
1010try :
@@ -84,6 +84,22 @@ def collect_testcases(
8484 LoadErrors = [],
8585 )
8686
87+ # 检查是否启用 RUN_ALL_CASES 模式
88+ run_all_cases = os .getenv ("TESTSOLAR_TTP_RUN_ALL_CASES" , "" ).lower () in ["1" , "true" ]
89+ if run_all_cases :
90+ logger .info ("[Load] 执行一条命令运行当前项目的所有用例" )
91+ load_result .Tests .append (TestCase (Name = "." , Attributes = {}))
92+ reporter = FileReporter (Path (entry_param .FileReportPath ))
93+ reporter .report_load_result (load_result )
94+ return
95+
96+ # 检查是否启用 FILEMODE 模式
97+ file_mode = os .getenv ("TESTSOLAR_TTP_FILEMODE" , "" ).lower () in ["1" , "true" ]
98+ if file_mode :
99+ logger .info ("[Load] 以文件模式运行当前用例" )
100+ collect_testcases_file_mode (entry_param , load_result )
101+ return
102+
87103 valid_selectors , load_errors = filter_invalid_selector_path (
88104 workspace = entry_param .ProjectPath ,
89105 selectors = entry_param .TestSelectors ,
@@ -116,11 +132,11 @@ def collect_testcases(
116132 append_extra_args (args )
117133
118134 args .extend (testcase_list )
119- print (f"[Load] try to collect testcases: { args } " )
135+ logger . info (f"[Load] try to collect testcases: { args } " )
120136 _ , captured_stderr , exit_code = pytest_main_with_output (args = args , plugin = my_plugin )
121137 if exit_code != 0 :
122138 # 若加载用例失败,则将本批次的用例结果统一作为loaderror上报,并将标准错误流作为用例错误日志上报
123- print (f"[Warn][Load] collect testcases exit_code: { exit_code } " )
139+ logger . warning (f"[Warn][Load] collect testcases exit_code: { exit_code } " )
124140 if len (my_plugin .collected ) == 0 and len (my_plugin .errors .items ()) == 0 :
125141 for selector in valid_selectors :
126142 load_result .LoadErrors .append (
@@ -147,13 +163,199 @@ def collect_testcases(
147163 )
148164 )
149165
150- print (f"[Load] collect testcase count: { len (load_result .Tests )} " )
151- print (f"[Load] collect load error count: { len (load_result .LoadErrors )} " )
166+ logger . info (f"[Load] collect testcase count: { len (load_result .Tests )} " )
167+ logger . info (f"[Load] collect load error count: { len (load_result .LoadErrors )} " )
152168
153169 reporter = FileReporter (Path (entry_param .FileReportPath ))
154170 reporter .report_load_result (load_result )
155171
156172
173+ def collect_testcases_file_mode (entry_param : EntryParam , load_result : LoadResult ) -> None :
174+ """
175+ 文件模式:只解析到文件层级,不解析里面的类和方法
176+ 扫描传进来的testselector对应的文件:
177+ - 如果是tests,则扫描tests目录下所有pytest相关的测试文件
178+ - 如果是.,则扫描整个项目目录下符合pytest规范的文件
179+ - 排除隐藏目录、文件和缓存文件
180+ """
181+ valid_selectors , load_errors = filter_invalid_selector_path (
182+ workspace = entry_param .ProjectPath ,
183+ selectors = entry_param .TestSelectors ,
184+ )
185+
186+ load_result .LoadErrors .extend (load_errors )
187+
188+ # 收集所有测试文件
189+ test_files = set ()
190+
191+ for selector in valid_selectors :
192+ # 移除数据驱动部分
193+ if CASE_DRIVE_SEPARATOR in selector :
194+ selector = selector .split (CASE_DRIVE_SEPARATOR )[0 ]
195+
196+ # 提取路径部分(去掉查询参数)
197+ path_part , _ , _ = selector .partition ("?" )
198+
199+ # 构建完整路径
200+ project_path_obj = Path (entry_param .ProjectPath )
201+ full_path = project_path_obj / path_part
202+
203+ if full_path .is_file ():
204+ # 如果是文件,直接添加
205+ if _is_pytest_test_file (str (full_path )):
206+ test_files .add (path_part )
207+ elif full_path .is_dir ():
208+ # 如果是目录,扫描目录下的所有测试文件
209+ discovered_files = _scan_pytest_files (str (full_path ), entry_param .ProjectPath )
210+ test_files .update (discovered_files )
211+ elif path_part == "." :
212+ # 如果是根目录,扫描整个项目
213+ discovered_files = _scan_pytest_files (entry_param .ProjectPath , entry_param .ProjectPath )
214+ test_files .update (discovered_files )
215+ else :
216+ # 尝试转换为pytest路径处理(兼容原有逻辑)
217+ pytest_path = selector_to_pytest (test_selector = selector )
218+ if pytest_path :
219+ # 提取文件路径部分(去掉类名和方法名)
220+ file_path = pytest_path .split ("::" )[0 ]
221+ full_file_path = project_path_obj / file_path
222+ if full_file_path .exists () and _is_pytest_test_file (str (full_file_path )):
223+ test_files .add (file_path )
224+
225+ # 为每个测试文件创建一个测试用例
226+ project_path_obj = Path (entry_param .ProjectPath )
227+ for file_path in sorted (test_files ):
228+ # 检查文件是否存在
229+ full_file_path = project_path_obj / file_path
230+ if full_file_path .exists ():
231+ # 将文件路径转换为选择器格式
232+ load_result .Tests .append (TestCase (Name = file_path , Attributes = {}))
233+ else :
234+ load_result .LoadErrors .append (
235+ LoadError (
236+ name = file_path ,
237+ message = f"Test file not found: { full_file_path } " ,
238+ )
239+ )
240+
241+ print (f"[Load] File mode - collected { len (load_result .Tests )} test files" )
242+ print (f"[Load] File mode - load error count: { len (load_result .LoadErrors )} " )
243+
244+ reporter = FileReporter (Path (entry_param .FileReportPath ))
245+ reporter .report_load_result (load_result )
246+
247+
248+ def _is_pytest_test_file (file_path : str ) -> bool :
249+ """
250+ 判断文件是否是pytest测试文件
251+ pytest默认的测试文件命名规则:
252+ 1. test_*.py - 以test_开头的.py文件
253+ 2. *_test.py - 以_test.py结尾的.py文件(但不能只是_test.py)
254+ """
255+ file_path_obj = Path (file_path )
256+ if file_path_obj .suffix != ".py" :
257+ return False
258+
259+ filename = file_path_obj .name
260+
261+ # 规则1: 以test_开头
262+ if filename .startswith ("test_" ):
263+ return True
264+
265+ # 规则2: 以_test.py结尾,但不能只是_test.py
266+ if filename .endswith ("_test.py" ) and filename != "_test.py" :
267+ return True
268+
269+ return False
270+
271+
272+ def _scan_pytest_files (scan_path : str , project_path : str ) -> set [str ]:
273+ """
274+ 扫描指定路径下的所有pytest测试文件
275+ 排除隐藏目录、文件和缓存文件
276+ """
277+ test_files = set ()
278+
279+ # 需要排除的目录和文件模式
280+ exclude_dirs = {
281+ "__pycache__" ,
282+ ".pytest_cache" ,
283+ ".git" ,
284+ ".svn" ,
285+ ".hg" ,
286+ "node_modules" ,
287+ ".venv" ,
288+ "venv" ,
289+ ".env" ,
290+ "env" ,
291+ ".tox" ,
292+ "build" ,
293+ "dist" ,
294+ ".coverage" ,
295+ "htmlcov" ,
296+ ".mypy_cache" ,
297+ ".idea" ,
298+ ".vscode" ,
299+ }
300+
301+ exclude_patterns = {
302+ ".pyc" ,
303+ ".pyo" ,
304+ ".pyd" ,
305+ ".so" ,
306+ ".egg-info" ,
307+ ".coverage" ,
308+ ".DS_Store" ,
309+ "Thumbs.db" ,
310+ }
311+
312+ try :
313+ scan_path_obj = Path (scan_path )
314+ project_path_obj = Path (project_path )
315+
316+ def should_exclude_dir (dir_path : Path ) -> bool :
317+ """判断是否应该排除目录"""
318+ dir_name = dir_path .name
319+ return (
320+ dir_name .startswith ("." ) # 隐藏目录
321+ or dir_name in exclude_dirs # 特定排除目录
322+ or dir_name .startswith ("__pycache__" ) # Python缓存目录
323+ )
324+
325+ def scan_directory (current_path : Path ) -> None :
326+ """递归扫描目录"""
327+ try :
328+ for item in current_path .iterdir ():
329+ if item .is_dir ():
330+ # 检查是否应该排除此目录
331+ if not should_exclude_dir (item ):
332+ scan_directory (item )
333+ elif item .is_file ():
334+ # 跳过隐藏文件和特定扩展名的文件
335+ if item .name .startswith ("." ):
336+ continue
337+
338+ if any (item .name .endswith (pattern ) for pattern in exclude_patterns ):
339+ continue
340+
341+ # 检查是否是pytest测试文件
342+ if _is_pytest_test_file (str (item )):
343+ # 计算相对于项目根目录的路径
344+ rel_path = item .relative_to (project_path_obj )
345+ # 统一使用正斜杠
346+ rel_path_str = str (rel_path ).replace ("\\ " , "/" )
347+ test_files .add (rel_path_str )
348+ except (OSError , PermissionError ) as e :
349+ print (f"[Warning] Failed to access directory { current_path } : { e } " )
350+
351+ scan_directory (scan_path_obj )
352+
353+ except (OSError , PermissionError ) as e :
354+ print (f"[Warning] Failed to scan directory { scan_path } : { e } " )
355+
356+ return test_files
357+
358+
157359def show_workspace_files (workdir : str ) -> None :
158360 print ()
159361 print (f"Workspace [{ workdir } ] files:" )
0 commit comments