Spaces:
Running
Running
File size: 3,351 Bytes
5d06dbc 4cf7fc5 a64c042 4cf7fc5 b066699 4cf7fc5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | from typing import Any, Callable
from unittest.mock import patch
import pytest
from langchain.tools import ToolRuntime
from langchain_core.tools import BaseTool, StructuredTool
from multi_agent_sdlc.system.process import ProcessResult
from multi_agent_sdlc.tools.shared.execution import run_docker_compose
from multi_agent_sdlc.tools.shared.models import DockerComposeOperation
from multi_agent_sdlc.workflow.state import DevState
def get_tool_function(
tool: BaseTool,
) -> Callable[..., Any]:
if not isinstance(tool, StructuredTool):
raise TypeError(f"Expected StructuredTool, got {type(tool).__name__}")
if tool.func is None:
raise TypeError(f"Tool {tool.name} does not have a synchronous function")
return tool.func
@pytest.mark.parametrize(
("operation", "expected_command"),
[
(
DockerComposeOperation.UP,
[
"docker",
"compose",
"up",
"-d",
"--build",
"--wait",
"--wait-timeout",
"60",
],
),
(
DockerComposeOperation.DOWN,
[
"docker",
"compose",
"down",
],
),
(
DockerComposeOperation.BUILD,
[
"docker",
"compose",
"build",
],
),
(
DockerComposeOperation.CONFIG,
[
"docker",
"compose",
"config",
"--quiet",
],
),
(
DockerComposeOperation.PS,
[
"docker",
"compose",
"ps",
],
),
],
)
def test_run_docker_compose_executes_expected_command(
operation: DockerComposeOperation,
expected_command: list[str],
tool_runtime: ToolRuntime[DevState],
process_result: ProcessResult,
) -> None:
function = get_tool_function(run_docker_compose)
with patch(
"multi_agent_sdlc.tools.shared.execution.execute_process",
return_value=process_result,
) as execute_process_mock:
result = function(
operation=operation,
runtime=tool_runtime,
)
assert result is process_result
execute_process_mock.assert_called_once_with(
expected_command,
project_directory=tool_runtime.state["project_directory"],
timeout_seconds=120,
)
def test_run_docker_compose_uses_custom_timeout(
tool_runtime: ToolRuntime[DevState],
process_result: ProcessResult,
) -> None:
function = get_tool_function(run_docker_compose)
with patch(
"multi_agent_sdlc.tools.shared.execution.execute_process",
return_value=process_result,
) as execute_process_mock:
result = function(
operation=DockerComposeOperation.BUILD,
runtime=tool_runtime,
timeout_seconds=300,
)
assert result is process_result
execute_process_mock.assert_called_once_with(
[
"docker",
"compose",
"build",
],
project_directory=tool_runtime.state["project_directory"],
timeout_seconds=300,
)
|