From 5f091440d99e562f61ef8346e39bbd7978d15053 Mon Sep 17 00:00:00 2001 From: Sapandeep Singh <110763985+sapandeep31@users.noreply.github.com> Date: Thu, 10 Sep 2026 11:27:51 +0530 Subject: [PATCH 1/3] fix(eos_ai): normalize LLM URLs, improve error resilience, and add unit test suite --- ebuild/eos_ai/llm_integration.py | 129 ++++++-- tests/unit/test_eos_ai_llm.py | 515 +++++++++++++++++++++++++++++++ 2 files changed, 622 insertions(+), 22 deletions(-) create mode 100644 tests/unit/test_eos_ai_llm.py diff --git a/ebuild/eos_ai/llm_integration.py b/ebuild/eos_ai/llm_integration.py index 57bbb094..32e07592 100644 --- a/ebuild/eos_ai/llm_integration.py +++ b/ebuild/eos_ai/llm_integration.py @@ -72,14 +72,40 @@ def __init__( self.timeout = timeout if provider == "ollama": - self.base_url = base_url or self.OLLAMA_URL + self.base_url = (base_url or self.OLLAMA_URL).rstrip("/") elif provider == "openai": - self.base_url = base_url or self.OPENAI_URL + self.base_url = (base_url or self.OPENAI_URL).rstrip("/") self.api_key = api_key or os.environ.get("OPENAI_API_KEY", "") elif provider == "custom": - self.base_url = base_url or "" + self.base_url = (base_url or "").rstrip("/") self.api_key = api_key or os.environ.get("EOS_LLM_API_KEY", "") + @staticmethod + def _normalize_ollama_url(base_url: str) -> str: + """Normalize Ollama base URL to generation endpoint.""" + url = (base_url or "").rstrip("/") + if url.endswith("/api/generate"): + return url + return f"{url}/api/generate" + + @staticmethod + def _normalize_openai_url(base_url: str) -> str: + """Normalize OpenAI-compatible base URL to chat completions endpoint. + + Handles: + - "https://api.openai.com" -> "https://api.openai.com/v1/chat/completions" + - "https://api.openai.com/" -> "https://api.openai.com/v1/chat/completions" + - "https://api.openai.com/v1" -> "https://api.openai.com/v1/chat/completions" + - "https://api.openai.com/v1/" -> "https://api.openai.com/v1/chat/completions" + - "http://localhost:8000/v1/chat/completions" -> unchanged + """ + url = (base_url or "").rstrip("/") + if url.endswith("/chat/completions"): + return url + if url.endswith("/v1"): + return f"{url}/chat/completions" + return f"{url}/v1/chat/completions" + @classmethod def auto(cls) -> "LLMClient": """Auto-detect available LLM provider. @@ -87,7 +113,7 @@ def auto(cls) -> "LLMClient": Priority: 1. Ollama running locally 2. OpenAI API key in environment - 3. EOS_LLM_API_KEY + EOS_LLM_URL in environment + 3. EOS_LLM_URL in environment (EOS_LLM_API_KEY optional) 4. None (returns a client that will fail gracefully) """ # Try Ollama @@ -100,9 +126,9 @@ def auto(cls) -> "LLMClient": return cls(provider="openai", model="gpt-4o-mini", api_key=openai_key) # Try custom - custom_key = os.environ.get("EOS_LLM_API_KEY", "") custom_url = os.environ.get("EOS_LLM_URL", "") - if custom_key and custom_url: + if custom_url: + custom_key = os.environ.get("EOS_LLM_API_KEY", "") model = os.environ.get("EOS_LLM_MODEL", "default") return cls(provider="custom", model=model, api_key=custom_key, base_url=custom_url) @@ -110,14 +136,19 @@ def auto(cls) -> "LLMClient": # No provider available return cls(provider="none", model="none") - @staticmethod - def _check_ollama() -> bool: - """Check if Ollama is running locally.""" + @classmethod + def _check_ollama(cls, base_url: Optional[str] = None) -> bool: + """Check if Ollama is running at target base_url or localhost default.""" + target_url = (base_url or cls.OLLAMA_URL).rstrip("/") + if target_url.endswith("/api/generate"): + tags_url = target_url.rsplit("/", 1)[0] + "/tags" + elif target_url.endswith("/api/tags"): + tags_url = target_url + else: + tags_url = f"{target_url}/api/tags" + try: - req = urllib.request.Request( - f"{LLMClient.OLLAMA_URL}/api/tags", - method="GET", - ) + req = urllib.request.Request(tags_url, method="GET") with urllib.request.urlopen(req, timeout=3) as resp: return resp.status == 200 except (urllib.error.URLError, OSError, TimeoutError): @@ -128,9 +159,11 @@ def is_available(self) -> bool: if self.provider == "none": return False if self.provider == "ollama": - return self._check_ollama() - if self.provider in ("openai", "custom"): + return self._check_ollama(self.base_url) + if self.provider == "openai": return bool(self.api_key and self.base_url) + if self.provider == "custom": + return bool(self.base_url) return False def analyze(self, prompt: str, system: str = "") -> LLMResponse: @@ -163,6 +196,43 @@ def analyze(self, prompt: str, system: str = "") -> LLMResponse: return self._call_ollama(prompt, system) else: return self._call_openai_compat(prompt, system) + except urllib.error.HTTPError as e: + err_details = "" + try: + raw_err = e.read().decode("utf-8") + try: + err_json = json.loads(raw_err) + if isinstance(err_json, dict): + if "error" in err_json: + inner = err_json["error"] + err_details = inner.get("message", str(inner)) if isinstance(inner, dict) else str(inner) + elif "message" in err_json: + err_details = str(err_json["message"]) + except (json.JSONDecodeError, ValueError): + pass + if not err_details and raw_err.strip(): + err_details = raw_err.strip()[:200] + except Exception: + pass + msg = f"HTTP {e.code}: {err_details}" if err_details else f"HTTP {e.code}: {e.reason}" + return LLMResponse( + text="", model=self.model, provider=self.provider, + success=False, error=msg, + ) + except urllib.error.URLError as e: + if isinstance(e.reason, TimeoutError): + error_msg = f"Request timed out after {self.timeout}s" + else: + error_msg = f"Connection error: {e.reason}" + return LLMResponse( + text="", model=self.model, provider=self.provider, + success=False, error=error_msg, + ) + except TimeoutError: + return LLMResponse( + text="", model=self.model, provider=self.provider, + success=False, error=f"Request timed out after {self.timeout}s", + ) except Exception as e: return LLMResponse( text="", model=self.model, provider=self.provider, @@ -171,7 +241,7 @@ def analyze(self, prompt: str, system: str = "") -> LLMResponse: def _call_ollama(self, prompt: str, system: str) -> LLMResponse: """Call Ollama local API.""" - url = f"{self.base_url}/api/generate" + url = self._normalize_ollama_url(self.base_url or self.OLLAMA_URL) payload = { "model": self.model, "prompt": prompt, @@ -199,7 +269,7 @@ def _call_ollama(self, prompt: str, system: str) -> LLMResponse: def _call_openai_compat(self, prompt: str, system: str) -> LLMResponse: """Call OpenAI-compatible chat completions API.""" - url = f"{self.base_url}/v1/chat/completions" + url = self._normalize_openai_url(self.base_url or self.OPENAI_URL) payload = { "model": self.model, "messages": [ @@ -213,19 +283,34 @@ def _call_openai_compat(self, prompt: str, system: str) -> LLMResponse: data = json.dumps(payload).encode("utf-8") headers = { "Content-Type": "application/json", - "Authorization": f"Bearer {self.api_key}", } + if self.api_key: + headers["Authorization"] = f"Bearer {self.api_key}" + req = urllib.request.Request(url, data=data, headers=headers, method="POST") with urllib.request.urlopen(req, timeout=self.timeout) as resp: body = json.loads(resp.read().decode("utf-8")) - choice = body.get("choices", [{}])[0] - message = choice.get("message", {}) - usage = body.get("usage", {}) + if isinstance(body, dict) and "error" in body and not body.get("choices"): + err_data = body["error"] + err_msg = err_data.get("message", str(err_data)) if isinstance(err_data, dict) else str(err_data) + return LLMResponse( + text="", + model=self.model, + provider=self.provider, + success=False, + error=err_msg, + ) + + choices = body.get("choices") or [] + first_choice = choices[0] if choices else {} + message = first_choice.get("message", {}) if isinstance(first_choice, dict) else {} + content = message.get("content", "") if isinstance(message, dict) else "" + usage = body.get("usage", {}) if isinstance(body.get("usage"), dict) else {} return LLMResponse( - text=message.get("content", ""), + text=content or "", model=body.get("model", self.model), provider=self.provider, tokens_used=usage.get("total_tokens", 0), diff --git a/tests/unit/test_eos_ai_llm.py b/tests/unit/test_eos_ai_llm.py new file mode 100644 index 00000000..81bfb7a2 --- /dev/null +++ b/tests/unit/test_eos_ai_llm.py @@ -0,0 +1,515 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2026 EoS Project + +"""Unit tests for ebuild EoS AI LLM integration. + +Tests LLMClient initialization, base URL normalization, provider auto-detection, +availability checks, Ollama/OpenAI dispatch payloads, error resilience, and +integration with EosHardwareAnalyzer. +""" + +from __future__ import annotations + +import io +import json +import urllib.error +import urllib.request +from unittest.mock import MagicMock, patch + +import pytest + +from ebuild.eos_ai.eos_hw_analyzer import EosHardwareAnalyzer, HardwareProfile, PeripheralInfo +from ebuild.eos_ai.llm_integration import LLMClient, LLMResponse + + +class MockHTTPResponse: + """Mock urllib response object.""" + + def __init__(self, data: dict | bytes, status: int = 200): + self.status = status + if isinstance(data, dict): + self._raw = json.dumps(data).encode("utf-8") + else: + self._raw = data + + def read(self) -> bytes: + return self._raw + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + pass + + +@pytest.mark.ebuild +class TestLLMClientInitAndConfig: + """Tests for LLMClient initialization and configuration.""" + + def test_default_init(self): + client = LLMClient() + assert client.provider == "ollama" + assert client.model == "llama3" + assert client.base_url == "http://localhost:11434" + assert client.timeout == 120 + + def test_init_strips_trailing_slash_from_base_url(self): + client_ollama = LLMClient(provider="ollama", base_url="http://localhost:11434/") + assert client_ollama.base_url == "http://localhost:11434" + + client_openai = LLMClient( + provider="openai", + base_url="https://api.openai.com/v1/", + api_key="sk-test", + ) + assert client_openai.base_url == "https://api.openai.com/v1" + + def test_custom_provider_init(self): + client = LLMClient( + provider="custom", + model="mistral-7b", + base_url="http://192.168.1.100:8000/v1/", + api_key="local-key", + timeout=30, + ) + assert client.provider == "custom" + assert client.model == "mistral-7b" + assert client.base_url == "http://192.168.1.100:8000/v1" + assert client.api_key == "local-key" + assert client.timeout == 30 + + def test_get_provider_info(self): + assert LLMClient(provider="none").get_provider_info() == "No LLM provider configured" + + ollama = LLMClient(provider="ollama", model="llama3", base_url="http://localhost:11434") + assert "Ollama (local) — model: llama3" in ollama.get_provider_info() + + openai = LLMClient(provider="openai", model="gpt-4o") + assert openai.get_provider_info() == "OpenAI — model: gpt-4o" + + custom = LLMClient(provider="custom", model="qwen", base_url="http://localhost:8000/v1") + assert "custom — model: qwen — http://localhost:8000/v1" in custom.get_provider_info() + + +@pytest.mark.ebuild +class TestURLNormalization: + """Tests for URL normalization across Ollama and OpenAI-compatible endpoints.""" + + def test_normalize_ollama_urls(self): + expected = "http://localhost:11434/api/generate" + assert LLMClient._normalize_ollama_url("http://localhost:11434") == expected + assert LLMClient._normalize_ollama_url("http://localhost:11434/") == expected + assert LLMClient._normalize_ollama_url("http://localhost:11434/api/generate") == expected + + def test_normalize_openai_urls(self): + expected = "https://api.openai.com/v1/chat/completions" + assert LLMClient._normalize_openai_url("https://api.openai.com") == expected + assert LLMClient._normalize_openai_url("https://api.openai.com/") == expected + assert LLMClient._normalize_openai_url("https://api.openai.com/v1") == expected + assert LLMClient._normalize_openai_url("https://api.openai.com/v1/") == expected + + expected_local = "http://localhost:8000/v1/chat/completions" + assert LLMClient._normalize_openai_url("http://localhost:8000/v1") == expected_local + assert LLMClient._normalize_openai_url("http://localhost:8000/v1/chat/completions") == expected_local + + +@pytest.mark.ebuild +class TestAutoDetectionAndAvailability: + """Tests for provider auto-detection priority and availability checking.""" + + def test_auto_detect_prefers_ollama_if_online(self): + with patch.object(LLMClient, "_check_ollama", return_value=True): + client = LLMClient.auto() + assert client.provider == "ollama" + assert client.model == "llama3" + + def test_auto_detect_falls_back_to_openai_if_ollama_offline(self, monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-key-12345") + with patch.object(LLMClient, "_check_ollama", return_value=False): + client = LLMClient.auto() + assert client.provider == "openai" + assert client.model == "gpt-4o-mini" + assert client.api_key == "sk-test-key-12345" + + def test_auto_detect_falls_back_to_custom_url(self, monkeypatch): + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + monkeypatch.setenv("EOS_LLM_URL", "http://localhost:8000/v1") + monkeypatch.setenv("EOS_LLM_MODEL", "qwen-coder") + with patch.object(LLMClient, "_check_ollama", return_value=False): + client = LLMClient.auto() + assert client.provider == "custom" + assert client.model == "qwen-coder" + assert client.base_url == "http://localhost:8000/v1" + + def test_auto_detect_returns_none_when_no_provider(self, monkeypatch): + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + monkeypatch.delenv("EOS_LLM_URL", raising=False) + with patch.object(LLMClient, "_check_ollama", return_value=False): + client = LLMClient.auto() + assert client.provider == "none" + assert client.model == "none" + assert client.is_available() is False + + def test_is_available_logic(self): + # none provider + none_client = LLMClient(provider="none") + assert none_client.is_available() is False + + # ollama provider calls check_ollama with target base_url + with patch.object(LLMClient, "_check_ollama") as mock_check: + mock_check.return_value = True + ollama = LLMClient(provider="ollama", base_url="http://remote-ollama:11434") + assert ollama.is_available() is True + mock_check.assert_called_once_with("http://remote-ollama:11434") + + # openai requires key and base_url + openai_no_key = LLMClient(provider="openai", api_key="") + assert openai_no_key.is_available() is False + openai_with_key = LLMClient(provider="openai", api_key="sk-test") + assert openai_with_key.is_available() is True + + # custom provider works with base_url even without key (local vLLM/Ollama) + custom_no_key = LLMClient(provider="custom", base_url="http://localhost:8000/v1", api_key="") + assert custom_no_key.is_available() is True + + custom_empty_url = LLMClient(provider="custom", base_url="", api_key="key") + assert custom_empty_url.is_available() is False + + def test_check_ollama_handles_tags_endpoint(self): + # Verify request is made to /api/tags + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse({"models": []}, status=200) + assert LLMClient._check_ollama("http://localhost:11434/") is True + req = mock_urlopen.call_args[0][0] + assert req.full_url == "http://localhost:11434/api/tags" + + # Verify network failure returns False gracefully + with patch("urllib.request.urlopen", side_effect=urllib.error.URLError("Refused")): + assert LLMClient._check_ollama("http://localhost:11434") is False + + +@pytest.mark.ebuild +class TestOllamaDispatch: + """Tests for Ollama API call construction and response parsing.""" + + def test_ollama_successful_call(self): + client = LLMClient(provider="ollama", model="llama3", base_url="http://localhost:11434/") + mock_response_body = { + "model": "llama3", + "response": "Recommendation: Enable UART and CAN.", + "eval_count": 84, + "done": True, + } + + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse(mock_response_body) + + resp = client.analyze(prompt="STM32H7 board", system="Hardware system prompt") + + assert resp.success is True + assert resp.text == "Recommendation: Enable UART and CAN." + assert resp.model == "llama3" + assert resp.provider == "ollama" + assert resp.tokens_used == 84 + assert resp.error == "" + + # Check outbound request + req = mock_urlopen.call_args[0][0] + assert req.full_url == "http://localhost:11434/api/generate" + assert req.headers["Content-type"] == "application/json" + + payload = json.loads(req.data.decode("utf-8")) + assert payload["model"] == "llama3" + assert payload["prompt"] == "STM32H7 board" + assert payload["system"] == "Hardware system prompt" + assert payload["stream"] is False + + +@pytest.mark.ebuild +class TestOpenAICompatDispatch: + """Tests for OpenAI-compatible API call construction and response parsing.""" + + def test_openai_successful_call(self): + client = LLMClient( + provider="openai", + model="gpt-4o", + api_key="sk-test-secret", + base_url="https://api.openai.com/v1/", + ) + mock_response_body = { + "id": "chatcmpl-123", + "model": "gpt-4o", + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": "Recommended EOS config: enable Ethernet and I2C.", + }, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 50, + "completion_tokens": 30, + "total_tokens": 80, + }, + } + + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse(mock_response_body) + + resp = client.analyze(prompt="Analyze NRF52840", system="System prompt") + + assert resp.success is True + assert resp.text == "Recommended EOS config: enable Ethernet and I2C." + assert resp.model == "gpt-4o" + assert resp.provider == "openai" + assert resp.tokens_used == 80 + assert resp.error == "" + + # Check outbound request URL & headers + req = mock_urlopen.call_args[0][0] + assert req.full_url == "https://api.openai.com/v1/chat/completions" + assert req.headers["Authorization"] == "Bearer sk-test-secret" + assert req.headers["Content-type"] == "application/json" + + payload = json.loads(req.data.decode("utf-8")) + assert payload["model"] == "gpt-4o" + assert payload["temperature"] == 0.3 + assert payload["max_tokens"] == 4096 + assert payload["messages"] == [ + {"role": "system", "content": "System prompt"}, + {"role": "user", "content": "Analyze NRF52840"}, + ] + + def test_custom_endpoint_omits_bearer_header_when_api_key_empty(self): + client = LLMClient( + provider="custom", + model="local-model", + base_url="http://localhost:8000/v1", + api_key="", + ) + mock_response = { + "choices": [{"message": {"content": "Local output"}}], + "usage": {"total_tokens": 10}, + } + + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse(mock_response) + resp = client.analyze("Hardware prompt") + + assert resp.success is True + assert resp.text == "Local output" + + req = mock_urlopen.call_args[0][0] + assert req.full_url == "http://localhost:8000/v1/chat/completions" + assert "Authorization" not in req.headers + + +@pytest.mark.ebuild +class TestErrorResilienceAndEdgeCases: + """Tests for edge cases, malformed payloads, and network error handling.""" + + def test_analyze_when_provider_none(self): + client = LLMClient(provider="none") + resp = client.analyze("Test prompt") + assert resp.success is False + assert "No LLM provider available" in resp.error + assert resp.text == "" + + def test_empty_choices_handled_safely_without_index_error(self): + client = LLMClient(provider="openai", api_key="sk-test") + empty_choices_body = {"id": "test", "choices": [], "usage": {}} + + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse(empty_choices_body) + resp = client.analyze("Prompt") + + assert resp.success is True + assert resp.text == "" + + def test_json_error_payload_in_response(self): + client = LLMClient(provider="openai", api_key="sk-test") + error_payload = { + "error": { + "message": "The model `gpt-4o-unknown` does not exist", + "type": "invalid_request_error", + } + } + + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse(error_payload) + resp = client.analyze("Prompt") + + assert resp.success is False + assert "The model `gpt-4o-unknown` does not exist" in resp.error + + def test_http_error_decodes_json_message(self): + client = LLMClient(provider="openai", api_key="sk-test") + err_json = json.dumps({"error": {"message": "Incorrect API key provided"}}).encode("utf-8") + http_err = urllib.error.HTTPError( + url="https://api.openai.com/v1/chat/completions", + code=401, + msg="Unauthorized", + hdrs={}, # type: ignore[arg-type] + fp=io.BytesIO(err_json), + ) + + with patch("urllib.request.urlopen", side_effect=http_err): + resp = client.analyze("Prompt") + + assert resp.success is False + assert "HTTP 401: Incorrect API key provided" in resp.error + + def test_http_error_fallback_when_plain_text(self): + client = LLMClient(provider="openai", api_key="sk-test") + http_err = urllib.error.HTTPError( + url="https://api.openai.com/v1/chat/completions", + code=502, + msg="Bad Gateway", + hdrs={}, # type: ignore[arg-type] + fp=io.BytesIO(b"502 Bad Gateway"), + ) + + with patch("urllib.request.urlopen", side_effect=http_err): + resp = client.analyze("Prompt") + + assert resp.success is False + assert "HTTP 502: 502 Bad Gateway" in resp.error + + def test_network_connection_error(self): + client = LLMClient(provider="ollama", base_url="http://localhost:11434") + url_err = urllib.error.URLError(reason="Connection refused") + + with patch("urllib.request.urlopen", side_effect=url_err): + resp = client.analyze("Prompt") + + assert resp.success is False + assert "Connection error: Connection refused" in resp.error + + def test_timeout_error_direct(self): + client = LLMClient(provider="ollama", timeout=45) + + with patch("urllib.request.urlopen", side_effect=TimeoutError()): + resp = client.analyze("Prompt") + + assert resp.success is False + assert "Request timed out after 45s" in resp.error + + def test_timeout_error_wrapped_in_urlerror(self): + client = LLMClient(provider="openai", api_key="sk-test", timeout=60) + url_err = urllib.error.URLError(reason=TimeoutError()) + + with patch("urllib.request.urlopen", side_effect=url_err): + resp = client.analyze("Prompt") + + assert resp.success is False + assert "Request timed out after 60s" in resp.error + + +@pytest.mark.ebuild +class TestHardwareAnalyzerIntegrationWithLLM: + """Integration test between EosHardwareAnalyzer and LLMClient.""" + + def test_analyzer_enriches_profile_when_llm_suggests_peripherals(self): + analyzer = EosHardwareAnalyzer() + base_profile = HardwareProfile( + mcu="STM32F4", + arch="arm", + core="cortex-m4", + confidence=0.8, + ) + base_profile.peripherals.append(PeripheralInfo(name="UART1", peripheral_type="uart")) + + mock_client = MagicMock(spec=LLMClient) + mock_client.is_available.return_value = True + mock_client.provider = "ollama" + mock_client.analyze.return_value = LLMResponse( + text="Recommend enabling ethernet and spi for this device.", + model="llama3", + provider="ollama", + success=True, + ) + + analyzer._llm_client = mock_client + enriched = analyzer.analyze_with_llm(base_profile) + + assert enriched.has_peripheral("uart") + assert enriched.has_peripheral("ethernet") + assert enriched.has_peripheral("spi") + assert "llm_analyzed:ollama" in enriched.features + assert enriched.confidence == 0.9 + + def test_analyzer_unchanged_when_llm_fails(self): + analyzer = EosHardwareAnalyzer() + base_profile = HardwareProfile( + mcu="STM32F4", + arch="arm", + core="cortex-m4", + confidence=0.8, + ) + + mock_client = MagicMock(spec=LLMClient) + mock_client.is_available.return_value = True + mock_client.analyze.return_value = LLMResponse( + text="", + model="llama3", + provider="ollama", + success=False, + error="Connection refused", + ) + + analyzer._llm_client = mock_client + result = analyzer.analyze_with_llm(base_profile) + + assert result.confidence == 0.8 + assert "llm_analyzed" not in "".join(result.features) + + def test_analyze_uses_default_system_prompt_when_none_provided(self): + client = LLMClient(provider="ollama") + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse({"response": "ok"}) + resp = client.analyze("Test prompt") + assert resp.success is True + + req = mock_urlopen.call_args[0][0] + payload = json.loads(req.data.decode("utf-8")) + assert "embedded systems hardware expert" in payload["system"] + + def test_http_error_with_top_level_message_field(self): + client = LLMClient(provider="openai", api_key="sk-test") + err_json = json.dumps({"message": "Rate limit exceeded"}).encode("utf-8") + http_err = urllib.error.HTTPError( + url="https://api.openai.com/v1/chat/completions", + code=429, + msg="Too Many Requests", + hdrs={}, # type: ignore[arg-type] + fp=io.BytesIO(err_json), + ) + + with patch("urllib.request.urlopen", side_effect=http_err): + resp = client.analyze("Prompt") + assert resp.success is False + assert "HTTP 429: Rate limit exceeded" in resp.error + + def test_check_ollama_endpoint_variants(self): + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse({"models": []}) + + # URL already ending in /api/generate + assert LLMClient._check_ollama("http://localhost:11434/api/generate") is True + req = mock_urlopen.call_args[0][0] + assert req.full_url == "http://localhost:11434/api/tags" + + # URL already ending in /api/tags + assert LLMClient._check_ollama("http://localhost:11434/api/tags") is True + req = mock_urlopen.call_args[0][0] + assert req.full_url == "http://localhost:11434/api/tags" + + def test_analyze_catches_generic_exception(self): + client = LLMClient(provider="ollama") + with patch("urllib.request.urlopen", side_effect=ValueError("Unexpected internal failure")): + resp = client.analyze("Prompt") + assert resp.success is False + assert "Unexpected internal failure" in resp.error From b7b7b152977210d8cf7a039bdcf8c744972425ea Mon Sep 17 00:00:00 2001 From: Sapandeep Singh <110763985+sapandeep31@users.noreply.github.com> Date: Thu, 10 Sep 2026 11:33:49 +0530 Subject: [PATCH 2/3] feat(eos_ai): support standard OPENAI_BASE_URL, OLLAMA_HOST, and JSON validation --- ebuild/eos_ai/llm_integration.py | 36 +++++++++++++++++++++++++------- tests/unit/test_eos_ai_llm.py | 31 +++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 7 deletions(-) diff --git a/ebuild/eos_ai/llm_integration.py b/ebuild/eos_ai/llm_integration.py index 32e07592..9225ca2c 100644 --- a/ebuild/eos_ai/llm_integration.py +++ b/ebuild/eos_ai/llm_integration.py @@ -72,9 +72,13 @@ def __init__( self.timeout = timeout if provider == "ollama": - self.base_url = (base_url or self.OLLAMA_URL).rstrip("/") + default_url = os.environ.get("OLLAMA_HOST") or self.OLLAMA_URL + if default_url and not default_url.startswith(("http://", "https://")): + default_url = f"http://{default_url}" + self.base_url = (base_url or default_url).rstrip("/") elif provider == "openai": - self.base_url = (base_url or self.OPENAI_URL).rstrip("/") + default_url = os.environ.get("OPENAI_BASE_URL") or self.OPENAI_URL + self.base_url = (base_url or default_url).rstrip("/") self.api_key = api_key or os.environ.get("OPENAI_API_KEY", "") elif provider == "custom": self.base_url = (base_url or "").rstrip("/") @@ -111,19 +115,25 @@ def auto(cls) -> "LLMClient": """Auto-detect available LLM provider. Priority: - 1. Ollama running locally - 2. OpenAI API key in environment + 1. Ollama running locally or at OLLAMA_HOST + 2. OpenAI API key in environment (respecting OPENAI_BASE_URL) 3. EOS_LLM_URL in environment (EOS_LLM_API_KEY optional) 4. None (returns a client that will fail gracefully) """ # Try Ollama - if cls._check_ollama(): - return cls(provider="ollama", model="llama3") + ollama_url = os.environ.get("OLLAMA_HOST") or cls.OLLAMA_URL + if ollama_url and not ollama_url.startswith(("http://", "https://")): + ollama_url = f"http://{ollama_url}" + if cls._check_ollama(ollama_url): + model = os.environ.get("OLLAMA_MODEL", "llama3") + return cls(provider="ollama", model=model, base_url=ollama_url) # Try OpenAI openai_key = os.environ.get("OPENAI_API_KEY", "") if openai_key: - return cls(provider="openai", model="gpt-4o-mini", api_key=openai_key) + openai_base = os.environ.get("OPENAI_BASE_URL") + model = os.environ.get("OPENAI_MODEL", "gpt-4o-mini") + return cls(provider="openai", model=model, api_key=openai_key, base_url=openai_base) # Try custom custom_url = os.environ.get("EOS_LLM_URL", "") @@ -259,6 +269,12 @@ def _call_ollama(self, prompt: str, system: str) -> LLMResponse: with urllib.request.urlopen(req, timeout=self.timeout) as resp: body = json.loads(resp.read().decode("utf-8")) + if not isinstance(body, dict): + return LLMResponse( + text="", model=self.model, provider="ollama", + success=False, error="Invalid JSON response: expected object", + ) + return LLMResponse( text=body.get("response", ""), model=body.get("model", self.model), @@ -292,6 +308,12 @@ def _call_openai_compat(self, prompt: str, system: str) -> LLMResponse: with urllib.request.urlopen(req, timeout=self.timeout) as resp: body = json.loads(resp.read().decode("utf-8")) + if not isinstance(body, dict): + return LLMResponse( + text="", model=self.model, provider=self.provider, + success=False, error="Invalid JSON response: expected object", + ) + if isinstance(body, dict) and "error" in body and not body.get("choices"): err_data = body["error"] err_msg = err_data.get("message", str(err_data)) if isinstance(err_data, dict) else str(err_data) diff --git a/tests/unit/test_eos_ai_llm.py b/tests/unit/test_eos_ai_llm.py index 81bfb7a2..0dd07a6b 100644 --- a/tests/unit/test_eos_ai_llm.py +++ b/tests/unit/test_eos_ai_llm.py @@ -513,3 +513,34 @@ def test_analyze_catches_generic_exception(self): resp = client.analyze("Prompt") assert resp.success is False assert "Unexpected internal failure" in resp.error + + def test_openai_respects_openai_base_url_env_var(self, monkeypatch): + monkeypatch.setenv("OPENAI_BASE_URL", "https://api.groq.com/openai/v1/") + monkeypatch.setenv("OPENAI_API_KEY", "gsk-test") + client = LLMClient(provider="openai") + assert client.base_url == "https://api.groq.com/openai/v1" + + with patch.object(LLMClient, "_check_ollama", return_value=False): + auto_client = LLMClient.auto() + assert auto_client.provider == "openai" + assert auto_client.base_url == "https://api.groq.com/openai/v1" + + def test_ollama_respects_ollama_host_env_var(self, monkeypatch): + monkeypatch.setenv("OLLAMA_HOST", "192.168.1.50:11434/") + client = LLMClient(provider="ollama") + assert client.base_url == "http://192.168.1.50:11434" + + def test_dispatch_non_dict_json_response(self): + client = LLMClient(provider="ollama") + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse(b'["unexpected", "list"]') + resp = client.analyze("Prompt") + assert resp.success is False + assert "Invalid JSON response: expected object" in resp.error + + openai_client = LLMClient(provider="openai", api_key="sk-test") + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse(b'["unexpected", "list"]') + resp = openai_client.analyze("Prompt") + assert resp.success is False + assert "Invalid JSON response: expected object" in resp.error From 047c11180ae22bbf2726664b8f8d95b7f3fa2284 Mon Sep 17 00:00:00 2001 From: Sapandeep Singh <110763985+sapandeep31@users.noreply.github.com> Date: Thu, 10 Sep 2026 14:20:03 +0530 Subject: [PATCH 3/3] fix(eos_ai): address review findings on empty choices, env isolation, and scheme normalization --- docs/ai-input-formats.md | 21 +++++++++++--- ebuild/eos_ai/llm_integration.py | 49 +++++++++++++++++++++++--------- tests/unit/test_eos_ai_llm.py | 44 +++++++++++++++++++++++++++- 3 files changed, 95 insertions(+), 19 deletions(-) diff --git a/docs/ai-input-formats.md b/docs/ai-input-formats.md index bc96be72..028f7d04 100644 --- a/docs/ai-input-formats.md +++ b/docs/ai-input-formats.md @@ -100,12 +100,25 @@ ebuild analyze design.kicad_sch --llm-provider ollama --llm-model llama3 ebuild analyze design.kicad_sch --llm-provider openai --llm-model gpt-4o ``` -**LLM Provider Auto-Detection:** -1. **Ollama** (local, free) — checks `http://localhost:11434` -2. **OpenAI** — uses `OPENAI_API_KEY` env var -3. **Custom** — uses `EOS_LLM_API_KEY` + `EOS_LLM_URL` + `EOS_LLM_MODEL` env vars +**LLM Provider Auto-Detection Priority:** +1. **Ollama** (local, free) — checks `OLLAMA_HOST` (defaults to `http://localhost:11434`), using `OLLAMA_MODEL` (defaults to `llama3`) +2. **OpenAI** — detected via `OPENAI_API_KEY` (supports custom proxies via `OPENAI_BASE_URL`, model via `OPENAI_MODEL`) +3. **Custom** — detected via `EOS_LLM_URL` (supports keyless local inference like vLLM/llama.cpp; optional `EOS_LLM_API_KEY`, model via `EOS_LLM_MODEL`) 4. **None** — works without LLM (rule engine only) +**Environment Variables Reference:** + +| Variable | Provider | Default | Description | +|----------|----------|---------|-------------| +| `OLLAMA_HOST` | Ollama | `http://localhost:11434` | Ollama server URL or `host:port` | +| `OLLAMA_MODEL` | Ollama | `llama3` | Model name for Ollama generation | +| `OPENAI_API_KEY` | OpenAI | *(none)* | API key for OpenAI-compatible endpoint | +| `OPENAI_BASE_URL` | OpenAI | `https://api.openai.com` | Base URL for OpenAI API or reverse proxy | +| `OPENAI_MODEL` | OpenAI | `gpt-4o-mini` | Model name for OpenAI completions | +| `EOS_LLM_URL` | Custom | *(none)* | Base URL for custom OpenAI-compatible server (vLLM, etc.) | +| `EOS_LLM_API_KEY` | Custom | *(optional)* | API key for custom endpoint if authentication is required | +| `EOS_LLM_MODEL` | Custom | `default` | Model name for custom endpoint | + --- ## NOT Supported (Today) diff --git a/ebuild/eos_ai/llm_integration.py b/ebuild/eos_ai/llm_integration.py index 9225ca2c..7cb0f9a2 100644 --- a/ebuild/eos_ai/llm_integration.py +++ b/ebuild/eos_ai/llm_integration.py @@ -30,7 +30,7 @@ import urllib.error import urllib.request from dataclasses import dataclass -from typing import Optional +from typing import Any, Optional @dataclass @@ -73,9 +73,8 @@ def __init__( if provider == "ollama": default_url = os.environ.get("OLLAMA_HOST") or self.OLLAMA_URL - if default_url and not default_url.startswith(("http://", "https://")): - default_url = f"http://{default_url}" - self.base_url = (base_url or default_url).rstrip("/") + raw_url = base_url or default_url + self.base_url = self._ensure_scheme(raw_url).rstrip("/") elif provider == "openai": default_url = os.environ.get("OPENAI_BASE_URL") or self.OPENAI_URL self.base_url = (base_url or default_url).rstrip("/") @@ -84,6 +83,22 @@ def __init__( self.base_url = (base_url or "").rstrip("/") self.api_key = api_key or os.environ.get("EOS_LLM_API_KEY", "") + @staticmethod + def _ensure_scheme(url: str) -> str: + """Ensure a URL string has an http:// or https:// scheme prefix.""" + if not url: + return "" + if url.startswith(("http://", "https://")): + return url + return f"http://{url}" + + @staticmethod + def _error_message(payload: Any) -> str: + """Extract a human-readable error message from an API error payload.""" + if isinstance(payload, dict): + return str(payload.get("message", payload)) + return str(payload) + @staticmethod def _normalize_ollama_url(base_url: str) -> str: """Normalize Ollama base URL to generation endpoint.""" @@ -121,9 +136,8 @@ def auto(cls) -> "LLMClient": 4. None (returns a client that will fail gracefully) """ # Try Ollama - ollama_url = os.environ.get("OLLAMA_HOST") or cls.OLLAMA_URL - if ollama_url and not ollama_url.startswith(("http://", "https://")): - ollama_url = f"http://{ollama_url}" + ollama_host = os.environ.get("OLLAMA_HOST") or cls.OLLAMA_URL + ollama_url = cls._ensure_scheme(ollama_host).rstrip("/") if cls._check_ollama(ollama_url): model = os.environ.get("OLLAMA_MODEL", "llama3") return cls(provider="ollama", model=model, base_url=ollama_url) @@ -214,15 +228,14 @@ def analyze(self, prompt: str, system: str = "") -> LLMResponse: err_json = json.loads(raw_err) if isinstance(err_json, dict): if "error" in err_json: - inner = err_json["error"] - err_details = inner.get("message", str(inner)) if isinstance(inner, dict) else str(inner) + err_details = self._error_message(err_json["error"]) elif "message" in err_json: err_details = str(err_json["message"]) except (json.JSONDecodeError, ValueError): pass if not err_details and raw_err.strip(): err_details = raw_err.strip()[:200] - except Exception: + except (OSError, UnicodeDecodeError, AttributeError): pass msg = f"HTTP {e.code}: {err_details}" if err_details else f"HTTP {e.code}: {e.reason}" return LLMResponse( @@ -314,9 +327,8 @@ def _call_openai_compat(self, prompt: str, system: str) -> LLMResponse: success=False, error="Invalid JSON response: expected object", ) - if isinstance(body, dict) and "error" in body and not body.get("choices"): - err_data = body["error"] - err_msg = err_data.get("message", str(err_data)) if isinstance(err_data, dict) else str(err_data) + if "error" in body and not body.get("choices"): + err_msg = self._error_message(body["error"]) return LLMResponse( text="", model=self.model, @@ -331,8 +343,17 @@ def _call_openai_compat(self, prompt: str, system: str) -> LLMResponse: content = message.get("content", "") if isinstance(message, dict) else "" usage = body.get("usage", {}) if isinstance(body.get("usage"), dict) else {} + if not choices or not content: + return LLMResponse( + text="", + model=body.get("model", self.model), + provider=self.provider, + success=False, + error="Upstream returned no completion choices", + ) + return LLMResponse( - text=content or "", + text=content, model=body.get("model", self.model), provider=self.provider, tokens_used=usage.get("total_tokens", 0), diff --git a/tests/unit/test_eos_ai_llm.py b/tests/unit/test_eos_ai_llm.py index 0dd07a6b..b1ee724c 100644 --- a/tests/unit/test_eos_ai_llm.py +++ b/tests/unit/test_eos_ai_llm.py @@ -42,6 +42,23 @@ def __exit__(self, exc_type, exc_val, exc_tb): pass +@pytest.fixture(autouse=True) +def clean_ambient_llm_env(monkeypatch): + """Ensure tests run in an isolated environment without ambient LLM env vars.""" + env_vars = [ + "OLLAMA_HOST", + "OLLAMA_MODEL", + "OPENAI_BASE_URL", + "OPENAI_MODEL", + "OPENAI_API_KEY", + "EOS_LLM_URL", + "EOS_LLM_API_KEY", + "EOS_LLM_MODEL", + ] + for var in env_vars: + monkeypatch.delenv(var, raising=False) + + @pytest.mark.ebuild class TestLLMClientInitAndConfig: """Tests for LLMClient initialization and configuration.""" @@ -64,6 +81,10 @@ def test_init_strips_trailing_slash_from_base_url(self): ) assert client_openai.base_url == "https://api.openai.com/v1" + def test_init_normalizes_scheme_for_explicit_base_url_without_scheme(self): + client = LLMClient(provider="ollama", base_url="192.168.1.50:11434") + assert client.base_url == "http://192.168.1.50:11434" + def test_custom_provider_init(self): client = LLMClient( provider="custom", @@ -326,7 +347,8 @@ def test_empty_choices_handled_safely_without_index_error(self): mock_urlopen.return_value = MockHTTPResponse(empty_choices_body) resp = client.analyze("Prompt") - assert resp.success is True + assert resp.success is False + assert resp.error == "Upstream returned no completion choices" assert resp.text == "" def test_json_error_payload_in_response(self): @@ -466,6 +488,26 @@ def test_analyzer_unchanged_when_llm_fails(self): assert result.confidence == 0.8 assert "llm_analyzed" not in "".join(result.features) + def test_analyzer_unchanged_when_llm_returns_empty_choices(self): + analyzer = EosHardwareAnalyzer() + base_profile = HardwareProfile( + mcu="STM32F4", + arch="arm", + core="cortex-m4", + confidence=0.8, + ) + + real_client = LLMClient(provider="openai", api_key="sk-test") + empty_choices_body = {"id": "test", "choices": [], "usage": {}} + + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.return_value = MockHTTPResponse(empty_choices_body) + analyzer._llm_client = real_client + result = analyzer.analyze_with_llm(base_profile) + + assert result.confidence == 0.8 + assert not any("llm_analyzed" in f for f in result.features) + def test_analyze_uses_default_system_prompt_when_none_provided(self): client = LLMClient(provider="ollama") with patch("urllib.request.urlopen") as mock_urlopen: