"""Tests for LDRClient HTTP client."""
import pytest
from unittest.mock import MagicMock, Mock, patch
from local_deep_research.api.client import LDRClient, quick_query
class TestLDRClientInit:
"""Tests for LDRClient initialization."""
def test_default_base_url(self):
"""Test default base URL is localhost:5000."""
client = LDRClient()
assert client.base_url == "http://localhost:5000"
def test_custom_base_url(self):
"""Test custom base URL."""
client = LDRClient(base_url="http://example.com:8080")
assert client.base_url == "http://example.com:8080"
def test_initial_state(self):
"""Test initial state of client."""
client = LDRClient()
assert client.csrf_token is None
assert client.logged_in is False
assert client.username is None
def test_creates_session(self):
"""Test that a session is created."""
client = LDRClient()
assert client.session is not None
class TestLDRClientLogin:
"""Tests for LDRClient login."""
def test_login_success(self):
"""Test successful login."""
client = LDRClient()
# Mock the session responses
login_page_response = Mock()
login_page_response.text = (
''
)
login_response = Mock()
login_response.status_code = 200
csrf_response = Mock()
csrf_response.status_code = 200
csrf_response.json.return_value = {"csrf_token": "api_csrf_token"}
client.session = Mock()
client.session.get.side_effect = [login_page_response, csrf_response]
client.session.post.return_value = login_response
result = client.login("testuser", "testpass")
assert result is True
assert client.logged_in is True
assert client.username == "testuser"
assert client.csrf_token == "api_csrf_token"
def test_login_no_csrf_in_page(self):
"""Test login failure when CSRF token not found in page."""
client = LDRClient()
login_page_response = Mock()
login_page_response.text = "No CSRF here"
client.session = Mock()
client.session.get.return_value = login_page_response
result = client.login("testuser", "testpass")
assert result is False
assert client.logged_in is False
def test_login_http_error(self):
"""Test login failure on HTTP error."""
client = LDRClient()
login_page_response = Mock()
login_page_response.text = (
''
)
login_response = Mock()
login_response.status_code = 401 # Unauthorized
client.session = Mock()
client.session.get.return_value = login_page_response
client.session.post.return_value = login_response
result = client.login("testuser", "wrongpass")
assert result is False
def test_login_csrf_fetch_fails(self):
"""Test login succeeds even if CSRF fetch fails."""
client = LDRClient()
login_page_response = Mock()
login_page_response.text = (
''
)
login_response = Mock()
login_response.status_code = 200
csrf_response = Mock()
csrf_response.status_code = 500 # Failed
client.session = Mock()
client.session.get.side_effect = [login_page_response, csrf_response]
client.session.post.return_value = login_response
result = client.login("testuser", "testpass")
# Should still succeed but without API CSRF token
assert result is True
assert client.logged_in is True
assert client.csrf_token is None
def test_login_exception(self):
"""Test login handles exceptions."""
client = LDRClient()
client.session = Mock()
client.session.get.side_effect = Exception("Network error")
result = client.login("testuser", "testpass")
assert result is False
class TestLDRClientApiHeaders:
"""Tests for LDRClient API headers."""
def test_api_headers_with_token(self):
"""Test API headers when CSRF token is set."""
client = LDRClient()
client.csrf_token = "test_token"
headers = client._api_headers()
assert headers == {"X-CSRF-Token": "test_token"}
def test_api_headers_without_token(self):
"""Test API headers when no CSRF token."""
client = LDRClient()
headers = client._api_headers()
assert headers == {}
class TestLDRClientQuickResearch:
"""Tests for LDRClient quick_research."""
def test_requires_login(self):
"""Test that quick_research requires login."""
client = LDRClient()
with pytest.raises(RuntimeError, match="Not logged in"):
client.quick_research("test query")
def test_starts_research(self):
"""Test that research is started."""
client = LDRClient()
client.logged_in = True
start_response = Mock()
start_response.status_code = 200
start_response.json.return_value = {"research_id": "test-id"}
status_response = Mock()
status_response.status_code = 200
status_response.json.return_value = {"status": "completed"}
result_response = Mock()
result_response.status_code = 200
result_response.json.return_value = {"summary": "Test summary"}
client.session = Mock()
client.session.post.return_value = start_response
client.session.get.side_effect = [status_response, result_response]
result = client.quick_research("test query")
assert result == {"summary": "Test summary"}
def test_returns_research_id_when_not_waiting(self):
"""Test returning research_id when not waiting."""
client = LDRClient()
client.logged_in = True
start_response = Mock()
start_response.status_code = 200
start_response.json.return_value = {"research_id": "test-id"}
client.session = Mock()
client.session.post.return_value = start_response
result = client.quick_research("test query", wait_for_result=False)
assert result == {"research_id": "test-id"}
def test_raises_on_start_error(self):
"""Test raising error on start failure."""
client = LDRClient()
client.logged_in = True
error_response = Mock()
error_response.status_code = 500
error_response.json.return_value = [{"message": "Server error"}]
client.session = Mock()
client.session.post.return_value = error_response
with pytest.raises(RuntimeError, match="Failed to start research"):
client.quick_research("test query")
def test_raises_on_no_research_id(self):
"""Test raising error when no research_id returned."""
client = LDRClient()
client.logged_in = True
response = Mock()
response.status_code = 200
response.json.return_value = {} # No research_id
client.session = Mock()
client.session.post.return_value = response
with pytest.raises(RuntimeError, match="No research ID"):
client.quick_research("test query")
class TestLDRClientWaitForResearch:
"""Tests for LDRClient wait_for_research."""
def test_returns_result_on_completion(self):
"""Test returning result when research completes."""
client = LDRClient()
status_response = Mock()
status_response.status_code = 200
status_response.json.return_value = {"status": "completed"}
result_response = Mock()
result_response.status_code = 200
result_response.json.return_value = {"summary": "Result"}
client.session = Mock()
client.session.get.side_effect = [status_response, result_response]
result = client.wait_for_research("test-id", timeout=10)
assert result == {"summary": "Result"}
def test_raises_on_failure(self):
"""Test raising error on research failure."""
client = LDRClient()
status_response = Mock()
status_response.status_code = 200
status_response.json.return_value = {
"status": "failed",
"error": "Test error",
}
client.session = Mock()
client.session.get.return_value = status_response
with pytest.raises(RuntimeError, match="Research failed"):
client.wait_for_research("test-id", timeout=10)
def test_raises_on_timeout(self):
"""Test raising error on timeout."""
client = LDRClient()
status_response = Mock()
status_response.status_code = 200
status_response.json.return_value = {"status": "in_progress"}
client.session = Mock()
client.session.get.return_value = status_response
with pytest.raises(RuntimeError, match="timed out"):
client.wait_for_research("test-id", timeout=1)
def test_raises_on_result_fetch_error(self):
"""Test raising error when result fetch fails."""
client = LDRClient()
status_response = Mock()
status_response.status_code = 200
status_response.json.return_value = {"status": "completed"}
result_response = Mock()
result_response.status_code = 500
client.session = Mock()
client.session.get.side_effect = [status_response, result_response]
with pytest.raises(RuntimeError, match="Failed to get results"):
client.wait_for_research("test-id", timeout=10)
class TestLDRClientGetSettings:
"""Tests for LDRClient get_settings."""
def test_requires_login(self):
"""Test that get_settings requires login."""
client = LDRClient()
with pytest.raises(RuntimeError, match="Not logged in"):
client.get_settings()
def test_returns_settings(self):
"""Test returning settings."""
client = LDRClient()
client.logged_in = True
response = Mock()
response.status_code = 200
response.json.return_value = {"llm": {"model": "test"}}
client.session = Mock()
client.session.get.return_value = response
result = client.get_settings()
assert result == {"llm": {"model": "test"}}
def test_raises_on_error(self):
"""Test raising error on failure."""
client = LDRClient()
client.logged_in = True
response = Mock()
response.status_code = 500
client.session = Mock()
client.session.get.return_value = response
with pytest.raises(RuntimeError, match="Failed to get settings"):
client.get_settings()
class TestLDRClientUpdateSetting:
"""Tests for LDRClient update_setting."""
def test_requires_login(self):
"""Test that update_setting requires login."""
client = LDRClient()
with pytest.raises(RuntimeError, match="Not logged in"):
client.update_setting("key", "value")
def test_returns_true_on_success(self):
"""Test returning True on success."""
client = LDRClient()
client.logged_in = True
client.csrf_token = "token"
response = Mock()
response.status_code = 200
client.session = Mock()
client.session.put.return_value = response
result = client.update_setting("llm.model", "new_model")
assert result is True
def test_returns_false_on_failure(self):
"""Test returning False on failure."""
client = LDRClient()
client.logged_in = True
response = Mock()
response.status_code = 400
client.session = Mock()
client.session.put.return_value = response
result = client.update_setting("llm.model", "new_model")
assert result is False
class TestLDRClientGetHistory:
"""Tests for LDRClient get_history."""
def test_requires_login(self):
"""Test that get_history requires login."""
client = LDRClient()
with pytest.raises(RuntimeError, match="Not logged in"):
client.get_history()
def test_returns_list(self):
"""Test returning history list."""
client = LDRClient()
client.logged_in = True
response = Mock()
response.status_code = 200
response.json.return_value = [{"query": "test"}]
client.session = Mock()
client.session.get.return_value = response
result = client.get_history()
assert result == [{"query": "test"}]
def test_handles_dict_response(self):
"""Test handling dict response with history key."""
client = LDRClient()
client.logged_in = True
response = Mock()
response.status_code = 200
response.json.return_value = {"history": [{"query": "test"}]}
client.session = Mock()
client.session.get.return_value = response
result = client.get_history()
assert result == [{"query": "test"}]
def test_handles_dict_response_with_items(self):
"""Test handling dict response with items key."""
client = LDRClient()
client.logged_in = True
response = Mock()
response.status_code = 200
response.json.return_value = {"items": [{"query": "test"}]}
client.session = Mock()
client.session.get.return_value = response
result = client.get_history()
assert result == [{"query": "test"}]
def test_raises_on_error(self):
"""Test raising error on failure."""
client = LDRClient()
client.logged_in = True
response = Mock()
response.status_code = 500
client.session = Mock()
client.session.get.return_value = response
with pytest.raises(RuntimeError, match="Failed to get history"):
client.get_history()
class TestLDRClientLogout:
"""Tests for LDRClient logout."""
def test_logout_clears_state(self):
"""Test that logout clears client state."""
client = LDRClient()
client.logged_in = True
client.username = "testuser"
client.csrf_token = "token"
client.session = Mock()
client.logout()
assert client.logged_in is False
assert client.username is None
assert client.csrf_token is None
def test_logout_posts_when_logged_in(self):
"""Test that logout posts to server when logged in."""
client = LDRClient()
client.logged_in = True
client.session = Mock()
client.logout()
client.session.post.assert_called_once()
def test_logout_skips_post_when_not_logged_in(self):
"""Test that logout doesn't post when not logged in."""
client = LDRClient()
client.logged_in = False
client.session = Mock()
client.logout()
client.session.post.assert_not_called()
class TestLDRClientContextManager:
"""Tests for LDRClient context manager."""
def test_enter_returns_client(self):
"""Test that __enter__ returns the client."""
client = LDRClient()
result = client.__enter__()
assert result is client
def test_exit_calls_logout(self):
"""Test that __exit__ calls logout."""
client = LDRClient()
client.session = Mock()
with patch.object(client, "logout") as mock_logout:
client.__exit__(None, None, None)
mock_logout.assert_called_once()
class TestLDRClientBenchmarks:
"""Tests for LDRClient benchmark methods."""
def test_submit_benchmark(self):
"""Test submitting benchmark results."""
client = LDRClient()
with patch(
"local_deep_research.api.client.Benchmark_results"
) as mock_class:
mock_benchmarks = Mock()
mock_benchmarks.add_result.return_value = True
mock_class.return_value = mock_benchmarks
result = client.submit_benchmark(
model="test-model",
hardware="test-hardware",
accuracy_focused=90.0,
accuracy_source=85.0,
avg_time_per_question=30.0,
context_window=4096,
temperature=0.7,
ldr_version="1.0.0",
date_tested="2024-01-01",
)
assert result is True
mock_benchmarks.add_result.assert_called_once()
def test_get_benchmarks_all(self):
"""Test getting all benchmarks."""
client = LDRClient()
with patch(
"local_deep_research.api.client.Benchmark_results"
) as mock_class:
mock_benchmarks = Mock()
mock_benchmarks.get_all.return_value = [{"model": "test"}]
mock_class.return_value = mock_benchmarks
result = client.get_benchmarks()
assert result == [{"model": "test"}]
mock_benchmarks.get_all.assert_called_once()
def test_get_benchmarks_best_only(self):
"""Test getting best benchmarks only."""
client = LDRClient()
with patch(
"local_deep_research.api.client.Benchmark_results"
) as mock_class:
mock_benchmarks = Mock()
mock_benchmarks.get_best.return_value = [{"model": "best"}]
mock_class.return_value = mock_benchmarks
result = client.get_benchmarks(best_only=True)
assert result == [{"model": "best"}]
mock_benchmarks.get_best.assert_called_once()
class TestQuickQuery:
"""Tests for quick_query function."""
def test_returns_summary(self):
"""Test that quick_query returns summary."""
with patch(
"local_deep_research.api.client.LDRClient"
) as mock_client_class:
mock_client = MagicMock()
mock_client.login.return_value = True
mock_client.quick_research.return_value = {
"summary": "Test summary"
}
mock_client_class.return_value.__enter__ = Mock(
return_value=mock_client
)
mock_client_class.return_value.__exit__ = Mock(return_value=False)
result = quick_query("user", "pass", "test query")
assert result == "Test summary"
def test_raises_on_login_failure(self):
"""Test raising error on login failure."""
with patch(
"local_deep_research.api.client.LDRClient"
) as mock_client_class:
mock_client = MagicMock()
mock_client.login.return_value = False
mock_client_class.return_value.__enter__ = Mock(
return_value=mock_client
)
mock_client_class.return_value.__exit__ = Mock(return_value=False)
with pytest.raises(RuntimeError, match="Login failed"):
quick_query("user", "wrongpass", "test query")
def test_uses_custom_base_url(self):
"""Test using custom base URL."""
with patch(
"local_deep_research.api.client.LDRClient"
) as mock_client_class:
mock_client = MagicMock()
mock_client.login.return_value = True
mock_client.quick_research.return_value = {"summary": "Test"}
mock_client_class.return_value.__enter__ = Mock(
return_value=mock_client
)
mock_client_class.return_value.__exit__ = Mock(return_value=False)
quick_query("user", "pass", "query", base_url="http://custom:8080")
mock_client_class.assert_called_once_with("http://custom:8080")
def test_returns_no_summary_available(self):
"""Test returning default when no summary."""
with patch(
"local_deep_research.api.client.LDRClient"
) as mock_client_class:
mock_client = MagicMock()
mock_client.login.return_value = True
mock_client.quick_research.return_value = {} # No summary
mock_client_class.return_value.__enter__ = Mock(
return_value=mock_client
)
mock_client_class.return_value.__exit__ = Mock(return_value=False)
result = quick_query("user", "pass", "query")
assert result == "No summary available"