Coverage for tests / unit / utils / async_tool_executor / conftest.py: 91%

1"""Shared fixtures for AsyncToolExecutor tests.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Iterator 

6from dataclasses import dataclass, field 

7from typing import Any 

8 

9import pytest 

10 

11from lintro.models.core.tool_result import ToolResult 

12from lintro.utils.async_tool_executor import AsyncToolExecutor 

13 

14 

15@dataclass 

16class MockToolDefinition: 

17 """Mock tool definition for testing. 

18 

19 Attributes: 

20 name: Name of the tool. 

21 conflicts_with: List of tools this tool conflicts with. 

22 """ 

23 

24 name: str = "mock_tool" 

25 conflicts_with: list[str] = field(default_factory=list) 

26 

27 

28@dataclass 

29class MockToolPlugin: 

30 """Mock tool plugin for testing. 

31 

32 Attributes: 

33 definition: Tool definition mock. 

34 check_result: Result to return from check(). 

35 fix_result: Result to return from fix(). 

36 check_called: Whether check() was called. 

37 fix_called: Whether fix() was called. 

38 delay: Optional delay in seconds before returning result. 

39 """ 

40 

41 definition: MockToolDefinition = field(default_factory=MockToolDefinition) 

42 check_result: ToolResult | None = None 

43 fix_result: ToolResult | None = None 

44 check_called: bool = field(default=False, init=False) 

45 fix_called: bool = field(default=False, init=False) 

46 delay: float = 0.0 

47 

48 def __post_init__(self) -> None: 

49 """Initialize default results if not provided.""" 

50 if self.check_result is None: 

51 self.check_result = ToolResult( 

52 name=self.definition.name, 

53 success=True, 

54 output="check output", 

55 issues_count=0, 

56 ) 

57 if self.fix_result is None: 

58 self.fix_result = ToolResult( 

59 name=self.definition.name, 

60 success=True, 

61 output="fix output", 

62 issues_count=0, 

63 ) 

64 

65 def check( 

66 self, 

67 paths: list[str], 

68 options: dict[str, Any] | None = None, 

69 ) -> ToolResult: 

70 """Mock check method. 

71 

72 Args: 

73 paths: Paths to check. 

74 options: Check options. 

75 

76 Returns: 

77 ToolResult: The configured check result. 

78 """ 

79 self.check_called = True 

80 if self.delay > 0: 

81 import time 

82 

83 time.sleep(self.delay) 

84 # check_result is guaranteed non-None by __post_init__ 

85 assert self.check_result is not None 

86 return self.check_result 

87 

88 def fix( 

89 self, 

90 paths: list[str], 

91 options: dict[str, Any] | None = None, 

92 ) -> ToolResult: 

93 """Mock fix method. 

94 

95 Args: 

96 paths: Paths to fix. 

97 options: Fix options. 

98 

99 Returns: 

100 ToolResult: The configured fix result. 

101 """ 

102 self.fix_called = True 

103 if self.delay > 0: 

104 import time 

105 

106 time.sleep(self.delay) 

107 # fix_result is guaranteed non-None by __post_init__ 

108 assert self.fix_result is not None 

109 return self.fix_result 

110 

111 

112@pytest.fixture 

113def executor() -> Iterator[AsyncToolExecutor]: 

114 """Create an AsyncToolExecutor for testing. 

115 

116 Yields: 

117 AsyncToolExecutor: The executor instance. 

118 """ 

119 exec_instance = AsyncToolExecutor(max_workers=2) 

120 yield exec_instance 

121 exec_instance.shutdown()