Function main

scripts/test_mcp_interactive.py:323–408

()

def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("binary")
    parser.add_argument(
        "--scenario",
        choices=("initialize", "invalid-index", "roundtrip", "advanced"),
        required=True,
    )
    parser.add_argument("--repo-path", required=True)
    parser.add_argument("--response-timeout", type=float, default=45.0)
    parser.add_argument("--exit-timeout", type=float, default=15.0)
    args = parser.parse_args()

    try:
        command = mcp_command(args.binary)
        process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=0,
        )
    except (OSError, SmokeFailure) as error:
        print(f"FAIL: could not start MCP server: {error}", file=sys.stderr)
        return 1
    assert process.stdout is not None
    assert process.stderr is not None
    responses: "queue.Queue[dict[str, Any]]" = queue.Queue()
    transcript: list[dict[str, Any]] = []
    stderr_chunks: list[bytes] = []
    stdout_thread = threading.Thread(
        target=read_json_lines,
        args=(process.stdout, responses, transcript),
        daemon=True,
    )
    stderr_thread = threading.Thread(
        target=drain, args=(process.stderr, stderr_chunks), daemon=True
    )
    stdout_thread.start()
    stderr_thread.start()

    try:
        run_scenario(
            process,
            responses,
            args.scenario,
            args.repo_path,
            args.response_timeout,
        )
        assert process.stdin is not None
        process.stdin.close()
        try:
            return_code = process.wait(timeout=args.exit_timeout)
        except subprocess.TimeoutExpired as error:
            raise SmokeFailure("MCP server did not exit after interactive stdin EOF") from error
        if return_code != 0:
            raise SmokeFailure(f"MCP server exited nonzero after completed session: {return_code}")
        stdout_thread.join(timeout=2)