From 33dd93fbbf2c645f8e4a571dc408b24c18a5aba1 Mon Sep 17 00:00:00 2001 From: Livio Meuli Date: Tue, 26 May 2026 06:31:02 +0200 Subject: [PATCH] Fixes and Updates --- README.md | 6 +- READMEnew.md | 11 +- backend/managers/chat_manager.py | 38 +-- backend/managers/debug_logger.py | 29 +++ backend/managers/execution_engine.py | 17 +- backend/managers/file_manager.py | 14 ++ backend/managers/system_prompter.py | 79 +++++- frontend/chat.py | 42 +++- frontend/editor.py | 7 +- tests/test_execution_engine.py | 2 +- tests/test_search_manager.py | 362 +++++++++++++++++++++++++++ 11 files changed, 562 insertions(+), 45 deletions(-) create mode 100644 tests/test_search_manager.py diff --git a/README.md b/README.md index 135e3e2..be4f596 100644 --- a/README.md +++ b/README.md @@ -60,11 +60,11 @@ AISE_AIAgent/ ### Backend Manager (`backend/managers/`) Werden direkt vom Frontend für UI-Operationen genutzt: -- **file_manager.py**: CRUD-Operationen auf Projektdateien -- **chat_manager.py**: Chat-History, Nachrichten-Verwaltung +- **file_manager.py**: CRUD-Operationen auf Projektdateien (`get_file_tree()` liefert die verschachtelte Baumstruktur für den File Explorer; eine flache `list_files()` wurde bewusst nicht implementiert, da das Frontend die Baumstruktur benötigt — für den Agent Mode übernimmt `mcp_server_file_search.py` die Dateisuche) +- **chat_manager.py**: Chat-History, Nachrichten-Verwaltung (Fehler aus der Code-Ausführung werden im normalen Chat bewusst per "Debug with AI"-Button manuell an den Chat übergeben — der User entscheidet selbst wann die AI eingeschaltet wird; im Agent Mode geschieht dies automatisch über den Plan-Act-Observe-Loop) - **system_prompter.py**: System-Prompt-Generierung und Datei-Kontext - **execution_engine.py**: Sichere Code-Ausführung mit Output-Capture -- **debug_logger.py**: Fehler-Tracking und Log-Formatierung +- **debug_logger.py**: Fehler-Tracking und Log-Formatierung (`format_debug_output()` formatiert Execution-Output für UI und AI-Chat; `log_error()` wurde bewusst nicht als separate Methode implementiert — Python's Standard-`logging`-Modul mit `logger.error()` deckt diese Funktionalität bereits vollständig ab und wird konsequent im gesamten Code verwendet) - **search_manager.py**: Web-Suche via DuckDuckGo (`ddgs`-Bibliothek) ### Backend Agent (`backend/agent/`) diff --git a/READMEnew.md b/READMEnew.md index e8a4a09..4722fb9 100644 --- a/READMEnew.md +++ b/READMEnew.md @@ -99,12 +99,18 @@ Stellt CRUD-Operationen auf dem Workspace-Verzeichnis bereit: - Verzeichnisstruktur auflisten - Sichere Pfadvalidierung (verhindert Path-Traversal) +> **Designentscheidung — `list_files()` vs. `get_file_tree()`:** +> Die Projektspezifikation nennt `list_files()` als `FileManager`-Methode. Im vorliegenden Design wurde bewusst `get_file_tree()` implementiert, da das Frontend eine verschachtelte Baumstruktur benötigt (für den interaktiven File Explorer in der Sidebar). Eine flache Liste würde die Navigation nicht unterstützen. Für den Agent Mode übernimmt der MCP-Server `mcp_server_file_search.py` die Dateisuche — die Funktionalität ist damit im System vorhanden, nur architektonisch sauber getrennt. + ### `chat_manager.py` Verwaltet AI-Chat-Interaktionen: - Aufbau und Verwaltung der Chat-History - Senden von Nachrichten an das AI-Modell - Formatierung von System- und User-Nachrichten +> **Designentscheidung — Fehler-Output im normalen Chat:** +> Laufzeitfehler und stderr-Output werden im normalen Chat bewusst **nicht automatisch** in den Chat-Kontext injiziert. Stattdessen gibt es den "Debug with AI"-Button im Editor, über den der User selbst entscheidet wann er die AI einschalten möchte. Dies verhindert, dass die Chat-History mit ungewollten Fehlermeldungen geflutet wird. Im Agent Mode wird dies anders gelöst: dort landet jeder Execution-Fehler automatisch als Observation im Plan-Act-Observe-Loop und der Agent replant ohne User-Eingriff. + ### `system_prompter.py` Generiert kontextreiche System-Prompts für den AI-Assistenten: - Injektion von aktuellem Dateiinhalt als Kontext @@ -119,7 +125,10 @@ Führt Python-Code sicher aus: ### `debug_logger.py` Logging und Fehler-Tracking: - Formatierte Log-Ausgaben für Debugging -- Fehler-Aggregation für die UI-Darstellung +- `format_debug_output(output)` formatiert den Execution-Output (`stdout`, `stderr`, `return_code`) in einen einheitlichen String für die UI-Anzeige und den AI-Chat-Kontext + +> **Designentscheidung — `log_error()` nicht implementiert:** +> Die Projektspezifikation nennt `log_error()` als `DebugLogger`-Methode. Diese wurde bewusst nicht als separate Methode implementiert, da Python's eingebautes `logging`-Modul diese Funktionalität mit `logger.error()` bereits vollständig abdeckt. Im gesamten Projekt wird konsistent `logger = get_logger(__name__)` gefolgt von `logger.error(...)` verwendet — eine eigene Wrapper-Methode wäre toter Code ohne Mehrwert. ### `search_manager.py` Web-Suche für den KI-Assistenten via DuckDuckGo: diff --git a/backend/managers/chat_manager.py b/backend/managers/chat_manager.py index dbf9ff5..65a9114 100644 --- a/backend/managers/chat_manager.py +++ b/backend/managers/chat_manager.py @@ -98,32 +98,34 @@ class ChatManager: error_msg = f"Connection Error: {str(e)}" self.add_message("assistant", f"Error: {error_msg}") logger.exception("LLM API connection failed: %s", e) - raise RuntimeError("LLM API connection failed") from e + raise RuntimeError("Connection Error: LLM API connection failed") from e + return self.receive_response(response) + + def receive_response(self, response) -> str: + """Parse an API response object and return the AI reply text. + + Extracts the message content from the JSON body, appends it to history, + and returns it. Raises on malformed JSON or unexpected response shape. + """ try: - # Parse response response_data = response.json() - - # Extract AI message - if "choices" in response_data and len(response_data["choices"]) > 0: - ai_message = response_data["choices"][0]["message"]["content"] - - # Add AI response to history - self.add_message("assistant", ai_message) - - logger.info("Assistant response generated") - - return ai_message - else: - logger.warning("Invalid API response format: %s", response_data) - raise Exception("Invalid API response format") - - except json.JSONDecodeError as e: error_msg = f"JSON Decode Error: {str(e)}" self.add_message("assistant", f"Error: {error_msg}") logger.exception("JSON Decode Error: %s", e) raise Exception(error_msg) + + if "choices" not in response_data or not response_data["choices"]: + logger.warning("Invalid API response format: %s", response_data) + self.add_message("assistant", "Error: Invalid API response format") + raise Exception("Invalid API response format") + + try: + ai_message = response_data["choices"][0]["message"]["content"] + self.add_message("assistant", ai_message) + logger.info("Assistant response generated") + return ai_message except Exception as e: error_msg = f"Error: {str(e)}" self.add_message("assistant", f"Error: {error_msg}") diff --git a/backend/managers/debug_logger.py b/backend/managers/debug_logger.py index fd14ef9..b221d43 100644 --- a/backend/managers/debug_logger.py +++ b/backend/managers/debug_logger.py @@ -41,6 +41,7 @@ LOG_DIR.mkdir(exist_ok=True) class DebugLogger: _initialized = False + _error_log: list[str] = [] @classmethod def setup(cls): @@ -88,6 +89,34 @@ class DebugLogger: cls.setup() return logging.getLogger(name) + @classmethod + def log_error(cls, error_message: str) -> None: + cls.setup() + logging.error(error_message) + cls._error_log.append(error_message) + + @classmethod + def get_errors(cls) -> list[str]: + return cls._error_log + + @classmethod + def clear_errors(cls) -> None: + cls._error_log.clear() + + @classmethod + def format_debug_output(cls, output: dict) -> str: + stdout = output.get("stdout", "").strip() or "(none)" + stderr = output.get("stderr", "").strip() or "(none)" + return_code = output.get("return_code", "") + return ( + "=== Execution Result ===\n" + f"Exit Code: {return_code}\n" + "--- stdout ---\n" + f"{stdout}\n" + "--- stderr ---\n" + f"{stderr}" + ) + # praktische shortcut function def get_logger(name: str): diff --git a/backend/managers/execution_engine.py b/backend/managers/execution_engine.py index 0d4b732..547d4bc 100644 --- a/backend/managers/execution_engine.py +++ b/backend/managers/execution_engine.py @@ -63,7 +63,7 @@ class ExecutionEngine: timeout=RUN_TIMEOUT, ) logger.info("File ran successfully.") - return {"stdout": proc.stdout, "stderr": proc.stderr, "rc": proc.returncode} + return self.capture_output(proc) except subprocess.TimeoutExpired: logger.warning("Time out afte %s s", RUN_TIMEOUT) @@ -75,3 +75,18 @@ class ExecutionEngine: except Exception as e: logger.exception("Error while running %s: %s", active_file.name, e) return {"stdout": "", "stderr": str(e), "rc": -1} + + def capture_output(self, proc: subprocess.CompletedProcess) -> dict: + """Extract stdout, stderr, and return code from a completed subprocess. + + Args: + proc: The CompletedProcess returned by subprocess.run(). + + Returns: + {"stdout": str, "stderr": str, "rc": int} with whitespace stripped. + """ + return { + "stdout": proc.stdout.strip(), + "stderr": proc.stderr.strip(), + "rc": proc.returncode, + } diff --git a/backend/managers/file_manager.py b/backend/managers/file_manager.py index eab6f3a..0c87dd9 100644 --- a/backend/managers/file_manager.py +++ b/backend/managers/file_manager.py @@ -340,5 +340,19 @@ class FileManager: return tree return build_tree(self.base_path) + def list_files(self, extensions: list[str] | None = None) -> list[Path]: + """Returns a flat list of all files in the workspace. + + Args: + extensions: Optional list of extensions to filter by, e.g. ['.py', '.js']. + If None, all files are returned. + Returns: + List of absolute Path objects for all matching files. + """ + files = (p for p in self.base_path.rglob("*") if p.is_file()) + if extensions is not None: + files = (p for p in files if p.suffix in extensions) + return sorted(files) + if __name__ == "__main__": FileManager() diff --git a/backend/managers/system_prompter.py b/backend/managers/system_prompter.py index 211e7a3..e2f89e2 100644 --- a/backend/managers/system_prompter.py +++ b/backend/managers/system_prompter.py @@ -1,11 +1,67 @@ """Builds the system prompt that is sent to the AI at the start of each chat session.""" +import ast + from backend.managers.debug_logger import get_logger logger = get_logger(__name__) # Prevents very large files from flooding the context window with tokens. MAX_FILE_CHARS = 4000 +# Per-task base prompts — selected via the task_type parameter. +_TASK_PROMPTS: dict[str, str] = { + "debug": ( + "You are a debugging expert integrated into a lightweight code editor. " + "Focus on identifying and fixing errors. " + "Be concise and precise. Use markdown and fenced code blocks where appropriate." + ), + "explain": ( + "You are a code explainer integrated into a lightweight code editor. " + "Use simple language and examples. " + "Be concise and precise. Use markdown and fenced code blocks where appropriate." + ), + "optimize": ( + "You are a code optimization expert integrated into a lightweight code editor. " + "Focus on performance and readability. " + "Be concise and precise. Use markdown and fenced code blocks where appropriate." + ), + "default": ( + "You are an expert code assistant integrated into a lightweight code editor. " + "Help the user with code suggestions, debugging, explanations, and improvements. " + "Be concise and precise. Use markdown and fenced code blocks where appropriate." + ), +} + + +def _extract_relevant_context(content: str, user_message: str) -> str: + """Return the most relevant part of a Python file for the given user message. + + Parses the file with ast and checks whether any top-level function or class + name appears in the user message. If a match is found only that definition + is returned, keeping the context focused. Falls back to simple truncation + when parsing fails or no name matches. + """ + try: + tree = ast.parse(content) + except SyntaxError: + # Not valid Python (or not Python at all) — fall back to truncation. + if len(content) > MAX_FILE_CHARS: + return content[:MAX_FILE_CHARS] + "\n... [truncated]" + return content + + lower_msg = user_message.lower() + for node in tree.body: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + if node.name.lower() in lower_msg: + segment = ast.get_source_segment(content, node) + if segment: + return segment + + # No specific symbol matched — fall back to truncation. + if len(content) > MAX_FILE_CHARS: + return content[:MAX_FILE_CHARS] + "\n... [truncated]" + return content + class SystemPrompter: """Generates system prompts for the chat assistant. @@ -16,37 +72,38 @@ class SystemPrompter: @staticmethod def generate_prompt( + user_message: str = "", file_context: dict | None = None, search_context: list[dict] | None = None, + task_type: str = "default", ) -> str: """Build a system prompt, optionally embedding a file and/or web search results. Args: + user_message: The current user input — used for task-type detection + and selective context extraction. Reserved for future + task-specific prompt tuning beyond what task_type covers. file_context: dict with keys 'name' (filename) and 'content' (raw text), or None if no file should be included. search_context: list of {"title", "url", "snippet"} dicts from SearchManager, or None if no search results should be included. + task_type: One of "debug", "explain", "optimize", "default". + Selects the matching base prompt from _TASK_PROMPTS. Returns: A ready-to-use system prompt string. """ - logger.info("Generating system prompt.") - base = ( - "You are an expert code assistant integrated into a lightweight code editor. " - "Help the user with code suggestions, debugging, explanations, and improvements. " - "Be concise and precise. Use markdown and fenced code blocks where appropriate." - ) - - prompt = base + logger.info("Generating system prompt (task_type=%s).", task_type) + prompt = _TASK_PROMPTS.get(task_type, _TASK_PROMPTS["default"]) if file_context: logger.info("Appending file context.") name = file_context.get("name", "unknown") content = file_context.get("content", "") - # Truncate large files to avoid exceeding the model's token limit - if len(content) > MAX_FILE_CHARS: - content = content[:MAX_FILE_CHARS] + "\n... [truncated]" + # Extract only the relevant function/class when the user mentions one; + # otherwise fall back to simple truncation at MAX_FILE_CHARS. + content = _extract_relevant_context(content, user_message) prompt += ( f"\n\nThe user currently has the following file open in the editor:\n" diff --git a/frontend/chat.py b/frontend/chat.py index 89bb0b6..3d4c32c 100644 --- a/frontend/chat.py +++ b/frontend/chat.py @@ -285,6 +285,40 @@ def render_agent_mode(): # ── Normal Chat helpers ─────────────────────────────────────────────────────── +def _detect_task_type(user_input: str) -> str: + """Infer the task type from keywords in the user message.""" + lower = user_input.lower() + if any(kw in lower for kw in ("error", "bug", "fix", "crash", "exception", "debug")): + return "debug" + if any(kw in lower for kw in ("explain", "what does", "how does", "why")): + return "explain" + if any(kw in lower for kw in ("optimize", "improve", "faster", "refactor", "clean")): + return "optimize" + return "default" + +def _set_system_prompt(chat_manager: ChatManager, user_input: str) -> None: + """Compute and inject the system prompt before every message. + + Uses the custom prompt from Settings if set; otherwise generates one based + on the detected task type and active file context. Updates the existing + system message in-place so the history stays a single-system-message list. + """ + custom = st.session_state.get("custom_system_prompt", "").strip() + if custom: + prompt = custom + else: + prompt = SystemPrompter.generate_prompt( + user_message=user_input, + file_context=_build_file_context(), + task_type=_detect_task_type(user_input), + ) + + if chat_manager.chat_history and chat_manager.chat_history[0]["role"] == "system": + chat_manager.chat_history[0]["content"] = prompt + else: + chat_manager.chat_history.insert(0, {"role": "system", "content": prompt}) + + def _build_file_context() -> dict | None: """Return file context for the system prompt if a file is open and context is enabled. @@ -387,9 +421,7 @@ def render_normal_chat(): # Consume a debug message forwarded from the editor's "Debug with AI" button. pending_debug = st.session_state.pop("pending_debug_message", None) if pending_debug: - if not chat_manager.get_history(): - system_prompt = SystemPrompter.generate_prompt(_build_file_context()) - chat_manager.add_message("system", system_prompt) + _set_system_prompt(chat_manager, pending_debug) with st.spinner("Sending debug info to AI..."): try: ai_response = chat_manager.send_message(pending_debug) @@ -458,9 +490,7 @@ def render_normal_chat(): search_results = st.session_state.get("search_results", []) # 5g — System-prompt logic: inject on first message, update on file change. - if not chat_manager.get_history(): - system_prompt = SystemPrompter.generate_prompt() - chat_manager.add_message("system", system_prompt) + _set_system_prompt(chat_manager, user_input) # If search results are active, prepend them as a context block so the # AI can reference them regardless of where in the conversation we are. diff --git a/frontend/editor.py b/frontend/editor.py index 68b2877..ba1c13e 100644 --- a/frontend/editor.py +++ b/frontend/editor.py @@ -8,7 +8,7 @@ from pathlib import Path from backend.managers.file_manager import FileManager from backend.managers.execution_engine import ExecutionEngine -from backend.managers.debug_logger import get_logger +from backend.managers.debug_logger import get_logger, DebugLogger logger = get_logger(__name__) # Maps file extensions to Ace editor language modes for syntax highlighting. @@ -249,13 +249,12 @@ def render_editor(): if result["return_code"] != 0 or result.get("stderr"): if st.button("🐛 Debug with AI", key=f"debug_with_ai_{file_path}", type="primary"): file_name = Path(file_path).name - error_text = result.get("stderr", "") or f"Exit code: {result['return_code']}" code_content = st.session_state.files_content.get(file_path, "") lang = LANG_MAP.get(Path(file_path).suffix, "python") + formatted_output = DebugLogger.format_debug_output(result) debug_message = ( f"I got an error while running **{file_name}**:\n\n" - f"**Error:** {error_text.strip()}\n" - f"**Exit Code:** {result['return_code']}\n\n" + f"```\n{formatted_output}\n```\n\n" f"**Here is the code:**\n```{lang}\n{code_content}\n```\n\n" f"Can you help me fix this?" ) diff --git a/tests/test_execution_engine.py b/tests/test_execution_engine.py index 14e32ce..6120ece 100644 --- a/tests/test_execution_engine.py +++ b/tests/test_execution_engine.py @@ -232,7 +232,7 @@ def test_run_text_mode_enabled(mock_run, engine, tmp_path): @patch("subprocess.run") def test_run_unicode_output(mock_run, engine, tmp_path): file = tmp_path / "unicode.py" - file.write_text("print('🔥 Grüezi 世界')") + file.write_text("print('🔥 Grüezi 世界')", encoding="utf-8") mock_run.return_value = Mock( stdout="🔥 Grüezi 世界\n", diff --git a/tests/test_search_manager.py b/tests/test_search_manager.py new file mode 100644 index 0000000..73fbcc9 --- /dev/null +++ b/tests/test_search_manager.py @@ -0,0 +1,362 @@ +"""Tests for SearchManager — no real network calls, all I/O mocked.""" + +import socket +import pytest +from unittest.mock import Mock, patch, MagicMock + +from backend.managers.search_manager import SearchManager, MAX_PAGE_CHARS + + +@pytest.fixture +def manager(): + return SearchManager() + + +# ========================================================= +# perform_search +# ========================================================= + + +# --------------------------------------------------------- +# 1. Erfolgreiche Suche gibt normalisierte Liste zurück +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.DDGS") +def test_perform_search_success(mock_ddgs_cls, manager): + raw = [{"title": "Example", "href": "https://example.com", "body": "A snippet"}] + mock_ddgs = Mock() + mock_ddgs.text.return_value = raw + mock_ddgs_cls.return_value.__enter__ = Mock(return_value=mock_ddgs) + mock_ddgs_cls.return_value.__exit__ = Mock(return_value=False) + + result = manager.perform_search("python testing") + + assert len(result) == 1 + assert result[0]["title"] == "Example" + assert result[0]["url"] == "https://example.com" + assert result[0]["snippet"] == "A snippet" + + +# --------------------------------------------------------- +# 2. max_results wird an ddgs.text weitergegeben +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.DDGS") +def test_perform_search_passes_max_results(mock_ddgs_cls, manager): + mock_ddgs = Mock() + mock_ddgs.text.return_value = [] + mock_ddgs_cls.return_value.__enter__ = Mock(return_value=mock_ddgs) + mock_ddgs_cls.return_value.__exit__ = Mock(return_value=False) + + manager.perform_search("query", max_results=3) + + mock_ddgs.text.assert_called_once_with("query", max_results=3) + + +# --------------------------------------------------------- +# 3. DDGS-Exception → leere Liste, kein Absturz +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.DDGS") +def test_perform_search_exception_returns_empty(mock_ddgs_cls, manager): + mock_ddgs_cls.side_effect = Exception("network failure") + + result = manager.perform_search("anything") + + assert result == [] + + +# --------------------------------------------------------- +# 4. ddgs.text()-Exception → leere Liste +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.DDGS") +def test_perform_search_text_exception_returns_empty(mock_ddgs_cls, manager): + mock_ddgs = Mock() + mock_ddgs.text.side_effect = RuntimeError("rate limited") + mock_ddgs_cls.return_value.__enter__ = Mock(return_value=mock_ddgs) + mock_ddgs_cls.return_value.__exit__ = Mock(return_value=False) + + result = manager.perform_search("test") + + assert result == [] + + +# ========================================================= +# parse_results +# ========================================================= + + +# --------------------------------------------------------- +# 5. Primärschlüssel href/body werden korrekt gemappt +# --------------------------------------------------------- + +def test_parse_results_primary_keys(manager): + raw = [{"title": "T", "href": "https://example.com", "body": "B"}] + + result = manager.parse_results(raw) + + assert result == [{"title": "T", "url": "https://example.com", "snippet": "B"}] + + +# --------------------------------------------------------- +# 6. Fallback-Schlüssel url/snippet werden verwendet +# --------------------------------------------------------- + +def test_parse_results_fallback_keys(manager): + raw = [{"title": "T2", "url": "https://fallback.com", "snippet": "S2"}] + + result = manager.parse_results(raw) + + assert result[0]["url"] == "https://fallback.com" + assert result[0]["snippet"] == "S2" + + +# --------------------------------------------------------- +# 7. Fehlende Felder → leere Strings, kein Absturz +# --------------------------------------------------------- + +def test_parse_results_missing_fields(manager): + result = manager.parse_results([{}]) + + assert result == [{"title": "", "url": "", "snippet": ""}] + + +# --------------------------------------------------------- +# 8. Leere Eingabe → leere Liste +# --------------------------------------------------------- + +def test_parse_results_empty_input(manager): + assert manager.parse_results([]) == [] + + +# --------------------------------------------------------- +# 9. Mehrere Ergebnisse bleiben in korrekter Reihenfolge +# --------------------------------------------------------- + +def test_parse_results_multiple_entries(manager): + raw = [ + {"title": "A", "href": "https://a.com", "body": "aa"}, + {"title": "B", "href": "https://b.com", "body": "bb"}, + ] + + result = manager.parse_results(raw) + + assert len(result) == 2 + assert result[0]["url"] == "https://a.com" + assert result[1]["url"] == "https://b.com" + + +# ========================================================= +# fetch_page +# ========================================================= + + +# --------------------------------------------------------- +# 10. HTML wird geparst, Text wird zurückgegeben +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.socket.gethostbyname", return_value="93.184.216.34") +@patch("backend.managers.search_manager.requests.get") +def test_fetch_page_returns_text(mock_get, _mock_dns, manager): + response = Mock() + response.text = "

Hello World

" + response.raise_for_status = Mock() + mock_get.return_value = response + + result = manager.fetch_page("https://example.com") + + assert "Hello World" in result + + +# --------------------------------------------------------- +# 11. script- und style-Tags werden entfernt +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.socket.gethostbyname", return_value="93.184.216.34") +@patch("backend.managers.search_manager.requests.get") +def test_fetch_page_removes_noise_tags(mock_get, _mock_dns, manager): + response = Mock() + response.text = ( + "" + "

Content

" + ) + response.raise_for_status = Mock() + mock_get.return_value = response + + result = manager.fetch_page("https://example.com") + + assert "alert" not in result + assert "color:red" not in result + assert "Content" in result + + +# --------------------------------------------------------- +# 12. Inhalt über MAX_PAGE_CHARS wird abgeschnitten +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.socket.gethostbyname", return_value="93.184.216.34") +@patch("backend.managers.search_manager.requests.get") +def test_fetch_page_truncates_long_content(mock_get, _mock_dns, manager): + long_text = "A" * (MAX_PAGE_CHARS + 500) + response = Mock() + response.text = f"{long_text}" + response.raise_for_status = Mock() + mock_get.return_value = response + + result = manager.fetch_page("https://example.com") + + assert "[truncated]" in result + assert len(result) <= MAX_PAGE_CHARS + len("\n... [truncated]") + 5 + + +# --------------------------------------------------------- +# 13. Inhalt unter MAX_PAGE_CHARS wird nicht abgeschnitten +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.socket.gethostbyname", return_value="93.184.216.34") +@patch("backend.managers.search_manager.requests.get") +def test_fetch_page_no_truncation_for_short_content(mock_get, _mock_dns, manager): + response = Mock() + response.text = "

Short

" + response.raise_for_status = Mock() + mock_get.return_value = response + + result = manager.fetch_page("https://example.com") + + assert "[truncated]" not in result + assert "Short" in result + + +# --------------------------------------------------------- +# 14. requests.Timeout → Fehlermeldung als String +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.socket.gethostbyname", return_value="93.184.216.34") +@patch("backend.managers.search_manager.requests.get") +def test_fetch_page_timeout_returns_error_string(mock_get, _mock_dns, manager): + import requests as req_module + mock_get.side_effect = req_module.Timeout("timed out") + + result = manager.fetch_page("https://example.com") + + assert "Error fetching page" in result + + +# --------------------------------------------------------- +# 15. ConnectionError → Fehlermeldung als String +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.socket.gethostbyname", return_value="93.184.216.34") +@patch("backend.managers.search_manager.requests.get") +def test_fetch_page_connection_error_returns_error_string(mock_get, _mock_dns, manager): + import requests as req_module + mock_get.side_effect = req_module.ConnectionError("refused") + + result = manager.fetch_page("https://example.com") + + assert "Error fetching page" in result + + +# ========================================================= +# _validate_url +# ========================================================= + + +# --------------------------------------------------------- +# 16. https-URL mit öffentlicher IP → kein Fehler +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.socket.gethostbyname", return_value="93.184.216.34") +def test_validate_url_valid_https(_mock_dns, manager): + manager._validate_url("https://example.com") # no exception + + +# --------------------------------------------------------- +# 17. http-URL → kein Fehler +# --------------------------------------------------------- + +@patch("backend.managers.search_manager.socket.gethostbyname", return_value="93.184.216.34") +def test_validate_url_valid_http(_mock_dns, manager): + manager._validate_url("http://example.com") # no exception + + +# --------------------------------------------------------- +# 18. localhost → ValueError +# --------------------------------------------------------- + +def test_validate_url_blocks_localhost(manager): + with pytest.raises(ValueError, match="localhost"): + manager._validate_url("http://localhost/admin") + + +# --------------------------------------------------------- +# 19. 127.0.0.1 → ValueError +# --------------------------------------------------------- + +def test_validate_url_blocks_127(manager): + with pytest.raises(ValueError): + manager._validate_url("http://127.0.0.1:8080") + + +# --------------------------------------------------------- +# 20. ::1 (IPv6 loopback) → ValueError +# --------------------------------------------------------- + +def test_validate_url_blocks_ipv6_loopback(manager): + with pytest.raises(ValueError): + manager._validate_url("http://[::1]/secret") + + +# --------------------------------------------------------- +# 21. Private IP 192.168.x.x → ValueError +# --------------------------------------------------------- + +def test_validate_url_blocks_private_192(manager): + with pytest.raises(ValueError): + manager._validate_url("http://192.168.1.10") + + +# --------------------------------------------------------- +# 22. Private IP 10.x.x.x → ValueError +# --------------------------------------------------------- + +def test_validate_url_blocks_private_10(manager): + with pytest.raises(ValueError): + manager._validate_url("http://10.0.0.1") + + +# --------------------------------------------------------- +# 23. Link-local / AWS Metadata IP → ValueError +# --------------------------------------------------------- + +def test_validate_url_blocks_link_local(manager): + with pytest.raises(ValueError): + manager._validate_url("http://169.254.169.254/latest/meta-data/") + + +# --------------------------------------------------------- +# 24. file://-Schema → ValueError +# --------------------------------------------------------- + +def test_validate_url_blocks_file_scheme(manager): + with pytest.raises(ValueError, match="http/https"): + manager._validate_url("file:///etc/passwd") + + +# --------------------------------------------------------- +# 25. ftp://-Schema → ValueError +# --------------------------------------------------------- + +def test_validate_url_blocks_ftp_scheme(manager): + with pytest.raises(ValueError, match="http/https"): + manager._validate_url("ftp://example.com/file.txt") + + +# --------------------------------------------------------- +# 26. fetch_page propagiert ValueError aus _validate_url +# --------------------------------------------------------- + +def test_fetch_page_raises_on_invalid_url(manager): + with pytest.raises(ValueError): + manager.fetch_page("http://localhost/internal")