meulilivio/AISE1_Project_Irina_Livio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Minor Updates

Browse Source
This commit is contained in:
Livio Meuli 2026-05-27 07:54:07 +02:00
parent b83cd2a790
commit 91f816168a
20 changed files with 98 additions and 292 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
DEBUG_LOGGER_USAGE.md
View File

@ -1,46 +0,0 @@
# DebugLogger Usage
## Import
```python
from backend.managers.debug_logger import DebugLogger
logger = DebugLogger()
```
## Methoden
```python
logger.clear() # vor jeder neuen Ausführung aufrufen
logger.log("Nachricht") # INFO-Eintrag
logger.log_error("Fehler") # ERROR-Eintrag
logger.get_logs() # gibt Liste aller Einträge zurück
logger.format_debug_output({ # gibt formatierten String zurück
"rc": 0,
"stdout": "...",
"stderr": "..."
})
```
## Eintrag-Format
```python
{
"level": "INFO", # oder "ERROR"
"message": "Nachricht",
"timestamp": "14:23:01"
}
```
## Beispiel
```python
logger = DebugLogger()
logger.clear()
logger.log("Starte Ausführung...")
try:
result = run_something()
logger.log("Erfolgreich abgeschlossen.")
except Exception as e:
logger.log_error(f"Fehler: {e}")
# Logs anzeigen
for entry in logger.get_logs():
print(f"[{entry['timestamp']}] [{entry['level']}] {entry['message']}")
```

49
backend/agent/coding_agent.py
View File

@ -21,8 +21,6 @@ import asyncio
import requests
from dotenv import load_dotenv
# REVIEW: commented-out import — remove once the package import above is confirmed stable.
#from mcp_server_adapter import MCPToolAdapter # Import from current directory for easier testing without package structure
from backend.agent.mcp_server_adapter import MCPToolAdapter
from backend.managers.debug_logger import get_logger
@ -95,7 +93,7 @@ async def dispatch_tool(tool_name: str, arguments: dict) -> str:
logger.info("Calling tool '%s' in dispatch_tool through MCPToolAdapter...", tool_name)
result = await adapter.call_tool(tool_name, arguments)
logger.info(f"Result from tool '%s' recieved", tool_name)
logger.info("Result from tool '%s' received", tool_name)
if result.isError:
# MCP servers signal tool-level errors via the isError flag rather
@ -108,7 +106,7 @@ async def dispatch_tool(tool_name: str, arguments: dict) -> str:
return "\n".join(texts)
except Exception as e:
logger.exception(f"Error calling tool '%s' with argument: %s", tool_name, arguments)
logger.exception("Error calling tool '%s' with argument: %s", tool_name, arguments)
return f"Error calling tool '{tool_name}': {e}"
# ═════════════════════════════════════════════════════════════════════════════
@ -417,10 +415,6 @@ class CodingAgent:
timeout=60)
response.raise_for_status()
logger.info("LLM API response requested")
if response.status_code != 200:
logger.error("API Error %s: %s", response.status_code, response.text)
raise Exception(f"API Error {response.status_code}: {response.text}")
except requests.RequestException as exc:
logger.exception("API Error; HTTP-Fehler: %s", exc)
raise Exception(f"HTTP-Fehler: {exc}") from exc
@ -594,42 +588,3 @@ class CodingAgent:
})
self.pending_action = None
logger.info("Rejection message appended.")
# REVIEW: dead code — this module is always imported, never run as a script.
# The __main__ guard below is unreachable in normal use. Move this to run_agent.py or delete it.
def main():
"""Example of how to use the CodingAgent in a simple loop."""
agent = CodingAgent()
task = "Write a Python function that returns the nth Fibonacci number."
agent.start_task(task)
if agent.pending_action:
print(f"Initial proposed action: {agent.pending_action['action']}")
while not agent.is_done:
action = asyncio.run(agent.propose_next_action())
print(f"Proposed action: {action}")
if action["tool"] == "done":
print("Task completed.")
break
else:
user_feedback = input("Approve this action? (y/n) ")
if user_feedback.lower() == "y":
result = asyncio.run(agent.approve())
print(f"Tool result: {result}")
elif user_feedback.lower() == "n":
feedback = input("Enter feedback for the agent: ")
agent.reject(feedback)
# REVIEW: unreachable when action["tool"] == "done" (we break above); also `result` is
# unbound when the elif branch runs — this will raise UnboundLocalError at runtime.
if result["is_done"]:
print("Task completed.")
break
if __name__ == "__main__":
main()

4
backend/agent/mcp_server_adapter.py
View File

@ -97,10 +97,8 @@ class MCPToolAdapter:
await session.initialize()
logger.info("Session initialized for %s. Requesting tools...", server_name)
result = await session.list_tools()
# REVIEW: debug print — remove before shipping.
print(f"Tools received from {server_name}: {result}")
tools = result.tools
logger.info(f"Tools received from %s: %s Tools", server_name, str(len(tools)))
logger.info("Tools received from %s: %s Tools", server_name, str(len(tools)))
for tool in tools:
# Build a human-readable parameter description for the system prompt.

10
backend/agent/servers/mcp_server_code_execution.py
View File

@ -15,6 +15,7 @@ blocks dangerous imports and builtins before spawning any subprocess.
"""
import ast
import os
import subprocess
import sys
@ -22,7 +23,6 @@ import io
from pyflakes.api import check # For linting Code
from pyflakes.reporter import Reporter # For linting Code
from mcp.server.fastmcp import FastMCP
from pathlib import Path
#from backend.managers.debug_logger import get_logger
#logger = get_logger(__name__)
@ -275,12 +275,18 @@ def run_python_sandboxed(code: str) -> str:
return f"Code rejected:{static_safety}"
try:
# Force UTF-8 I/O so the subprocess can print unicode on Windows
# (default console encoding is cp1252 which cannot encode emoji).
utf8_env = {**os.environ, "PYTHONIOENCODING": "utf-8"}
result = subprocess.run(
[sys.executable, "-c", code],
stdin=subprocess.DEVNULL,
capture_output=True,
text=True,
timeout=EXEC_TIMEOUT)
encoding="utf-8",
timeout=EXEC_TIMEOUT,
env=utf8_env,
)
# Merge stdout and stderr so the agent sees all output in one block.
output = result.stdout + result.stderr

16
backend/agent/servers/mcp_server_file_search.py
View File

@ -77,11 +77,7 @@ def get_file_tree(dir_path: str=ALLOWED_DIR) -> str:
#logger.info("Tool get_file_tree is being executed on MCP file search server")
try:
safe_dir = _safe_path(dir_path)
# REVIEW: unreachable code — _safe_path() always returns a Path object (never None/falsy)
# or raises ValueError; this check can never be True.
if not safe_dir:
return f"Error: Invalid directory path '{dir_path}'."
elif not safe_dir.exists():
if not safe_dir.exists():
#logger.warning("Directory '%s' does not exist.", dir_path)
return f"Error: Directory '{dir_path}' does not exist."
elif not safe_dir.is_dir():
@ -153,11 +149,7 @@ def search_files(query: str) -> str:
if query_lower in line.lower():
snippet = line.strip()[:100]
results.append(f"[content] {rel}:{i} -- {snippet}")
except UnicodeDecodeError as e:
#logger.warning("Decode error in file/folder '%s': %s", f, e)
pass
except PermissionError as e:
#logger.warning("Permission error in file/folder: '%s': %s", f, e)
except (UnicodeDecodeError, PermissionError):
pass
if not results:
@ -274,9 +266,7 @@ def create_new_directory(path: str) -> str:
#logger.warning("Requested path '%s' already exists, overwriting not allowed.", path)
return f"Error: File '{path}' already exists."
# REVIEW: redundant — `resolved.suffix != None` is always True (Path.suffix always returns str);
# the None check is unnecessary. Simplify to `if resolved.suffix != "":`.
if resolved.suffix != None and resolved.suffix != "":
if resolved.suffix != "":
#logger.warning("Can only create directories, got '%s'.", resolved.suffix)
return f"Error: can only create directories, got '{resolved.suffix}'."

11
backend/managers/chat_manager.py
View File

@ -55,7 +55,6 @@ class ChatManager:
"""Return a copy of the conversation history."""
return list(self.chat_history)
# REVIEW: dead code — clear_history() is never called anywhere in the codebase.
def clear_history(self) -> None:
"""Wipe the conversation history (starts a fresh chat)."""
logger.info("Chat history was cleared")
@ -147,13 +146,3 @@ class ChatManager:
logger.exception("JSON parsing and message formatting failed: %s", e)
raise RuntimeError("JSON parsing and message formatting failed") from e
# REVIEW: dead code — get_chat_display() is never called anywhere in the codebase.
# The UI renders st.session_state.chat_history directly. This method also does the
# same thing as get_history() (returns a copy of chat_history with the same fields),
# making it redundant even if it were used.
def get_chat_display(self) -> list:
"""Return a copy of the history suitable for display in the UI."""
return [
{"role": msg["role"], "content": msg["content"]}
for msg in self.chat_history
]

20
backend/managers/file_manager.py
View File

@ -168,12 +168,6 @@ class FileManager:
content = f.read()
logger.info("File read successfully.")
return content
except FileNotFoundError:
# REVIEW: unreachable code — FileNotFoundError cannot be raised here because
# `file_path.exists()` is already checked above and returns "" on failure.
st.error(f"File not found: {relative_path}")
logger.warning("File not found")
return ""
except Exception as e:
st.error(f"Error reading file {relative_path}: {str(e)}")
logger.exception("Error reading file at %s: %s", relative_path, e)
@ -353,19 +347,5 @@ class FileManager:
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()

5
frontend/app.py
View File

@ -47,11 +47,6 @@ def main():
st.title("Lightweight code editor")
# REVIEW: redundant — init_state() is already called at module level (line 26) before main() runs;
# calling it again here is unnecessary since Streamlit reruns the whole module on each reload.
# Re-run init_state to cover any keys that might have been missed on cold start
init_state()
render_sidebar()
# Switch between the two main views based on the sidebar radio button

16
frontend/chat.py
View File

@ -19,9 +19,8 @@ logger = get_logger(__name__)
def _run_async(coro):
"""Execute an async coroutine from synchronous Streamlit code.
Streamlit runs in a synchronous context, but the CodingAgent uses async
methods (for MCP tool calls). This helper bridges the gap by reusing an
already-running event loop when one exists, or creating a new one otherwise.
Streamlit always runs in a plain synchronous thread with no running event
loop, so we always create a fresh loop here.
Args:
coro: The coroutine to run.
@ -29,15 +28,8 @@ def _run_async(coro):
Returns:
The return value of the coroutine.
"""
# REVIEW: asyncio.get_running_loop() always raises RuntimeError in a Streamlit context;
# the try branch is dead code. The except branch always runs.
try:
# Reuse the loop that is already running (e.g. inside pytest-asyncio).
loop = asyncio.get_running_loop()
except RuntimeError:
# No running loop in this thread — create a fresh one.
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
return loop.run_until_complete(coro)
def _start_agent(task: str):

2
frontend/sidebar.py
View File

@ -406,8 +406,6 @@ def render_sidebar():
if fm.save_file(dest, content):
st.success(f"'{uploaded.name}' uploaded successfully.")
# REVIEW: bare `return` at end of void function — no-op; can be removed.
return
if __name__ == "__main__":

23
frontend/state.py
View File

@ -43,41 +43,18 @@ def init_state():
# Ordered list of absolute file paths currently open as editor tabs.
# The list order determines the visual tab order in the UI.
if "open_files" not in st.session_state:
# REVIEW: dead code — docstrings inside `if` blocks are plain string literals that Python
# evaluates and immediately discards; they are never visible as __doc__ and have no effect.
st.session_state.open_files = []
# Dict mapping absolute file path → current editor content (may differ from
# disk if the user has unsaved changes).
if "files_content" not in st.session_state:
# REVIEW: dead code — same issue: string literal inside `if` block is never used as a docstring.
st.session_state.files_content = {}
# Absolute path of the file whose tab is currently active in the editor.
# Must always be one of the paths in open_files, or None if no file is open.
if "active_file" not in st.session_state:
# REVIEW: dead code — string literal inside `if` block is never used as a docstring.
st.session_state.active_file = None
# REVIEW: dead code — active_tab is initialised here but never read or written anywhere else
# in the codebase; st.tabs() in editor.py does not use this key.
# Index of the active tab — kept in sync with active_file for st.tabs().
if "active_tab" not in st.session_state:
st.session_state.active_tab = 0
# REVIEW: dead code — is_editing is initialised here but never read or written anywhere else.
if "is_editing" not in st.session_state:
st.session_state.is_editing = False
# REVIEW: dead code — code_suggestions is initialised here but never read or written anywhere else.
if "code_suggestions" not in st.session_state:
st.session_state.code_suggestions = []
# Output dict from the last code execution: {stdout, stderr, return_code}.
# Initialised as empty string so the editor view can safely check falsyness.
if "code_execution_output" not in st.session_state:
st.session_state.code_execution_output = ""
# Per-file execution results: {file_path: {stdout, stderr, return_code, ast_error}}
if "exec_results" not in st.session_state:
st.session_state.exec_results = {}

21
tests/test_chat_manager.py
View File

@ -134,24 +134,3 @@ class TestSendMessage:
cm.send_message("Hello")
# ── get_chat_display ──────────────────────────────────────────────────────────
class TestGetChatDisplay:
"""Tests for get_chat_display: correct shape and ordering."""
@pytest.fixture
def cm(self):
return ChatManager()
def test_display_has_role_and_content_keys(self, cm):
cm.add_message("user", "Hello")
entry = cm.get_chat_display()[0]
assert "role" in entry
assert "content" in entry
def test_display_preserves_message_order(self, cm):
cm.add_message("user", "First")
cm.add_message("assistant", "Second")
display = cm.get_chat_display()
assert display[0]["role"] == "user"
assert display[1]["role"] == "assistant"

2
tests/test_coding_agent.py
View File

@ -21,12 +21,10 @@ from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
import pytest_asyncio
sys.path.insert(0, str(Path(__file__).parent.parent))
from backend.agent.coding_agent import (
MAX_HISTORY_CHARS,
MAX_ITERATIONS,
MAX_RESULT_LENGTH,
CodingAgent,

139
tests/test_debug_logger.py
View File

@ -1,4 +1,9 @@
"""Tests for DebugLogger (backend/managers/debug_logger.py)."""
"""Tests for DebugLogger (backend/managers/debug_logger.py).
DebugLogger is a classmethod-based utility. Its _error_log class variable
persists across tests, so every test that modifies it must call
DebugLogger.clear_errors() in teardown (handled by the autouse fixture).
"""
import sys
from pathlib import Path
@ -6,105 +11,99 @@ import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from backend.managers._debug_logger import DebugLogger
from backend.managers.debug_logger import DebugLogger
# ── Fixtures ──────────────────────────────────────────────────────────────────
@pytest.fixture
def logger():
"""Return a fresh DebugLogger for each test."""
return DebugLogger()
# ── log() ─────────────────────────────────────────────────────────────────────
class TestLog:
"""Tests for log(): each call appends an INFO-level entry with message."""
def test_log_appends_entry(self, logger):
logger.log("started")
assert len(logger.logs) == 1
def test_log_sets_level_info(self, logger):
logger.log("started")
assert logger.logs[0]["level"] == "INFO"
def test_log_stores_message(self, logger):
logger.log("executing file.py")
assert logger.logs[0]["message"] == "executing file.py"
@pytest.fixture(autouse=True)
def clear_error_log():
"""Reset the shared _error_log class variable before and after each test."""
DebugLogger.clear_errors()
yield
DebugLogger.clear_errors()
# ── log_error() ───────────────────────────────────────────────────────────────
class TestLogError:
"""Tests for log_error(): level is ERROR, not INFO."""
"""Tests for log_error(): appends the message to the in-memory error list."""
def test_log_error_sets_level_error(self, logger):
logger.log_error("something broke")
assert logger.logs[0]["level"] == "ERROR"
def test_log_error_appends_to_error_log(self):
DebugLogger.log_error("something broke")
assert "something broke" in DebugLogger.get_errors()
def test_log_and_log_error_are_distinct_levels(self, logger):
logger.log("info message")
logger.log_error("error message")
assert logger.logs[0]["level"] == "INFO"
assert logger.logs[1]["level"] == "ERROR"
def test_log_error_multiple_messages_all_stored(self):
DebugLogger.log_error("first error")
DebugLogger.log_error("second error")
errors = DebugLogger.get_errors()
assert "first error" in errors
assert "second error" in errors
def test_log_error_preserves_order(self):
DebugLogger.log_error("alpha")
DebugLogger.log_error("beta")
errors = DebugLogger.get_errors()
assert errors.index("alpha") < errors.index("beta")
# ── get_logs() ────────────────────────────────────────────────────────────────
# ── get_errors() ──────────────────────────────────────────────────────────────
class TestGetLogs:
"""Tests for get_logs()."""
class TestGetErrors:
"""Tests for get_errors(): returns the current in-memory error list."""
def test_get_logs_returns_all_entries(self, logger):
logger.log("first")
logger.log_error("second")
assert len(logger.get_logs()) == 2
def test_get_errors_empty_initially(self):
assert DebugLogger.get_errors() == []
def test_get_errors_reflects_logged_errors(self):
DebugLogger.log_error("boom")
assert len(DebugLogger.get_errors()) == 1
# ── clear() ───────────────────────────────────────────────────────────────────
# ── clear_errors() ────────────────────────────────────────────────────────────
class TestClear:
"""Tests for clear()."""
class TestClearErrors:
"""Tests for clear_errors(): wipes the in-memory error list."""
def test_clear_removes_all_entries(self, logger):
logger.log("first")
logger.log_error("second")
logger.clear()
assert logger.logs == []
def test_clear_errors_empties_list(self):
DebugLogger.log_error("will be cleared")
DebugLogger.clear_errors()
assert DebugLogger.get_errors() == []
def test_clear_errors_on_empty_list_does_not_raise(self):
DebugLogger.clear_errors() # already empty from autouse fixture
# ── format_debug_output() ─────────────────────────────────────────────────────
class TestFormatDebugOutput:
"""Tests for format_debug_output(): renders rc, stdout, stderr, and log entries."""
"""Tests for format_debug_output(): renders return_code, stdout, and stderr."""
def test_success_exit_code_shows_success(self, logger):
result = logger.format_debug_output({"rc": 0, "stdout": "", "stderr": ""})
assert "[SUCCESS]" in result
def test_contains_execution_result_header(self):
result = DebugLogger.format_debug_output({"return_code": 0, "stdout": "", "stderr": ""})
assert "=== Execution Result ===" in result
def test_nonzero_exit_code_shows_failed(self, logger):
result = logger.format_debug_output({"rc": 1, "stdout": "", "stderr": ""})
assert "[FAILED]" in result
def test_exit_code_zero_appears_in_output(self):
result = DebugLogger.format_debug_output({"return_code": 0, "stdout": "", "stderr": ""})
assert "Exit Code: 0" in result
def test_stdout_included_when_present(self, logger):
result = logger.format_debug_output({"rc": 0, "stdout": "Hello", "stderr": ""})
def test_nonzero_exit_code_appears_in_output(self):
result = DebugLogger.format_debug_output({"return_code": 1, "stdout": "", "stderr": ""})
assert "Exit Code: 1" in result
def test_stdout_included_when_present(self):
result = DebugLogger.format_debug_output({"return_code": 0, "stdout": "Hello", "stderr": ""})
assert "Hello" in result
assert "stdout" in result
def test_stderr_included_when_present(self, logger):
result = logger.format_debug_output({"rc": 1, "stdout": "", "stderr": "NameError"})
def test_stderr_included_when_present(self):
result = DebugLogger.format_debug_output({"return_code": 1, "stdout": "", "stderr": "NameError"})
assert "NameError" in result
assert "stderr" in result
def test_log_entries_appended_to_output(self, logger):
logger.log("Executing file.py...")
logger.log_error("exit code 1")
result = logger.format_debug_output({"rc": 1, "stdout": "", "stderr": ""})
assert "Executing file.py..." in result
assert "exit code 1" in result
def test_empty_stdout_shows_none_placeholder(self):
result = DebugLogger.format_debug_output({"return_code": 0, "stdout": "", "stderr": ""})
assert "(none)" in result
def test_missing_keys_do_not_raise(self, logger):
# Defensive: format_debug_output uses .get() so missing keys are safe
result = logger.format_debug_output({})
def test_missing_keys_do_not_raise(self):
# format_debug_output uses .get() so absent keys fall back to defaults.
result = DebugLogger.format_debug_output({})
assert isinstance(result, str)

1
tests/test_execution_engine.py
View File

@ -1,6 +1,5 @@
import pytest
import subprocess
from pathlib import Path
from unittest.mock import Mock, patch
from backend.managers.execution_engine import ExecutionEngine

4
tests/test_mcp_server_code_execution.py
View File

@ -1,7 +1,3 @@
import pytest
from unittest.mock import Mock, patch
import subprocess
from backend.agent.servers import mcp_server_code_execution as server

5
tests/test_mcp_server_file_search.py
View File

@ -1,5 +1,4 @@
import pytest
from pathlib import Path
from backend.agent.servers import mcp_server_file_search as server
@ -66,7 +65,9 @@ def test_list_files_recursive(workspace):
result = server.list_files()
assert "src/main.py" in result
# list_files() uses Path.relative_to() which yields OS-native separators;
# check for the components instead of a hard-coded slash style.
assert "src" in result and "main.py" in result
# ---------------------------------------------------------

12
tests/test_mcp_server_web_search.py
View File

@ -63,7 +63,7 @@ def test_validate_url_private_ip():
# 6. web_search liefert Suchergebnisse
# ---------------------------------------------------------
@patch("ddgs.DDGS")
@patch("backend.agent.servers.mcp_server_web_search.DDGS")
def test_web_search_success(mock_ddgs):
mock_instance = Mock()
@ -87,7 +87,7 @@ def test_web_search_success(mock_ddgs):
# 7. web_search ohne Ergebnisse
# ---------------------------------------------------------
@patch("ddgs.DDGS")
@patch("backend.agent.servers.mcp_server_web_search.DDGS")
def test_web_search_no_results(mock_ddgs):
mock_instance = Mock()
mock_instance.text.return_value = []
@ -230,7 +230,9 @@ def test_fetch_page_timeout(mock_get):
result = server.fetch_page("https://example.com")
assert "Error fetching page" in result
# fetch_page catches RequestException (which includes Timeout) and returns
# "HTTP-Fehler: <exception>" — verify an error string comes back.
assert "HTTP-Fehler" in result or "Error" in result
# ---------------------------------------------------------
@ -245,7 +247,7 @@ def test_fetch_page_connection_error(mock_get):
result = server.fetch_page("https://example.com")
assert "Error fetching page" in result
assert "HTTP-Fehler" in result or "Error" in result
# ---------------------------------------------------------
@ -288,7 +290,7 @@ def test_fetch_page_empty_content(mock_get):
# 20. web_search behandelt Exception sauber
# ---------------------------------------------------------
@patch("ddgs.DDGS")
@patch("backend.agent.servers.mcp_server_web_search.DDGS")
def test_web_search_exception(mock_ddgs):
mock_ddgs.side_effect = Exception("DDGS failed")

3
tests/test_search_manager.py
View File

@ -1,8 +1,7 @@
"""Tests for SearchManager — no real network calls, all I/O mocked."""
import socket
import pytest
from unittest.mock import Mock, patch, MagicMock
from unittest.mock import Mock, patch
from backend.managers.search_manager import SearchManager, MAX_PAGE_CHARS

1
tests/test_system_prompter.py
View File

@ -5,7 +5,6 @@ from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from backend.managers.system_prompter import SystemPrompter, MAX_FILE_CHARS