"""Unit tests for grok-cli-mcp utility functions."""
import json
import os
from unittest.mock import patch
import pytest
from grok_cli_mcp.types import GrokMessage
from grok_cli_mcp.utils import (
ENV_GROK_API_KEY,
ENV_GROK_CLI_PATH,
_collect_assistant_text,
_extract_json_from_text,
_require_api_key,
_resolve_grok_path,
)
class TestResolveGrokPath:
"""Tests for _resolve_grok_path()."""
def test_explicit_env_override(self):
"""Test that explicit env var takes priority."""
with patch.dict(os.environ, {ENV_GROK_CLI_PATH: "/custom/path/grok"}):
with patch("os.path.exists", return_value=True):
result = _resolve_grok_path()
assert result == "/custom/path/grok"
def test_default_homebrew_path(self):
"""Test fallback to default homebrew path."""
with patch.dict(os.environ, {}, clear=True):
with patch("os.path.exists") as mock_exists:
mock_exists.return_value = True
with patch("shutil.which", return_value=None):
result = _resolve_grok_path()
assert result == "/opt/homebrew/bin/grok"
def test_path_lookup(self):
"""Test fallback to PATH lookup."""
with patch.dict(os.environ, {}, clear=True):
with patch("os.path.exists", return_value=False):
with patch("shutil.which", return_value="/usr/local/bin/grok"):
result = _resolve_grok_path()
assert result == "/usr/local/bin/grok"
def test_fallback_to_grok(self):
"""Test final fallback when nothing is found."""
with patch.dict(os.environ, {}, clear=True):
with patch("os.path.exists", return_value=False):
with patch("shutil.which", return_value=None):
result = _resolve_grok_path()
assert result == "grok"
class TestRequireApiKey:
"""Tests for _require_api_key()."""
def test_api_key_present(self):
"""Test when API key is set."""
with patch.dict(os.environ, {ENV_GROK_API_KEY: "test-key"}):
result = _require_api_key()
assert result == "test-key"
def test_api_key_missing(self):
"""Test error when API key is not set."""
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(ValueError, match="GROK_API_KEY is not set"):
_require_api_key()
class TestExtractJsonFromText:
"""Tests for _extract_json_from_text()."""
def test_simple_object(self):
"""Test basic JSON object extraction."""
text = '{"role": "assistant", "content": "Hello"}'
result = _extract_json_from_text(text)
assert result == {"role": "assistant", "content": "Hello"}
def test_simple_array(self):
"""Test basic JSON array extraction."""
text = '[{"role": "assistant", "content": "Test"}]'
result = _extract_json_from_text(text)
assert isinstance(result, list)
assert len(result) == 1
assert result[0]["content"] == "Test"
def test_json_with_surrounding_noise(self):
"""Test JSON extraction with surrounding text."""
text = '''Some debug output here
{"role": "assistant", "content": "Response"}
More text after'''
result = _extract_json_from_text(text)
assert result["role"] == "assistant"
assert result["content"] == "Response"
def test_nested_json(self):
"""Test extraction of nested JSON."""
text = '{"role": "assistant", "content": {"text": "Nested", "type": "text"}}'
result = _extract_json_from_text(text)
assert result["content"]["text"] == "Nested"
def test_invalid_json(self):
"""Test error on invalid JSON."""
with pytest.raises(ValueError, match="No JSON-like content"):
_extract_json_from_text("This is not JSON at all")
def test_empty_string(self):
"""Test error on empty string."""
with pytest.raises(ValueError, match="No JSON-like content"):
_extract_json_from_text("")
class TestCollectAssistantText:
"""Tests for _collect_assistant_text()."""
def test_simple_string_content(self):
"""Test with simple string content."""
messages = [
GrokMessage(role="assistant", content="Hello world"),
GrokMessage(role="user", content="Ignored"),
]
result = _collect_assistant_text(messages)
assert result == "Hello world"
def test_multiple_assistants(self):
"""Test concatenation of multiple assistant messages."""
messages = [
GrokMessage(role="assistant", content="First"),
GrokMessage(role="assistant", content="Second"),
]
result = _collect_assistant_text(messages)
assert "First" in result
assert "Second" in result
def test_structured_content_blocks(self):
"""Test with structured content blocks."""
messages = [
GrokMessage(
role="assistant",
content=[
{"type": "text", "text": "Block 1"},
{"type": "text", "text": "Block 2"},
],
)
]
result = _collect_assistant_text(messages)
assert "Block 1" in result
assert "Block 2" in result
def test_dict_with_text_field(self):
"""Test with dict content containing text field."""
messages = [GrokMessage(role="assistant", content={"text": "Content"})]
result = _collect_assistant_text(messages)
assert result == "Content"
def test_fallback_to_json_stringify(self):
"""Test fallback for unrecognized content structure."""
messages = [GrokMessage(role="assistant", content={"custom": "field"})]
result = _collect_assistant_text(messages)
# Should contain JSON representation
assert "custom" in result
def test_empty_messages(self):
"""Test with no messages."""
result = _collect_assistant_text([])
assert result == ""
def test_only_user_messages(self):
"""Test with only user messages (no assistant)."""
messages = [
GrokMessage(role="user", content="User message"),
]
result = _collect_assistant_text(messages)
assert result == ""
# Note: Integration tests for _run_grok() would require actual Grok CLI
# and are better suited for a separate test suite marked with @pytest.mark.integration