Coverage for tests / unit / utils / async_tool_executor / test_exceptions.py: 100%

1"""Tests for exception handling in parallel execution.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from typing import Any, cast 

7 

8from assertpy import assert_that 

9 

10from lintro.enums.action import Action 

11from lintro.models.core.tool_result import ToolResult 

12from lintro.plugins.base import BaseToolPlugin 

13from lintro.utils.async_tool_executor import AsyncToolExecutor 

14 

15from .conftest import MockToolDefinition, MockToolPlugin 

16 

17 

18def test_exception_in_tool_creates_failed_result( 

19 executor: AsyncToolExecutor, 

20) -> None: 

21 """Test that exceptions in tools create failed results. 

22 

23 Args: 

24 executor: AsyncToolExecutor fixture. 

25 """ 

26 

27 def raising_check( 

28 paths: list[str], 

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

30 ) -> ToolResult: 

31 raise RuntimeError("Test exception") 

32 

33 tool = MockToolPlugin(definition=MockToolDefinition(name="raise_tool")) 

34 # Use object.__setattr__ to bypass dataclass method assignment restriction 

35 object.__setattr__(tool, "check", raising_check) 

36 

37 async def run_test() -> Any: 

38 # MockToolPlugin implements the same interface as BaseToolPlugin for testing 

39 return await executor.run_tools_parallel( 

40 tools=cast(list[tuple[str, BaseToolPlugin]], [("raise_tool", tool)]), 

41 paths=["."], 

42 action=Action.CHECK, 

43 ) 

44 

45 results = asyncio.run(run_test()) 

46 

47 assert_that(results).is_length(1) 

48 name, result = results[0] 

49 assert_that(name).is_equal_to("raise_tool") 

50 assert_that(result.success).is_false() 

51 assert_that(result.output).contains("Parallel execution failed")