Coverage for tests / unit / utils / test_timeout_utils.py: 98%

1"""Unit tests for timeout utilities.""" 

2 

3import subprocess 

4from typing import Any 

5from unittest.mock import Mock 

6 

7import pytest 

8from assertpy import assert_that 

9 

10from lintro.tools.core.timeout_utils import ( 

11 create_timeout_result, 

12 get_timeout_value, 

13 run_subprocess_with_timeout, 

14) 

15 

16 

17class MockDefinition: 

18 """Mock definition for testing.""" 

19 

20 def __init__(self, name: str) -> None: 

21 """Initialize mock definition. 

22 

23 Args: 

24 name: Tool name. 

25 """ 

26 self.name = name 

27 

28 

29class MockTool: 

30 """Mock tool for testing timeout utilities.""" 

31 

32 def __init__(self, name: str = "test_tool", default_timeout: int = 300) -> None: 

33 """Initialize mock tool. 

34 

35 Args: 

36 name: Tool name for testing. 

37 default_timeout: Default timeout value in seconds. 

38 """ 

39 self.definition = MockDefinition(name) 

40 self._default_timeout = default_timeout 

41 self.options: dict[str, Any] = {} 

42 

43 def _run_subprocess( 

44 self, 

45 cmd: list[str], 

46 timeout: int | None = None, 

47 cwd: str | None = None, 

48 ) -> tuple[bool, str]: 

49 """Mock subprocess runner. 

50 

51 Args: 

52 cmd: Command to run. 

53 timeout: Optional timeout value. 

54 cwd: Optional working directory. 

55 

56 Returns: 

57 tuple[bool, str]: Success status and output. 

58 """ 

59 return True, "success" 

60 

61 

62def test_get_timeout_value_with_option() -> None: 

63 """Test getting timeout value when set in options.""" 

64 tool = MockTool() 

65 tool.options["timeout"] = 60 

66 

67 assert_that(get_timeout_value(tool)).is_equal_to(60) 

68 

69 

70def test_get_timeout_value_with_default() -> None: 

71 """Test getting timeout value when using tool default.""" 

72 tool = MockTool(default_timeout=45) 

73 

74 assert_that(get_timeout_value(tool)).is_equal_to(45) 

75 

76 

77def test_get_timeout_value_with_custom_default() -> None: 

78 """Test getting timeout value with custom default parameter.""" 

79 tool = MockTool() 

80 

81 assert_that(get_timeout_value(tool, 120)).is_equal_to(120) 

82 

83 

84def test_create_timeout_result() -> None: 

85 """Test creating a timeout result object.""" 

86 tool = MockTool("pytest") 

87 

88 result = create_timeout_result(tool, 30, ["pytest", "test"]) 

89 

90 assert_that(result.success).is_false() 

91 assert_that(result.output).contains( 

92 "pytest execution timed out (30s limit exceeded)", 

93 ) 

94 assert_that(result.issues_count).is_equal_to(1) 

95 assert_that(result.issues).is_empty() 

96 assert_that(result.timed_out).is_true() 

97 assert_that(result.timeout_seconds).is_equal_to(30) 

98 

99 

100def test_run_subprocess_with_timeout_success() -> None: 

101 """Test successful subprocess execution with timeout.""" 

102 tool = MockTool() 

103 tool._run_subprocess = Mock(return_value=(True, "output")) # type: ignore[method-assign] 

104 

105 success, output = run_subprocess_with_timeout(tool, ["echo", "test"]) 

106 

107 assert_that(success).is_true() 

108 assert_that(output).is_equal_to("output") 

109 tool._run_subprocess.assert_called_once_with( 

110 cmd=["echo", "test"], 

111 timeout=None, 

112 cwd=None, 

113 ) 

114 

115 

116def test_run_subprocess_with_timeout_exception() -> None: 

117 """Test subprocess timeout exception handling.""" 

118 tool = MockTool() 

119 

120 # Mock subprocess to raise TimeoutExpired 

121 def mock_run_subprocess( 

122 cmd: list[str], 

123 timeout: int | None = None, 

124 cwd: str | None = None, 

125 ) -> tuple[bool, str]: 

126 raise subprocess.TimeoutExpired( 

127 cmd=["slow", "command"], 

128 timeout=10, 

129 output="timeout occurred", 

130 ) 

131 

132 tool._run_subprocess = mock_run_subprocess # type: ignore[method-assign] 

133 

134 with pytest.raises(subprocess.TimeoutExpired) as exc_info: 

135 run_subprocess_with_timeout(tool, ["slow", "command"], timeout=10) 

136 

137 # Verify the exception has enhanced message 

138 assert_that(str(exc_info.value.output)).contains("test_tool execution timed out") 

139 assert_that(str(exc_info.value.output)).contains("(10s limit exceeded)")