"""Tests for the xAI Grok Build CLI adapter (detect / build / failure / env forwarding).""" from __future__ import annotations import os import subprocess from pathlib import Path from unittest.mock import MagicMock, patch import pytest from integrations.llm_cli.binary_resolver import npm_prefix_bin_dirs from integrations.llm_cli.grok_cli import ( GrokCLIAdapter, _classify_grok_auth_from_probe, _fallback_grok_paths, _has_explicit_grok_auth_env, parse_grok_models_output, ) from tests.integrations.llm_cli.testing_helpers import write_fake_runnable_cli_bin _SUBPROCESS_RUN = "integrations.llm_cli.grok_cli.subprocess.run" _WHICH = "integrations.llm_cli.binary_resolver.shutil.which" def _posix_path_set(paths: list[str]) -> set[str]: return {Path(p).as_posix() for p in paths} def _version_proc(version: str = "0.1.0") -> MagicMock: m = MagicMock() m.returncode = 0 m.stdout = f"grok {version}\n" m.stderr = "" return m def _auth_proc(returncode: int = 0, stdout: str = "ok", stderr: str = "") -> MagicMock: m = MagicMock() m.returncode = returncode m.stdout = stdout m.stderr = stderr return m # --------------------------------------------------------------------------- # Auth classification from probe output # --------------------------------------------------------------------------- def test_classify_auth_logged_in_string() -> None: logged_in, detail = _classify_grok_auth_from_probe( 0, "You are logged in with grok.com.\n\nDefault model: grok-build\n", "" ) assert logged_in is True assert "authenticated" in detail.lower() def test_classify_auth_exit0_is_authenticated() -> None: logged_in, detail = _classify_grok_auth_from_probe(0, "some other output", "") assert logged_in is True assert "authenticated" in detail.lower() def test_classify_auth_unauthorized_string() -> None: logged_in, detail = _classify_grok_auth_from_probe(1, "", "Error: Unauthorized") assert logged_in is False assert "grok login" in detail.lower() or "xai_api_key" in detail.lower() def test_classify_auth_401_in_output() -> None: logged_in, detail = _classify_grok_auth_from_probe(1, "", "HTTP 401 Unauthorized") assert logged_in is False def test_classify_auth_not_logged_in_string() -> None: logged_in, detail = _classify_grok_auth_from_probe(1, "not logged in", "") assert logged_in is False def test_classify_auth_network_error_is_unclear() -> None: logged_in, detail = _classify_grok_auth_from_probe(1, "", "network unreachable") assert logged_in is None assert "network" in detail.lower() def test_classify_auth_unknown_nonzero_is_unclear() -> None: logged_in, detail = _classify_grok_auth_from_probe(2, "", "something weird") assert logged_in is None assert "unclear" in detail.lower() def test_classify_auth_api_key_invalid() -> None: logged_in, detail = _classify_grok_auth_from_probe(1, "", "api key is invalid") assert logged_in is False # --------------------------------------------------------------------------- # _has_explicit_grok_auth_env # --------------------------------------------------------------------------- def test_has_explicit_auth_env_with_key() -> None: with patch.dict(os.environ, {"XAI_API_KEY": "xai-test"}, clear=False): assert _has_explicit_grok_auth_env() == "XAI_API_KEY" def test_has_explicit_auth_env_empty_key() -> None: with patch.dict(os.environ, {"XAI_API_KEY": ""}, clear=False): assert _has_explicit_grok_auth_env() is None def test_has_explicit_auth_env_missing_key() -> None: env = {k: v for k, v in os.environ.items() if k != "XAI_API_KEY"} with patch.dict(os.environ, env, clear=True): assert _has_explicit_grok_auth_env() is None # --------------------------------------------------------------------------- # detect() # --------------------------------------------------------------------------- @patch(_SUBPROCESS_RUN) @patch(_WHICH) def test_detect_logged_in_via_probe(mock_which: MagicMock, mock_run: MagicMock) -> None: mock_which.return_value = "/usr/bin/grok" mock_run.side_effect = [_version_proc(), _auth_proc(returncode=0, stdout="ok")] with patch.dict(os.environ, {"XAI_API_KEY": "", "GROK_CLI_BIN": ""}, clear=False): probe = GrokCLIAdapter().detect() assert probe.installed is True assert probe.logged_in is True assert probe.bin_path == "/usr/bin/grok" assert probe.version == "0.1.0" @patch(_SUBPROCESS_RUN) @patch(_WHICH) def test_detect_logged_in_via_api_key_fallback(mock_which: MagicMock, mock_run: MagicMock) -> None: """XAI_API_KEY promotes to authenticated even when the probe result is unclear.""" mock_which.return_value = "/usr/bin/grok" mock_run.side_effect = [_version_proc(), _auth_proc(returncode=1, stderr="network unreachable")] with patch.dict(os.environ, {"XAI_API_KEY": "xai-test", "GROK_CLI_BIN": ""}, clear=False): probe = GrokCLIAdapter().detect() assert probe.installed is True assert probe.logged_in is True assert "XAI_API_KEY" in probe.detail @patch(_SUBPROCESS_RUN) @patch(_WHICH) def test_detect_not_authenticated(mock_which: MagicMock, mock_run: MagicMock) -> None: mock_which.return_value = "/usr/bin/grok" mock_run.side_effect = [_version_proc(), _auth_proc(returncode=1, stderr="Error: Unauthorized")] with patch.dict(os.environ, {"XAI_API_KEY": "", "GROK_CLI_BIN": ""}, clear=False): probe = GrokCLIAdapter().detect() assert probe.installed is True assert probe.logged_in is False @patch(_SUBPROCESS_RUN) @patch(_WHICH) def test_detect_api_key_does_not_override_explicit_false( mock_which: MagicMock, mock_run: MagicMock ) -> None: """XAI_API_KEY must not promote logged_in when the probe explicitly returned False. The key itself may be what was rejected; masking that defers the failure to invocation time with a confusing 300s timeout instead of a clear error. """ mock_which.return_value = "/usr/bin/grok" mock_run.side_effect = [_version_proc(), _auth_proc(returncode=1, stderr="Error: Unauthorized")] with patch.dict(os.environ, {"XAI_API_KEY": "xai-invalid", "GROK_CLI_BIN": ""}, clear=False): probe = GrokCLIAdapter().detect() assert probe.logged_in is False @patch(_SUBPROCESS_RUN) @patch(_WHICH) def test_detect_auth_unclear_on_network_error(mock_which: MagicMock, mock_run: MagicMock) -> None: mock_which.return_value = "/usr/bin/grok" mock_run.side_effect = [_version_proc(), _auth_proc(returncode=1, stderr="connection refused")] with patch.dict(os.environ, {"XAI_API_KEY": "", "GROK_CLI_BIN": ""}, clear=False): probe = GrokCLIAdapter().detect() assert probe.installed is True assert probe.logged_in is None @patch(_SUBPROCESS_RUN) @patch(_WHICH) def test_detect_auth_probe_timeout_is_unclear(mock_which: MagicMock, mock_run: MagicMock) -> None: mock_which.return_value = "/usr/bin/grok" mock_run.side_effect = [ _version_proc(), subprocess.TimeoutExpired(cmd=["/usr/bin/grok", "models"], timeout=10.0), ] with patch.dict(os.environ, {"XAI_API_KEY": "", "GROK_CLI_BIN": ""}, clear=False): probe = GrokCLIAdapter().detect() assert probe.installed is True assert probe.logged_in is None assert "timed out" in probe.detail.lower() @patch(_SUBPROCESS_RUN) @patch(_WHICH) def test_detect_api_key_fallback_overrides_probe_timeout( mock_which: MagicMock, mock_run: MagicMock ) -> None: mock_which.return_value = "/usr/bin/grok" mock_run.side_effect = [ _version_proc(), subprocess.TimeoutExpired(cmd=["/usr/bin/grok", "models"], timeout=10.0), ] with patch.dict(os.environ, {"XAI_API_KEY": "xai-key", "GROK_CLI_BIN": ""}, clear=False): probe = GrokCLIAdapter().detect() assert probe.logged_in is True assert "XAI_API_KEY" in probe.detail @patch("integrations.llm_cli.grok_cli._fallback_grok_paths", return_value=[]) @patch(_WHICH, return_value=None) def test_detect_not_installed(_mock_which: MagicMock, _mock_fallback: MagicMock) -> None: with patch.dict(os.environ, {"GROK_CLI_BIN": ""}, clear=False): probe = GrokCLIAdapter().detect() assert probe.installed is False assert probe.logged_in is None assert probe.bin_path is None assert "not found" in probe.detail.lower() @patch(_SUBPROCESS_RUN) @patch(_WHICH) def test_detect_version_command_fails(mock_which: MagicMock, mock_run: MagicMock) -> None: mock_which.return_value = "/usr/bin/grok" m = MagicMock() m.returncode = 1 m.stdout = "" m.stderr = "some error\n" mock_run.return_value = m with patch.dict(os.environ, {"GROK_CLI_BIN": ""}, clear=False): probe = GrokCLIAdapter().detect() assert probe.installed is False assert probe.logged_in is None @patch(_SUBPROCESS_RUN) @patch(_WHICH) def test_detect_version_timeout(mock_which: MagicMock, mock_run: MagicMock) -> None: mock_which.return_value = "/usr/bin/grok" mock_run.side_effect = subprocess.TimeoutExpired( cmd=["/usr/bin/grok", "--version"], timeout=5.0 ) with patch.dict(os.environ, {"GROK_CLI_BIN": ""}, clear=False): probe = GrokCLIAdapter().detect() assert probe.installed is False assert probe.logged_in is None assert "--version" in probe.detail # --------------------------------------------------------------------------- # build() # --------------------------------------------------------------------------- @patch(_WHICH, return_value="/usr/bin/grok") def test_build_basic_invocation(_mock_which: MagicMock) -> None: with patch.dict(os.environ, {"GROK_CLI_BIN": ""}, clear=False): inv = GrokCLIAdapter().build(prompt="explain this alert", model=None, workspace="") assert inv.argv[0] == "/usr/bin/grok" assert "-p" in inv.argv assert "explain this alert" in inv.argv assert "--output-format" in inv.argv assert "plain" in inv.argv assert "--no-auto-update" not in inv.argv # Prompt is delivered as the -p argument, not stdin. assert inv.stdin is None assert inv.timeout_sec == 300.0 @patch(_WHICH, return_value="/usr/bin/grok") def test_build_never_auto_approves(_mock_which: MagicMock) -> None: """OpenSRE uses Grok as a text responder; it must not auto-run Grok's own tools.""" with patch.dict(os.environ, {"GROK_CLI_BIN": ""}, clear=False): inv = GrokCLIAdapter().build(prompt="p", model=None, workspace="") assert "--always-approve" not in inv.argv @patch(_WHICH, return_value="/usr/bin/grok") def test_build_adds_model_flag(_mock_which: MagicMock) -> None: with patch.dict(os.environ, {"GROK_CLI_BIN": ""}, clear=False): inv = GrokCLIAdapter().build(prompt="p", model="grok-build-0.1", workspace="") assert "-m" in inv.argv idx = inv.argv.index("-m") assert inv.argv[idx + 1] == "grok-build-0.1" @patch(_WHICH, return_value="/usr/bin/grok") def test_build_omits_model_flag_when_empty(_mock_which: MagicMock) -> None: with patch.dict(os.environ, {"GROK_CLI_BIN": ""}, clear=False): inv = GrokCLIAdapter().build(prompt="p", model="", workspace="") assert "-m" not in inv.argv @patch(_WHICH, return_value="/usr/bin/grok") def test_build_uses_provided_workspace(_mock_which: MagicMock) -> None: workspace = "/my/project" with patch.dict(os.environ, {"GROK_CLI_BIN": ""}, clear=False): inv = GrokCLIAdapter().build(prompt="p", model=None, workspace=workspace) assert Path(inv.cwd) == Path(workspace) @patch(_WHICH, return_value="/usr/bin/grok") def test_build_sets_no_color_env(_mock_which: MagicMock) -> None: with patch.dict(os.environ, {"GROK_CLI_BIN": ""}, clear=False): inv = GrokCLIAdapter().build(prompt="p", model=None, workspace="") assert inv.env is not None assert inv.env.get("NO_COLOR") == "1" @patch("integrations.llm_cli.grok_cli._fallback_grok_paths", return_value=[]) @patch(_WHICH, return_value=None) def test_build_raises_when_binary_not_found( _mock_which: MagicMock, _mock_fallback: MagicMock ) -> None: with ( patch.dict(os.environ, {"GROK_CLI_BIN": ""}, clear=False), pytest.raises(RuntimeError, match="Grok Build CLI not found"), ): GrokCLIAdapter().build(prompt="p", model=None, workspace="") # --------------------------------------------------------------------------- # parse / explain_failure # --------------------------------------------------------------------------- def test_parse_returns_stripped_stdout() -> None: result = GrokCLIAdapter().parse(stdout=" hello world \n", stderr="", returncode=0) assert result == "hello world" def test_parse_raises_on_empty_stdout() -> None: """Empty stdout on exit 0 must raise rather than return a silent blank response.""" with pytest.raises(RuntimeError, match="empty output"): GrokCLIAdapter().parse(stdout="", stderr="", returncode=0) def test_parse_raises_on_whitespace_only_stdout() -> None: with pytest.raises(RuntimeError, match="empty output"): GrokCLIAdapter().parse(stdout=" \n ", stderr="", returncode=0) def test_parse_surfaces_stderr_via_explain_failure() -> None: """When stdout is empty, stderr is surfaced through explain_failure in the error message.""" with pytest.raises(RuntimeError) as exc_info: GrokCLIAdapter().parse(stdout="", stderr="401 Unauthorized", returncode=1) assert "401" in str(exc_info.value) or "auth" in str(exc_info.value).lower() def test_explain_failure_includes_returncode_and_stderr() -> None: msg = GrokCLIAdapter().explain_failure(stdout="", stderr="boom", returncode=1) assert "1" in msg assert "boom" in msg def test_explain_failure_maps_auth_errors() -> None: msg = GrokCLIAdapter().explain_failure(stdout="", stderr="401 Unauthorized", returncode=1) assert "grok login" in msg.lower() or "xai_api_key" in msg.lower() def test_explain_failure_falls_back_to_stdout() -> None: msg = GrokCLIAdapter().explain_failure(stdout="some output", stderr="", returncode=2) assert "some output" in msg def test_explain_failure_maps_quota_via_shared_helper() -> None: """Quota/rate-limit errors get an actionable hint from the shared classifier.""" msg = GrokCLIAdapter().explain_failure( stdout="", stderr="Error 429: rate limit exceeded", returncode=1 ) assert "quota or rate limit" in msg.lower() def test_auth_hint_mentions_login_and_api_key() -> None: adapter = GrokCLIAdapter() assert "grok login" in adapter.auth_hint assert "XAI_API_KEY" in adapter.auth_hint # --------------------------------------------------------------------------- # GROK_CLI_BIN env override # --------------------------------------------------------------------------- @patch(_SUBPROCESS_RUN) def test_detect_uses_grok_cli_bin_env(mock_run: MagicMock, tmp_path: Path) -> None: fake_bin = write_fake_runnable_cli_bin(tmp_path, "my-grok") mock_run.side_effect = [_version_proc(), _auth_proc(returncode=0, stdout="ok")] with patch.dict( os.environ, {"GROK_CLI_BIN": str(fake_bin), "XAI_API_KEY": ""}, clear=False, ): probe = GrokCLIAdapter().detect() assert probe.bin_path == str(fake_bin) assert probe.installed is True # --------------------------------------------------------------------------- # Registry # --------------------------------------------------------------------------- def test_grok_cli_registry_entry() -> None: from integrations.llm_cli.registry import get_cli_provider_registration reg = get_cli_provider_registration("grok-cli") assert reg is not None assert reg.model_env_key == "GROK_CLI_MODEL" assert reg.adapter_factory().name == "grok-cli" # --------------------------------------------------------------------------- # Subprocess env forwarding — XAI_API_KEY must be scoped to the Grok subprocess # --------------------------------------------------------------------------- def test_xai_key_forwarded_via_build() -> None: """XAI_API_KEY is forwarded explicitly by build(), not via the blanket prefix allowlist.""" with ( patch.dict( os.environ, { "XAI_API_KEY": "xai-forward-me", "XAI_BASE_URL": "https://proxy.example.com", "GROK_CLI_BIN": "", }, clear=False, ), patch(_WHICH, return_value="/usr/bin/grok"), ): inv = GrokCLIAdapter().build(prompt="p", model=None, workspace="") assert inv.env is not None assert inv.env["XAI_API_KEY"] == "xai-forward-me" assert inv.env["XAI_BASE_URL"] == "https://proxy.example.com" def test_xai_key_not_in_blanket_subprocess_env() -> None: """XAI_API_KEY must NOT be forwarded via the global prefix allowlist (would leak to others).""" from integrations.llm_cli.subprocess_env import build_cli_subprocess_env with patch.dict(os.environ, {"XAI_API_KEY": "xai-secret"}, clear=False): env = build_cli_subprocess_env(None) assert "XAI_API_KEY" not in env # --------------------------------------------------------------------------- # Fallback paths # --------------------------------------------------------------------------- def test_fallback_paths_linux() -> None: npm_prefix_bin_dirs.cache_clear() with ( patch("integrations.llm_cli.binary_resolver.sys.platform", "linux"), patch.dict(os.environ, {"npm_config_prefix": "/custom/npm"}, clear=False), ): paths = _fallback_grok_paths() normalized = _posix_path_set(paths) assert "/custom/npm/bin/grok" in normalized # --------------------------------------------------------------------------- # parse_grok_models_output # --------------------------------------------------------------------------- def test_parse_models_output_typical() -> None: text = ( "You are logged in with grok.com.\n\n" "Default model: grok-build\n\n" "Available models:\n" " - grok-composer-2.5-fast\n" " * grok-build (default)\n" ) assert parse_grok_models_output(text) == ["grok-composer-2.5-fast", "grok-build"] def test_parse_models_output_empty_string() -> None: assert parse_grok_models_output("") == [] def test_parse_models_output_no_available_models_section() -> None: assert parse_grok_models_output("You are logged in.\n\nDefault model: grok-build\n") == [] def test_parse_models_output_stops_at_non_list_line() -> None: text = ( "Available models:\n" " - grok-fast\n" " - grok-slow\n" "\n" "Some trailing info that is not a model.\n" " - grok-phantom\n" ) # Blank line terminates the section; grok-phantom must not be included. assert parse_grok_models_output(text) == ["grok-fast", "grok-slow"] def test_parse_models_output_strips_parenthetical_annotation() -> None: text = "Available models:\n * grok-build (default)\n" assert parse_grok_models_output(text) == ["grok-build"] def test_parse_models_output_model_id_with_parenthesis_in_name() -> None: # A model whose ID itself contains '(' should be truncated at the first '(' # per the current split("(")[0] behaviour — this test documents that contract. text = "Available models:\n - grok-weird(beta)\n" assert parse_grok_models_output(text) == ["grok-weird"] def test_parse_models_output_section_header_case_insensitive() -> None: text = "AVAILABLE MODELS:\n - grok-x\n" assert parse_grok_models_output(text) == ["grok-x"]