Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
268 changes: 228 additions & 40 deletions covermcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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,
)
83 changes: 83 additions & 0 deletions test/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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