Fixes and Updates
This commit is contained in:
Livio Meuli
parent
efdd9351b3
commit
33dd93fbbf
@ -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/`)
|
||||
|
||||
11
READMEnew.md
11
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:
|
||||
|
||||
@ -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}")
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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,
|
||||
}
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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?"
|
||||
)
|
||||
|
||||
@ -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",
|
||||
|
||||
362
tests/test_search_manager.py
Normal file
362
tests/test_search_manager.py
Normal file
@ -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 = "<html><body><h1>Hello World</h1></body></html>"
|
||||
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 = (
|
||||
"<html><head><style>body{color:red}</style></head>"
|
||||
"<body><script>alert('x')</script><p>Content</p></body></html>"
|
||||
)
|
||||
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"<html><body>{long_text}</body></html>"
|
||||
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 = "<html><body><p>Short</p></body></html>"
|
||||
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")
|
||||
Loading…
x
Reference in New Issue
Block a user