Chat interface not found at {html_path}
", - status_code=404, - mimetype="text/html" - ) - except Exception as e: - logging.error(f'Error serving chat web: {str(e)}') - return func.HttpResponse( - f"{str(e)}
", - status_code=500, - mimetype="text/html" - ) + html_path = Path(__file__).resolve().parent.parent / "chat-web" / "index.html" + content, status_code, headers = serve_static_file(html_path, "text/html", use_cache_headers=True) + + return func.HttpResponse( + content, + status_code=status_code, + mimetype="text/html", + headers=headers + ) @app.route(route="chat-web/chat.js", methods=["GET"], auth_level=func.AuthLevel.ANONYMOUS) def serve_chat_js(req: func.HttpRequest) -> func.HttpResponse: """Serve the chat JavaScript file""" - try: - js_path = Path(__file__).resolve().parent.parent / "chat-web" / "chat.js" - - if js_path.exists(): - with open(js_path, 'r', encoding='utf-8') as f: - js_content = f.read() - - return func.HttpResponse( - js_content, - status_code=200, - mimetype="application/javascript", - headers={ - "Cache-Control": "no-store, no-cache, must-revalidate, max-age=0", - "Pragma": "no-cache", - "Expires": "0" - } - ) - else: - return func.HttpResponse( - f"// Error: JavaScript file not found at {js_path}", - status_code=404, - mimetype="application/javascript" - ) - except Exception as e: - logging.error(f'Error serving chat.js: {str(e)}') - return func.HttpResponse( - f"// Error: {str(e)}", - status_code=500, - mimetype="application/javascript" - ) + js_path = Path(__file__).resolve().parent.parent / "chat-web" / "chat.js" + content, status_code, headers = serve_static_file(js_path, "application/javascript", use_cache_headers=True) + + return func.HttpResponse( + content, + status_code=status_code, + mimetype="application/javascript", + headers=headers + ) diff --git a/shared/http_utils.py b/shared/http_utils.py new file mode 100644 index 000000000..1da19fab0 --- /dev/null +++ b/shared/http_utils.py @@ -0,0 +1,182 @@ +"""Common validation and serving utilities for HTTP endpoints. + +Provides reusable validation functions and file serving utilities to reduce +duplication across Azure Functions endpoints. +""" +from typing import List, Dict, Any, Optional, Tuple +from pathlib import Path +import logging + +_LOGGER = logging.getLogger(__name__) + + +def validate_messages(messages: Any) -> Tuple[bool, Optional[str]]: + """Validate chat messages format. + + Args: + messages: The messages to validate (should be list of dicts) + + Returns: + Tuple of (is_valid, error_message) + - If valid: (True, None) + - If invalid: (False, "error description") + + Expected format: + [{"role": "user|assistant|system", "content": "..."}] + """ + if not messages: + return False, "No messages provided" + + if not isinstance(messages, list): + return False, "Messages must be a list" + + for idx, msg in enumerate(messages): + if not isinstance(msg, dict): + return False, f"Message {idx} must be a dict" + + if 'role' not in msg: + return False, f"Message {idx} missing 'role' field" + + if 'content' not in msg: + return False, f"Message {idx} missing 'content' field" + + # Validate role is one of the expected values + valid_roles = {'user', 'assistant', 'system'} + roles_str = ", ".join(sorted(valid_roles)) + if msg['role'] not in valid_roles: + return False, f"Message {idx} has invalid role '{msg['role']}'. Expected one of: {roles_str}" + + return True, None + + +def create_cors_headers( + allow_origin: str = "*", + allow_methods: str = "POST, GET, OPTIONS", + allow_headers: str = "Content-Type" +) -> Dict[str, str]: + """Create standard CORS headers for HTTP responses. + + Args: + allow_origin: Allowed origins (default: "*") + allow_methods: Allowed HTTP methods (default: "POST, GET, OPTIONS") + allow_headers: Allowed headers (default: "Content-Type") + + Returns: + Dict of CORS headers + """ + return { + "Access-Control-Allow-Origin": allow_origin, + "Access-Control-Allow-Methods": allow_methods, + "Access-Control-Allow-Headers": allow_headers, + } + + +def create_no_cache_headers() -> Dict[str, str]: + """Create headers that prevent caching. + + Useful for serving dynamic content that should always be fresh. + + Returns: + Dict of cache-control headers + """ + return { + "Cache-Control": "no-store, no-cache, must-revalidate, max-age=0", + "Pragma": "no-cache", + "Expires": "0", + } + + +def validate_provider_choice( + provider_choice: Optional[str], + model_override: Optional[str] = None +) -> Tuple[bool, Optional[str], Optional[Dict[str, Any]]]: + """Validate provider choice and model override. + + Args: + provider_choice: The requested provider ('auto', 'openai', 'azure', 'local', 'lora') + model_override: Optional model path/name + + Returns: + Tuple of (is_valid, error_message, hints) + - If valid: (True, None, None) + - If invalid: (False, "error description", {"hint": "...", "key": "value"}) + """ + if not provider_choice: + return True, None, None + + provider_lower = provider_choice.lower() + valid_providers = {'auto', 'openai', 'azure', 'local', 'lora', 'lmstudio'} + + if provider_lower not in valid_providers: + return False, f"Invalid provider '{provider_choice}'", { + "hint": f"Valid providers: {', '.join(sorted(valid_providers))}", + "requested": provider_choice + } + + # LoRA provider requires model path + if provider_lower == 'lora' and not model_override: + return False, "LoRA provider requires model path", { + "hint": "Provide 'model' in request body (e.g., data_out/lora_training/lora_adapter)", + "provider": provider_choice + } + + return True, None, None + + +def serve_static_file( + file_path: Path, + mimetype: str, + use_cache_headers: bool = False +) -> Tuple[Optional[str], int, Dict[str, str]]: + """Serve a static file with appropriate headers. + + Args: + file_path: Path to the file to serve + mimetype: MIME type for the response (e.g., 'text/html', 'application/javascript') + use_cache_headers: Whether to add no-cache headers (default: False for better caching) + + Returns: + Tuple of (content, status_code, headers) + - On success: (file_content, 200, headers_dict) + - On error: (error_message, error_code, {}) + + Example: + content, status, headers = serve_static_file( + Path("chat-web/index.html"), + "text/html", + use_cache_headers=True + ) + return func.HttpResponse(content, status_code=status, mimetype=mimetype, headers=headers) + """ + try: + if not file_path.exists(): + error_msg = f"File not found: {file_path}" + if mimetype.startswith("text/html"): + error_msg = f"{error_msg}
" + elif mimetype.startswith("application/javascript"): + error_msg = f"// Error: {error_msg}" + else: + error_msg = f"Error: {error_msg}" + + return error_msg, 404, {} + + with open(file_path, 'r', encoding='utf-8') as f: + content = f.read() + + headers = {} + if use_cache_headers: + headers.update(create_no_cache_headers()) + + return content, 200, headers + + except Exception as e: + _LOGGER.error(f"Error serving file {file_path}: {e}") + + error_msg = f"Internal error: {str(e)}" + if mimetype.startswith("text/html"): + error_msg = f"{error_msg}
" + elif mimetype.startswith("application/javascript"): + error_msg = f"// Error: {error_msg}" + + return error_msg, 500, {} + diff --git a/shared/import_helpers.py b/shared/import_helpers.py new file mode 100644 index 000000000..045f5c0b5 --- /dev/null +++ b/shared/import_helpers.py @@ -0,0 +1,104 @@ +"""Utilities for defensive imports with graceful fallbacks. + +This module provides helpers to reduce boilerplate when importing optional +dependencies that may not be available in all runtime environments. +""" +from typing import Any, Callable, Dict, Optional, Tuple +import logging + +_LOGGER = logging.getLogger(__name__) + + +def safe_import( + module_path: str, + *, + import_names: Optional[Tuple[str, ...]] = None, + fallback_factory: Optional[Callable[[str], Any]] = None, + log_failure: bool = True, +) -> Any: + """Safely import a module or specific names, with optional fallbacks. + + Args: + module_path: Full module path (e.g., 'shared.sql_engine') + import_names: Optional tuple of specific names to import from the module + (e.g., ('sql_health', 'engine_stats')). If None, returns the module. + fallback_factory: Optional callable that receives the import name and returns + a fallback function/object. Called once per import_name on failure. + log_failure: Whether to log import failures (default: True) + + Returns: + - If import_names is None: the imported module, or None on failure + - If import_names is provided: dict mapping names to imported values or fallbacks + + Examples: + # Import entire module + sql_engine = safe_import('shared.sql_engine') + + # Import specific functions with fallbacks + funcs = safe_import( + 'shared.sql_engine', + import_names=('sql_health', 'engine_stats'), + fallback_factory=lambda name: lambda: {"enabled": False, "error": f"{name}_import_failed"} + ) + sql_health = funcs['sql_health'] + engine_stats = funcs['engine_stats'] + """ + try: + # Import the module + parts = module_path.split('.') + module = __import__(module_path, fromlist=parts[-1:] if len(parts) > 1 else []) + + if import_names is None: + # Return the whole module + return module + + # Extract specific names + result: Dict[str, Any] = {} + for name in import_names: + if hasattr(module, name): + result[name] = getattr(module, name) + elif fallback_factory: + result[name] = fallback_factory(name) + else: + result[name] = None + + return result + + except Exception as e: + if log_failure: + _LOGGER.info(f"[safe_import] Failed to import {module_path}: {e}") + + if import_names is None: + # No specific names requested, return None + return None + + # Build fallback dict + result: Dict[str, Any] = {} + for name in import_names: + if fallback_factory: + result[name] = fallback_factory(name) + else: + result[name] = None + + return result + + +def create_stub_function(name: str, error_key: str = "error") -> Callable[..., Dict[str, Any]]: + """Create a stub function that returns a dict indicating unavailability. + + Args: + name: The function name (used in error message) + error_key: Key name for the error field (default: 'error') + + Returns: + A function that accepts any args/kwargs and returns an error dict + + Example: + sql_health = create_stub_function('sql_health') + # sql_health() returns {"enabled": False, "error": "sql_health_unavailable"} + """ + def stub(*args, **kwargs) -> Dict[str, Any]: + return {"enabled": False, error_key: f"{name}_unavailable"} + + stub.__name__ = name + return stub diff --git a/talk-to-ai/src/chat_providers.py b/talk-to-ai/src/chat_providers.py index e5361907c..a38a88ff0 100644 --- a/talk-to-ai/src/chat_providers.py +++ b/talk-to-ai/src/chat_providers.py @@ -103,6 +103,34 @@ class BaseChatProvider: def complete(self, messages: List[RoleMessage], stream: bool = True) -> Iterable[str] | str: raise NotImplementedError + @staticmethod + def _handle_openai_streaming_response(response) -> Generator[str, None, None]: + """Extract content from OpenAI-style streaming response. + + Common helper for OpenAI, LMStudio, and other OpenAI-compatible providers. + Handles the standard streaming chunk format with resilient error handling. + """ + for chunk in response: + try: + delta = chunk.choices[0].delta + if delta and delta.content: + yield delta.content + except Exception: + # Be resilient to SDK shape changes + pass + + @staticmethod + def _handle_openai_non_streaming_response(response) -> str: + """Extract content from OpenAI-style non-streaming response. + + Common helper for OpenAI, LMStudio, and other OpenAI-compatible providers. + Handles the standard completion format with resilient error handling. + """ + try: + return response.choices[0].message.content or "" + except Exception: + return "" + class LoraLocalProvider(BaseChatProvider): """Provider for local inference with LoRA adapters. @@ -382,37 +410,18 @@ def __init__(self, model: str, api_key: Optional[str] = None, temperature: float self.max_output_tokens = max_output_tokens def complete(self, messages: List[RoleMessage], stream: bool = True) -> Iterable[str] | str: + resp = self.client.chat.completions.create( + model=self.model, + messages=messages, + temperature=self.temperature, + max_tokens=self.max_output_tokens, + stream=stream, + ) + if stream: - resp = self.client.chat.completions.create( - model=self.model, - messages=messages, - temperature=self.temperature, - max_tokens=self.max_output_tokens, - stream=True, - ) - - def gen() -> Generator[str, None, None]: - for chunk in resp: - try: - delta = chunk.choices[0].delta - if delta and delta.content: - yield delta.content - except Exception: - # Be resilient to SDK shape changes - pass - return gen() + return self._handle_openai_streaming_response(resp) else: - resp = self.client.chat.completions.create( - model=self.model, - messages=messages, - temperature=self.temperature, - max_tokens=self.max_output_tokens, - stream=False, - ) - try: - return resp.choices[0].message.content or "" - except Exception: - return "" + return self._handle_openai_non_streaming_response(resp) class LMStudioProvider(BaseChatProvider): @@ -431,36 +440,18 @@ def __init__(self, base_url: str = "http://127.0.0.1:1234/v1", model: str = "loc self.max_output_tokens = max_output_tokens def complete(self, messages: List[RoleMessage], stream: bool = True) -> Iterable[str] | str: + resp = self.client.chat.completions.create( + model=self.model, + messages=messages, + temperature=self.temperature, + max_tokens=self.max_output_tokens, + stream=stream, + ) + if stream: - resp = self.client.chat.completions.create( - model=self.model, - messages=messages, - temperature=self.temperature, - max_tokens=self.max_output_tokens, - stream=True, - ) - - def gen() -> Generator[str, None, None]: - for chunk in resp: - try: - delta = chunk.choices[0].delta - if delta and delta.content: - yield delta.content - except Exception: - pass - return gen() + return self._handle_openai_streaming_response(resp) else: - resp = self.client.chat.completions.create( - model=self.model, - messages=messages, - temperature=self.temperature, - max_tokens=self.max_output_tokens, - stream=False, - ) - try: - return resp.choices[0].message.content or "" - except Exception: - return "" + return self._handle_openai_non_streaming_response(resp) class AzureOpenAIProvider(BaseChatProvider): diff --git a/tests/test_http_utils.py b/tests/test_http_utils.py new file mode 100644 index 000000000..1c34657ad --- /dev/null +++ b/tests/test_http_utils.py @@ -0,0 +1,240 @@ +"""Test suite for http_utils module. + +This test validates HTTP utility functions for validation and file serving. +""" +import sys +from pathlib import Path +import tempfile +import os + +# Add shared to path +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from shared.http_utils import ( + validate_messages, + create_cors_headers, + create_no_cache_headers, + validate_provider_choice, + serve_static_file +) + + +def test_validate_messages_success(): + """Test that valid messages pass validation.""" + messages = [ + {"role": "user", "content": "Hello"}, + {"role": "assistant", "content": "Hi there!"}, + {"role": "system", "content": "You are helpful"} + ] + + is_valid, error = validate_messages(messages) + + assert is_valid is True + assert error is None + print("✓ Valid messages pass validation") + + +def test_validate_messages_empty(): + """Test that empty messages fail validation.""" + is_valid, error = validate_messages([]) + + assert is_valid is False + assert "No messages" in error + print("✓ Empty messages fail validation") + + +def test_validate_messages_not_list(): + """Test that non-list input fails validation.""" + is_valid, error = validate_messages("not a list") + + assert is_valid is False + assert "must be a list" in error + print("✓ Non-list input fails validation") + + +def test_validate_messages_missing_role(): + """Test that message without role fails validation.""" + messages = [ + {"content": "Hello"} + ] + + is_valid, error = validate_messages(messages) + + assert is_valid is False + assert "role" in error + print("✓ Message without role fails validation") + + +def test_validate_messages_missing_content(): + """Test that message without content fails validation.""" + messages = [ + {"role": "user"} + ] + + is_valid, error = validate_messages(messages) + + assert is_valid is False + assert "content" in error + print("✓ Message without content fails validation") + + +def test_validate_messages_invalid_role(): + """Test that message with invalid role fails validation.""" + messages = [ + {"role": "invalid_role", "content": "Hello"} + ] + + is_valid, error = validate_messages(messages) + + assert is_valid is False + assert "invalid role" in error.lower() + print("✓ Invalid role fails validation") + + +def test_create_cors_headers(): + """Test CORS headers creation.""" + headers = create_cors_headers() + + assert "Access-Control-Allow-Origin" in headers + assert "Access-Control-Allow-Methods" in headers + assert "Access-Control-Allow-Headers" in headers + assert headers["Access-Control-Allow-Origin"] == "*" + print("✓ CORS headers created correctly") + + +def test_create_cors_headers_custom(): + """Test CORS headers with custom values.""" + headers = create_cors_headers( + allow_origin="https://example.com", + allow_methods="GET, POST", + allow_headers="X-Custom-Header" + ) + + assert headers["Access-Control-Allow-Origin"] == "https://example.com" + assert headers["Access-Control-Allow-Methods"] == "GET, POST" + assert headers["Access-Control-Allow-Headers"] == "X-Custom-Header" + print("✓ Custom CORS headers created correctly") + + +def test_create_no_cache_headers(): + """Test no-cache headers creation.""" + headers = create_no_cache_headers() + + assert "Cache-Control" in headers + assert "no-cache" in headers["Cache-Control"] + assert "Pragma" in headers + assert "Expires" in headers + print("✓ No-cache headers created correctly") + + +def test_validate_provider_choice_valid(): + """Test that valid provider choices pass validation.""" + for provider in ['auto', 'openai', 'azure', 'local', 'lmstudio']: + is_valid, error, hints = validate_provider_choice(provider) + assert is_valid is True, f"Provider {provider} should be valid" + assert error is None + assert hints is None + + print("✓ Valid provider choices pass validation") + + +def test_validate_provider_choice_invalid(): + """Test that invalid provider fails validation.""" + is_valid, error, hints = validate_provider_choice("invalid_provider") + + assert is_valid is False + assert "Invalid provider" in error + assert hints is not None + assert "hint" in hints + print("✓ Invalid provider fails validation") + + +def test_validate_provider_choice_lora_without_model(): + """Test that LoRA without model path fails validation.""" + is_valid, error, hints = validate_provider_choice("lora", model_override=None) + + assert is_valid is False + assert "LoRA" in error + assert "model path" in error + assert hints is not None + print("✓ LoRA without model path fails validation") + + +def test_validate_provider_choice_lora_with_model(): + """Test that LoRA with model path passes validation.""" + is_valid, error, hints = validate_provider_choice("lora", model_override="/path/to/adapter") + + assert is_valid is True + assert error is None + assert hints is None + print("✓ LoRA with model path passes validation") + + +def test_serve_static_file_success(): + """Test serving an existing file.""" + # Create a temporary file + with tempfile.NamedTemporaryFile(mode='w', suffix='.html', delete=False, encoding='utf-8') as f: + f.write("Test") + temp_path = Path(f.name) + + try: + content, status, headers = serve_static_file(temp_path, "text/html", use_cache_headers=False) + + assert status == 200 + assert "" in content + assert "Test" in content + print("✓ Serving existing file works") + finally: + os.unlink(temp_path) + + +def test_serve_static_file_with_cache_headers(): + """Test serving file with cache headers.""" + # Create a temporary file + with tempfile.NamedTemporaryFile(mode='w', suffix='.js', delete=False, encoding='utf-8') as f: + f.write("console.log('test');") + temp_path = Path(f.name) + + try: + content, status, headers = serve_static_file(temp_path, "application/javascript", use_cache_headers=True) + + assert status == 200 + assert "console.log" in content + assert "Cache-Control" in headers + assert "no-cache" in headers["Cache-Control"] + print("✓ Serving file with cache headers works") + finally: + os.unlink(temp_path) + + +def test_serve_static_file_not_found(): + """Test serving non-existent file.""" + fake_path = Path("/nonexistent/file.html") + content, status, headers = serve_static_file(fake_path, "text/html") + + assert status == 404 + assert "not found" in content.lower() + print("✓ Serving non-existent file returns 404") + + +if __name__ == "__main__": + print("Testing http_utils module...\n") + + test_validate_messages_success() + test_validate_messages_empty() + test_validate_messages_not_list() + test_validate_messages_missing_role() + test_validate_messages_missing_content() + test_validate_messages_invalid_role() + test_create_cors_headers() + test_create_cors_headers_custom() + test_create_no_cache_headers() + test_validate_provider_choice_valid() + test_validate_provider_choice_invalid() + test_validate_provider_choice_lora_without_model() + test_validate_provider_choice_lora_with_model() + test_serve_static_file_success() + test_serve_static_file_with_cache_headers() + test_serve_static_file_not_found() + + print("\n✅ All http_utils tests passed!") diff --git a/tests/test_import_helpers.py b/tests/test_import_helpers.py new file mode 100644 index 000000000..b96013dbd --- /dev/null +++ b/tests/test_import_helpers.py @@ -0,0 +1,155 @@ +"""Test suite for import_helpers module. + +This test validates that the defensive import utilities work correctly +and provide proper fallback behavior. +""" +import sys +from pathlib import Path + +# Add shared to path +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from shared.import_helpers import safe_import, create_stub_function + + +def test_safe_import_module_success(): + """Test that safe_import returns module when import succeeds.""" + # Import a known module + result = safe_import('json') + + assert result is not None + assert hasattr(result, 'loads') + assert hasattr(result, 'dumps') + print("✓ safe_import returns module on success") + + +def test_safe_import_module_failure(): + """Test that safe_import returns None when module doesn't exist.""" + result = safe_import('nonexistent_module_xyz123', log_failure=False) + + assert result is None + print("✓ safe_import returns None for missing module") + + +def test_safe_import_with_names_success(): + """Test importing specific names from a module.""" + result = safe_import('json', import_names=('loads', 'dumps')) + + assert 'loads' in result + assert 'dumps' in result + assert callable(result['loads']) + assert callable(result['dumps']) + print("✓ safe_import extracts specific names successfully") + + +def test_safe_import_with_names_and_fallback(): + """Test that fallback factory is used when import fails.""" + def my_fallback(name): + return f"fallback_{name}" + + result = safe_import( + 'nonexistent_module', + import_names=('func1', 'func2'), + fallback_factory=my_fallback, + log_failure=False + ) + + assert result['func1'] == "fallback_func1" + assert result['func2'] == "fallback_func2" + print("✓ safe_import uses fallback_factory on failure") + + +def test_safe_import_partial_failure(): + """Test behavior when some names exist and others don't.""" + # json has 'loads' but not 'nonexistent_func' + result = safe_import( + 'json', + import_names=('loads', 'nonexistent_func'), + fallback_factory=lambda name: f"fallback_{name}" + ) + + assert callable(result['loads']) # Real function + assert result['nonexistent_func'] == "fallback_nonexistent_func" # Fallback + print("✓ safe_import handles partial availability correctly") + + +def test_create_stub_function(): + """Test that stub function returns proper error dict.""" + stub = create_stub_function('my_func') + + result = stub() + + assert isinstance(result, dict) + assert result['enabled'] == False + assert result['error'] == 'my_func_unavailable' + assert stub.__name__ == 'my_func' + print("✓ create_stub_function creates proper stub") + + +def test_create_stub_function_with_args(): + """Test that stub function accepts any arguments.""" + stub = create_stub_function('func_with_args') + + # Should work with any args/kwargs + result1 = stub(1, 2, 3) + result2 = stub(x=1, y=2) + result3 = stub(1, 2, z=3) + + assert all(r['enabled'] == False for r in [result1, result2, result3]) + print("✓ Stub function accepts arbitrary arguments") + + +def test_create_stub_function_custom_error_key(): + """Test creating stub with custom error key.""" + stub = create_stub_function('my_func', error_key='failure_reason') + + result = stub() + + assert 'failure_reason' in result + assert result['failure_reason'] == 'my_func_unavailable' + print("✓ create_stub_function accepts custom error key") + + +def test_safe_import_real_world_pattern(): + """Test the pattern used in function_app.py.""" + # Simulate importing sql_health and engine_stats with fallbacks + result = safe_import( + 'nonexistent.sql_engine', + import_names=('sql_health', 'engine_stats'), + fallback_factory=create_stub_function, + log_failure=False + ) + + sql_health = result['sql_health'] + engine_stats = result['engine_stats'] + + # Both should be callable stubs + assert callable(sql_health) + assert callable(engine_stats) + + # Both should return error dicts + health_result = sql_health() + stats_result = engine_stats() + + assert health_result['enabled'] == False + assert 'sql_health_unavailable' in health_result['error'] + assert stats_result['enabled'] == False + assert 'engine_stats_unavailable' in stats_result['error'] + + print("✓ Real-world pattern from function_app.py works correctly") + + +if __name__ == "__main__": + print("Testing import_helpers module...\n") + + test_safe_import_module_success() + test_safe_import_module_failure() + test_safe_import_with_names_success() + test_safe_import_with_names_and_fallback() + test_safe_import_partial_failure() + test_create_stub_function() + test_create_stub_function_with_args() + test_create_stub_function_custom_error_key() + test_safe_import_real_world_pattern() + + print("\n✅ All import_helpers tests passed!") diff --git a/tests/test_provider_response_handling.py b/tests/test_provider_response_handling.py new file mode 100644 index 000000000..79baed77b --- /dev/null +++ b/tests/test_provider_response_handling.py @@ -0,0 +1,120 @@ +"""Test suite for refactored provider response handling. + +This test validates that the extracted helper methods in BaseChatProvider +maintain the same behavior as the original inline implementations. +""" +import sys +from pathlib import Path +from unittest.mock import Mock + +# Add talk-to-ai to path +sys.path.insert(0, str(Path(__file__).parent.parent / "talk-to-ai" / "src")) + +from chat_providers import BaseChatProvider + + +def test_handle_openai_streaming_response(): + """Test that streaming response handler correctly extracts content.""" + # Create mock streaming response with typical OpenAI chunk structure + mock_chunk1 = Mock() + mock_chunk1.choices = [Mock()] + mock_chunk1.choices[0].delta = Mock() + mock_chunk1.choices[0].delta.content = "Hello" + + mock_chunk2 = Mock() + mock_chunk2.choices = [Mock()] + mock_chunk2.choices[0].delta = Mock() + mock_chunk2.choices[0].delta.content = " world" + + mock_chunk3 = Mock() + mock_chunk3.choices = [Mock()] + mock_chunk3.choices[0].delta = Mock() + mock_chunk3.choices[0].delta.content = None # Empty content should be skipped + + mock_response = [mock_chunk1, mock_chunk2, mock_chunk3] + + # Test the helper method + result = list(BaseChatProvider._handle_openai_streaming_response(mock_response)) + + assert result == ["Hello", " world"] + print("✓ Streaming response handler works correctly") + + +def test_handle_openai_streaming_response_resilience(): + """Test that streaming handler is resilient to malformed chunks.""" + # Create response with some malformed chunks + mock_chunk1 = Mock() + mock_chunk1.choices = [Mock()] + mock_chunk1.choices[0].delta = Mock() + mock_chunk1.choices[0].delta.content = "Good" + + # Malformed chunk - accessing delta.content will raise AttributeError + mock_chunk2 = Mock() + mock_chunk2.choices = [Mock()] + mock_chunk2.choices[0].delta = Mock(spec=[]) # Empty spec means no attributes + + mock_chunk3 = Mock() + mock_chunk3.choices = [Mock()] + mock_chunk3.choices[0].delta = Mock() + mock_chunk3.choices[0].delta.content = " data" + + mock_response = [mock_chunk1, mock_chunk2, mock_chunk3] + + # Should not raise, should skip malformed chunk + result = list(BaseChatProvider._handle_openai_streaming_response(mock_response)) + + # Both good chunks should be returned (malformed chunk is skipped or yields nothing) + assert "Good" in result + assert " data" in result + print("✓ Streaming handler is resilient to malformed chunks") + + +def test_handle_openai_non_streaming_response(): + """Test that non-streaming response handler correctly extracts content.""" + # Create mock non-streaming response + mock_response = Mock() + mock_response.choices = [Mock()] + mock_response.choices[0].message = Mock() + mock_response.choices[0].message.content = "Complete response" + + result = BaseChatProvider._handle_openai_non_streaming_response(mock_response) + + assert result == "Complete response" + print("✓ Non-streaming response handler works correctly") + + +def test_handle_openai_non_streaming_response_empty(): + """Test that non-streaming handler returns empty string for None content.""" + mock_response = Mock() + mock_response.choices = [Mock()] + mock_response.choices[0].message = Mock() + mock_response.choices[0].message.content = None + + result = BaseChatProvider._handle_openai_non_streaming_response(mock_response) + + assert result == "" + print("✓ Non-streaming handler handles None content") + + +def test_handle_openai_non_streaming_response_resilience(): + """Test that non-streaming handler is resilient to malformed response.""" + # Malformed response - missing structure + mock_response = Mock() + # No choices attribute - should be caught by exception handler + + result = BaseChatProvider._handle_openai_non_streaming_response(mock_response) + + assert result == "" + print("✓ Non-streaming handler is resilient to errors") + + +if __name__ == "__main__": + print("Testing refactored provider response handling...\n") + + test_handle_openai_streaming_response() + test_handle_openai_streaming_response_resilience() + test_handle_openai_non_streaming_response() + test_handle_openai_non_streaming_response_empty() + test_handle_openai_non_streaming_response_resilience() + + print("\n✅ All provider response handling tests passed!")