mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
## Description `headroom wrap opencode` currently routes `headroom/*` models only to ordinary Anthropic or OpenAI backends. A user with a GitHub Copilot subscription cannot point those `headroom/*` requests at the Copilot seat while keeping Headroom compression and stats, even though Headroom already has the validated subscription resolver, the proxy seed path, and the OpenCode provider route needed to do it. This PR adds `--copilot-subscription` to the OpenCode wrap command. It reuses the existing Copilot subscription token resolver, passes the validated endpoint and token seed into the existing proxy startup path, rejects unsupported runtime modes, and treats any non-empty Copilot API token as a private session seed so token-only sessions do not reuse a shared proxy. The generated OpenCode provider still targets the local proxy, and subscription secrets stay out of OpenCode config, environment, and terminal output. Closes #2441 ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [x] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - Add `headroom wrap opencode --copilot-subscription` in `headroom/cli/wrap.py`. - Reuse the existing validated Copilot subscription resolver through one small required-resolution helper shared with the dedicated Copilot wrapper. - Pass the resolved endpoint and token seed into `_ensure_proxy()` as `openai_api_url`, `copilot_api_token`, `copilot_refresh_oauth_token`, and `copilot_api_token_expires_at`. - Reject `--copilot-subscription` with `--no-proxy`, `--prepare-only`, and translated backends before proxy or OpenCode launch. - Validate subscription mode before snapshotting OpenCode config, so rejected invocations don't create stale backups. - Scrub inherited Copilot proxy seed variables from the OpenCode child environment. - Treat any non-empty Copilot API token as a private session seed so token-only sessions do not reuse shared or persistent proxies. - Add focused OpenCode and persistent-proxy coverage for seed handoff, guard failures, direct-token isolation, secret non-disclosure, and unchanged non-subscription behavior. - Leave `CHANGELOG.md` untouched because Headroom generates changelog entries from conventional commits. ## Testing - [x] Unit tests pass (`uv run pytest tests/test_cli/test_wrap_opencode.py tests/test_cli/test_wrap_persistent.py tests/test_cli/test_wrap_copilot.py -q`, `106 passed in 136.65s`) - [x] Linting passes (`uv run ruff check headroom/cli/wrap.py tests/test_cli/test_wrap_opencode.py tests/test_cli/test_wrap_persistent.py`) - [ ] Type checking passes (`uv run mypy headroom`) - [x] New tests added for new functionality when applicable - [ ] Manual testing performed ### Test Output ```text Targeted subscription tests pass: OpenCode `7 passed, 37 deselected in 0.36s`, persistent proxy `2 passed, 29 deselected in 0.34s`, and dedicated Copilot `11 passed, 20 deselected in 0.39s`. Coverage includes inherited resolver-input env scrubbing, HEADROOM_BACKEND rejection, no-backup-on-rejection, OpenCode-only scrub scoping, and private-proxy teardown on config-injection failure. The full focused command `uv run pytest tests/test_cli/test_wrap_opencode.py tests/test_cli/test_wrap_persistent.py tests/test_cli/test_wrap_copilot.py -q` passed with `106 passed in 136.65s`. Ruff check and format check pass. ``` ## Real Behavior Proof - Environment: Windows, `uv` development environment, local CLI tests, no live Copilot seat on this host - Exact command / steps: run the focused OpenCode and persistent-proxy tests with a mocked `CopilotSubscriptionTokenResolution`, then capture the proof rows for seed handoff, direct-token isolation, guards, and secret non-disclosure - Observed result: Targeted subscription and proxy-seed tests pass, including OpenCode-only resolver-input env scrubbing and private-proxy teardown on config-injection failure; the full focused command passed with `106 passed in 136.65s`; Ruff check and format check pass. - Not tested: live Copilot subscription seat run on this host ## 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 - [ ] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my feature works - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Additional Notes - Feature approval comes from the open `enhancement` label on https://github.com/headroomlabs-ai/headroom/issues/2441. - Keep the final live-backend claim behind manual owner proof. Local CLI tests can prove config, guard, secret, and proxy-seed behavior, but they cannot prove a real Copilot seat on this host. - `CHANGELOG.md` remains untouched because Headroom's release pipeline generates changelog entries from conventional commits.
This commit is contained in:
Rod Boev
GitHub
parent
4aac068814
commit
9089e7f7d3
3 changed files with 343 additions and 31 deletions
|
|
@ -58,6 +58,12 @@ from headroom.agent_savings import (
|
|||
apply_agent_savings_env_defaults,
|
||||
)
|
||||
from headroom.copilot_auth import (
|
||||
_API_TOKEN_ENV_VARS,
|
||||
_API_TOKEN_EXPIRES_AT_ENV_VAR,
|
||||
_COPILOT_OAUTH_TOKEN_ENV_VARS,
|
||||
_GENERIC_GITHUB_TOKEN_ENV_VARS,
|
||||
_REFRESH_OAUTH_TOKEN_ENV_VAR,
|
||||
CopilotSubscriptionTokenResolution,
|
||||
has_oauth_auth,
|
||||
resolve_client_bearer_token,
|
||||
resolve_copilot_api_url,
|
||||
|
|
@ -167,8 +173,15 @@ from .main import main
|
|||
|
||||
_COPILOT_PROXY_SEED_ENV_VARS = (
|
||||
"GITHUB_COPILOT_API_TOKEN",
|
||||
"GITHUB_COPILOT_REFRESH_OAUTH_TOKEN",
|
||||
"GITHUB_COPILOT_API_TOKEN_EXPIRES_AT",
|
||||
_REFRESH_OAUTH_TOKEN_ENV_VAR,
|
||||
_API_TOKEN_EXPIRES_AT_ENV_VAR,
|
||||
)
|
||||
_COPILOT_SUBSCRIPTION_LAUNCH_SECRET_ENV_VARS = (
|
||||
*_API_TOKEN_ENV_VARS,
|
||||
_REFRESH_OAUTH_TOKEN_ENV_VAR,
|
||||
_API_TOKEN_EXPIRES_AT_ENV_VAR,
|
||||
*_COPILOT_OAUTH_TOKEN_ENV_VARS,
|
||||
*_GENERIC_GITHUB_TOKEN_ENV_VARS,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -177,6 +190,11 @@ def _scrub_copilot_proxy_seed_env(env: dict[str, str]) -> None:
|
|||
env.pop(key, None)
|
||||
|
||||
|
||||
def _scrub_copilot_subscription_launch_env(env: dict[str, str]) -> None:
|
||||
for key in _COPILOT_SUBSCRIPTION_LAUNCH_SECRET_ENV_VARS:
|
||||
env.pop(key, None)
|
||||
|
||||
|
||||
def _read_text(path: Path) -> str:
|
||||
"""Read a text file as UTF-8, falling back to the system locale encoding."""
|
||||
return fsutil.read_text(path)
|
||||
|
|
@ -3982,8 +4000,10 @@ def _ensure_proxy(
|
|||
) -> tuple[subprocess.Popen | None, int]:
|
||||
"""Start or verify proxy. Returns (process_handle, actual_port)."""
|
||||
helpers = _live_wrap_module()
|
||||
copilot_subscription_seed_requested = bool(copilot_refresh_oauth_token) or (
|
||||
copilot_api_token_expires_at is not None
|
||||
copilot_subscription_seed_requested = (
|
||||
bool(copilot_api_token)
|
||||
or bool(copilot_refresh_oauth_token)
|
||||
or copilot_api_token_expires_at is not None
|
||||
)
|
||||
# --no-proxy reuses an already-running proxy, so backend/region/provider
|
||||
# flags (which only apply when we start one) would be silently dropped.
|
||||
|
|
@ -3999,7 +4019,7 @@ def _ensure_proxy(
|
|||
)
|
||||
if isolated_copilot_subscription_proxy:
|
||||
click.echo(
|
||||
" Copilot subscription refresh seeds are session-specific; "
|
||||
" Copilot subscription seeds are session-specific; "
|
||||
"starting a dedicated local proxy instance for this wrap session."
|
||||
)
|
||||
if not isolated_copilot_subscription_proxy and manifest is not None:
|
||||
|
|
@ -5401,6 +5421,17 @@ def unwrap_claude(
|
|||
# =============================================================================
|
||||
|
||||
|
||||
def _require_copilot_subscription_resolution() -> CopilotSubscriptionTokenResolution:
|
||||
resolution = resolve_subscription_bearer_token_details()
|
||||
if resolution is None:
|
||||
raise click.ClickException(
|
||||
"GitHub Copilot subscription mode requires a reusable GitHub/Copilot bearer "
|
||||
"token, but none could be resolved. Run `headroom copilot-auth login` first, or set "
|
||||
"GITHUB_COPILOT_TOKEN / GITHUB_COPILOT_GITHUB_TOKEN."
|
||||
)
|
||||
return resolution
|
||||
|
||||
|
||||
@wrap.command(context_settings={"ignore_unknown_options": True})
|
||||
@_rtk_option
|
||||
@click.option(
|
||||
|
|
@ -5543,7 +5574,8 @@ def copilot(
|
|||
copilot_proxy_token: str | None = None
|
||||
copilot_refresh_oauth_token: str | None = None
|
||||
copilot_api_token_expires_at: float | None = None
|
||||
subscription_resolution = None
|
||||
client_bearer: str | None = None
|
||||
subscription_resolution: CopilotSubscriptionTokenResolution | None = None
|
||||
if _should_use_copilot_oauth(
|
||||
backend=effective_backend,
|
||||
provider_type=provider_type,
|
||||
|
|
@ -5551,10 +5583,8 @@ def copilot(
|
|||
force_subscription=subscription,
|
||||
):
|
||||
if subscription:
|
||||
subscription_resolution = resolve_subscription_bearer_token_details()
|
||||
client_bearer = (
|
||||
subscription_resolution.token if subscription_resolution is not None else None
|
||||
)
|
||||
subscription_resolution = _require_copilot_subscription_resolution()
|
||||
client_bearer = subscription_resolution.token
|
||||
else:
|
||||
client_bearer = resolve_client_bearer_token()
|
||||
if not client_bearer:
|
||||
|
|
@ -7646,6 +7676,11 @@ def openclaw(
|
|||
help="Enable code graph indexing via codebase-memory-mcp (optional)",
|
||||
)
|
||||
@click.option("--no-proxy", is_flag=True, help="Skip proxy startup (use existing proxy)")
|
||||
@click.option(
|
||||
"--copilot-subscription",
|
||||
is_flag=True,
|
||||
help="Route headroom/* models through the authenticated GitHub Copilot subscription",
|
||||
)
|
||||
@click.option("--learn", is_flag=True, help="Enable live traffic learning")
|
||||
@click.option("--memory", is_flag=True, help="Enable persistent cross-session memory")
|
||||
@click.option(
|
||||
|
|
@ -7664,6 +7699,7 @@ def opencode(
|
|||
no_serena: bool,
|
||||
code_graph: bool,
|
||||
no_proxy: bool,
|
||||
copilot_subscription: bool,
|
||||
learn: bool,
|
||||
memory: bool,
|
||||
backend: str | None,
|
||||
|
|
@ -7690,7 +7726,28 @@ def opencode(
|
|||
headroom wrap opencode --no-serena # Skip Serena MCP registration
|
||||
headroom wrap opencode --port 9999 # Custom proxy port
|
||||
headroom wrap opencode --backend anyllm --anyllm-provider groq
|
||||
headroom wrap opencode --copilot-subscription # Use a GitHub Copilot subscription
|
||||
"""
|
||||
subscription_resolution = None
|
||||
if copilot_subscription:
|
||||
effective_backend = backend or os.environ.get("HEADROOM_BACKEND")
|
||||
if effective_backend not in (None, "", "anthropic"):
|
||||
raise click.ClickException(
|
||||
"--copilot-subscription cannot be combined with translated backends "
|
||||
"such as anyllm or litellm-*; use the anthropic backend."
|
||||
)
|
||||
if no_proxy:
|
||||
raise click.ClickException(
|
||||
"--copilot-subscription cannot be combined with --no-proxy because "
|
||||
"it requires a private seeded proxy."
|
||||
)
|
||||
if prepare_only:
|
||||
raise click.ClickException(
|
||||
"--copilot-subscription cannot be combined with --prepare-only because "
|
||||
"it requires a running private seeded proxy."
|
||||
)
|
||||
subscription_resolution = _require_copilot_subscription_resolution()
|
||||
|
||||
# Snapshot OpenCode config.json BEFORE any wrap-time mutation so
|
||||
# `headroom unwrap opencode` can restore the user's pre-wrap state.
|
||||
_opencode_config_file, _opencode_backup_file = opencode_config_paths()
|
||||
|
|
@ -7770,33 +7827,44 @@ def opencode(
|
|||
backend=backend,
|
||||
anyllm_provider=anyllm_provider,
|
||||
region=region,
|
||||
openai_api_url=(subscription_resolution.api_url if subscription_resolution else None),
|
||||
copilot_api_token=(subscription_resolution.token if subscription_resolution else None),
|
||||
copilot_refresh_oauth_token=(
|
||||
subscription_resolution.refresh_oauth_token if subscription_resolution else None
|
||||
),
|
||||
copilot_api_token_expires_at=(
|
||||
subscription_resolution.api_token_expires_at if subscription_resolution else None
|
||||
),
|
||||
)
|
||||
|
||||
# If the proxy fell back to a different port, move our marker so
|
||||
# cleanup tracking stays accurate and update MCP config.
|
||||
if actual_port != port:
|
||||
_unregister_proxy_client(port)
|
||||
_register_proxy_client(actual_port)
|
||||
if not no_mcp:
|
||||
from headroom.mcp_registry import OpencodeRegistrar
|
||||
try:
|
||||
# If the proxy fell back to a different port, move our marker so
|
||||
# cleanup tracking stays accurate and update MCP config.
|
||||
if actual_port != port:
|
||||
_unregister_proxy_client(port)
|
||||
_register_proxy_client(actual_port)
|
||||
if not no_mcp:
|
||||
from headroom.mcp_registry import OpencodeRegistrar
|
||||
|
||||
_setup_headroom_mcp(OpencodeRegistrar(), actual_port, verbose=verbose, force=True)
|
||||
_setup_headroom_mcp(OpencodeRegistrar(), actual_port, verbose=verbose, force=True)
|
||||
|
||||
env, env_vars_display = _build_opencode_launch_env(
|
||||
actual_port, os.environ, project=_project_name_from_cwd(), include_mcp=not no_mcp
|
||||
)
|
||||
|
||||
# Inject Headroom provider into OpenCode config so traffic routes through proxy.
|
||||
inject_opencode_provider_config(actual_port)
|
||||
if memory:
|
||||
mem_dir = Path.cwd() / ".headroom"
|
||||
_inject_memory_mcp_config(
|
||||
os.environ.get("USER", os.environ.get("USERNAME", "default")),
|
||||
launch_environ = os.environ.copy()
|
||||
if subscription_resolution is not None:
|
||||
_scrub_copilot_subscription_launch_env(launch_environ)
|
||||
env, env_vars_display = _build_opencode_launch_env(
|
||||
actual_port, launch_environ, project=_project_name_from_cwd(), include_mcp=not no_mcp
|
||||
)
|
||||
|
||||
# Proxy already started by _ensure_proxy above; tell _launch_tool to
|
||||
# skip duplicate startup.
|
||||
try:
|
||||
# Inject Headroom provider into OpenCode config so traffic routes through proxy.
|
||||
inject_opencode_provider_config(actual_port)
|
||||
if memory:
|
||||
mem_dir = Path.cwd() / ".headroom"
|
||||
_inject_memory_mcp_config(
|
||||
os.environ.get("USER", os.environ.get("USERNAME", "default")),
|
||||
)
|
||||
|
||||
# Proxy already started by _ensure_proxy above; tell _launch_tool to
|
||||
# skip duplicate startup.
|
||||
_launch_tool(
|
||||
binary=opencode_bin,
|
||||
args=opencode_args,
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ from click.testing import CliRunner
|
|||
|
||||
from headroom.cli import wrap as wrap_mod
|
||||
from headroom.cli.main import main
|
||||
from headroom.copilot_auth import CopilotSubscriptionTokenResolution
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
|
|
@ -32,11 +33,225 @@ def _set_test_home(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|||
monkeypatch.delenv("OPENCODE_CONFIG", raising=False)
|
||||
|
||||
|
||||
def _subscription_resolution() -> CopilotSubscriptionTokenResolution:
|
||||
return CopilotSubscriptionTokenResolution(
|
||||
token="copilot-api-secret",
|
||||
source="test",
|
||||
confidence="test",
|
||||
api_url="https://api.githubcopilot.com",
|
||||
token_fingerprint="sha256:test",
|
||||
refresh_oauth_token="copilot-refresh-secret",
|
||||
api_token_expires_at=123.5,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Wrap opencode
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_wrap_opencode_copilot_subscription_handoffs_seed_after_actual_port(
|
||||
runner: CliRunner,
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
monkeypatch.setenv("GITHUB_COPILOT_API_TOKEN", "inherited-api-secret")
|
||||
monkeypatch.setenv("GITHUB_COPILOT_REFRESH_OAUTH_TOKEN", "inherited-refresh-secret")
|
||||
monkeypatch.setenv("GITHUB_COPILOT_API_TOKEN_EXPIRES_AT", "999.0")
|
||||
monkeypatch.setenv("GITHUB_COPILOT_TOKEN", "inherited-seat-token")
|
||||
monkeypatch.setenv("GITHUB_COPILOT_GITHUB_TOKEN", "inherited-github-token")
|
||||
monkeypatch.setenv("COPILOT_GITHUB_TOKEN", "inherited-alt-github-token")
|
||||
monkeypatch.setenv("COPILOT_PROVIDER_BEARER_TOKEN", "inherited-provider-bearer")
|
||||
monkeypatch.setenv("GH_TOKEN", "inherited-gh-token")
|
||||
monkeypatch.setenv("GITHUB_TOKEN", "inherited-github-pat")
|
||||
captured: dict[str, object] = {}
|
||||
|
||||
def fake_ensure_proxy(*args, **kwargs): # noqa: ANN002, ANN003
|
||||
captured["ensure"] = kwargs
|
||||
return None, 9010
|
||||
|
||||
def fake_launch_tool(**kwargs): # noqa: ANN003
|
||||
captured["launch"] = kwargs
|
||||
|
||||
with (
|
||||
patch.object(wrap_mod.shutil, "which", return_value="opencode"),
|
||||
patch.object(
|
||||
wrap_mod,
|
||||
"_require_copilot_subscription_resolution",
|
||||
return_value=_subscription_resolution(),
|
||||
),
|
||||
patch.object(wrap_mod, "_ensure_proxy", side_effect=fake_ensure_proxy),
|
||||
patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool),
|
||||
):
|
||||
result = runner.invoke(
|
||||
main,
|
||||
[
|
||||
"wrap",
|
||||
"opencode",
|
||||
"--copilot-subscription",
|
||||
"--no-rtk",
|
||||
"--no-mcp",
|
||||
"--no-serena",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
ensure = captured["ensure"]
|
||||
assert ensure["openai_api_url"] == "https://api.githubcopilot.com"
|
||||
assert ensure["copilot_api_token"] == "copilot-api-secret"
|
||||
assert ensure["copilot_refresh_oauth_token"] == "copilot-refresh-secret"
|
||||
assert ensure["copilot_api_token_expires_at"] == 123.5
|
||||
launch = captured["launch"]
|
||||
assert launch["port"] == 9010
|
||||
assert "copilot-api-secret" not in result.output
|
||||
assert "copilot-api-secret" not in str(launch["env"])
|
||||
assert "copilot-refresh-secret" not in str(launch["env"])
|
||||
assert "copilot-api-secret" not in launch["env"]["OPENCODE_CONFIG_CONTENT"]
|
||||
assert "GITHUB_COPILOT_API_TOKEN" not in launch["env"]
|
||||
assert "GITHUB_COPILOT_REFRESH_OAUTH_TOKEN" not in launch["env"]
|
||||
assert "GITHUB_COPILOT_API_TOKEN_EXPIRES_AT" not in launch["env"]
|
||||
assert "GITHUB_COPILOT_TOKEN" not in launch["env"]
|
||||
assert "GITHUB_COPILOT_GITHUB_TOKEN" not in launch["env"]
|
||||
assert "COPILOT_GITHUB_TOKEN" not in launch["env"]
|
||||
assert "COPILOT_PROVIDER_BEARER_TOKEN" not in launch["env"]
|
||||
assert "GH_TOKEN" not in launch["env"]
|
||||
assert "GITHUB_TOKEN" not in launch["env"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"extra_args, message",
|
||||
[
|
||||
(["--no-proxy"], "--no-proxy"),
|
||||
(["--prepare-only"], "--prepare-only"),
|
||||
(["--backend", "anyllm"], "translated backends"),
|
||||
],
|
||||
)
|
||||
def test_wrap_opencode_copilot_subscription_rejects_incompatible_modes(
|
||||
runner: CliRunner,
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
extra_args: list[str],
|
||||
message: str,
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
config_file = tmp_path / ".config" / "opencode" / "opencode.json"
|
||||
config_file.parent.mkdir(parents=True)
|
||||
config_file.write_text("{}", encoding="utf-8")
|
||||
with patch.object(wrap_mod, "_ensure_proxy", side_effect=AssertionError("proxy launched")):
|
||||
result = runner.invoke(
|
||||
main,
|
||||
["wrap", "opencode", "--copilot-subscription", "--no-rtk", "--no-mcp", *extra_args],
|
||||
)
|
||||
assert result.exit_code == 1
|
||||
assert message in result.output
|
||||
assert not config_file.with_name("opencode.json.headroom-backup").exists()
|
||||
|
||||
|
||||
def test_wrap_opencode_copilot_subscription_rejects_headroom_backend_env(
|
||||
runner: CliRunner,
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
monkeypatch.setenv("HEADROOM_BACKEND", "anyllm")
|
||||
with patch.object(wrap_mod, "_ensure_proxy", side_effect=AssertionError("proxy launched")):
|
||||
result = runner.invoke(
|
||||
main,
|
||||
["wrap", "opencode", "--copilot-subscription", "--no-rtk", "--no-mcp"],
|
||||
)
|
||||
assert result.exit_code == 1
|
||||
assert "translated backends" in result.output
|
||||
|
||||
|
||||
def test_wrap_opencode_copilot_subscription_requires_login_before_launch(
|
||||
runner: CliRunner,
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
with (
|
||||
patch.object(
|
||||
wrap_mod,
|
||||
"resolve_subscription_bearer_token_details",
|
||||
return_value=None,
|
||||
),
|
||||
patch.object(wrap_mod, "_ensure_proxy", side_effect=AssertionError("proxy launched")),
|
||||
):
|
||||
result = runner.invoke(
|
||||
main,
|
||||
["wrap", "opencode", "--copilot-subscription", "--no-rtk", "--no-mcp"],
|
||||
)
|
||||
assert result.exit_code == 1
|
||||
assert "headroom copilot-auth login" in result.output
|
||||
|
||||
|
||||
def test_wrap_opencode_copilot_subscription_cleans_up_proxy_on_config_failure(
|
||||
runner: CliRunner,
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
|
||||
class _FakeProxy:
|
||||
def __init__(self) -> None:
|
||||
self.terminated = False
|
||||
self.wait_timeout: float | None = None
|
||||
|
||||
def poll(self) -> None:
|
||||
return None
|
||||
|
||||
def terminate(self) -> None:
|
||||
self.terminated = True
|
||||
|
||||
def wait(self, timeout: float | None = None) -> int:
|
||||
self.wait_timeout = timeout
|
||||
return 0
|
||||
|
||||
proxy = _FakeProxy()
|
||||
|
||||
with (
|
||||
patch.object(wrap_mod.shutil, "which", return_value="opencode"),
|
||||
patch.object(
|
||||
wrap_mod,
|
||||
"_require_copilot_subscription_resolution",
|
||||
return_value=_subscription_resolution(),
|
||||
),
|
||||
patch.object(wrap_mod, "_ensure_proxy", return_value=(proxy, 9010)),
|
||||
patch.object(wrap_mod, "_register_proxy_client"),
|
||||
patch.object(wrap_mod, "_unregister_proxy_client"),
|
||||
patch.object(wrap_mod, "_live_proxy_clients", return_value=[]),
|
||||
patch.object(
|
||||
wrap_mod,
|
||||
"inject_opencode_provider_config",
|
||||
side_effect=RuntimeError("config write failed"),
|
||||
),
|
||||
patch.object(wrap_mod, "_launch_tool", side_effect=AssertionError("launch should not run")),
|
||||
):
|
||||
result = runner.invoke(
|
||||
main,
|
||||
[
|
||||
"wrap",
|
||||
"opencode",
|
||||
"--copilot-subscription",
|
||||
"--no-rtk",
|
||||
"--no-mcp",
|
||||
"--no-serena",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert isinstance(result.exception, RuntimeError)
|
||||
assert str(result.exception) == "config write failed"
|
||||
assert proxy.terminated is True
|
||||
assert proxy.wait_timeout == 5
|
||||
|
||||
|
||||
def test_wrap_opencode_sets_config_content_env(
|
||||
runner: CliRunner,
|
||||
tmp_path: Path,
|
||||
|
|
|
|||
|
|
@ -417,6 +417,35 @@ def test_ensure_proxy_starts_isolated_ephemeral_proxy_for_copilot_subscription_s
|
|||
assert calls[1][2]["copilot_api_token_expires_at"] == 456.5
|
||||
|
||||
|
||||
def test_ensure_proxy_isolates_copilot_subscription_seed_with_api_token_only(monkeypatch) -> None:
|
||||
calls: list[object] = []
|
||||
|
||||
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: None)
|
||||
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: port == 8787)
|
||||
monkeypatch.setattr(
|
||||
wrap_cli,
|
||||
"_find_available_port",
|
||||
lambda start_port, **kw: calls.append(("find_port", start_port)) or 8788,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
wrap_cli,
|
||||
"_start_proxy",
|
||||
lambda *args, **kwargs: calls.append(("start", args, kwargs)),
|
||||
)
|
||||
|
||||
proc, actual_port = wrap_cli._ensure_proxy(
|
||||
8787,
|
||||
False,
|
||||
copilot_api_token="direct-token-only",
|
||||
)
|
||||
|
||||
assert proc is None
|
||||
assert actual_port == 8788
|
||||
assert calls[0] == ("find_port", 8788)
|
||||
assert calls[1][0] == "start"
|
||||
assert calls[1][2]["copilot_api_token"] == "direct-token-only"
|
||||
|
||||
|
||||
def test_ensure_proxy_starts_isolated_ephemeral_proxy_when_subscription_seed_targets_persistent_port(
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue