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

1"""Tests for on_result callback functionality.""" 

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_on_result_callback_invoked(executor: AsyncToolExecutor) -> None: 

19 """Test that on_result callback is invoked for each tool. 

20 

21 Args: 

22 executor: AsyncToolExecutor fixture. 

23 """ 

24 callback_results: list[tuple[str, ToolResult]] = [] 

25 

26 def on_result(name: str, result: ToolResult) -> None: 

27 callback_results.append((name, result)) 

28 

29 # MockToolPlugin implements the same interface as BaseToolPlugin for testing 

30 tools = cast( 

31 list[tuple[str, BaseToolPlugin]], 

32 [ 

33 ("tool1", MockToolPlugin(definition=MockToolDefinition(name="tool1"))), 

34 ("tool2", MockToolPlugin(definition=MockToolDefinition(name="tool2"))), 

35 ], 

36 ) 

37 

38 async def run_test() -> Any: 

39 return await executor.run_tools_parallel( 

40 tools, 

41 paths=["."], 

42 action=Action.CHECK, 

43 on_result=on_result, 

44 ) 

45 

46 asyncio.run(run_test()) 

47 

48 assert_that(callback_results).is_length(2) 

49 callback_names = [name for name, _ in callback_results] 

50 assert_that(callback_names).contains("tool1", "tool2") 

51 

52 

53def test_on_result_receives_correct_args(executor: AsyncToolExecutor) -> None: 

54 """Test on_result callback receives correct arguments. 

55 

56 Args: 

57 executor: AsyncToolExecutor fixture. 

58 """ 

59 received_args: list[tuple[str, bool]] = [] 

60 

61 def on_result(name: str, result: ToolResult) -> None: 

62 received_args.append((name, result.success)) 

63 

64 fail_result = ToolResult( 

65 name="fail_tool", 

66 success=False, 

67 output="", 

68 issues_count=1, 

69 ) 

70 # MockToolPlugin implements the same interface as BaseToolPlugin for testing 

71 tools = cast( 

72 list[tuple[str, BaseToolPlugin]], 

73 [ 

74 ( 

75 "success_tool", 

76 MockToolPlugin(definition=MockToolDefinition(name="success_tool")), 

77 ), 

78 ( 

79 "fail_tool", 

80 MockToolPlugin( 

81 definition=MockToolDefinition(name="fail_tool"), 

82 check_result=fail_result, 

83 ), 

84 ), 

85 ], 

86 ) 

87 

88 async def run_test() -> Any: 

89 return await executor.run_tools_parallel( 

90 tools, 

91 paths=["."], 

92 action=Action.CHECK, 

93 on_result=on_result, 

94 ) 

95 

96 asyncio.run(run_test()) 

97 

98 result_dict = dict(received_args) 

99 assert_that(result_dict["success_tool"]).is_true() 

100 assert_that(result_dict["fail_tool"]).is_false()