# -*- 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()