mirror of
https://github.com/TencentCloud/Octop.git
synced 2026-10-21 06:29:24 +08:00
2209cdc90f
Replace orcakit-harness-agent / harness-* with octop-harness, octop-gateway, octop-memory, and octop-browser, and align docs, UI copy, and generated paths. Co-authored-by: jubaoliang <jubaoliang@tencent.com> Co-authored-by: Cursor <cursoragent@cursor.com>
264 lines
8.6 KiB
Python
264 lines
8.6 KiB
Python
"""Unit tests for OpenCode Go session-header wiring (preset-only)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from octop.infra.agents.providers.opencode_session import (
|
|
OPENCODE_SESSION_HEADER,
|
|
ensure_opencode_session_header,
|
|
is_opencode_base_url,
|
|
is_opencode_preset,
|
|
session_header_for_provider,
|
|
)
|
|
from octop.infra.agents.providers.probe import build_probe_chat_model
|
|
from octop.infra.agents.providers.store import ProviderStore
|
|
|
|
GO_PRESET_ID = "opencode-go-openai"
|
|
GO_PRESET_NAME = "OpenCode Go (Compatible)"
|
|
GO_ANTHROPIC_ID = "opencode-go-anthropic"
|
|
GO_ANTHROPIC_NAME = "OpenCode Go (Anthropic)"
|
|
ZEN_PRESET_ID = "opencode-zen-openai"
|
|
ZEN_PRESET_NAME = "OpenCode Zen (Compatible)"
|
|
ZEN_ANTHROPIC_ID = "opencode-zen-anthropic"
|
|
ZEN_ANTHROPIC_NAME = "OpenCode Zen (Anthropic)"
|
|
GO_OPENAI_URL = "https://opencode.ai/zen/go/v1"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("name", "expected"),
|
|
[
|
|
(GO_PRESET_ID, True),
|
|
(GO_PRESET_NAME, True),
|
|
(GO_ANTHROPIC_ID, True),
|
|
(GO_ANTHROPIC_NAME, True),
|
|
(ZEN_PRESET_ID, True),
|
|
(ZEN_PRESET_NAME, True),
|
|
(ZEN_ANTHROPIC_ID, True),
|
|
(ZEN_ANTHROPIC_NAME, True),
|
|
("my-custom-go", False),
|
|
("", False),
|
|
(None, False),
|
|
],
|
|
)
|
|
def test_is_opencode_preset(name: str | None, expected: bool) -> None:
|
|
assert is_opencode_preset(name) is expected
|
|
|
|
|
|
def test_ensure_injects_throwaway_session_for_go_preset() -> None:
|
|
headers = ensure_opencode_session_header(GO_PRESET_ID, None)
|
|
assert headers[OPENCODE_SESSION_HEADER]
|
|
assert len(headers[OPENCODE_SESSION_HEADER]) == 32
|
|
|
|
|
|
def test_ensure_preserves_existing_header() -> None:
|
|
headers = ensure_opencode_session_header(GO_ANTHROPIC_ID, {OPENCODE_SESSION_HEADER: "custom"})
|
|
assert headers[OPENCODE_SESSION_HEADER] == "custom"
|
|
|
|
|
|
def test_ensure_leaves_unrelated_provider_untouched() -> None:
|
|
headers = ensure_opencode_session_header(
|
|
"custom",
|
|
{"Authorization": "Bearer x"},
|
|
base_url="https://api.example.com/v1",
|
|
)
|
|
assert headers == {"Authorization": "Bearer x"}
|
|
assert OPENCODE_SESSION_HEADER not in headers
|
|
|
|
|
|
def test_ensure_injects_for_custom_provider_on_opencode_host() -> None:
|
|
headers = ensure_opencode_session_header(
|
|
"my-relay",
|
|
None,
|
|
base_url="https://proxy.opencode.ai/v1",
|
|
)
|
|
assert headers[OPENCODE_SESSION_HEADER]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("base_url", "expected"),
|
|
[
|
|
("https://opencode.ai/zen/v1", True),
|
|
("https://opencode.ai/zen/go/v1", True),
|
|
("https://OPENCODE.AI/zen", True),
|
|
("https://proxy.opencode.ai/v1", True),
|
|
("https://go.opencode.ai", True),
|
|
("https://evil-opencode.ai/zen/go", False),
|
|
("https://opencode.ai.evil.example/v1", False),
|
|
("https://api.example.com/v1", False),
|
|
("", False),
|
|
(None, False),
|
|
],
|
|
)
|
|
def test_is_opencode_base_url(base_url: str | None, expected: bool) -> None:
|
|
assert is_opencode_base_url(base_url) is expected
|
|
|
|
|
|
def test_session_header_for_provider() -> None:
|
|
assert session_header_for_provider(GO_PRESET_NAME) == OPENCODE_SESSION_HEADER
|
|
assert session_header_for_provider("openai") is None
|
|
assert session_header_for_provider("custom", GO_OPENAI_URL) == OPENCODE_SESSION_HEADER
|
|
assert session_header_for_provider("custom", "https://api.example.com/v1") is None
|
|
|
|
|
|
def _probe_row(**overrides: Any) -> SimpleNamespace:
|
|
data: dict[str, Any] = {
|
|
"name": GO_PRESET_NAME,
|
|
"kind": "openai",
|
|
"base_url": GO_OPENAI_URL,
|
|
"api_key": "sk-test",
|
|
"extra_json": None,
|
|
"get_models": lambda: [{"id": "glm-5.2", "name": "GLM-5.2"}],
|
|
}
|
|
data.update(overrides)
|
|
return SimpleNamespace(**data)
|
|
|
|
|
|
def test_probe_chat_model_injects_throwaway_session_header() -> None:
|
|
with patch("octop_harness.llm.factory.build_chat_model") as mock_build:
|
|
mock_build.return_value = object()
|
|
build_probe_chat_model(_probe_row(), model_id="glm-5.2")
|
|
|
|
provider = mock_build.call_args[0][0]
|
|
assert OPENCODE_SESSION_HEADER in provider.headers
|
|
assert len(provider.headers[OPENCODE_SESSION_HEADER]) == 32
|
|
|
|
|
|
def test_probe_chat_model_keeps_user_session_header() -> None:
|
|
row = _probe_row(extra_json=json.dumps({"headers": {OPENCODE_SESSION_HEADER: "mine"}}))
|
|
with patch("octop_harness.llm.factory.build_chat_model") as mock_build:
|
|
mock_build.return_value = object()
|
|
build_probe_chat_model(row, model_id="glm-5.2")
|
|
|
|
provider = mock_build.call_args[0][0]
|
|
assert provider.headers[OPENCODE_SESSION_HEADER] == "mine"
|
|
|
|
|
|
def test_probe_injects_for_custom_provider_on_go_url() -> None:
|
|
row = _probe_row(name="my-relay")
|
|
with patch("octop_harness.llm.factory.build_chat_model") as mock_build:
|
|
mock_build.return_value = object()
|
|
build_probe_chat_model(row, model_id="glm-5.2")
|
|
|
|
provider = mock_build.call_args[0][0]
|
|
assert OPENCODE_SESSION_HEADER in provider.headers
|
|
|
|
|
|
class _FakeRepo:
|
|
def __init__(self, rows: list[SimpleNamespace]) -> None:
|
|
self._rows = rows
|
|
self.updates: list[dict[str, Any]] = []
|
|
|
|
def list_all(self) -> list[SimpleNamespace]:
|
|
return list(self._rows)
|
|
|
|
def update(self, provider_id: int, **kwargs: Any) -> None:
|
|
self.updates.append({"provider_id": provider_id, **kwargs})
|
|
|
|
|
|
def _row_with_extra(
|
|
extra_json: str | None,
|
|
*,
|
|
name: str = GO_PRESET_NAME,
|
|
base_url: str = GO_OPENAI_URL,
|
|
) -> SimpleNamespace:
|
|
models_json = json.dumps([{"id": "glm-5.2", "name": "GLM-5.2", "enabled": True}])
|
|
return SimpleNamespace(
|
|
id=7,
|
|
name=name,
|
|
kind="openai",
|
|
base_url=base_url,
|
|
api_key="sk-test",
|
|
extra_json=extra_json,
|
|
models_json=models_json,
|
|
enabled=True,
|
|
get_models=lambda: json.loads(models_json),
|
|
)
|
|
|
|
|
|
def test_store_sets_session_header_for_go_preset() -> None:
|
|
store = ProviderStore(_FakeRepo([_row_with_extra(None)])) # type: ignore[arg-type]
|
|
|
|
configs = store.build_harness_configs()
|
|
|
|
assert len(configs) == 1
|
|
assert configs[0].session_header == OPENCODE_SESSION_HEADER
|
|
assert OPENCODE_SESSION_HEADER not in configs[0].headers
|
|
|
|
|
|
def test_store_matches_preset_id() -> None:
|
|
row = _row_with_extra(None, name=GO_PRESET_ID)
|
|
configs = ProviderStore(_FakeRepo([row])).build_harness_configs() # type: ignore[arg-type]
|
|
assert configs[0].session_header == OPENCODE_SESSION_HEADER
|
|
|
|
|
|
def test_store_sets_session_header_for_zen_preset() -> None:
|
|
row = _row_with_extra(None, name=ZEN_PRESET_NAME, base_url="https://opencode.ai/zen/v1")
|
|
configs = ProviderStore(_FakeRepo([row])).build_harness_configs() # type: ignore[arg-type]
|
|
assert configs[0].session_header == OPENCODE_SESSION_HEADER
|
|
|
|
|
|
def test_store_does_not_write_provider_row() -> None:
|
|
repo = _FakeRepo([_row_with_extra(None)])
|
|
store = ProviderStore(repo) # type: ignore[arg-type]
|
|
|
|
store.build_harness_configs()
|
|
|
|
assert repo.updates == []
|
|
|
|
|
|
def test_store_sets_session_header_for_custom_provider_on_opencode_host() -> None:
|
|
row = _row_with_extra(None, name="custom-go")
|
|
configs = ProviderStore(_FakeRepo([row])).build_harness_configs() # type: ignore[arg-type]
|
|
|
|
assert configs and configs[0].session_header == OPENCODE_SESSION_HEADER
|
|
|
|
|
|
def test_store_skips_custom_provider_on_other_host() -> None:
|
|
row = _row_with_extra(None, name="custom", base_url="https://api.example.com/v1")
|
|
configs = ProviderStore(_FakeRepo([row])).build_harness_configs() # type: ignore[arg-type]
|
|
|
|
assert configs and configs[0].session_header is None
|
|
|
|
|
|
def test_knowledge_remote_embedding_sends_session_header() -> None:
|
|
from octop.infra.knowledge import embed as embed_module
|
|
|
|
provider = SimpleNamespace(
|
|
name=GO_PRESET_NAME,
|
|
base_url=GO_OPENAI_URL,
|
|
api_key="sk-test",
|
|
extra_json=None,
|
|
)
|
|
services = SimpleNamespace(
|
|
settings_repo=SimpleNamespace(
|
|
get=lambda key: {
|
|
"knowledge_embedding_backend": "remote",
|
|
"knowledge_embedding_model": "embed-1",
|
|
"knowledge_embedding_provider_id": "3",
|
|
}.get(key)
|
|
),
|
|
provider_repo=SimpleNamespace(get=lambda _pid: provider),
|
|
)
|
|
seen: dict[str, str] = {}
|
|
|
|
def fake_batched(
|
|
_client: Any,
|
|
_base: str,
|
|
headers: dict[str, str],
|
|
_model: str,
|
|
_texts: list[str],
|
|
) -> list[list[float]]:
|
|
seen.update(headers)
|
|
return [[0.1]]
|
|
|
|
with patch.object(embed_module, "_embed_remote_batched", fake_batched):
|
|
embed_module.embed_knowledge_texts(services, ["hello"])
|
|
|
|
assert seen[OPENCODE_SESSION_HEADER]
|