Coverage for lintro / tools / core / timeout_utils.py: 100%

1"""Shared timeout handling utilities for tool implementations. 

2 

3This module provides standardized timeout handling across different tools, 

4ensuring consistent behavior and error messages for subprocess timeouts. 

5""" 

6 

7import subprocess # nosec B404 - used safely with shell disabled 

8from dataclasses import dataclass, field 

9from typing import Any 

10 

11from loguru import logger 

12 

13 

14@dataclass 

15class TimeoutResult: 

16 """Timeout result structure.""" 

17 

18 success: bool 

19 output: str 

20 issues_count: int 

21 issues: list[Any] = field(default_factory=list) 

22 timed_out: bool = True 

23 timeout_seconds: int = 0 

24 

25 

26def run_subprocess_with_timeout( 

27 tool: Any, 

28 cmd: list[str], 

29 timeout: int | None = None, 

30 cwd: str | None = None, 

31 tool_name: str | None = None, 

32) -> tuple[bool, str]: 

33 """Run a subprocess command with timeout handling. 

34 

35 This is a wrapper around tool._run_subprocess that provides consistent 

36 timeout error handling and messaging across different tools. 

37 

38 Args: 

39 tool: Tool instance with _run_subprocess method. 

40 cmd: Command to run. 

41 timeout: Timeout in seconds. If None, uses tool's default timeout. 

42 cwd: Working directory for command execution. 

43 tool_name: Name of the tool for error messages. If None, uses tool.name. 

44 

45 Returns: 

46 tuple[bool, str]: (success, output) where success is True if command 

47 succeeded without timeout, and output contains command output or 

48 timeout error message. 

49 

50 Raises: 

51 subprocess.TimeoutExpired: If command times out (re-raised with context). 

52 """ 

53 tool_name = tool_name or tool.definition.name 

54 

55 try: 

56 success, output = tool._run_subprocess(cmd=cmd, timeout=timeout, cwd=cwd) 

57 return bool(success), str(output) 

58 except subprocess.TimeoutExpired as e: 

59 # Re-raise with more context for the calling tool 

60 actual_timeout = timeout or tool.options.get("timeout", tool._default_timeout) 

61 timeout_msg = ( 

62 f"{tool_name} execution timed out ({actual_timeout}s limit exceeded).\n\n" 

63 "This may indicate:\n" 

64 " - Large codebase taking too long to process\n" 

65 " - Need to increase timeout via --tool-options timeout=N\n" 

66 " - Command hanging due to external dependencies\n" 

67 ) 

68 logger.warning(timeout_msg) 

69 

70 # Create a new TimeoutExpired with enhanced message 

71 raise subprocess.TimeoutExpired( 

72 cmd=cmd, 

73 timeout=actual_timeout, 

74 output=timeout_msg, 

75 ) from e 

76 

77 

78def get_timeout_value(tool: Any, default_timeout: int | None = None) -> int: 

79 """Get timeout value from tool options with fallback to default. 

80 

81 Args: 

82 tool: Tool instance with options. 

83 default_timeout: Default timeout if not specified in options. 

84 

85 Returns: 

86 int: Timeout value in seconds. 

87 """ 

88 if default_timeout is None: 

89 default_timeout = getattr(tool, "_default_timeout", 300) 

90 

91 return int(tool.options.get("timeout", default_timeout)) 

92 

93 

94def create_timeout_result( 

95 tool: Any, 

96 timeout: int, 

97 cmd: list[str] | None = None, 

98 tool_name: str | None = None, 

99) -> TimeoutResult: 

100 """Create a standardized timeout result. 

101 

102 Args: 

103 tool: Tool instance. 

104 timeout: Timeout value that was exceeded. 

105 cmd: Optional command that timed out. 

106 tool_name: Optional tool name override. 

107 

108 Returns: 

109 TimeoutResult: Result dataclass with timeout information. 

110 """ 

111 tool_name = tool_name or tool.definition.name 

112 

113 return TimeoutResult( 

114 success=False, 

115 output=( 

116 f"{tool_name} execution timed out ({timeout}s limit exceeded).\n\n" 

117 "This may indicate:\n" 

118 " - Large codebase taking too long to process\n" 

119 " - Need to increase timeout via --tool-options timeout=N\n" 

120 " - Command hanging due to external dependencies\n" 

121 ), 

122 issues_count=1, # Count timeout as execution failure 

123 issues=[], 

124 timed_out=True, 

125 timeout_seconds=timeout, 

126 )