Coverage for /home/crpier/Projects/snektest/snektest/collection.py: 79%

1import asyncio 

2from collections.abc import Callable 

3from importlib.machinery import ModuleSpec 

4from importlib.util import module_from_spec, spec_from_file_location 

5from inspect import getmembers, isfunction 

6from pathlib import Path 

7from sys import modules 

8from types import FunctionType 

9from typing import TypeGuard, cast 

10 

11from pydantic import ValidationError 

12 

13from snektest.annotations import PyFilePath, validate_PyFilePath 

14from snektest.models import CollectionError, FilterItem, TestName 

15from snektest.utils import ( 

16 get_test_function_markers, 

17 get_test_function_params, 

18 is_test_function, 

19) 

20 

21TEST_FILE_PREFIX = "test_" 

22 

23FuncTestEntry = tuple[TestName, FunctionType] 

24TestsQueue = asyncio.Queue[FuncTestEntry] 

25 

26 

27def load_tests_from_file( # noqa: PLR0913 

28 file_path: PyFilePath, 

29 filter_item: FilterItem, 

30 queue: TestsQueue, 

31 loop: asyncio.AbstractEventLoop, 

32 *, 

33 mark: str | None = None, 

34 spec_loader: Callable[..., object] = spec_from_file_location, 

35) -> None: 

36 """Load and queue tests from a single Python file.""" 

37 module_name = ".".join(file_path.with_suffix("").parts) 

38 if module_name in modules: 

39 module = modules[module_name] 

40 else: 

41 spec = spec_loader(module_name, file_path) 

42 spec_value = cast("ModuleSpec", spec) 

43 loader = getattr(spec_value, "loader", None) 

44 if loader is None: 

45 msg = f"Could not load spec from {file_path}" 

46 raise CollectionError(msg) 

47 

48 module = module_from_spec(spec_value) 

49 modules[module_name] = module 

50 loader.exec_module(module) 

51 

52 runnable_functions = [func for _, func in getmembers(module, isfunction)] 

53 runnable_functions = filter(is_test_function, runnable_functions) 

54 if filter_item.function_name: 

55 runnable_functions = filter( 

56 lambda func: func.__name__ == filter_item.function_name, runnable_functions 

57 ) 

58 

59 if mark is not None: 

60 runnable_functions = filter( 

61 lambda func: mark in get_test_function_markers(func), runnable_functions 

62 ) 

63 

64 for func in runnable_functions: 

65 for param_names in get_test_function_params(func): 

66 if filter_item.params and filter_item.params != param_names: 

67 continue 

68 test_name = TestName( 

69 file_path=file_path, func_name=func.__name__, params_part=param_names 

70 ) 

71 _ = loop.call_soon_threadsafe(queue.put_nowait, (test_name, func)) 

72 

73 

74def generate_file_list(filter_item: FilterItem) -> list[PyFilePath]: 

75 """Generate a list of valid file paths for given filter item.""" 

76 

77 def path_is_runnable(file_path: Path) -> TypeGuard[PyFilePath]: 

78 if not file_path.name.startswith(TEST_FILE_PREFIX): 

79 return False 

80 try: 

81 file_path = validate_PyFilePath(file_path) 

82 except ValidationError: 

83 return False 

84 return True 

85 

86 if filter_item.file_path.is_dir(): 

87 paths = [ 

88 dirpath / name 

89 for dirpath, _, filenames in filter_item.file_path.walk() 

90 for name in filenames 

91 ] 

92 else: 

93 paths = [filter_item.file_path] 

94 

95 return [path for path in paths if path_is_runnable(path)] 

96 

97 

98def load_tests_from_filters( 

99 filter_items: list[FilterItem], 

100 queue: TestsQueue, 

101 loop: asyncio.AbstractEventLoop, 

102 *, 

103 mark: str | None = None, 

104 exception_holder: list[BaseException] | None = None, 

105) -> None: 

106 """Load tests from all filter items and populate the queue. 

107 

108 Args: 

109 filter_items: List of filter items to load tests from 

110 queue: Queue to populate with tests 

111 loop: Event loop for thread-safe queue operations 

112 exception_holder: Optional list to store exception if one occurs during collection 

113 """ 

114 try: 

115 for filter_item in filter_items: 

116 file_paths = generate_file_list(filter_item) 

117 for file_path in file_paths: 

118 load_tests_from_file( 

119 file_path=file_path, 

120 filter_item=filter_item, 

121 queue=queue, 

122 loop=loop, 

123 mark=mark, 

124 ) 

125 except BaseException as e: 

126 if exception_holder is not None: 

127 if isinstance(e, CollectionError): 

128 exception_holder.append(e) 

129 else: 

130 collection_error = CollectionError(f"Error during collection: {e}") 

131 collection_error.__cause__ = e 

132 exception_holder.append(collection_error) 

133 finally: 

134 _ = loop.call_soon_threadsafe(queue.shutdown)