headroom/tests/test_cli/test_wrap_kimi.py
kaz eac49656a1
feat(wrap): add headroom wrap kimi for Kimi CLI (#1426)
## Description

Adds `headroom wrap kimi`, routing Kimi CLI through the Headroom proxy.

Kimi CLI speaks an OpenAI-compatible `/chat/completions` API (its
`kosong` backend wraps `AsyncOpenAI`) and lets the base URL be
overridden via `KIMI_BASE_URL`. This wrapper points it at the local
proxy. Kimi's own OAuth bearer is forwarded upstream unchanged, so —
unlike the Copilot subscription path — no extra login or token exchange
is needed.

## Type of Change

- [x] New feature (non-breaking change that adds functionality)

## Changes Made

- `headroom/providers/kimi/`: new slice; `build_launch_env` sets
`KIMI_BASE_URL` with the per-project base-URL prefix, mirroring the
aider/vibe slices.
- `headroom/cli/wrap.py`: `kimi` subcommand; falls back to the
`kimi-cli` binary when `kimi` is not on `PATH`; `--kimi-api-url`
overrides the upstream coding endpoint (default
`https://api.kimi.com/coding/v1`).
- `tests/test_cli/test_wrap_kimi.py`: 8 tests for the wrap command.
- `README.md`: Kimi CLI row in the agent-compatibility matrix.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ pytest tests/test_cli/test_wrap_kimi.py -q
........                                                                 [100%]
8 passed in 0.36s

$ ruff check headroom/providers/kimi headroom/cli/wrap.py tests/test_cli/test_wrap_kimi.py
All checks passed!

$ ruff format --check headroom/providers/kimi headroom/cli/wrap.py tests/test_cli/test_wrap_kimi.py
4 files already formatted
```

## Real Behavior Proof

- Environment: macOS; Kimi CLI (`kimi` / `kimi-cli`); `headroom proxy`
started with `--openai-api-url https://api.kimi.com/coding/v1`.
- Exact command / steps: start `headroom proxy --port 8787
--openai-api-url https://api.kimi.com/coding/v1`, then `curl -s
http://localhost:8787/v1/chat/completions` with the Kimi OAuth bearer
and a one-line `kimi-for-coding` chat request (`"Reply with exactly:
PONG"`).
- Observed result: `HTTP 200`; `choices[0].message.content == "PONG"`
from `kimi-for-coding`; the OAuth bearer was forwarded and accepted
upstream; the per-project path `/p/<name>/v1/chat/completions` also
returned `HTTP 200`.
- Not tested: Windows/Linux PATH discovery; the `--learn` / `--memory`
live paths beyond flag wiring.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes

## Additional Notes

- `ruff check` and `ruff format --check` pass locally; `mypy` was run on
the new `headroom/providers/kimi` slice only (clean), so the full-tree
`mypy headroom` box is left unchecked and is left to CI.
- The slice deliberately reuses `codex.proxy_base_url` and
`with_project_prefix`, identical to the aider/vibe wrappers, so
per-project savings attribution works without Kimi sending custom
headers.
- Kimi's separate search/fetch services are out of scope for
`KIMI_BASE_URL` and continue to hit Kimi directly; only the LLM
`/chat/completions` traffic is compressed.

Co-authored-by: JerrettDavis <mxjerrett@gmail.com>
2026-07-15 21:40:58 +00:00

214 lines
7.4 KiB
Python

"""Tests for `headroom wrap kimi` command."""
from __future__ import annotations
from pathlib import Path
from typing import Any
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from headroom.cli import wrap as wrap_mod
from headroom.cli.main import main
@pytest.fixture
def runner() -> CliRunner:
return CliRunner()
def test_wrap_kimi_launch(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Kimi launches with correct configuration."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
captured: dict[str, Any] = {}
def fake_launch_tool(**kwargs: Any) -> None: # noqa: ANN003
captured.update(kwargs)
with patch.object(wrap_mod.shutil, "which", return_value="kimi"):
with patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool):
with patch.object(wrap_mod, "_project_name_from_cwd", return_value=None):
result = runner.invoke(
main, ["wrap", "kimi", "--port", "9000", "--", "-m", "kimi-for-coding"]
)
assert result.exit_code == 0, result.output
env = captured["env"]
assert isinstance(env, dict)
assert captured["tool_label"] == "KIMI"
assert captured["agent_type"] == "kimi"
assert captured["args"] == ("-m", "kimi-for-coding")
assert captured["openai_api_url"] == "https://api.kimi.com/coding/v1"
assert env["KIMI_BASE_URL"] == "http://127.0.0.1:9000/v1"
def test_wrap_kimi_with_project_name(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Project name is encoded in KIMI_BASE_URL when run from a project directory."""
project_dir = tmp_path / "my-project"
project_dir.mkdir()
monkeypatch.chdir(project_dir)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
captured: dict[str, Any] = {}
def fake_launch_tool(**kwargs: Any) -> None: # noqa: ANN003
captured.update(kwargs)
with patch.object(wrap_mod.shutil, "which", return_value="kimi"):
with patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool):
result = runner.invoke(main, ["wrap", "kimi", "--port", "7000"])
assert result.exit_code == 0, result.output
env = captured["env"]
assert env["KIMI_BASE_URL"] == "http://127.0.0.1:7000/p/my-project/v1"
def test_wrap_kimi_cli_fallback(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Falls back to the `kimi-cli` binary when `kimi` is not on PATH."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
captured: dict[str, Any] = {}
def fake_launch_tool(**kwargs: Any) -> None: # noqa: ANN003
captured.update(kwargs)
def fake_which(name: str) -> str | None:
return "/usr/local/bin/kimi-cli" if name == "kimi-cli" else None
with patch.object(wrap_mod.shutil, "which", side_effect=fake_which):
with patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool):
with patch.object(wrap_mod, "_project_name_from_cwd", return_value=None):
result = runner.invoke(main, ["wrap", "kimi"])
assert result.exit_code == 0, result.output
assert captured["binary"] == "/usr/local/bin/kimi-cli"
def test_wrap_kimi_not_found(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Error message when neither kimi nor kimi-cli is found."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
with patch.object(wrap_mod.shutil, "which", return_value=None):
result = runner.invoke(main, ["wrap", "kimi"])
assert result.exit_code == 1
assert "Error: 'kimi' (or 'kimi-cli') not found in PATH" in result.output
assert "https://github.com/MoonshotAI/kimi-cli" in result.output
def test_wrap_kimi_custom_port(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Custom --port is passed to _launch_tool and appears in KIMI_BASE_URL."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
captured: dict[str, Any] = {}
def fake_launch_tool(**kwargs: Any) -> None: # noqa: ANN003
captured.update(kwargs)
with patch.object(wrap_mod.shutil, "which", return_value="kimi"):
with patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool):
with patch.object(wrap_mod, "_project_name_from_cwd", return_value=None):
result = runner.invoke(main, ["wrap", "kimi", "--port", "9999"])
assert result.exit_code == 0, result.output
assert captured["port"] == 9999
assert captured["env"]["KIMI_BASE_URL"] == "http://127.0.0.1:9999/v1"
def test_wrap_kimi_custom_api_url(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""--kimi-api-url overrides the upstream endpoint passed to _launch_tool."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
captured: dict[str, Any] = {}
def fake_launch_tool(**kwargs: Any) -> None: # noqa: ANN003
captured.update(kwargs)
with patch.object(wrap_mod.shutil, "which", return_value="kimi"):
with patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool):
with patch.object(wrap_mod, "_project_name_from_cwd", return_value=None):
result = runner.invoke(
main,
["wrap", "kimi", "--kimi-api-url", "https://api.moonshot.ai/v1"],
)
assert result.exit_code == 0, result.output
assert captured["openai_api_url"] == "https://api.moonshot.ai/v1"
def test_wrap_kimi_no_proxy(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""--no-proxy flag prevents proxy startup."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
captured: dict[str, Any] = {}
def fake_launch_tool(**kwargs: Any) -> None: # noqa: ANN003
captured.update(kwargs)
with patch.object(wrap_mod.shutil, "which", return_value="kimi"):
with patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool):
with patch.object(wrap_mod, "_project_name_from_cwd", return_value=None):
result = runner.invoke(main, ["wrap", "kimi", "--no-proxy"])
assert result.exit_code == 0, result.output
assert captured["no_proxy"] is True
def test_wrap_kimi_learn_memory(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""--learn and --memory flags are passed to _launch_tool."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
captured: dict[str, Any] = {}
def fake_launch_tool(**kwargs: Any) -> None: # noqa: ANN003
captured.update(kwargs)
with patch.object(wrap_mod.shutil, "which", return_value="kimi"):
with patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool):
with patch.object(wrap_mod, "_project_name_from_cwd", return_value=None):
result = runner.invoke(main, ["wrap", "kimi", "--learn", "--memory"])
assert result.exit_code == 0, result.output
assert captured["learn"] is True
assert captured["memory"] is True