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

Merge pull request 'feature/backend_implemation' (#4) from feature/backend_implemation into main

Browse Source 
Reviewed-on: meulilivio/AISE1_Project#4
This commit is contained in:
Livio Meuli 2026-04-09 14:20:36 +02:00
commit a45eaae84a
8 changed files with 506 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
backend/__init__.py
View File

@ -0,0 +1,4 @@
"""Backend - AI agents, managers, and utilities"""
from backend.managers import ChatManager
__all__ = ["ChatManager"]

5
backend/managers/__init__.py
View File

@ -0,0 +1,5 @@
"""Backend Managers - Business logic for UI components"""
from backend.managers.chat_manager import ChatManager
from backend.managers.system_prompter import SystemPrompter
__all__ = ["ChatManager", "SystemPrompter"]

97
backend/managers/chat_manager.py
View File

@ -0,0 +1,97 @@
"""Chat Manager - Handles chat history and AI communication"""
import os
from dotenv import load_dotenv
import requests
import json
load_dotenv()
class ChatManager:
def __init__(self):
self.api_host = os.getenv("HOST")
self.api_port = os.getenv("PORT")
self.api_key = os.getenv("API_KEY")
self.model = os.getenv("MODEL")
# API endpoint URL (OpenAI-compatible format)
self.api_url = f"http://{self.api_host}:{self.api_port}/v1/chat/completions"
# Chat history stored in memory
self.chat_history = []
def add_message(self, role: str, content: str) -> None:
self.chat_history.append({"role": role, "content": content})
def get_history(self) -> list:
return self.chat_history
def clear_history(self) -> None:
self.chat_history = []
def send_message(self, user_message: str) -> str:
# Add user message to history
self.add_message("user", user_message)
try:
# Prepare request to OpenAI-compatible API
headers = {
"Content-Type": "application/json",
}
# Add API key if available
if self.api_key and self.api_key != "EMPTY":
headers["Authorization"] = f"Bearer {self.api_key}"
payload = {
"model": self.model,
"messages": self.chat_history,
"temperature": 0.7,
"max_tokens": 2000,
"stream": False,
}
# Make API request
response = requests.post(
self.api_url, headers=headers, json=payload, timeout=30
)
# Check if request was successful
if response.status_code != 200:
error_msg = f"API Error {response.status_code}: {response.text}"
raise Exception(error_msg)
# 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)
return ai_message
else:
raise Exception("Invalid API response format")
except requests.exceptions.RequestException as e:
error_msg = f"Connection Error: {str(e)}"
# Add error message to history so user sees it
self.add_message("assistant", f"Error: {error_msg}")
raise Exception(error_msg)
except json.JSONDecodeError as e:
error_msg = f"JSON Decode Error: {str(e)}"
self.add_message("assistant", f"Error: {error_msg}")
raise Exception(error_msg)
except Exception as e:
error_msg = f"Error: {str(e)}"
self.add_message("assistant", f"Error: {error_msg}")
raise Exception(error_msg)
def get_chat_display(self) -> list:
return [
{"role": msg["role"], "content": msg["content"]}
for msg in self.chat_history
]

38
backend/managers/system_prompter.py
View File

@ -0,0 +1,38 @@
"""System Prompter - Builds system prompts with optional file context"""
MAX_FILE_CHARS = 4000 # Limit file context to avoid token overflow
class SystemPrompter:
@staticmethod
def generate_prompt(file_context: dict | None = None) -> str:
"""Build a system prompt, optionally embedding a file's content.
Args:
file_context: dict with keys 'name' and 'content', or None.
Returns:
A system prompt string.
"""
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."
)
if file_context:
name = file_context.get("name", "unknown")
content = file_context.get("content", "")
# Truncate large files to avoid exceeding token limits
if len(content) > MAX_FILE_CHARS:
content = content[:MAX_FILE_CHARS] + "\n... [truncated]"
file_section = (
f"\n\nThe user currently has the following file open in the editor:\n"
f"<file name=\"{name}\">\n"
f"<code>\n{content}\n</code>\n"
f"</file>\n"
f"Refer to this file when answering questions about the code."
)
return base + file_section
return base

36
frontend/chat.py
View File

@ -1,8 +1,10 @@
import streamlit as st
from backend.managers.chat_manager import ChatManager
from backend.managers.system_prompter import SystemPrompter
def render_chat():
st.subheader("Chat with AI Assistant")
chat_section = st.container()
setup_section = st.container()
@ -10,21 +12,37 @@ def render_chat():
if st.session_state.chat_history:
for message in st.session_state.chat_history:
st.markdown(f"**{message['role'].capitalize()}:** {message['content']}")
# Clear the input field before the widget is rendered (Streamlit requirement)
if st.session_state.get("_clear_chat_input"):
st.session_state.chat_input = ""
st.session_state._clear_chat_input = False
user_input = st.text_input("Type your message here:", key="chat_input")
if st.button("Send", key="send_button") and user_input:
chat_manager = st.session_state.chat_manager
# Inject system prompt on the first message
if not chat_manager.get_history():
system_prompt = SystemPrompter.generate_prompt()
chat_manager.add_message("system", system_prompt)
st.session_state.chat_history.append({"role": "user", "content": user_input})
# Here you would typically call your AI assistant to get a response
# For demonstration, we'll just echo the user's message
ai_response = f"Echo: {user_input}"
try:
ai_response = chat_manager.send_message(user_input)
except Exception as e:
ai_response = f"Error: {e}"
st.session_state.chat_history.append({"role": "assistant", "content": ai_response})
st.session_state.chat_input = "" # Clear input after sending
st.session_state._clear_chat_input = True # Clear input on next rerun
st.rerun()
with setup_section:
st.info("This is where you can set up your AI assistant. For now, this section is just a placeholder.")
st.toggle("Use debug system prompt", key="use_system_prompt", value=True)
# Here you could add options to configure the AI assistant, such as selecting a model, setting parameters, etc.

6
frontend/state.py
View File

@ -1,8 +1,12 @@
import streamlit as st
from backend.managers.chat_manager import ChatManager
def init_state():
# Sidebar state initialization
# Chat manager (persists across reruns)
if "chat_manager" not in st.session_state:
st.session_state.chat_manager = ChatManager()
# Editor state initialization
if "open_files" not in st.session_state:

242
tests/test_chat_manager.py
View File

@ -0,0 +1,242 @@
"""Test script for ChatManager - Pytest compatible tests"""
import sys
from pathlib import Path
import pytest
from unittest.mock import patch, MagicMock
# Add project root to Python path
sys.path.insert(0, str(Path(__file__).parent.parent))
from backend.managers.chat_manager import ChatManager
class TestChatManager:
"""Test suite for ChatManager functionality."""
@pytest.fixture
def chat_manager(self):
return ChatManager()
def test_initialization(self, chat_manager):
"""Test that ChatManager initializes correctly."""
assert chat_manager.api_url is not None
assert chat_manager.model is not None
assert chat_manager.chat_history == []
def test_add_message(self, chat_manager):
"""Test adding messages to chat history."""
chat_manager.add_message("user", "Hello")
assert len(chat_manager.chat_history) == 1
assert chat_manager.chat_history[0]["role"] == "user"
assert chat_manager.chat_history[0]["content"] == "Hello"
def test_get_history(self, chat_manager):
"""Test retrieving chat history."""
chat_manager.add_message("user", "Hello")
chat_manager.add_message("assistant", "Hi there!")
history = chat_manager.get_history()
assert len(history) == 2
assert history[0]["role"] == "user"
assert history[1]["role"] == "assistant"
def test_clear_history(self, chat_manager):
"""Test clearing chat history."""
chat_manager.add_message("user", "Hello")
assert len(chat_manager.chat_history) == 1
chat_manager.clear_history()
assert len(chat_manager.chat_history) == 0
def test_send_message_integration(self, chat_manager):
"""
Integration test for sending message to AI.
This test actually communicates with the API.
"""
try:
# Send a simple test message
response = chat_manager.send_message("Hello, what is 2+2?")
# Verify response is not empty
assert isinstance(response, str)
assert len(response) > 0
# Verify message was added to history
assert len(chat_manager.chat_history) == 2 # user + assistant
assert chat_manager.chat_history[0]["role"] == "user"
assert chat_manager.chat_history[1]["role"] == "assistant"
print(f"API Test Passed")
print(f"Response: {response}")
except Exception as e:
# If API is not reachable, mark as skipped
pytest.skip(f"API not reachable: {str(e)}")
def test_multiple_messages(self, chat_manager):
"""Test sending multiple messages in a conversation."""
try:
# Send first message
response1 = chat_manager.send_message("What is your name?")
assert len(response1) > 0
# Send follow-up message
response2 = chat_manager.send_message("Tell me more")
assert len(response2) > 0
# Verify full conversation is in history
assert len(chat_manager.chat_history) == 4 # 2 user + 2 assistant
print(f"Conversation Test Passed")
print(f"Messages: {len(chat_manager.chat_history)}")
except Exception as e:
pytest.skip(f"API not reachable: {str(e)}")
class TestChatManagerSendMessage:
"""Unit tests for send_message using mocked HTTP requests."""
@pytest.fixture
def chat_manager(self):
return ChatManager()
def _mock_response(self, content="AI reply", status_code=200):
mock = MagicMock()
mock.status_code = status_code
mock.json.return_value = {
"choices": [{"message": {"role": "assistant", "content": content}}]
}
mock.text = "error text"
return mock
def test_send_message_adds_user_message_to_history(self, chat_manager):
with patch("requests.post", return_value=self._mock_response()):
chat_manager.send_message("Hello")
assert chat_manager.chat_history[0] == {"role": "user", "content": "Hello"}
def test_send_message_adds_assistant_response_to_history(self, chat_manager):
with patch("requests.post", return_value=self._mock_response("Hi there")):
chat_manager.send_message("Hello")
assert chat_manager.chat_history[1] == {"role": "assistant", "content": "Hi there"}
def test_send_message_returns_ai_content(self, chat_manager):
with patch("requests.post", return_value=self._mock_response("Answer")):
response = chat_manager.send_message("Question")
assert response == "Answer"
def test_send_message_history_grows_with_each_call(self, chat_manager):
with patch("requests.post", return_value=self._mock_response()):
chat_manager.send_message("First")
chat_manager.send_message("Second")
assert len(chat_manager.chat_history) == 4 # 2 user + 2 assistant
def test_send_message_connection_error_raises(self, chat_manager):
import requests
with patch("requests.post", side_effect=requests.exceptions.ConnectionError("refused")):
with pytest.raises(Exception, match="Connection Error"):
chat_manager.send_message("Hello")
def test_send_message_api_error_status_raises(self, chat_manager):
mock = self._mock_response(status_code=500)
with patch("requests.post", return_value=mock):
with pytest.raises(Exception, match="API Error 500"):
chat_manager.send_message("Hello")
def test_send_message_empty_choices_raises(self, chat_manager):
mock = MagicMock()
mock.status_code = 200
mock.json.return_value = {"choices": []}
with patch("requests.post", return_value=mock):
with pytest.raises(Exception, match="Invalid API response format"):
chat_manager.send_message("Hello")
def test_send_message_missing_choices_key_raises(self, chat_manager):
mock = MagicMock()
mock.status_code = 200
mock.json.return_value = {}
with patch("requests.post", return_value=mock):
with pytest.raises(Exception):
chat_manager.send_message("Hello")
def test_send_message_timeout_raises(self, chat_manager):
import requests
with patch("requests.post", side_effect=requests.exceptions.Timeout()):
with pytest.raises(Exception):
chat_manager.send_message("Hello")
class TestChatManagerGetChatDisplay:
"""Tests for get_chat_display()."""
@pytest.fixture
def chat_manager(self):
return ChatManager()
def test_empty_history_returns_empty_list(self, chat_manager):
assert chat_manager.get_chat_display() == []
def test_display_contains_role_and_content_keys(self, chat_manager):
chat_manager.add_message("user", "Hello")
display = chat_manager.get_chat_display()
assert "role" in display[0]
assert "content" in display[0]
def test_display_preserves_message_order(self, chat_manager):
chat_manager.add_message("user", "First")
chat_manager.add_message("assistant", "Second")
display = chat_manager.get_chat_display()
assert display[0]["role"] == "user"
assert display[1]["role"] == "assistant"
def test_display_matches_history(self, chat_manager):
chat_manager.add_message("user", "Hi")
chat_manager.add_message("assistant", "Hello!")
assert chat_manager.get_chat_display() == chat_manager.get_history()
def test_system_message_included_in_display(self, chat_manager):
chat_manager.add_message("system", "You are a helper.")
display = chat_manager.get_chat_display()
assert display[0]["role"] == "system"
def test_chat_manager_demo():
"""Demo test - Shows interactive chat (can be run manually)."""
print("\n" + "=" * 60)
print("ChatManager Demo - Interactive Test")
print("=" * 60 + "\n")
chat_manager = ChatManager()
print(f"Connected to API: {chat_manager.api_url}")
print(f"Model: {chat_manager.model}\n")
# Demo conversation
test_messages = ["Hello! What can you do?", "Tell me a joke", "What is Python?"]
print("Starting conversation...\n")
for message in test_messages:
print(f"User: {message}")
try:
response = chat_manager.send_message(message)
print(f"Assistant: {response}\n")
except Exception as e:
print(f"Error: {str(e)}\n")
pytest.skip(f"API not reachable: {str(e)}")
# Display full chat history
print("=" * 60)
print("Chat History:")
print("=" * 60)
for msg in chat_manager.get_history():
print(f"{msg['role'].upper()}: {msg['content']}\n")
if __name__ == "__main__":
# Run with: pytest tests/test_chat_manager.py -v -s
pytest.main([__file__, "-v", "-s"])

88
tests/test_system_prompter.py Normal file
View File

@ -0,0 +1,88 @@
"""Tests for SystemPrompter."""
import sys
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
class TestSystemPrompterBasePrompt:
"""Tests for generate_prompt() without file context."""
def test_returns_non_empty_string(self):
prompt = SystemPrompter.generate_prompt()
assert isinstance(prompt, str)
assert len(prompt) > 0
def test_describes_code_assistant(self):
prompt = SystemPrompter.generate_prompt()
assert "code assistant" in prompt.lower()
def test_contains_no_file_xml_tag(self):
prompt = SystemPrompter.generate_prompt()
assert "<file" not in prompt
assert "<code>" not in prompt
def test_none_equals_no_argument(self):
assert SystemPrompter.generate_prompt(file_context=None) == SystemPrompter.generate_prompt()
class TestSystemPrompterWithFileContext:
"""Tests for generate_prompt() with file_context provided."""
def test_includes_filename(self):
prompt = SystemPrompter.generate_prompt(file_context={"name": "main.py", "content": ""})
assert "main.py" in prompt
def test_includes_file_content(self):
prompt = SystemPrompter.generate_prompt(file_context={"name": "app.py", "content": "x = 42"})
assert "x = 42" in prompt
def test_uses_xml_file_tag(self):
prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": "pass"})
assert "<file" in prompt
def test_uses_xml_code_tag(self):
prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": "pass"})
assert "<code>" in prompt
def test_with_context_is_longer_than_base(self):
base = SystemPrompter.generate_prompt()
with_ctx = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": "x=1"})
assert len(with_ctx) > len(base)
def test_missing_name_key_uses_unknown(self):
prompt = SystemPrompter.generate_prompt(file_context={"content": "some code"})
assert "unknown" in prompt
def test_missing_content_key_does_not_raise(self):
prompt = SystemPrompter.generate_prompt(file_context={"name": "empty.py"})
assert "empty.py" in prompt
class TestSystemPrompterTruncation:
"""Tests for file content truncation."""
def test_large_file_is_truncated(self):
large = "a" * (MAX_FILE_CHARS + 500)
prompt = SystemPrompter.generate_prompt(file_context={"name": "big.py", "content": large})
assert "[truncated]" in prompt
def test_small_file_is_not_truncated(self):
content = "print('hello')"
prompt = SystemPrompter.generate_prompt(file_context={"name": "small.py", "content": content})
assert "[truncated]" not in prompt
assert content in prompt
def test_file_exactly_at_limit_is_not_truncated(self):
content = "x" * MAX_FILE_CHARS
prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": content})
assert "[truncated]" not in prompt
def test_file_one_over_limit_is_truncated(self):
content = "x" * (MAX_FILE_CHARS + 1)
prompt = SystemPrompter.generate_prompt(file_context={"name": "f.py", "content": content})
assert "[truncated]" in prompt