diff --git a/covermcp/server.py b/covermcp/server.py index 440a7de..931d8c2 100644 --- a/covermcp/server.py +++ b/covermcp/server.py @@ -258,6 +258,110 @@ def create_options() -> dict: } # fmt: skip +def _build_tool_args(subcommand: str, **kwargs: Any) -> list[str]: + """Process kwargs from the wrapper tool into a list of CLI strings. + + Args: + subcommand: The dcover subcommand (e.g., "create", "issues"). + **kwargs: Tool-specific keyword arguments. + + Returns: + A list of formatted command-line arguments. + """ + tool_args = [] + if subcommand == "create": + entry_points = kwargs.get("entry_points") + if entry_points: + tool_args.extend([x.strip() for x in entry_points if x.strip()]) + + elif subcommand == "refactor": + if kwargs.get("dry_run"): + tool_args.append("--dry-run") + + elif subcommand == "issues": + if (limit := kwargs.get("limit")) is not None: + tool_args.extend(["--limit", str(limit)]) + if (skip := kwargs.get("skip")) is not None: + tool_args.extend(["--skip", str(skip)]) + if kwargs.get("prompt"): + tool_args.append("--prompt") + if (cover_json := kwargs.get("cover_json")) is not None: + tool_args.extend(["--cover-json", cover_json]) + if kwargs.get("dry_run"): + tool_args.append("--dry-run") + return tool_args + + +async def _run_dcover_command( # noqa: PLR0913,PLR0917 + ctx: Context, + path: str | None, + subcommand: str, + passthrough_args: list[str] | None, + working_directory: Path, + dcover_timeout: int | None, + **kwargs: Any, +) -> object: + """Internal helper to execute any dcover command, stream output, and handle errors. + + Args: + ctx: MCP server context for logging. + path: Path to the dcover executable. + subcommand: The dcover subcommand to run (e.g., "create", "refactor"). + passthrough_args: A list of additional arguments from the user/LLM. + working_directory: The project directory to run in. + dcover_timeout: The maximum execution time in seconds. + **kwargs: Tool-specific keyword arguments (e.g., entry_points, dry_run). + + Returns: + A dictionary containing the execution result. + + Raises: + ToolError: If the command fails, times out, or dcover is not found. + """ + + # We'll assume that the path and directories exist + path = find_dcover_executable(path) + + # Build the core command + command = [path, subcommand, "--batch"] + + # Process kwargs from the wrapper tool into a list of CLI strings + tool_args = _build_tool_args(subcommand, **kwargs) + command.extend(tool_args) + + # Add passthrough arguments from the LLM + if passthrough_args: + await ctx.debug(f"{passthrough_args} provided by LLM") + command.extend(passthrough_args) + + # Add options from the environment variable + options = os.getenv(DIFFBLUE_COVER_OPTIONS) + if options: + await ctx.debug(f"{DIFFBLUE_COVER_OPTIONS} provided in environment variable") + command.extend(shlex.split(options)) + + cmd = " ".join(command) + await ctx.debug(f"Running: {cmd}") + await ctx.debug(f"Working directory: {working_directory}") + await ctx.debug(f"Timeout: {dcover_timeout}s") + + output_lines = [] + try: + for line in execute(command, working_directory, dcover_timeout): + output_lines.append(line) + await ctx.debug(line) + await ctx.report_progress(progress=len(output_lines)) + return { + "return_code": 0, + "status": "success", + "output": "\n".join(output_lines), + "command": command, + "working_directory": working_directory, + } + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: + raise ToolError(str(e), "\n".join(output_lines)) from e + + @mcp.tool() # Ignore "too many parameters for a method" and "too many positional arguments" check async def create( # noqa: PLR0913,PLR0917 @@ -314,46 +418,15 @@ async def create( # noqa: PLR0913,PLR0917 This tool requires a valid Diffblue Cover license. See: https://docs.diffblue.com/features/cover-cli/commands-and-arguments#create-tests """ - - # We'll assume that the path and directories exist - - path = find_dcover_executable(path) - - command = [path, "create", "--batch"] - - if args is not None and len(args) > 0: - await ctx.debug(f"{args} provided by LLM") - command.extend(args) - - options = os.getenv(DIFFBLUE_COVER_OPTIONS) - if options is not None and len(options) > 0: - await ctx.debug(f"{DIFFBLUE_COVER_OPTIONS} provided in environment variable") - command.extend(shlex.split(options)) - - entry_points = [x.strip() for x in entry_points if x.strip()] if entry_points else [] - if entry_points: - command.extend(entry_points) - - cmd = " ".join(command) - await ctx.debug(f"Running: {cmd}") - await ctx.debug(f"Working directory: {working_directory}") - await ctx.debug(f"Timeout: {dcover_timeout}s") - - output_lines = [] - try: - for line in execute(command, working_directory, dcover_timeout): - output_lines.append(line) - await ctx.debug(line) - await ctx.report_progress(progress=len(output_lines)) - return { - "return_code": 0, - "status": "success", - "output": "\n".join(output_lines), - "command": command, - "working_directory": working_directory, - } - except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: - raise ToolError(str(e), "\n".join(output_lines)) from e + return await _run_dcover_command( + ctx=ctx, + path=path, + subcommand="create", + passthrough_args=args, + working_directory=working_directory, + dcover_timeout=dcover_timeout, + entry_points=entry_points, + ) def find_dcover_executable(provided: str | None) -> str: @@ -391,3 +464,118 @@ def find_dcover_executable(provided: str | None) -> str: "Ensure dcover is in PATH, provide explicit path, " f"or set {DIFFBLUE_COVER_CLI} environment variable." ) + + +@mcp.tool() +# Ignore "too many parameters for a method" and "too many positional arguments" check +async def refactor( # noqa: PLR0913,PLR0917 + path: Annotated[str | None, "The path to the dcover executable"] = None, + working_directory: Annotated[Path, "The directory containing the project"] = DEFAULT_WORKING_DIRECTORY, + dcover_timeout: Annotated[ + int | None, "The maximum time in seconds to wait for dcover to create tests." + ] = DEFAULT_TIMEOUT, + dry_run: Annotated[bool, "Run preflight checks only (aliased as --preflight)"] = False, + args: Annotated[list[str] | None, "The additional options to pass to dcover refactor"] = None, + ctx: Annotated[Context | None, "The MCP Server Context"] = None, +) -> object: + """Invoke Diffblue Cover to refactor the project (aliased as 'fix-build'). + + This tool executes the `dcover refactor` command to apply automated + refactorings, such as fixing build issues or adding missing dependencies. + + Args: + path: Path to the dcover executable. If not provided, searches system PATH + and the DIFFBLUE_COVER_CLI environment variable. + working_directory: Root directory of the Java project to test. Defaults to + the current working directory. + dcover_timeout: Maximum execution time in seconds. Defaults to 600. Set to None + for no timeout (not recommended). + dry_run: If True, passes the '--dry-run' flag to check for readiness + without applying changes. + args: Additional arguments to pass to dcover. Defaults to None. + ctx: MCP server context for logging and progress reporting (auto-injected by FastMCP). + + Returns: + dict: Execution result containing: + - return_code (int): Exit code (0 for success) + - status (str): "success" if completed without errors + - output (str): Complete stdout/stderr from dcover + - command (list[str]): The exact command that was executed + - working_directory (Path): Directory where command was run + + Raises: + ToolError: If dcover executable not found, command fails, or timeout exceeded. + The error includes the partial output collected before failure. + """ + return await _run_dcover_command( + ctx=ctx, + path=path, + subcommand="refactor", + passthrough_args=args, + working_directory=working_directory, + dcover_timeout=dcover_timeout, + dry_run=dry_run, + ) + + +@mcp.tool() +# Ignore "too many parameters for a method" and "too many positional arguments" check +async def issues( # noqa: PLR0913,PLR0917 + path: Annotated[str | None, "The path to the dcover executable"] = None, + working_directory: Annotated[Path, "The directory containing the project"] = DEFAULT_WORKING_DIRECTORY, + dcover_timeout: Annotated[ + int | None, "The maximum time in seconds to wait for dcover to create tests." + ] = DEFAULT_TIMEOUT, + limit: Annotated[int | None, "Limit the number of issues to output"] = None, + skip: Annotated[int | None, "Skip the first N issues"] = None, + prompt: Annotated[bool, "Output suggested prompt for each actionable issue"] = False, + cover_json: Annotated[str | None, "Path to a JSON-formatted test-writing summary report"] = None, + dry_run: Annotated[bool, "Run preflight checks only (aliased as --preflight)"] = False, + args: Annotated[list[str] | None, "The additional options to pass to dcover issues"] = None, + ctx: Annotated[Context | None, "The MCP Server Context"] = None, +) -> object: + """Invoke Diffblue Cover to identify project issues. + + This tool executes the `dcover issues` command to output a prioritized + list of project issues that may prevent test generation. + + Args: + path: Path to the dcover executable. If not provided, searches system PATH + and the DIFFBLUE_COVER_CLI environment variable. + working_directory: Root directory of the Java project to test. Defaults to + the current working directory. + dcover_timeout: Maximum execution time in seconds. Defaults to 600. Set to None + for no timeout (not recommended). + limit: Limit the number of issues to output. + skip: Skip the first N issues from the report. + prompt: If True, outputs a suggested prompt for each actionable issue. + cover_json: Location of the JSON-formatted test-writing summary report. + dry_run: If True, passes the '--dry-run' flag to check for readiness. + args: Additional arguments to pass to dcover. Defaults to None. + ctx: MCP server context for logging and progress reporting (auto-injected by FastMCP). + + Returns: + dict: Execution result containing: + - return_code (int): Exit code (0 for success) + - status (str): "success" if completed without errors + - output (str): Complete stdout/stderr from dcover + - command (list[str]): The exact command that was executed + - working_directory (Path): Directory where command was run + + Raises: + ToolError: If dcover executable not found, command fails, or timeout exceeded. + The error includes the partial output collected before failure. + """ + return await _run_dcover_command( + ctx=ctx, + path=path, + subcommand="issues", + passthrough_args=args, + working_directory=working_directory, + dcover_timeout=dcover_timeout, + limit=limit, + skip=skip, + prompt=prompt, + cover_json=cover_json, + dry_run=dry_run, + ) diff --git a/test/test_server.py b/test/test_server.py index 925448f..cdc6201 100644 --- a/test/test_server.py +++ b/test/test_server.py @@ -173,3 +173,86 @@ def test_find_dcover_executable(monkeypatch): m.setattr(shutil, "which", lambda *args: None) with pytest.raises(ToolError): server.find_dcover_executable(None) + + +@pytest.mark.asyncio +async def test_dcover_refactor_command(mock_happy_path_execution): + async with Client(server.mcp) as client: + result = await client.call_tool("refactor", arguments={"path": Path("path", "to", "dcover")}) + assert result is not None + + mock_happy_path_execution.assert_called_once_with( + [str(Path("path", "to", "dcover")), "refactor", "--batch"], + Path(result.data["working_directory"]), + DEFAULT_TIMEOUT, + ) + assert "command" in result.data + assert "line 1" in result.data["output"] + assert "line 2" in result.data["output"] + + +@pytest.mark.asyncio +async def test_dcover_refactor_command_with_dry_run(mock_happy_path_execution): + async with Client(server.mcp) as client: + result = await client.call_tool("refactor", arguments={"path": Path("path", "to", "dcover"), "dry_run": True}) + assert result is not None + + mock_happy_path_execution.assert_called_once_with( + [str(Path("path", "to", "dcover")), "refactor", "--batch", "--dry-run"], + Path(result.data["working_directory"]), + DEFAULT_TIMEOUT, + ) + assert "command" in result.data + assert "--dry-run" in result.data["command"] + + +@pytest.mark.asyncio +async def test_dcover_issues_command(mock_happy_path_execution): + async with Client(server.mcp) as client: + result = await client.call_tool("issues", arguments={"path": Path("path", "to", "dcover")}) + assert result is not None + + mock_happy_path_execution.assert_called_once_with( + [str(Path("path", "to", "dcover")), "issues", "--batch"], + Path(result.data["working_directory"]), + DEFAULT_TIMEOUT, + ) + assert "command" in result.data + assert "line 1" in result.data["output"] + assert "line 2" in result.data["output"] + + +@pytest.mark.asyncio +async def test_dcover_issues_command_with_all_args(mock_happy_path_execution): + async with Client(server.mcp) as client: + arguments = { + "path": Path("path", "to", "dcover"), + "limit": 10, + "skip": 5, + "prompt": True, + "cover_json": "/tmp/report.json", + "dry_run": True, + } + result = await client.call_tool("issues", arguments=arguments) + assert result is not None + + expected_command = [ + str(Path("path", "to", "dcover")), + "issues", + "--batch", + "--limit", + "10", + "--skip", + "5", + "--prompt", + "--cover-json", + "/tmp/report.json", + "--dry-run", + ] + + mock_happy_path_execution.assert_called_once_with( + expected_command, + Path(result.data["working_directory"]), + DEFAULT_TIMEOUT, + ) + assert result.data["command"] == expected_command