""" Extended Tests for API Client Phase 20: API Client & Authentication - API Client Tests Tests authentication, session management, and API operations. """ import pytest from unittest.mock import patch, MagicMock class TestAuthentication: """Tests for authentication functionality""" @patch("local_deep_research.api.client.SafeSession") def test_login_success(self, mock_session_cls): """Test successful login flow""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session # Mock login page response with CSRF token mock_login_page = MagicMock() mock_login_page.text = """
""" # Mock login POST response mock_login_response = MagicMock() mock_login_response.status_code = 200 # Mock CSRF token endpoint mock_csrf_response = MagicMock() mock_csrf_response.status_code = 200 mock_csrf_response.json.return_value = {"csrf_token": "api_csrf_456"} mock_session.get.side_effect = [mock_login_page, mock_csrf_response] mock_session.post.return_value = mock_login_response client = LDRClient() 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_456" @patch("local_deep_research.api.client.SafeSession") def test_login_invalid_credentials(self, mock_session_cls): """Test login with invalid credentials""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session # Mock login page response mock_login_page = MagicMock() mock_login_page.text = '' # Mock failed login mock_login_response = MagicMock() mock_login_response.status_code = 401 mock_session.get.return_value = mock_login_page mock_session.post.return_value = mock_login_response client = LDRClient() result = client.login("baduser", "badpass") assert result is False assert client.logged_in is False @patch("local_deep_research.api.client.SafeSession") def test_login_csrf_token_extraction(self, mock_session_cls): """Test CSRF token extraction from login page""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session # HTML with various CSRF patterns mock_login_page = MagicMock() mock_login_page.text = """
""" mock_session.get.side_effect = [ mock_login_page, MagicMock( status_code=200, json=MagicMock(return_value={"csrf_token": "api_token"}), ), ] mock_session.post.return_value = MagicMock(status_code=200) client = LDRClient() client.login("user", "pass") # Verify the correct CSRF token was sent in login POST call_args = mock_session.post.call_args assert call_args[1]["data"]["csrf_token"] == "extracted_token_123" @patch("local_deep_research.api.client.SafeSession") def test_login_session_persistence(self, mock_session_cls): """Test session cookies persist after login""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_login_page = MagicMock() mock_login_page.text = '' mock_session.get.side_effect = [ mock_login_page, MagicMock( status_code=200, json=MagicMock(return_value={"csrf_token": "token"}), ), ] mock_session.post.return_value = MagicMock(status_code=200) client = LDRClient() client.login("user", "pass") # Session should be used for subsequent requests assert client.session is mock_session @patch("local_deep_research.api.client.SafeSession") def test_logout_session_cleanup(self, mock_session_cls): """Test logout cleans up session""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session client = LDRClient() client.logged_in = True client.csrf_token = "test_token" client.username = "testuser" client.logout() assert client.logged_in is False assert client.csrf_token is None assert client.username is None mock_session.close.assert_called_once() @patch("local_deep_research.api.client.SafeSession") def test_session_expiry_handling(self, mock_session_cls): """Test handling of expired session""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session # Simulate expired session response mock_session.get.return_value = MagicMock(status_code=401) client = LDRClient() client.logged_in = True with pytest.raises(RuntimeError): client.get_settings() @patch("local_deep_research.api.client.SafeSession") def test_login_no_csrf_token_in_page(self, mock_session_cls): """Test login fails gracefully when no CSRF token found""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session # Page without CSRF token mock_login_page = MagicMock() mock_login_page.text = ( '
' ) mock_session.get.return_value = mock_login_page client = LDRClient() result = client.login("user", "pass") assert result is False @patch("local_deep_research.api.client.SafeSession") def test_login_redirect_handling(self, mock_session_cls): """Test login handles redirects properly""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_login_page = MagicMock() mock_login_page.text = '' # 302 redirect after successful login mock_login_response = MagicMock() mock_login_response.status_code = 302 mock_csrf_response = MagicMock() mock_csrf_response.status_code = 200 mock_csrf_response.json.return_value = {"csrf_token": "api_token"} mock_session.get.side_effect = [mock_login_page, mock_csrf_response] mock_session.post.return_value = mock_login_response client = LDRClient() result = client.login("user", "pass") assert result is True @patch("local_deep_research.api.client.SafeSession") def test_login_html_parsing(self, mock_session_cls): """Test CSRF extraction from various HTML formats""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session # Test different HTML input formats html_formats = [ '', '', '', ] for html in html_formats: mock_login_page = MagicMock() mock_login_page.text = f"
{html}
" mock_session.get.side_effect = [ mock_login_page, MagicMock( status_code=200, json=MagicMock(return_value={"csrf_token": "api"}), ), ] mock_session.post.return_value = MagicMock(status_code=200) client = LDRClient() result = client.login("user", "pass") assert result is True @patch("local_deep_research.api.client.SafeSession") def test_login_error_extraction(self, mock_session_cls): """Test error message extraction on login failure""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_login_page = MagicMock() mock_login_page.text = '' mock_login_response = MagicMock() mock_login_response.status_code = 403 mock_session.get.return_value = mock_login_page mock_session.post.return_value = mock_login_response client = LDRClient() result = client.login("user", "pass") assert result is False class TestAPIOperations: """Tests for API operation methods""" @patch("local_deep_research.api.client.SafeSession") def test_api_get_request(self, mock_session_cls): """Test GET request to API""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_session.get.return_value = MagicMock( status_code=200, json=MagicMock(return_value={"data": "test"}) ) client = LDRClient() client.logged_in = True result = client.get_settings() assert result == {"data": "test"} @patch("local_deep_research.api.client.SafeSession") def test_api_post_request(self, mock_session_cls): """Test POST request to API""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_session.post.return_value = MagicMock( status_code=200, json=MagicMock(return_value={"research_id": "123"}) ) client = LDRClient() client.logged_in = True client.csrf_token = "test_token" result = client.quick_research("test query", wait_for_result=False) assert result == {"research_id": "123"} # Verify CSRF token was included call_args = mock_session.post.call_args assert call_args[1]["headers"]["X-CSRF-Token"] == "test_token" @patch("local_deep_research.api.client.SafeSession") def test_api_put_request(self, mock_session_cls): """Test PUT request to API""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_session.put.return_value = MagicMock(status_code=200) client = LDRClient() client.logged_in = True client.csrf_token = "test_token" result = client.update_setting("llm.model", "test-model") assert result is True mock_session.put.assert_called_once() @patch("local_deep_research.api.client.SafeSession") def test_api_error_handling(self, mock_session_cls): """Test API error handling""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_session.get.return_value = MagicMock(status_code=500) client = LDRClient() client.logged_in = True with pytest.raises(RuntimeError): client.get_settings() @patch("local_deep_research.api.client.SafeSession") @patch("time.sleep") def test_api_timeout_handling(self, mock_sleep, mock_session_cls): """Test timeout handling in wait_for_research""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session # Always return in_progress status mock_session.get.return_value = MagicMock( status_code=200, json=MagicMock(return_value={"status": "in_progress"}), ) client = LDRClient() client.logged_in = True with pytest.raises(RuntimeError, match="timed out"): client.wait_for_research("123", timeout=1) @patch("local_deep_research.api.client.SafeSession") def test_api_response_parsing(self, mock_session_cls): """Test API response JSON parsing""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session complex_response = { "history": [ {"id": 1, "query": "test1"}, {"id": 2, "query": "test2"}, ] } mock_session.get.return_value = MagicMock( status_code=200, json=MagicMock(return_value=complex_response) ) client = LDRClient() client.logged_in = True result = client.get_history() assert len(result) == 2 assert result[0]["query"] == "test1" @patch("local_deep_research.api.client.SafeSession") def test_api_headers_with_csrf(self, mock_session_cls): """Test API headers include CSRF token""" from local_deep_research.api.client import LDRClient client = LDRClient() client.csrf_token = "my_csrf_token" headers = client._api_headers() assert headers["X-CSRF-Token"] == "my_csrf_token" @patch("local_deep_research.api.client.SafeSession") def test_api_headers_without_csrf(self, mock_session_cls): """Test API headers when no CSRF token""" from local_deep_research.api.client import LDRClient client = LDRClient() client.csrf_token = None headers = client._api_headers() assert headers == {} @patch("local_deep_research.api.client.SafeSession") def test_not_logged_in_raises_error(self, mock_session_cls): """Test methods raise error when not logged in""" from local_deep_research.api.client import LDRClient client = LDRClient() client.logged_in = False with pytest.raises(RuntimeError, match="Not logged in"): client.get_settings() with pytest.raises(RuntimeError, match="Not logged in"): client.quick_research("test") with pytest.raises(RuntimeError, match="Not logged in"): client.get_history() @patch("local_deep_research.api.client.SafeSession") @patch("time.sleep") def test_wait_for_research_success(self, mock_sleep, mock_session_cls): """Test successful research completion""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session # First call returns in_progress, second returns completed mock_session.get.side_effect = [ MagicMock( status_code=200, json=MagicMock(return_value={"status": "in_progress"}), ), MagicMock( status_code=200, json=MagicMock(return_value={"status": "completed"}), ), MagicMock( status_code=200, json=MagicMock(return_value={"summary": "Research results"}), ), ] client = LDRClient() client.logged_in = True result = client.wait_for_research("123", timeout=30) assert result["summary"] == "Research results" @patch("local_deep_research.api.client.SafeSession") def test_wait_for_research_failure(self, mock_session_cls): """Test research failure handling""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_session.get.return_value = MagicMock( status_code=200, json=MagicMock( return_value={"status": "failed", "error": "LLM error"} ), ) client = LDRClient() client.logged_in = True with pytest.raises(RuntimeError, match="Research failed"): client.wait_for_research("123") class TestContextManager: """Tests for context manager functionality""" @patch("local_deep_research.api.client.SafeSession") def test_context_manager_enter(self, mock_session_cls): """Test context manager __enter__""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session with LDRClient() as client: assert isinstance(client, LDRClient) @patch("local_deep_research.api.client.SafeSession") def test_context_manager_exit_logout(self, mock_session_cls): """Test context manager __exit__ calls logout""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session with LDRClient() as client: client.logged_in = True mock_session.close.assert_called_once() class TestQuickQuery: """Tests for quick_query convenience function""" @patch("local_deep_research.api.client.LDRClient") def test_quick_query_success(self, mock_client_cls): """Test quick_query returns summary""" from local_deep_research.api.client import quick_query mock_client = MagicMock() mock_client_cls.return_value.__enter__ = MagicMock( return_value=mock_client ) mock_client_cls.return_value.__exit__ = MagicMock(return_value=False) mock_client.login.return_value = True mock_client.quick_research.return_value = {"summary": "Test summary"} result = quick_query("user", "pass", "test query") assert result == "Test summary" @patch("local_deep_research.api.client.LDRClient") def test_quick_query_login_failure(self, mock_client_cls): """Test quick_query raises on login failure""" from local_deep_research.api.client import quick_query mock_client = MagicMock() mock_client_cls.return_value.__enter__ = MagicMock( return_value=mock_client ) mock_client_cls.return_value.__exit__ = MagicMock(return_value=False) mock_client.login.return_value = False with pytest.raises(RuntimeError, match="Login failed"): quick_query("user", "pass", "test") class TestBenchmarkMethods: """Tests for benchmark-related methods""" @patch("local_deep_research.api.client.SafeSession") @patch("local_deep_research.api.client.Benchmark_results") def test_submit_benchmark(self, mock_benchmark_cls, mock_session_cls): """Test benchmark submission""" from local_deep_research.api.client import LDRClient mock_benchmark = MagicMock() mock_benchmark.add_result.return_value = True mock_benchmark_cls.return_value = mock_benchmark client = LDRClient() result = client.submit_benchmark( model="test-model", hardware="test-hw", accuracy_focused=85.0, accuracy_source=80.0, avg_time_per_question=30.0, context_window=32000, temperature=0.1, ldr_version="0.6.0", date_tested="2024-01-01", ) assert result is True mock_benchmark.add_result.assert_called_once() @patch("local_deep_research.api.client.SafeSession") @patch("local_deep_research.api.client.Benchmark_results") def test_get_benchmarks_all(self, mock_benchmark_cls, mock_session_cls): """Test getting all benchmarks""" from local_deep_research.api.client import LDRClient mock_benchmark = MagicMock() mock_benchmark.get_all.return_value = [{"model": "test"}] mock_benchmark_cls.return_value = mock_benchmark client = LDRClient() result = client.get_benchmarks(best_only=False) assert result == [{"model": "test"}] mock_benchmark.get_all.assert_called_once() @patch("local_deep_research.api.client.SafeSession") @patch("local_deep_research.api.client.Benchmark_results") def test_get_benchmarks_best_only( self, mock_benchmark_cls, mock_session_cls ): """Test getting best benchmarks only""" from local_deep_research.api.client import LDRClient mock_benchmark = MagicMock() mock_benchmark.get_best.return_value = [{"model": "best"}] mock_benchmark_cls.return_value = mock_benchmark client = LDRClient() result = client.get_benchmarks(best_only=True) assert result == [{"model": "best"}] mock_benchmark.get_best.assert_called_once() class TestHistoryHandling: """Tests for history retrieval""" @patch("local_deep_research.api.client.SafeSession") def test_get_history_dict_format(self, mock_session_cls): """Test history with dict response format""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_session.get.return_value = MagicMock( status_code=200, json=MagicMock(return_value={"history": [{"id": 1}]}), ) client = LDRClient() client.logged_in = True result = client.get_history() assert result == [{"id": 1}] @patch("local_deep_research.api.client.SafeSession") def test_get_history_items_format(self, mock_session_cls): """Test history with items key in response""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_session.get.return_value = MagicMock( status_code=200, json=MagicMock(return_value={"items": [{"id": 2}]}) ) client = LDRClient() client.logged_in = True result = client.get_history() assert result == [{"id": 2}] @patch("local_deep_research.api.client.SafeSession") def test_get_history_list_format(self, mock_session_cls): """Test history with list response format""" from local_deep_research.api.client import LDRClient mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_session.get.return_value = MagicMock( status_code=200, json=MagicMock(return_value=[{"id": 3}]) ) client = LDRClient() client.logged_in = True result = client.get_history() assert result == [{"id": 3}] class TestErrorResponseParsing: """Tests for error response JSON parsing (PR #2027). PR #2027 changed bare `except:` to `except Exception:` when parsing error JSON from failed API responses. """ def test_valid_json_error_parsed(self): """Valid JSON error response is parsed.""" import json error_body = json.dumps([{"message": "Bad request"}]) try: data = json.loads(error_body) if isinstance(data, list) and len(data) > 0: msg = data[0].get("message", "Unknown error") else: msg = str(data) except Exception: msg = error_body[:200] assert msg == "Bad request" def test_invalid_json_falls_back_to_text(self): """Invalid JSON falls back to raw text (caught by except Exception).""" error_body = "Not JSON at all" try: import json data = json.loads(error_body) msg = str(data) except Exception: msg = error_body[:200] assert msg == "Not JSON at all" def test_empty_response_falls_back(self): """Empty response body falls back to text.""" error_body = "" try: import json data = json.loads(error_body) msg = str(data) except Exception: msg = error_body[:200] assert msg == ""