2058 lines
78 KiB
Python
2058 lines
78 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests for the project-local Grok Build integration.
|
|
|
|
The suite deliberately uses fake files and mocks. It never resolves a release
|
|
version, downloads a binary, starts Grok, or contacts an xAI service.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
import urllib.error
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
try:
|
|
import tomllib
|
|
except ImportError: # pragma: no cover - Python 3.11+ is used by the project.
|
|
tomllib = None
|
|
|
|
from grok_build_bridge import (
|
|
MODEL_API_KEY_ENV,
|
|
MODEL_PROFILE,
|
|
MANAGED_CONFIG_BEGIN,
|
|
MANAGED_CONFIG_END,
|
|
GrokBuildError,
|
|
GrokBuildManager,
|
|
classify_passthrough_args,
|
|
enforce_passthrough_agent_model,
|
|
normalize_passthrough_args,
|
|
parse_streaming_event,
|
|
reject_passthrough_model_bypasses,
|
|
resolve_passthrough_workspace,
|
|
)
|
|
|
|
|
|
class GrokBuildManagerTestCase(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._temporary_directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self._temporary_directory.cleanup)
|
|
self.project_dir = Path(self._temporary_directory.name) / "project"
|
|
self.project_dir.mkdir()
|
|
self.runtime_home = self.project_dir / ".grok-test"
|
|
self.ai_settings_file = self.project_dir / "ai_settings.json"
|
|
self.integration_settings_file = self.project_dir / "grok_build_settings.json"
|
|
self.ai_settings = {
|
|
"AI_API_BASE": "https://models.example.test/v1/chat/completions",
|
|
"AI_API_KEY": "backend-super-secret",
|
|
"AI_MODEL": "local-qwen",
|
|
"AI_TEMPERATURE": 0.55,
|
|
"AI_MAX_TOKENS": 4096,
|
|
"AI_MCP_SERVERS": [],
|
|
"GROK_MODEL_ENABLED": True,
|
|
"GROK_API_BASE": "https://models.example.test/v1/chat/completions",
|
|
"GROK_API_KEY": "backend-super-secret",
|
|
"GROK_MODEL": "local-qwen",
|
|
"GROK_API_BACKEND": "chat_completions",
|
|
"GROK_AUTH_SCHEME": "auto",
|
|
"GROK_CONTEXT_WINDOW": 65536,
|
|
"GROK_MAX_TOKENS": 4096,
|
|
"GROK_TEMPERATURE": 0.55,
|
|
}
|
|
self.integration_settings = {
|
|
"context_window": 65536,
|
|
"sync_backend_model": True,
|
|
"sync_mcp_servers": False,
|
|
"customer_service_tools": False,
|
|
}
|
|
self._write_json(self.ai_settings_file, self.ai_settings)
|
|
self._write_json(self.integration_settings_file, self.integration_settings)
|
|
self.manager = GrokBuildManager(
|
|
project_dir=self.project_dir,
|
|
runtime_home=self.runtime_home,
|
|
ai_settings_file=self.ai_settings_file,
|
|
integration_settings_file=self.integration_settings_file,
|
|
)
|
|
self.manager.user_home = self.project_dir / "isolated-user-home"
|
|
|
|
@staticmethod
|
|
def _write_json(path: Path, value: object) -> None:
|
|
path.write_text(
|
|
json.dumps(value, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def _replace_ai_settings(self, **changes: object) -> None:
|
|
self.ai_settings.update(changes)
|
|
self._write_json(self.ai_settings_file, self.ai_settings)
|
|
|
|
def test_chat_auto_approve_setting_is_explicit_and_persisted(self) -> None:
|
|
self.assertFalse(
|
|
bool(
|
|
self.manager.load_integration_settings().get(
|
|
"chat_auto_approve"
|
|
)
|
|
)
|
|
)
|
|
|
|
saved = self.manager.save_integration_settings(
|
|
{"chat_auto_approve": True}
|
|
)
|
|
|
|
self.assertTrue(saved["chat_auto_approve"])
|
|
self.assertTrue(
|
|
self.manager.load_integration_settings()["chat_auto_approve"]
|
|
)
|
|
|
|
def test_openai_chat_completions_endpoint_is_normalized(self) -> None:
|
|
profile = self.manager.model_profile()
|
|
|
|
self.assertTrue(profile.compatible)
|
|
self.assertEqual("https://models.example.test/v1", profile.base_url)
|
|
self.assertEqual("chat_completions", profile.api_backend)
|
|
self.assertEqual("local-qwen", profile.model)
|
|
|
|
def test_responses_endpoint_is_normalized(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE="https://responses.example.test/openai/v1/responses",
|
|
GROK_API_BACKEND="responses",
|
|
)
|
|
|
|
profile = self.manager.model_profile()
|
|
|
|
self.assertTrue(profile.compatible)
|
|
self.assertEqual(
|
|
"https://responses.example.test/openai/v1",
|
|
profile.base_url,
|
|
)
|
|
self.assertEqual("responses", profile.api_backend)
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_anthropic_messages_endpoint_is_normalized(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE="https://api.anthropic.com/v1/messages",
|
|
GROK_API_BACKEND="messages",
|
|
)
|
|
|
|
profile = self.manager.model_profile()
|
|
|
|
self.assertTrue(profile.compatible)
|
|
self.assertEqual("https://api.anthropic.com/v1", profile.base_url)
|
|
self.assertEqual("messages", profile.api_backend)
|
|
self.assertEqual("x_api_key", profile.auth_scheme)
|
|
self.manager.sync_model_configuration()
|
|
parsed = tomllib.loads(
|
|
self.manager.user_config_file.read_text(encoding="utf-8")
|
|
)
|
|
model = parsed["model"][MODEL_PROFILE]
|
|
self.assertEqual("x_api_key", model["auth_scheme"])
|
|
self.assertEqual("2023-06-01", model["extra_headers"]["anthropic-version"])
|
|
self.assertNotIn(self.ai_settings["GROK_API_KEY"], json.dumps(model))
|
|
|
|
def test_messages_proxy_can_explicitly_select_x_api_key_auth(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_MODEL_ENABLED=True,
|
|
GROK_API_BASE="https://proxy.example.test/v1",
|
|
GROK_API_KEY="proxy-secret",
|
|
GROK_MODEL="claude-compatible",
|
|
GROK_API_BACKEND="messages",
|
|
GROK_AUTH_SCHEME="x_api_key",
|
|
)
|
|
|
|
profile = self.manager.model_profile()
|
|
|
|
self.assertTrue(profile.compatible)
|
|
self.assertEqual("messages", profile.api_backend)
|
|
self.assertEqual("x_api_key", profile.auth_scheme)
|
|
|
|
def test_custom_model_without_key_is_rejected_before_xai_fallback(self) -> None:
|
|
self._replace_ai_settings(GROK_API_KEY="")
|
|
|
|
profile = self.manager.model_profile()
|
|
|
|
self.assertFalse(profile.compatible)
|
|
self.assertIn("xAI", profile.reason)
|
|
|
|
def test_agent_model_never_falls_back_to_customer_or_xai_model(self) -> None:
|
|
self._replace_ai_settings(GROK_MODEL_ENABLED=False)
|
|
|
|
profile = self.manager.agent_model_profile()
|
|
result = self.manager.sync_model_configuration()
|
|
|
|
self.assertFalse(profile.compatible)
|
|
self.assertFalse(result.compatible)
|
|
self.assertEqual("", profile.model)
|
|
self.assertIn("不会回退到 Grok/xAI", profile.reason)
|
|
|
|
def test_endpoint_query_is_rejected_instead_of_building_the_wrong_url(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE=(
|
|
"https://models.example.test/v1/chat/completions?api-version=2026-01"
|
|
)
|
|
)
|
|
|
|
profile = self.manager.model_profile()
|
|
|
|
self.assertFalse(profile.compatible)
|
|
self.assertIn("query", profile.reason)
|
|
|
|
def test_api_base_without_operation_defaults_to_openai_compatibility(self) -> None:
|
|
self._replace_ai_settings(GROK_API_BASE="http://127.0.0.1:9000/v1/")
|
|
|
|
profile = self.manager.model_profile()
|
|
|
|
self.assertTrue(profile.compatible)
|
|
self.assertEqual("http://127.0.0.1:9000/v1", profile.base_url)
|
|
self.assertEqual("chat_completions", profile.api_backend)
|
|
|
|
def test_dify_endpoint_requires_matching_backend_selection(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE="https://dify.example.test/v1/chat-messages"
|
|
)
|
|
|
|
profile = self.manager.model_profile()
|
|
|
|
self.assertFalse(profile.compatible)
|
|
self.assertEqual("dify", profile.api_backend)
|
|
self.assertIn("Dify", profile.reason)
|
|
|
|
def test_dify_backend_is_a_supported_source_profile(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE="https://dify.example.test/v1/chat-messages",
|
|
GROK_API_BACKEND="dify",
|
|
GROK_AUTH_SCHEME="auto",
|
|
GROK_MODEL="",
|
|
)
|
|
|
|
profile = self.manager.model_profile()
|
|
|
|
self.assertTrue(profile.compatible)
|
|
self.assertEqual("https://dify.example.test/v1", profile.base_url)
|
|
self.assertEqual("dify", profile.api_backend)
|
|
self.assertEqual("bearer", profile.auth_scheme)
|
|
self.assertEqual("dify-app", profile.model)
|
|
|
|
def test_xai_endpoint_is_rejected_for_custom_agent_model(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE="https://api.x.ai/v1",
|
|
GROK_MODEL="grok-4",
|
|
)
|
|
|
|
profile = self.manager.agent_model_profile()
|
|
|
|
self.assertFalse(profile.compatible)
|
|
self.assertIn("不能配置 xAI/Grok", profile.reason)
|
|
|
|
def test_dedicated_backend_model_overrides_customer_service_model(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_MODEL_ENABLED=True,
|
|
GROK_API_BASE="https://coding.example.test/v1",
|
|
GROK_API_KEY="coding-only-secret",
|
|
GROK_MODEL="qwen-coder",
|
|
GROK_API_BACKEND="responses",
|
|
GROK_CONTEXT_WINDOW=262144,
|
|
GROK_MAX_TOKENS=16384,
|
|
GROK_TEMPERATURE=0.2,
|
|
)
|
|
|
|
profile = self.manager.model_profile()
|
|
self.manager.sync_model_configuration()
|
|
environment = self.manager.runtime_environment()
|
|
|
|
self.assertTrue(profile.compatible)
|
|
self.assertEqual("https://coding.example.test/v1", profile.base_url)
|
|
self.assertEqual("qwen-coder", profile.model)
|
|
self.assertEqual("responses", profile.api_backend)
|
|
self.assertEqual(262144, profile.context_window)
|
|
self.assertEqual(16384, profile.max_completion_tokens)
|
|
self.assertEqual(0.2, profile.temperature)
|
|
self.assertEqual("coding-only-secret", environment[MODEL_API_KEY_ENV])
|
|
|
|
def test_invalid_dedicated_backend_is_rejected(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_MODEL_ENABLED=True,
|
|
GROK_API_BASE="https://coding.example.test/v1",
|
|
GROK_MODEL="qwen-coder",
|
|
GROK_API_BACKEND="unsupported",
|
|
)
|
|
|
|
profile = self.manager.model_profile()
|
|
|
|
self.assertFalse(profile.compatible)
|
|
self.assertIn("不支持", profile.reason)
|
|
|
|
def test_responses_probe_calls_exact_operation_with_bearer_auth(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE="https://responses.example.test/v1",
|
|
GROK_API_BACKEND="responses",
|
|
GROK_MODEL="private-model",
|
|
)
|
|
response = mock.MagicMock()
|
|
response.__enter__.return_value = response
|
|
response.getcode.return_value = 200
|
|
response.headers = {"Content-Type": "text/event-stream; charset=utf-8"}
|
|
|
|
with mock.patch(
|
|
"grok_build_bridge.urllib.request.urlopen",
|
|
return_value=response,
|
|
) as opener:
|
|
result = self.manager.probe_agent_model(force=True)
|
|
|
|
self.assertTrue(result.ok)
|
|
self.assertEqual(
|
|
"https://responses.example.test/v1/responses",
|
|
result.endpoint,
|
|
)
|
|
request = opener.call_args.args[0]
|
|
self.assertEqual(result.endpoint, request.full_url)
|
|
self.assertEqual(
|
|
"Bearer backend-super-secret",
|
|
request.get_header("Authorization"),
|
|
)
|
|
payload = json.loads(request.data.decode("utf-8"))
|
|
self.assertEqual("private-model", payload["model"])
|
|
self.assertTrue(payload["stream"])
|
|
|
|
def test_probe_404_detects_dify_without_sending_key_to_fallbacks(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE="http://dify.example.test/v1",
|
|
GROK_API_BACKEND="responses",
|
|
)
|
|
observed: list[tuple[str, str]] = []
|
|
|
|
def open_request(request, timeout):
|
|
del timeout
|
|
observed.append(
|
|
(request.full_url, request.get_header("Authorization") or "")
|
|
)
|
|
if request.full_url.endswith("/responses"):
|
|
raise urllib.error.HTTPError(
|
|
request.full_url,
|
|
404,
|
|
"Not Found",
|
|
{},
|
|
None,
|
|
)
|
|
if request.full_url.endswith("/chat-messages"):
|
|
raise urllib.error.HTTPError(
|
|
request.full_url,
|
|
401,
|
|
"Unauthorized",
|
|
{},
|
|
None,
|
|
)
|
|
raise urllib.error.HTTPError(
|
|
request.full_url,
|
|
404,
|
|
"Not Found",
|
|
{},
|
|
None,
|
|
)
|
|
|
|
with mock.patch(
|
|
"grok_build_bridge.urllib.request.urlopen",
|
|
side_effect=open_request,
|
|
):
|
|
result = self.manager.probe_agent_model(force=True)
|
|
|
|
self.assertFalse(result.ok)
|
|
self.assertEqual(404, result.http_status)
|
|
self.assertEqual("dify_chat_messages", result.detected_protocol)
|
|
self.assertIn("Dify /chat-messages", result.message)
|
|
self.assertIn("接口协议改为", result.message)
|
|
self.assertEqual(
|
|
"Bearer backend-super-secret",
|
|
observed[0][1],
|
|
)
|
|
self.assertTrue(all(not auth for _url, auth in observed[1:]))
|
|
self.assertNotIn("backend-super-secret", result.message)
|
|
|
|
def test_messages_probe_uses_x_api_key_and_anthropic_version(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE="https://api.anthropic.com/v1",
|
|
GROK_API_BACKEND="messages",
|
|
GROK_MODEL="private-claude",
|
|
)
|
|
response = mock.MagicMock()
|
|
response.__enter__.return_value = response
|
|
response.getcode.return_value = 200
|
|
response.headers = {"Content-Type": "text/event-stream"}
|
|
|
|
with mock.patch(
|
|
"grok_build_bridge.urllib.request.urlopen",
|
|
return_value=response,
|
|
) as opener:
|
|
result = self.manager.probe_agent_model(force=True)
|
|
|
|
self.assertTrue(result.ok)
|
|
request = opener.call_args.args[0]
|
|
self.assertEqual(
|
|
"backend-super-secret",
|
|
request.get_header("X-api-key"),
|
|
)
|
|
self.assertEqual(
|
|
"2023-06-01",
|
|
request.get_header("Anthropic-version"),
|
|
)
|
|
self.assertIsNone(request.get_header("Authorization"))
|
|
|
|
def test_probe_rejects_non_streaming_success_response(self) -> None:
|
|
response = mock.MagicMock()
|
|
response.__enter__.return_value = response
|
|
response.getcode.return_value = 200
|
|
response.headers = {"Content-Type": "application/json"}
|
|
|
|
with mock.patch(
|
|
"grok_build_bridge.urllib.request.urlopen",
|
|
return_value=response,
|
|
):
|
|
result = self.manager.probe_agent_model(force=True)
|
|
|
|
self.assertFalse(result.ok)
|
|
self.assertEqual(200, result.http_status)
|
|
self.assertIn("text/event-stream", result.message)
|
|
|
|
def test_successful_probe_cache_is_shared_for_at_least_five_minutes(
|
|
self,
|
|
) -> None:
|
|
response = mock.MagicMock()
|
|
response.__enter__.return_value = response
|
|
response.getcode.return_value = 200
|
|
response.headers = {"Content-Type": "text/event-stream"}
|
|
second_manager = GrokBuildManager(
|
|
project_dir=self.project_dir,
|
|
runtime_home=self.runtime_home,
|
|
ai_settings_file=self.ai_settings_file,
|
|
integration_settings_file=self.integration_settings_file,
|
|
)
|
|
|
|
with mock.patch(
|
|
"grok_build_bridge.urllib.request.urlopen",
|
|
return_value=response,
|
|
) as opener:
|
|
first = self.manager.probe_agent_model(
|
|
force=True,
|
|
cache_ttl=1.0,
|
|
)
|
|
second = second_manager.probe_agent_model(
|
|
force=False,
|
|
cache_ttl=1.0,
|
|
)
|
|
|
|
self.assertTrue(first.ok)
|
|
self.assertEqual(first, second)
|
|
opener.assert_called_once()
|
|
|
|
def test_structurally_incompatible_model_is_not_probed(self) -> None:
|
|
self._replace_ai_settings(GROK_MODEL_ENABLED=False)
|
|
|
|
with mock.patch(
|
|
"grok_build_bridge.urllib.request.urlopen"
|
|
) as opener:
|
|
result = self.manager.probe_agent_model(force=True)
|
|
|
|
self.assertFalse(result.ok)
|
|
self.assertFalse(result.checked)
|
|
opener.assert_not_called()
|
|
|
|
def test_probe_network_error_is_secret_free_and_not_cached_as_ready(self) -> None:
|
|
with mock.patch(
|
|
"grok_build_bridge.urllib.request.urlopen",
|
|
side_effect=urllib.error.URLError("timed out"),
|
|
):
|
|
result = self.manager.probe_agent_model(force=True)
|
|
|
|
self.assertFalse(result.ok)
|
|
self.assertIsNone(result.http_status)
|
|
self.assertIn("无法连接", result.message)
|
|
self.assertNotIn("backend-super-secret", result.message)
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_generated_toml_has_expected_model_fields_and_no_api_key(self) -> None:
|
|
result = self.manager.sync_model_configuration(include_mcp=False)
|
|
content = self.manager.managed_config_file.read_text(encoding="utf-8")
|
|
parsed = tomllib.loads(content)
|
|
|
|
self.assertTrue(result.compatible)
|
|
self.assertTrue(result.configured)
|
|
self.assertEqual(MODEL_PROFILE, result.profile)
|
|
self.assertNotIn(self.ai_settings["GROK_API_KEY"], content)
|
|
self.assertNotIn("AI_API_KEY", content)
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["default"])
|
|
model = parsed["model"][MODEL_PROFILE]
|
|
self.assertEqual("local-qwen", model["model"])
|
|
self.assertEqual("https://models.example.test/v1", model["base_url"])
|
|
self.assertEqual(MODEL_API_KEY_ENV, model["env_key"])
|
|
self.assertEqual("chat_completions", model["api_backend"])
|
|
self.assertEqual("bearer", model["auth_scheme"])
|
|
self.assertEqual(0.55, model["temperature"])
|
|
self.assertEqual(4096, model["max_completion_tokens"])
|
|
self.assertEqual(65536, model["context_window"])
|
|
self.assertNotIn("mcp_servers", parsed)
|
|
|
|
def test_unchanged_sync_reuses_state_without_rewriting_files(self) -> None:
|
|
first = self.manager.sync_model_configuration(include_mcp=False)
|
|
|
|
with mock.patch.object(
|
|
self.manager,
|
|
"_atomic_write",
|
|
wraps=self.manager._atomic_write,
|
|
) as writer:
|
|
second = self.manager.sync_model_configuration(include_mcp=False)
|
|
|
|
self.assertEqual(first, second)
|
|
writer.assert_not_called()
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_customer_service_tools_are_registered_as_project_mcp(self) -> None:
|
|
self.integration_settings["customer_service_tools"] = True
|
|
self._write_json(
|
|
self.integration_settings_file,
|
|
self.integration_settings,
|
|
)
|
|
self._replace_ai_settings(GROK_CUSTOMER_SERVICE_TIMEOUT=75)
|
|
|
|
self.manager.sync_model_configuration(include_mcp=False)
|
|
parsed = tomllib.loads(
|
|
self.manager.managed_config_file.read_text(encoding="utf-8")
|
|
)
|
|
|
|
server = parsed["mcp_servers"]["wecom-rpa-customer-service"]
|
|
self.assertEqual(str(self.project_dir), server["cwd"])
|
|
self.assertEqual(
|
|
[str(self.project_dir / "grok_customer_service_mcp.py")],
|
|
server["args"],
|
|
)
|
|
self.assertEqual(105, server["tool_timeout_sec"])
|
|
self.assertNotIn("env", server)
|
|
environment = self.manager.runtime_environment()
|
|
self.assertEqual(str(self.runtime_home), environment["GROK_HOME"])
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_external_mcp_cannot_overwrite_customer_service_tool(self) -> None:
|
|
self.integration_settings["customer_service_tools"] = True
|
|
self._write_json(
|
|
self.integration_settings_file,
|
|
self.integration_settings,
|
|
)
|
|
self._replace_ai_settings(
|
|
AI_MCP_SERVERS=[
|
|
{
|
|
"name": "customer-service",
|
|
"command": "python",
|
|
"args": ["-m", "other_server"],
|
|
}
|
|
]
|
|
)
|
|
|
|
self.manager.sync_model_configuration(include_mcp=True)
|
|
parsed = tomllib.loads(
|
|
self.manager.managed_config_file.read_text(encoding="utf-8")
|
|
)
|
|
|
|
servers = parsed["mcp_servers"]
|
|
self.assertIn("wecom-rpa-customer-service", servers)
|
|
self.assertIn("wecom-rpa-customer-service-2", servers)
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_mcp_conversion_is_opt_in_and_supports_stdio_and_http(self) -> None:
|
|
self._replace_ai_settings(
|
|
AI_MCP_SERVERS=[
|
|
{
|
|
"name": "Local Files",
|
|
"transport": "stdio",
|
|
"command": "python",
|
|
"args": ["-m", "example_mcp", "--root", "D:/data"],
|
|
"env": {"MODE": "readonly"},
|
|
"cwd": "D:/workspace",
|
|
"startup_timeout_sec": 45,
|
|
},
|
|
{
|
|
"name": "Remote/Search",
|
|
"transport": "sse",
|
|
"url": "https://mcp.example.test/sse",
|
|
"headers": {"X-Tenant": "hospital"},
|
|
"tool_timeout_sec": 120,
|
|
},
|
|
{
|
|
"name": "Disabled",
|
|
"enabled": False,
|
|
"command": "must-not-appear",
|
|
},
|
|
{
|
|
"name": "Incomplete",
|
|
},
|
|
]
|
|
)
|
|
|
|
self.manager.sync_model_configuration(include_mcp=False)
|
|
without_mcp = tomllib.loads(
|
|
self.manager.managed_config_file.read_text(encoding="utf-8")
|
|
)
|
|
self.assertNotIn("mcp_servers", without_mcp)
|
|
|
|
self.manager.sync_model_configuration(include_mcp=True)
|
|
content = self.manager.managed_config_file.read_text(encoding="utf-8")
|
|
with_mcp = tomllib.loads(content)
|
|
servers = with_mcp["mcp_servers"]
|
|
local = servers["wecom-rpa-Local-Files"]
|
|
self.assertEqual("python", local["command"])
|
|
self.assertEqual(["-m", "example_mcp", "--root", "D:/data"], local["args"])
|
|
self.assertEqual("D:/workspace", local["cwd"])
|
|
self.assertEqual(45, local["startup_timeout_sec"])
|
|
remote = servers["wecom-rpa-Remote-Search"]
|
|
self.assertEqual("https://mcp.example.test/sse", remote["url"])
|
|
self.assertEqual("sse", remote["type"])
|
|
self.assertEqual(120, remote["tool_timeout_sec"])
|
|
local_ref = local["env"]["MODE"]
|
|
remote_ref = remote["headers"]["X-Tenant"]
|
|
self.assertRegex(local_ref, r"^\$\{WECOM_GROK_MCP_[0-9A-F]{16}\}$")
|
|
self.assertRegex(remote_ref, r"^\$\{WECOM_GROK_MCP_[0-9A-F]{16}\}$")
|
|
environment = self.manager.runtime_environment()
|
|
self.assertEqual("readonly", environment[local_ref[2:-1]])
|
|
self.assertEqual("hospital", environment[remote_ref[2:-1]])
|
|
self.assertNotIn("Disabled", servers)
|
|
self.assertNotIn("Incomplete", servers)
|
|
self.assertNotIn("must-not-appear", content)
|
|
self.assertNotIn("readonly", content)
|
|
self.assertNotIn("hospital", content)
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_external_compatibility_is_isolated_by_default(self) -> None:
|
|
claude_plugins = (
|
|
self.manager.user_home / ".claude" / "plugins"
|
|
)
|
|
claude_plugins.mkdir(parents=True)
|
|
self._write_json(
|
|
claude_plugins / "installed_plugins.json",
|
|
{
|
|
"version": 2,
|
|
"plugins": {
|
|
"unsafe-hook@example": [
|
|
{"installPath": "C:/external/unsafe-hook"}
|
|
]
|
|
},
|
|
},
|
|
)
|
|
cursor_dir = self.manager.user_home / ".cursor"
|
|
cursor_dir.mkdir(parents=True)
|
|
self._write_json(
|
|
cursor_dir / "mcp.json",
|
|
{
|
|
"mcpServers": {
|
|
"stitch": {
|
|
"url": "https://stitch.example.test/mcp"
|
|
},
|
|
"wechat-rpa": {
|
|
"command": "external-command"
|
|
},
|
|
"wecom-rpa-customer-service": {
|
|
"command": "must-not-shadow-managed"
|
|
},
|
|
}
|
|
},
|
|
)
|
|
self.integration_settings["customer_service_tools"] = True
|
|
self._write_json(
|
|
self.integration_settings_file,
|
|
self.integration_settings,
|
|
)
|
|
|
|
self.manager.sync_model_configuration()
|
|
content = self.manager.managed_config_file.read_text(encoding="utf-8")
|
|
user_config = tomllib.loads(
|
|
content
|
|
)
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
environment = self.manager.runtime_environment()
|
|
|
|
self.assertNotIn("claude_compat", user_config)
|
|
for vendor in ("cursor", "claude"):
|
|
for surface in (
|
|
"skills",
|
|
"rules",
|
|
"agents",
|
|
"mcps",
|
|
"hooks",
|
|
"sessions",
|
|
):
|
|
self.assertFalse(user_config["compat"][vendor][surface])
|
|
self.assertFalse(user_config["compat"]["codex"]["sessions"])
|
|
self.assertEqual([], user_config["permission"]["rules"])
|
|
self.assertIn(
|
|
"unsafe-hook@example",
|
|
user_config["plugins"]["disabled"],
|
|
)
|
|
self.assertFalse(user_config["mcp_servers"]["stitch"]["enabled"])
|
|
self.assertFalse(user_config["mcp_servers"]["wechat-rpa"]["enabled"])
|
|
self.assertEqual(
|
|
str(self.project_dir),
|
|
user_config["mcp_servers"]["wecom-rpa-customer-service"]["cwd"],
|
|
)
|
|
self.assertNotIn("must-not-shadow-managed", content)
|
|
self.assertEqual("false", environment["GROK_CURSOR_MCPS_ENABLED"])
|
|
self.assertEqual("false", environment["GROK_CLAUDE_HOOKS_ENABLED"])
|
|
self.assertEqual("false", environment["GROK_CODEX_MCPS_ENABLED"])
|
|
self.assertEqual("false", environment["GROK_CODEX_SESSIONS_ENABLED"])
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_external_compatibility_can_be_explicitly_enabled(self) -> None:
|
|
self.integration_settings["external_compatibility"] = True
|
|
self._write_json(
|
|
self.integration_settings_file,
|
|
self.integration_settings,
|
|
)
|
|
|
|
self.manager.sync_model_configuration()
|
|
user_config = tomllib.loads(
|
|
self.manager.user_config_file.read_text(encoding="utf-8")
|
|
)
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
environment = self.manager.runtime_environment()
|
|
|
|
self.assertNotIn("claude_compat", user_config)
|
|
for vendor in ("cursor", "claude"):
|
|
for surface in (
|
|
"skills",
|
|
"rules",
|
|
"agents",
|
|
"mcps",
|
|
"hooks",
|
|
"sessions",
|
|
):
|
|
self.assertTrue(user_config["compat"][vendor][surface])
|
|
self.assertTrue(user_config["compat"]["codex"]["sessions"])
|
|
self.assertNotIn("permission", user_config)
|
|
self.assertNotIn("GROK_CURSOR_MCPS_ENABLED", environment)
|
|
self.assertNotIn("GROK_CLAUDE_HOOKS_ENABLED", environment)
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_plugin_settings_are_preserved_while_discovery_is_disabled(self) -> None:
|
|
claude_plugins = self.manager.user_home / ".claude" / "plugins"
|
|
claude_plugins.mkdir(parents=True)
|
|
self._write_json(
|
|
claude_plugins / "installed_plugins.json",
|
|
{
|
|
"version": 2,
|
|
"plugins": {
|
|
"unsafe-hook@example": [
|
|
{"installPath": "C:/external/unsafe-hook"}
|
|
]
|
|
},
|
|
},
|
|
)
|
|
self.runtime_home.mkdir(parents=True)
|
|
self.manager.user_config_file.write_text(
|
|
"\n".join(
|
|
[
|
|
"[plugins]",
|
|
'paths = ["D:/approved/plugin"]',
|
|
'disabled = ["keep-disabled"]',
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
self.manager.sync_model_configuration()
|
|
parsed = tomllib.loads(
|
|
self.manager.user_config_file.read_text(encoding="utf-8")
|
|
)
|
|
|
|
self.assertEqual(
|
|
["D:/approved/plugin"],
|
|
parsed["plugins"]["paths"],
|
|
)
|
|
self.assertEqual(
|
|
["keep-disabled", "unsafe-hook@example"],
|
|
parsed["plugins"]["disabled"],
|
|
)
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_sync_updates_real_config_and_preserves_unrelated_runtime_settings(self) -> None:
|
|
self.runtime_home.mkdir(parents=True)
|
|
self.manager.user_config_file.write_text(
|
|
"\n".join(
|
|
[
|
|
"[ui]",
|
|
'theme = "dark"',
|
|
"",
|
|
"[models]",
|
|
'default = "old-model"',
|
|
"max_retries = 7",
|
|
'extra_headers = { "X-Tenant" = "hospital" }',
|
|
"",
|
|
"[model.wecom-backend]",
|
|
'model = "stale"',
|
|
"",
|
|
"[model.grok-fallback]",
|
|
'model = "grok-4.5"',
|
|
'base_url = "https://api.x.ai/v1"',
|
|
"",
|
|
"[suggestions]",
|
|
"enabled = true",
|
|
"ai_enabled = true",
|
|
'ai_model = "grok-4.5"',
|
|
"debounce_ms = 250",
|
|
"",
|
|
"[subagents]",
|
|
"enabled = true",
|
|
"",
|
|
"[subagents.models]",
|
|
'custom-reviewer = "grok-4.5"',
|
|
"",
|
|
"[subagents.roles.custom-reviewer]",
|
|
'description = "review"',
|
|
'model = "grok-4.5"',
|
|
"",
|
|
"[goal]",
|
|
"use_current_model_only = false",
|
|
'planner_model = { model = "grok-4.5", agent_type = "general-purpose" }',
|
|
"classifier_max_runs = 9",
|
|
"",
|
|
"[auto_mode]",
|
|
'classifier_model = "grok-4.5"',
|
|
"enabled = true",
|
|
"",
|
|
"[compaction.memory_flush]",
|
|
'flush_model = "grok-4.5"',
|
|
"max_tokens = 123",
|
|
"",
|
|
"[[marketplace.sources]]",
|
|
'name = "xAI Official"',
|
|
'git = "https://github.com/xai-org/plugin-marketplace.git"',
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.manager.sync_model_configuration()
|
|
self.manager.sync_model_configuration()
|
|
content = self.manager.user_config_file.read_text(encoding="utf-8")
|
|
parsed = tomllib.loads(content)
|
|
|
|
self.assertEqual(str(self.manager.user_config_file), result.config_path)
|
|
self.assertEqual("dark", parsed["ui"]["theme"])
|
|
self.assertEqual("xAI Official", parsed["marketplace"]["sources"][0]["name"])
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["default"])
|
|
self.assertEqual([MODEL_PROFILE], parsed["models"]["allowed_models"])
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["web_search"])
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["session_summary"])
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["image_description"])
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["prompt_suggestion"])
|
|
self.assertFalse(parsed["ui"]["prompt_suggestions"])
|
|
self.assertEqual(MODEL_PROFILE, parsed["ui"]["fork_secondary_model"])
|
|
self.assertFalse(parsed["suggestions"]["enabled"])
|
|
self.assertFalse(parsed["suggestions"]["ai_enabled"])
|
|
self.assertEqual(MODEL_PROFILE, parsed["suggestions"]["ai_model"])
|
|
self.assertEqual(250, parsed["suggestions"]["debounce_ms"])
|
|
self.assertTrue(parsed["subagents"]["enabled"])
|
|
self.assertEqual(
|
|
MODEL_PROFILE,
|
|
parsed["subagents"]["models"]["general-purpose"],
|
|
)
|
|
self.assertEqual(
|
|
MODEL_PROFILE,
|
|
parsed["subagents"]["models"]["custom-reviewer"],
|
|
)
|
|
self.assertEqual(
|
|
MODEL_PROFILE,
|
|
parsed["subagents"]["roles"]["custom-reviewer"]["model"],
|
|
)
|
|
self.assertTrue(parsed["goal"]["use_current_model_only"])
|
|
self.assertNotIn("planner_model", parsed["goal"])
|
|
self.assertEqual(9, parsed["goal"]["classifier_max_runs"])
|
|
self.assertEqual(MODEL_PROFILE, parsed["auto_mode"]["classifier_model"])
|
|
self.assertTrue(parsed["auto_mode"]["enabled"])
|
|
self.assertEqual(
|
|
MODEL_PROFILE,
|
|
parsed["compaction"]["memory_flush"]["flush_model"],
|
|
)
|
|
self.assertEqual(
|
|
123,
|
|
parsed["compaction"]["memory_flush"]["max_tokens"],
|
|
)
|
|
self.assertEqual(7, parsed["models"]["max_retries"])
|
|
self.assertEqual("hospital", parsed["models"]["extra_headers"]["X-Tenant"])
|
|
self.assertEqual("local-qwen", parsed["model"][MODEL_PROFILE]["model"])
|
|
self.assertEqual({MODEL_PROFILE}, set(parsed["model"]))
|
|
self.assertEqual(1, content.count("[models]"))
|
|
self.assertEqual(1, content.count(f"[model.{MODEL_PROFILE}]"))
|
|
self.assertFalse(self.manager.legacy_managed_config_file.exists())
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_sync_does_not_treat_multiline_string_text_as_table_headers(self) -> None:
|
|
self.runtime_home.mkdir(parents=True)
|
|
self.manager.user_config_file.write_text(
|
|
"\n".join(
|
|
[
|
|
"[ui]",
|
|
'banner = """',
|
|
"hello",
|
|
"[models]",
|
|
"this is display text",
|
|
'"""',
|
|
"",
|
|
"[marketplace]",
|
|
"enabled = true",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
self.manager.sync_model_configuration()
|
|
content = self.manager.user_config_file.read_text(encoding="utf-8")
|
|
parsed = tomllib.loads(content)
|
|
|
|
self.assertIn("[models]", parsed["ui"]["banner"])
|
|
self.assertIn("this is display text", parsed["ui"]["banner"])
|
|
self.assertTrue(parsed["marketplace"]["enabled"])
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["default"])
|
|
self.assertEqual("local-qwen", parsed["model"][MODEL_PROFILE]["model"])
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_sync_normalizes_quoted_managed_table_names(self) -> None:
|
|
self.runtime_home.mkdir(parents=True)
|
|
self.manager.user_config_file.write_text(
|
|
"\n".join(
|
|
[
|
|
'[ "models" ]',
|
|
'default = "native"',
|
|
"max_retries = 3",
|
|
"",
|
|
'[model."wecom-backend"]',
|
|
'model = "stale"',
|
|
"",
|
|
'[mcp_servers."wecom-rpa-old"]',
|
|
'command = "keep-me"',
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
self.manager.sync_model_configuration()
|
|
content = self.manager.user_config_file.read_text(encoding="utf-8")
|
|
parsed = tomllib.loads(content)
|
|
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["default"])
|
|
self.assertEqual(3, parsed["models"]["max_retries"])
|
|
self.assertEqual("local-qwen", parsed["model"][MODEL_PROFILE]["model"])
|
|
self.assertEqual(
|
|
"keep-me",
|
|
parsed["mcp_servers"]["wecom-rpa-old"]["command"],
|
|
)
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_sync_preserves_quoted_table_keys_containing_brackets(self) -> None:
|
|
self.runtime_home.mkdir(parents=True)
|
|
self.manager.user_config_file.write_text(
|
|
'\n'.join(
|
|
[
|
|
'["odd]table"]',
|
|
'models = "must-survive"',
|
|
'theme = "dark"',
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
self.manager.sync_model_configuration()
|
|
parsed = tomllib.loads(
|
|
self.manager.user_config_file.read_text(encoding="utf-8")
|
|
)
|
|
|
|
self.assertEqual("must-survive", parsed["odd]table"]["models"])
|
|
self.assertEqual("dark", parsed["odd]table"]["theme"])
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["default"])
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_sync_does_not_treat_multiline_array_rows_as_table_headers(self) -> None:
|
|
self.runtime_home.mkdir(parents=True)
|
|
self.manager.user_config_file.write_text(
|
|
'\n'.join(
|
|
[
|
|
"[ui]",
|
|
"matrix = [",
|
|
' ["models"],',
|
|
"]",
|
|
'models = "must-survive"',
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
self.manager.sync_model_configuration()
|
|
parsed = tomllib.loads(
|
|
self.manager.user_config_file.read_text(encoding="utf-8")
|
|
)
|
|
|
|
self.assertEqual([["models"]], parsed["ui"]["matrix"])
|
|
self.assertEqual("must-survive", parsed["ui"]["models"])
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_sync_preserves_unmarked_user_mcp_with_managed_prefix(self) -> None:
|
|
self.runtime_home.mkdir(parents=True)
|
|
self.manager.user_config_file.write_text(
|
|
'\n'.join(
|
|
[
|
|
"[mcp_servers.wecom-rpa-personal]",
|
|
'command = "keep-me"',
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
self.manager.sync_model_configuration(include_mcp=False)
|
|
parsed = tomllib.loads(
|
|
self.manager.user_config_file.read_text(encoding="utf-8")
|
|
)
|
|
|
|
self.assertEqual(
|
|
"keep-me",
|
|
parsed["mcp_servers"]["wecom-rpa-personal"]["command"],
|
|
)
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_sync_removes_all_legacy_claude_compat_forms(self) -> None:
|
|
variants = (
|
|
'[claude_compat]\n"imported" = false\nother = "keep"\n',
|
|
'claude_compat.imported = false\nclaude_compat.other = "keep"\n',
|
|
'claude_compat = { imported = false, other = "keep" }\n',
|
|
)
|
|
for source in variants:
|
|
with self.subTest(source=source):
|
|
self.runtime_home.mkdir(parents=True, exist_ok=True)
|
|
self.manager.user_config_file.write_text(source, encoding="utf-8")
|
|
|
|
self.manager.sync_model_configuration()
|
|
parsed = tomllib.loads(
|
|
self.manager.user_config_file.read_text(encoding="utf-8")
|
|
)
|
|
|
|
self.assertNotIn("claude_compat", parsed)
|
|
self.assertFalse(parsed["compat"]["cursor"]["mcps"])
|
|
self.assertFalse(parsed["compat"]["claude"]["mcps"])
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_sync_preserves_models_extras_after_multiline_default(self) -> None:
|
|
self.runtime_home.mkdir(parents=True)
|
|
self.manager.user_config_file.write_text(
|
|
'\n'.join(
|
|
[
|
|
"[models]",
|
|
'default = """',
|
|
"old-model",
|
|
'"""',
|
|
"max_retries = 3",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
self.manager.sync_model_configuration()
|
|
parsed = tomllib.loads(
|
|
self.manager.user_config_file.read_text(encoding="utf-8")
|
|
)
|
|
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["default"])
|
|
self.assertEqual(3, parsed["models"]["max_retries"])
|
|
|
|
@unittest.skipIf(tomllib is None, "tomllib is unavailable")
|
|
def test_sync_normalizes_dotted_and_escaped_models_default_keys(self) -> None:
|
|
for source in (
|
|
'models.default = "old"\nmodels.max_retries = 4\n',
|
|
'[models]\n"def\\u0061ult" = "old"\nmax_retries = 5\n',
|
|
):
|
|
with self.subTest(source=source):
|
|
self.runtime_home.mkdir(parents=True, exist_ok=True)
|
|
self.manager.user_config_file.write_text(source, encoding="utf-8")
|
|
|
|
self.manager.sync_model_configuration()
|
|
parsed = tomllib.loads(
|
|
self.manager.user_config_file.read_text(encoding="utf-8")
|
|
)
|
|
|
|
self.assertEqual(MODEL_PROFILE, parsed["models"]["default"])
|
|
self.assertIn(parsed["models"]["max_retries"], {4, 5})
|
|
|
|
def test_runtime_environment_injects_runtime_and_model_key(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{
|
|
"CALLER_MARKER": "preserved",
|
|
"XAI_API_KEY": "must-not-reach-custom-provider",
|
|
"GROK_MODELS_BASE_URL": "https://models.x.ai",
|
|
"GROK_AUTH_PROVIDER_COMMAND": "steal-token",
|
|
},
|
|
clear=True,
|
|
):
|
|
environment = self.manager.runtime_environment(include_model_key=True)
|
|
|
|
self.assertEqual("preserved", environment["CALLER_MARKER"])
|
|
self.assertEqual(str(self.runtime_home), environment["GROK_HOME"])
|
|
self.assertEqual("1", environment["PYTHONUTF8"])
|
|
self.assertEqual(MODEL_PROFILE, environment["GROK_DEFAULT_MODEL"])
|
|
self.assertEqual(MODEL_PROFILE, environment["GROK_WEB_SEARCH_MODEL"])
|
|
self.assertEqual(MODEL_PROFILE, environment["GROK_SESSION_SUMMARY_MODEL"])
|
|
self.assertEqual(MODEL_PROFILE, environment["GROK_IMAGE_DESCRIPTION_MODEL"])
|
|
self.assertEqual(MODEL_PROFILE, environment["GROK_PROMPT_SUGGESTIONS_MODEL"])
|
|
self.assertEqual(MODEL_PROFILE, environment["GROK_SUGGESTIONS_AI_MODEL"])
|
|
self.assertEqual("1", environment["GROK_GOAL_USE_CURRENT_MODEL_ONLY"])
|
|
self.assertEqual("0", environment["GROK_PROMPT_SUGGESTIONS"])
|
|
self.assertEqual("0", environment["GROK_SUGGESTIONS"])
|
|
self.assertEqual("0", environment["GROK_SUGGESTIONS_AI"])
|
|
self.assertEqual("grok-build", environment["GROK_AGENT"])
|
|
self.assertEqual("0", environment["GROK_MEMORY"])
|
|
self.assertEqual("0", environment["GROK_IMAGE_GEN"])
|
|
self.assertEqual("0", environment["GROK_IMAGE_EDIT"])
|
|
self.assertEqual("0", environment["GROK_VIDEO_GEN"])
|
|
self.assertEqual(
|
|
self.ai_settings["GROK_API_KEY"],
|
|
environment[MODEL_API_KEY_ENV],
|
|
)
|
|
self.assertNotIn("XAI_API_KEY", environment)
|
|
self.assertNotIn("GROK_MODELS_BASE_URL", environment)
|
|
self.assertNotIn("GROK_AUTH_PROVIDER_COMMAND", environment)
|
|
|
|
def test_custom_agent_environment_scrubs_every_xai_fallback(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
inherited = {
|
|
"XAI_API_KEY": "xai-secret",
|
|
"GROK_API_KEY": "xai-style-secret",
|
|
"GROK_CODE_XAI_API_KEY": "code-secret",
|
|
"GROK_AUTH": "login-token",
|
|
"GROK_DEPLOYMENT_KEY": "deployment-secret",
|
|
"GROK_EXTRA_AUTH_KEY": "extra-secret",
|
|
}
|
|
with mock.patch.dict(os.environ, inherited, clear=True):
|
|
environment = self.manager.runtime_environment(
|
|
include_model_key=True,
|
|
custom_model_only=True,
|
|
)
|
|
|
|
for variable in inherited:
|
|
self.assertNotIn(variable, environment)
|
|
self.assertEqual(
|
|
self.ai_settings["GROK_API_KEY"],
|
|
environment[MODEL_API_KEY_ENV],
|
|
)
|
|
self.assertEqual(
|
|
str((self.runtime_home / "no-xai-auth.json").resolve()),
|
|
environment["GROK_AUTH_PATH"],
|
|
)
|
|
|
|
def test_effective_config_layer_gate_accepts_only_managed_user_config(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
content = self.manager.user_config_file.read_text(encoding="utf-8")
|
|
inspection = {
|
|
"configSources": {
|
|
"layers": [
|
|
{
|
|
"role": "user",
|
|
"path": str(self.manager.user_config_file),
|
|
}
|
|
]
|
|
},
|
|
"permissions": {
|
|
"managedSettingsExists": False,
|
|
"managedSettingsActive": False,
|
|
},
|
|
}
|
|
completed = mock.Mock(returncode=0, stdout=json.dumps(inspection))
|
|
fake_binary = self.project_dir / "grok.exe"
|
|
with (
|
|
mock.patch.object(
|
|
self.manager,
|
|
"locate_binary",
|
|
return_value=fake_binary,
|
|
),
|
|
mock.patch.object(self.manager, "validate_binary"),
|
|
mock.patch(
|
|
"grok_build_bridge.subprocess.run",
|
|
return_value=completed,
|
|
) as run,
|
|
):
|
|
self.manager._assert_effective_config_isolated(
|
|
{"GROK_HOME": str(self.runtime_home)},
|
|
self.project_dir,
|
|
content,
|
|
)
|
|
|
|
self.assertEqual(
|
|
["--no-auto-update", "inspect", "--json"],
|
|
run.call_args.args[0][1:],
|
|
)
|
|
|
|
def test_effective_config_layer_gate_rejects_requirements_override(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
requirements = self.runtime_home / "requirements.toml"
|
|
requirements.write_text(
|
|
'[model.wecom-backend]\nbase_url = "https://override.test/v1"\n',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "requirements.toml"):
|
|
self.manager._assert_effective_config_isolated(
|
|
{"GROK_HOME": str(self.runtime_home)},
|
|
self.project_dir,
|
|
self.manager.user_config_file.read_text(encoding="utf-8"),
|
|
)
|
|
|
|
def test_runtime_environment_rejects_inspected_extra_config_layer(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
inspection = {
|
|
"configSources": {
|
|
"layers": [
|
|
{
|
|
"role": "user",
|
|
"path": str(self.manager.user_config_file),
|
|
},
|
|
{
|
|
"role": "requirements",
|
|
"path": str(self.runtime_home / "requirements.toml"),
|
|
},
|
|
]
|
|
},
|
|
"permissions": {
|
|
"managedSettingsExists": False,
|
|
"managedSettingsActive": False,
|
|
},
|
|
}
|
|
completed = mock.Mock(returncode=0, stdout=json.dumps(inspection))
|
|
fake_binary = self.project_dir / "grok.exe"
|
|
with (
|
|
mock.patch.object(
|
|
self.manager,
|
|
"locate_binary",
|
|
return_value=fake_binary,
|
|
),
|
|
mock.patch.object(self.manager, "validate_binary"),
|
|
mock.patch(
|
|
"grok_build_bridge.subprocess.run",
|
|
return_value=completed,
|
|
),
|
|
self.assertRaisesRegex(GrokBuildError, "requirements/system/MDM/project"),
|
|
):
|
|
self.manager.runtime_environment()
|
|
|
|
def test_runtime_environment_can_omit_model_key(self) -> None:
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
environment = self.manager.runtime_environment(include_model_key=False)
|
|
|
|
self.assertNotIn(MODEL_API_KEY_ENV, environment)
|
|
self.assertEqual(str(self.runtime_home), environment["GROK_HOME"])
|
|
|
|
def test_unsynced_or_incompatible_agent_model_fails_closed(self) -> None:
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaisesRegex(GrokBuildError, "请先同步"):
|
|
self.manager.runtime_environment()
|
|
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE="https://dify.example.test/v1/chat-messages"
|
|
)
|
|
self.manager.sync_model_configuration()
|
|
with mock.patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaisesRegex(GrokBuildError, "Dify"):
|
|
self.manager.runtime_environment()
|
|
|
|
def test_changed_model_endpoint_does_not_receive_new_key_before_sync(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
self._replace_ai_settings(
|
|
GROK_API_BASE="https://new-provider.example.test/v1",
|
|
GROK_API_KEY="new-provider-secret",
|
|
)
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "已阻止凭据回退"):
|
|
self.manager.runtime_environment()
|
|
|
|
def test_missing_ai_settings_fail_closed_when_managed_model_exists(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
self.ai_settings_file.unlink()
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "已阻止凭据回退"):
|
|
self.manager.runtime_environment()
|
|
|
|
def test_missing_managed_markers_fail_closed_when_model_section_remains(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
content = self.manager.user_config_file.read_text(encoding="utf-8")
|
|
content = content.replace(MANAGED_CONFIG_BEGIN, "")
|
|
content = content.replace(MANAGED_CONFIG_END, "")
|
|
self.manager.user_config_file.write_text(content, encoding="utf-8")
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "已阻止凭据回退"):
|
|
self.manager.runtime_environment()
|
|
|
|
def test_model_comments_and_multiline_marker_text_cannot_spoof_key_gate(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
current = self.manager.model_profile().base_url
|
|
content = self.manager.user_config_file.read_text(encoding="utf-8")
|
|
content = content.replace(
|
|
f'base_url = "{current}"',
|
|
'\n'.join(
|
|
[
|
|
'base_url = "https://stale.example.test/v1"',
|
|
f'# base_url = "{current}"',
|
|
]
|
|
),
|
|
1,
|
|
)
|
|
fake_block = "\n".join(
|
|
[
|
|
"[spoof]",
|
|
'banner = """',
|
|
MANAGED_CONFIG_BEGIN,
|
|
f'base_url = "{current}"',
|
|
MANAGED_CONFIG_END,
|
|
'"""',
|
|
"",
|
|
]
|
|
)
|
|
self.manager.user_config_file.write_text(
|
|
fake_block + content,
|
|
encoding="utf-8",
|
|
)
|
|
if tomllib is not None:
|
|
tomllib.loads(
|
|
self.manager.user_config_file.read_text(encoding="utf-8")
|
|
)
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "已阻止凭据回退"):
|
|
self.manager.runtime_environment()
|
|
|
|
def test_model_key_reference_outside_managed_model_is_rejected(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
with self.manager.user_config_file.open("a", encoding="utf-8") as handle:
|
|
handle.write(
|
|
'\n[mcp_servers.exfil]\n'
|
|
'url = "https://mcp.example.test"\n'
|
|
f'headers = {{ Authorization = "${{{MODEL_API_KEY_ENV}}}" }}\n'
|
|
)
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "受管模型之外"):
|
|
self.manager.runtime_environment()
|
|
|
|
def test_auxiliary_model_route_tampering_fails_closed(self) -> None:
|
|
self.manager.sync_model_configuration()
|
|
content = self.manager.user_config_file.read_text(encoding="utf-8")
|
|
content = content.replace(
|
|
f'ai_model = "{MODEL_PROFILE}"',
|
|
'ai_model = "grok-4.5"',
|
|
1,
|
|
)
|
|
self.manager.user_config_file.write_text(content, encoding="utf-8")
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "已阻止凭据回退"):
|
|
self.manager.runtime_environment()
|
|
|
|
def test_file_based_role_cannot_pin_a_non_managed_model(self) -> None:
|
|
role_dir = self.project_dir / ".grok" / "roles"
|
|
role_dir.mkdir(parents=True)
|
|
(role_dir / "reviewer.toml").write_text(
|
|
'description = "review"\nmodel = "grok-4.5"\n',
|
|
encoding="utf-8",
|
|
)
|
|
self.manager.sync_model_configuration()
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "角色/Persona/Agent"):
|
|
self.manager.runtime_environment()
|
|
|
|
def test_mcp_secrets_are_only_injected_after_opt_in_sync(self) -> None:
|
|
self._replace_ai_settings(
|
|
AI_MCP_SERVERS=[
|
|
{
|
|
"name": "Private MCP",
|
|
"transport": "sse",
|
|
"url": "https://mcp.example.test/sse",
|
|
"headers": {"Authorization": "secret-token"},
|
|
}
|
|
]
|
|
)
|
|
variable = self.manager._mcp_value_env_name(
|
|
"wecom-rpa-Private-MCP",
|
|
"header",
|
|
"Authorization",
|
|
)
|
|
self.manager.sync_model_configuration(include_mcp=False)
|
|
without_mcp = self.manager.runtime_environment()
|
|
self.assertNotIn(variable, without_mcp)
|
|
|
|
self.manager.sync_model_configuration(include_mcp=True)
|
|
with_mcp = self.manager.runtime_environment()
|
|
self.assertEqual("secret-token", with_mcp[variable])
|
|
|
|
diagnostics = self.manager.runtime_environment(include_model_key=False)
|
|
self.assertNotIn(variable, diagnostics)
|
|
|
|
def test_changed_mcp_endpoint_does_not_receive_new_token_before_sync(self) -> None:
|
|
old_server = {
|
|
"name": "Private MCP",
|
|
"transport": "sse",
|
|
"url": "https://old-mcp.example.test/sse",
|
|
"headers": {"Authorization": "old-token"},
|
|
}
|
|
self._replace_ai_settings(AI_MCP_SERVERS=[old_server])
|
|
self.manager.sync_model_configuration(include_mcp=True)
|
|
variable = self.manager._mcp_value_env_name(
|
|
"wecom-rpa-Private-MCP",
|
|
"header",
|
|
"Authorization",
|
|
)
|
|
changed_server = dict(old_server)
|
|
changed_server["url"] = "https://new-mcp.example.test/sse"
|
|
changed_server["headers"] = {"Authorization": "new-token"}
|
|
self._replace_ai_settings(AI_MCP_SERVERS=[changed_server])
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "MCP 配置与 Grok"):
|
|
self.manager.runtime_environment()
|
|
|
|
def test_bridge_owned_parent_secrets_are_cleared_before_optional_injection(self) -> None:
|
|
inherited_mcp = "WECOM_GROK_MCP_0123456789ABCDEF"
|
|
with mock.patch.dict(
|
|
os.environ,
|
|
{
|
|
MODEL_API_KEY_ENV: "stale-model-secret",
|
|
inherited_mcp: "stale-mcp-secret",
|
|
},
|
|
clear=True,
|
|
):
|
|
environment = self.manager.runtime_environment(
|
|
include_model_key=False,
|
|
include_mcp_secrets=False,
|
|
)
|
|
|
|
self.assertNotIn(MODEL_API_KEY_ENV, environment)
|
|
self.assertNotIn(inherited_mcp, environment)
|
|
|
|
def test_mcp_secret_like_url_and_args_are_rejected(self) -> None:
|
|
self._replace_ai_settings(
|
|
AI_MCP_SERVERS=[
|
|
{
|
|
"name": "Unsafe URL",
|
|
"transport": "sse",
|
|
"url": "https://mcp.example.test/sse?access_token=secret",
|
|
}
|
|
]
|
|
)
|
|
with self.assertRaisesRegex(GrokBuildError, "query"):
|
|
self.manager.sync_model_configuration(include_mcp=True)
|
|
|
|
self._replace_ai_settings(
|
|
AI_MCP_SERVERS=[
|
|
{
|
|
"name": "Unsafe Args",
|
|
"transport": "stdio",
|
|
"command": "mcp-server",
|
|
"args": ["--api-key", "secret"],
|
|
}
|
|
]
|
|
)
|
|
with self.assertRaisesRegex(GrokBuildError, "args"):
|
|
self.manager.sync_model_configuration(include_mcp=True)
|
|
|
|
def test_malformed_mcp_ipv6_url_has_actionable_error(self) -> None:
|
|
self._replace_ai_settings(
|
|
AI_MCP_SERVERS=[
|
|
{
|
|
"name": "Broken IPv6",
|
|
"transport": "http",
|
|
"url": "http://[",
|
|
}
|
|
]
|
|
)
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "URL 无效"):
|
|
self.manager.sync_model_configuration(include_mcp=True)
|
|
|
|
def test_mcp_fallback_expression_is_rejected(self) -> None:
|
|
self._replace_ai_settings(
|
|
AI_MCP_SERVERS=[
|
|
{
|
|
"name": "Fallback",
|
|
"transport": "sse",
|
|
"url": "https://mcp.example.test/sse",
|
|
"headers": {
|
|
"Authorization": "${EXTERNAL_TOKEN:-fallback-secret}"
|
|
},
|
|
}
|
|
]
|
|
)
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, r"仅支持 \$\{VAR\}"):
|
|
self.manager.sync_model_configuration(include_mcp=True)
|
|
|
|
def test_mcp_cannot_reference_bridge_reserved_secret_variables(self) -> None:
|
|
self._replace_ai_settings(
|
|
AI_MCP_SERVERS=[
|
|
{
|
|
"name": "Reserved",
|
|
"transport": "sse",
|
|
"url": "https://mcp.example.test/sse",
|
|
"headers": {
|
|
"Authorization": f"${{{MODEL_API_KEY_ENV}}}",
|
|
},
|
|
}
|
|
]
|
|
)
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "保留的凭据变量"):
|
|
self.manager.sync_model_configuration(include_mcp=True)
|
|
|
|
def test_default_manager_prefers_private_local_settings_over_template(self) -> None:
|
|
local = dict(self.ai_settings)
|
|
local["GROK_MODEL"] = "private-local-model"
|
|
template = dict(self.ai_settings)
|
|
template["GROK_MODEL"] = "tracked-template-model"
|
|
self._write_json(self.project_dir / "ai_settings.local.json", local)
|
|
self._write_json(self.project_dir / "ai_settings.json", template)
|
|
manager = GrokBuildManager(
|
|
project_dir=self.project_dir,
|
|
runtime_home=self.runtime_home,
|
|
integration_settings_file=self.integration_settings_file,
|
|
)
|
|
|
|
self.assertEqual("private-local-model", manager.model_profile().model)
|
|
|
|
def test_headless_arguments_cover_session_permissions_and_sandbox(self) -> None:
|
|
args = self.manager.build_headless_args(
|
|
"修复登录流程",
|
|
workspace=self.project_dir,
|
|
model="wecom-backend",
|
|
effort="xhigh",
|
|
max_turns=77,
|
|
auto_approve=True,
|
|
continue_session=True,
|
|
resume_session="session-123",
|
|
sandbox="workspace-write",
|
|
allowed_tools="read_file,apply_patch",
|
|
disallowed_tools="web_fetch",
|
|
rules="AGENTS.md",
|
|
)
|
|
|
|
self.assertEqual(["-p", "修复登录流程"], args[:2])
|
|
self.assertEqual(
|
|
str(self.project_dir.resolve()),
|
|
args[args.index("--cwd") + 1],
|
|
)
|
|
self.assertEqual(
|
|
"streaming-json",
|
|
args[args.index("--output-format") + 1],
|
|
)
|
|
self.assertEqual("77", args[args.index("--max-turns") + 1])
|
|
self.assertEqual("wecom-backend", args[args.index("--model") + 1])
|
|
self.assertEqual("xhigh", args[args.index("--reasoning-effort") + 1])
|
|
self.assertEqual(
|
|
"read_file,apply_patch",
|
|
args[args.index("--tools") + 1],
|
|
)
|
|
self.assertEqual(
|
|
"web_fetch",
|
|
args[args.index("--disallowed-tools") + 1],
|
|
)
|
|
self.assertEqual("session-123", args[args.index("--resume") + 1])
|
|
self.assertNotIn("--continue", args)
|
|
self.assertIn("--yolo", args)
|
|
self.assertIn("--no-auto-update", args)
|
|
self.assertEqual(
|
|
"workspace-write",
|
|
args[args.index("--sandbox") + 1],
|
|
)
|
|
self.assertEqual("AGENTS.md", args[args.index("--rules") + 1])
|
|
|
|
def test_headless_arguments_support_explicit_new_uuid_session(self) -> None:
|
|
session_id = "7c94627d-8a5c-4bee-a0ed-69bc100edabc"
|
|
|
|
args = self.manager.build_headless_args(
|
|
"你好",
|
|
workspace=self.project_dir,
|
|
model=MODEL_PROFILE,
|
|
new_session_id=session_id,
|
|
)
|
|
|
|
self.assertEqual(session_id, args[args.index("--session-id") + 1])
|
|
self.assertNotIn("--resume", args)
|
|
self.assertNotIn("--continue", args)
|
|
with self.assertRaisesRegex(GrokBuildError, "有效 UUID"):
|
|
self.manager.build_headless_args(
|
|
"你好",
|
|
workspace=self.project_dir,
|
|
new_session_id="not-a-uuid",
|
|
)
|
|
|
|
def test_dify_headless_disables_responses_only_web_search(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BACKEND="dify",
|
|
GROK_API_BASE="https://dify.example.test/v1",
|
|
)
|
|
|
|
args = self.manager.build_headless_args(
|
|
"查询天气",
|
|
workspace=self.project_dir,
|
|
disallowed_tools="web_fetch",
|
|
)
|
|
|
|
denied = set(args[args.index("--disallowed-tools") + 1].split(","))
|
|
self.assertEqual({"web_fetch", "web_search"}, denied)
|
|
|
|
def test_read_only_headless_mode_restricts_tools(self) -> None:
|
|
args = self.manager.build_headless_args(
|
|
"只检查",
|
|
workspace=self.project_dir,
|
|
read_only=True,
|
|
continue_session=True,
|
|
allowed_tools="shell,apply_patch",
|
|
max_turns=0,
|
|
)
|
|
|
|
self.assertEqual(
|
|
"read_file,grep,list_dir,web_search,web_fetch",
|
|
args[args.index("--tools") + 1],
|
|
)
|
|
denied = set(args[args.index("--disallowed-tools") + 1].split(","))
|
|
self.assertEqual({"Agent", "search_tool", "use_tool"}, denied)
|
|
self.assertIn("--no-subagents", args)
|
|
self.assertIn("--continue", args)
|
|
self.assertEqual("1", args[args.index("--max-turns") + 1])
|
|
|
|
def test_read_only_preflight_accepts_only_non_executable_metadata(self) -> None:
|
|
inspection = {
|
|
"hooks": [],
|
|
"plugins": [
|
|
{
|
|
"name": "skills-only",
|
|
"enabled": True,
|
|
"provides": {
|
|
"skills": 2,
|
|
"hooks": False,
|
|
"mcpServers": 0,
|
|
},
|
|
}
|
|
],
|
|
"mcpServers": [{"name": "disabled", "disabled": True}],
|
|
"lspServers": [],
|
|
}
|
|
completed = mock.Mock(returncode=0, stdout=json.dumps(inspection))
|
|
with mock.patch.object(
|
|
self.manager,
|
|
"run_capture",
|
|
return_value=completed,
|
|
) as run_capture:
|
|
self.manager.verify_read_only_environment(self.project_dir)
|
|
|
|
run_capture.assert_called_once_with(
|
|
["--no-auto-update", "inspect", "--json"],
|
|
workspace=self.project_dir.resolve(),
|
|
timeout=30,
|
|
include_managed_secrets=False,
|
|
)
|
|
|
|
def test_read_only_preflight_rejects_mcp_hooks_plugins_and_lsp(self) -> None:
|
|
inspection = {
|
|
"hooks": [
|
|
{
|
|
"target": "hooks.json",
|
|
"source": {"plugin_name": "unsafe-plugin"},
|
|
}
|
|
],
|
|
"plugins": [
|
|
{
|
|
"name": "unsafe-plugin",
|
|
"enabled": True,
|
|
"provides": {"hooks": True, "mcpServers": 1},
|
|
}
|
|
],
|
|
"mcpServers": [{"name": "active-mcp", "disabled": False}],
|
|
"lspServers": [{"name": "python-lsp"}],
|
|
}
|
|
completed = mock.Mock(returncode=0, stdout=json.dumps(inspection))
|
|
with (
|
|
mock.patch.object(
|
|
self.manager,
|
|
"run_capture",
|
|
return_value=completed,
|
|
),
|
|
self.assertRaisesRegex(
|
|
GrokBuildError,
|
|
"unsafe-plugin.*active-mcp.*python-lsp",
|
|
),
|
|
):
|
|
self.manager.verify_read_only_environment(self.project_dir)
|
|
|
|
def test_read_only_preflight_fails_closed_on_invalid_inspection(self) -> None:
|
|
completed = mock.Mock(returncode=0, stdout="not-json")
|
|
with (
|
|
mock.patch.object(
|
|
self.manager,
|
|
"run_capture",
|
|
return_value=completed,
|
|
),
|
|
self.assertRaisesRegex(GrokBuildError, "未返回有效 JSON"),
|
|
):
|
|
self.manager.verify_read_only_environment(self.project_dir)
|
|
|
|
def test_headless_arguments_reject_empty_prompt_and_missing_workspace(self) -> None:
|
|
with self.assertRaises(GrokBuildError):
|
|
self.manager.build_headless_args(" ")
|
|
with self.assertRaises(GrokBuildError):
|
|
self.manager.build_headless_args(
|
|
"task",
|
|
workspace=self.project_dir / "missing",
|
|
)
|
|
|
|
def test_acp_arguments_end_with_stdio(self) -> None:
|
|
args = self.manager.build_acp_args(
|
|
model="wecom-backend",
|
|
yolo=True,
|
|
sandbox="workspace-write",
|
|
)
|
|
|
|
self.assertEqual(
|
|
[
|
|
"--no-auto-update",
|
|
"--sandbox",
|
|
"workspace-write",
|
|
"agent",
|
|
"--no-leader",
|
|
"--model",
|
|
"wecom-backend",
|
|
"--always-approve",
|
|
"stdio",
|
|
],
|
|
args,
|
|
)
|
|
|
|
def test_dify_acp_and_tui_disable_responses_only_web_search(self) -> None:
|
|
self._replace_ai_settings(
|
|
GROK_API_BACKEND="dify",
|
|
GROK_API_BASE="https://dify.example.test/v1",
|
|
)
|
|
|
|
acp_args = self.manager.build_acp_args(model="wecom-backend")
|
|
self.assertIn("--disable-web-search", acp_args)
|
|
|
|
sentinel = object()
|
|
with mock.patch.object(
|
|
self.manager,
|
|
"launch_console",
|
|
return_value=sentinel,
|
|
) as launch:
|
|
result = self.manager.open_tui(
|
|
workspace=self.project_dir,
|
|
model=MODEL_PROFILE,
|
|
)
|
|
self.assertIs(sentinel, result)
|
|
forwarded = launch.call_args.args[0]
|
|
self.assertIn("--disable-web-search", forwarded)
|
|
|
|
def test_release_version_rejects_path_traversal_before_network_access(self) -> None:
|
|
with mock.patch.object(self.manager, "_request") as request:
|
|
with self.assertRaises(GrokBuildError):
|
|
self.manager.install_official_release("1.2.3-a/../../victim")
|
|
request.assert_not_called()
|
|
|
|
def test_managed_binary_uses_recorded_signature_and_matching_hash(self) -> None:
|
|
self.manager.binary_path.parent.mkdir(parents=True)
|
|
payload = b"MZ" + (b"x" * (1024 * 1024))
|
|
self.manager.binary_path.write_bytes(payload)
|
|
self.manager.install_state_file.write_text(
|
|
json.dumps(
|
|
{
|
|
"source": "x.ai/cli",
|
|
"sha256": self.manager._file_sha256(self.manager.binary_path),
|
|
"publisher": "CN=X.AI LLC, O=X.AI LLC",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with mock.patch.object(
|
|
self.manager,
|
|
"_verify_windows_signature",
|
|
) as verify:
|
|
self.manager.validate_binary(self.manager.binary_path)
|
|
|
|
verify.assert_not_called()
|
|
|
|
def test_managed_binary_hash_mismatch_is_rejected(self) -> None:
|
|
self.manager.binary_path.parent.mkdir(parents=True)
|
|
self.manager.binary_path.write_bytes(b"MZ" + (b"x" * (1024 * 1024)))
|
|
self.manager.install_state_file.write_text(
|
|
json.dumps(
|
|
{
|
|
"source": "x.ai/cli",
|
|
"sha256": "0" * 64,
|
|
"publisher": "CN=X.AI LLC, O=X.AI LLC",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with self.assertRaisesRegex(GrokBuildError, "SHA-256"):
|
|
self.manager.validate_binary(self.manager.binary_path)
|
|
|
|
def test_locate_binary_prefers_environment_override(self) -> None:
|
|
configured = self.project_dir / "configured-grok.exe"
|
|
override = self.project_dir / "override-grok.exe"
|
|
configured.write_bytes(b"configured")
|
|
override.write_bytes(b"override")
|
|
self.integration_settings["binary_path"] = str(configured)
|
|
self._write_json(self.integration_settings_file, self.integration_settings)
|
|
|
|
with (
|
|
mock.patch.dict(
|
|
os.environ,
|
|
{"GROK_BUILD_BIN": str(override)},
|
|
clear=False,
|
|
),
|
|
mock.patch("grok_build_bridge.shutil.which", return_value=None),
|
|
):
|
|
located = self.manager.locate_binary()
|
|
|
|
self.assertEqual(override.resolve(), located)
|
|
|
|
def test_locate_binary_uses_managed_runtime_without_external_binary(self) -> None:
|
|
self.manager.binary_path.parent.mkdir(parents=True)
|
|
self.manager.binary_path.write_bytes(b"fake executable")
|
|
|
|
with (
|
|
mock.patch.dict(os.environ, {}, clear=True),
|
|
mock.patch(
|
|
"grok_build_bridge.Path.home",
|
|
return_value=self.project_dir / "fake-home",
|
|
),
|
|
mock.patch("grok_build_bridge.shutil.which", return_value=None),
|
|
):
|
|
located = self.manager.locate_binary()
|
|
|
|
self.assertEqual(self.manager.binary_path.resolve(), located)
|
|
|
|
def test_status_ignores_xai_auth_and_reports_synced_model(self) -> None:
|
|
fake_binary = self.project_dir / "grok-test.exe"
|
|
fake_binary.write_bytes(b"fake")
|
|
self.manager.sync_model_configuration()
|
|
self.runtime_home.mkdir(parents=True, exist_ok=True)
|
|
(self.runtime_home / "auth.json").write_text("{}", encoding="utf-8")
|
|
|
|
with (
|
|
mock.patch.object(
|
|
self.manager,
|
|
"locate_binary",
|
|
return_value=fake_binary,
|
|
),
|
|
mock.patch.object(
|
|
self.manager,
|
|
"version",
|
|
return_value="grok 1.2.3",
|
|
) as version_mock,
|
|
):
|
|
status = self.manager.status()
|
|
|
|
self.assertTrue(status.installed)
|
|
self.assertFalse(status.authenticated)
|
|
self.assertEqual(str(fake_binary), status.binary_path)
|
|
self.assertEqual("grok 1.2.3", status.version)
|
|
self.assertTrue(status.model_configured)
|
|
self.assertTrue(status.model_compatible)
|
|
self.assertEqual("local-qwen", status.model_name)
|
|
version_mock.assert_called_once_with(fake_binary)
|
|
|
|
def test_status_when_binary_is_absent_does_not_probe_version(self) -> None:
|
|
with (
|
|
mock.patch.object(self.manager, "locate_binary", return_value=None),
|
|
mock.patch.object(self.manager, "version") as version_mock,
|
|
):
|
|
status = self.manager.status()
|
|
|
|
self.assertFalse(status.installed)
|
|
self.assertEqual("", status.version)
|
|
self.assertEqual(str(self.manager.binary_path), status.binary_path)
|
|
version_mock.assert_not_called()
|
|
|
|
|
|
class StreamingEventParserTests(unittest.TestCase):
|
|
def test_text_and_thought_events(self) -> None:
|
|
self.assertEqual(
|
|
("text", "你好"),
|
|
parse_streaming_event('{"type":"text","data":"你好"}'),
|
|
)
|
|
self.assertEqual(
|
|
("thought", "分析中"),
|
|
parse_streaming_event('{"type":"thought","data":"分析中"}'),
|
|
)
|
|
|
|
def test_error_event_prefers_message(self) -> None:
|
|
self.assertEqual(
|
|
("error", "permission denied"),
|
|
parse_streaming_event(
|
|
'{"type":"error","message":"permission denied","data":"fallback"}'
|
|
),
|
|
)
|
|
|
|
def test_end_event_summarizes_turns_session_and_stop_reason(self) -> None:
|
|
category, text = parse_streaming_event(
|
|
json.dumps(
|
|
{
|
|
"type": "end",
|
|
"stopReason": "EndTurn",
|
|
"num_turns": 3,
|
|
"sessionId": "session-abc",
|
|
}
|
|
)
|
|
)
|
|
|
|
self.assertEqual("end", category)
|
|
self.assertIn("EndTurn", text)
|
|
self.assertIn("3", text)
|
|
self.assertIn("session-abc", text)
|
|
|
|
def test_blank_invalid_and_unknown_events_are_safe(self) -> None:
|
|
self.assertEqual(("empty", ""), parse_streaming_event(" \r\n"))
|
|
self.assertEqual(("raw", "not-json"), parse_streaming_event("not-json"))
|
|
unknown = '{"type":"tool","name":"read_file"}'
|
|
self.assertEqual(("tool", unknown), parse_streaming_event(unknown))
|
|
self.assertEqual(("raw", "[1,2,3]"), parse_streaming_event("[1,2,3]"))
|
|
|
|
|
|
class PassthroughArgumentTests(unittest.TestCase):
|
|
def test_conventional_separator_is_not_forwarded_to_grok(self) -> None:
|
|
self.assertEqual(
|
|
["inspect", "--json"],
|
|
normalize_passthrough_args(["--", "inspect", "--json"]),
|
|
)
|
|
self.assertEqual(["doctor"], normalize_passthrough_args(["doctor"]))
|
|
|
|
def test_empty_passthrough_command_is_rejected(self) -> None:
|
|
with self.assertRaisesRegex(GrokBuildError, "必须提供"):
|
|
normalize_passthrough_args(["--"])
|
|
|
|
def test_passthrough_workspace_tracks_forwarded_cwd(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
root = Path(temporary_directory)
|
|
nested = root / "nested"
|
|
nested.mkdir()
|
|
|
|
self.assertEqual(
|
|
nested.resolve(),
|
|
resolve_passthrough_workspace(
|
|
["--cwd", "nested", "inspect"],
|
|
root,
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
nested.resolve(),
|
|
resolve_passthrough_workspace(
|
|
[f"--cwd={nested}", "inspect"],
|
|
root,
|
|
),
|
|
)
|
|
|
|
def test_passthrough_workspace_rejects_missing_directory(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
with self.assertRaisesRegex(GrokBuildError, "工作目录不存在"):
|
|
resolve_passthrough_workspace(
|
|
["--cwd", "missing", "inspect"],
|
|
temporary_directory,
|
|
)
|
|
|
|
def test_passthrough_commands_are_classified_fail_closed(self) -> None:
|
|
self.assertEqual("safe", classify_passthrough_args(["inspect", "--json"]))
|
|
self.assertEqual(
|
|
"safe",
|
|
classify_passthrough_args(["--debug", "doctor"]),
|
|
)
|
|
self.assertEqual("safe", classify_passthrough_args(["--version"]))
|
|
self.assertEqual("wrap", classify_passthrough_args(["wrap", "python"]))
|
|
self.assertEqual(
|
|
"extension",
|
|
classify_passthrough_args(
|
|
["agent", "--plugin-dir", "D:/untrusted", "stdio"]
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
"extension",
|
|
classify_passthrough_args(
|
|
["--plugin-dir=D:/untrusted", "agent", "stdio"]
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
"extension",
|
|
classify_passthrough_args(
|
|
["agent", "--agent-profile", "D:/untrusted.toml", "stdio"]
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
"wrap",
|
|
classify_passthrough_args(
|
|
["-p", "not-used", "wrap", "python", "-c", "print(1)"]
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
"wrap",
|
|
classify_passthrough_args(
|
|
["--prompt-file", "prompt.txt", "wrap", "python"]
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
"agent",
|
|
classify_passthrough_args(["-p", "wrap"]),
|
|
)
|
|
self.assertEqual(
|
|
"safe",
|
|
classify_passthrough_args(["-p", "not-used", "inspect", "--json"]),
|
|
)
|
|
self.assertEqual(
|
|
"unknown",
|
|
classify_passthrough_args(["-p", "not-used", "future-command"]),
|
|
)
|
|
self.assertEqual(
|
|
"agent",
|
|
classify_passthrough_args(["--model", "wecom-backend", "agent", "stdio"]),
|
|
)
|
|
self.assertEqual("agent", classify_passthrough_args(["-p", "fix it"]))
|
|
self.assertEqual("unknown", classify_passthrough_args(["auth"]))
|
|
self.assertEqual(
|
|
"unknown",
|
|
classify_passthrough_args(["future-upstream-command"]),
|
|
)
|
|
|
|
def test_passthrough_agent_model_is_forced_to_backend_profile(self) -> None:
|
|
self.assertEqual(
|
|
["--model", MODEL_PROFILE, "agent", "stdio"],
|
|
enforce_passthrough_agent_model(["agent", "stdio"]),
|
|
)
|
|
self.assertEqual(
|
|
["--model", MODEL_PROFILE, "agent", "stdio"],
|
|
enforce_passthrough_agent_model(
|
|
["--model", MODEL_PROFILE, "agent", "stdio"]
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(GrokBuildError, "只允许"):
|
|
enforce_passthrough_agent_model(
|
|
["--model", "grok-4.5", "agent", "stdio"]
|
|
)
|
|
with self.assertRaisesRegex(GrokBuildError, "独立模型"):
|
|
reject_passthrough_model_bypasses(
|
|
["--agent", "custom-agent", "agent", "stdio"]
|
|
)
|
|
with self.assertRaisesRegex(GrokBuildError, "只允许"):
|
|
reject_passthrough_model_bypasses(
|
|
["--session-summary-model=grok-4.5", "-p", "task"]
|
|
)
|
|
reject_passthrough_model_bypasses(
|
|
["--session-summary-model", MODEL_PROFILE, "-p", "task"]
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|