Merge branch 'MCP-setup' of https://gitea.fhgr.ch/meulilivio/AISE1_Project into MCP-setup
This commit is contained in:
commit
b98b74dad1
@ -240,10 +240,10 @@ def render_filetree_arborist(tree):
|
|||||||
selected = tree_view(
|
selected = tree_view(
|
||||||
data=data,
|
data=data,
|
||||||
icons={"open": "📂", "closed": "📁"},
|
icons={"open": "📂", "closed": "📁"},
|
||||||
height=200,
|
height=400,
|
||||||
selection=None,
|
selection=None,
|
||||||
select_internal_nodes=True, # allow clicking folder names, not just files
|
select_internal_nodes=True, # allow clicking folder names, not just files
|
||||||
open_by_default=False
|
open_by_default=True
|
||||||
)
|
)
|
||||||
|
|
||||||
return selected
|
return selected
|
||||||
|
|||||||
@ -0,0 +1,278 @@
|
|||||||
|
import pytest
|
||||||
|
from pathlib import Path
|
||||||
|
from backend.agent.servers import mcp_server_file_search as server
|
||||||
|
|
||||||
|
|
||||||
|
# =========================================================
|
||||||
|
# FIXTURES
|
||||||
|
# =========================================================
|
||||||
|
|
||||||
|
@pytest.fixture()
|
||||||
|
def workspace(tmp_path, monkeypatch):
|
||||||
|
"""
|
||||||
|
Erstellt einen isolierten Workspace für jeden Test.
|
||||||
|
"""
|
||||||
|
ws = tmp_path / "workspace"
|
||||||
|
ws.mkdir()
|
||||||
|
|
||||||
|
monkeypatch.setattr(server, "ALLOWED_DIR", ws)
|
||||||
|
|
||||||
|
return ws
|
||||||
|
|
||||||
|
|
||||||
|
# =========================================================
|
||||||
|
# BASIC TESTS (1–10)
|
||||||
|
# =========================================================
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 1. _safe_path erlaubt gültige Pfade
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_safe_path_valid(workspace):
|
||||||
|
result = server._safe_path("test.txt")
|
||||||
|
|
||||||
|
assert result == workspace / "test.txt"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 2. _safe_path blockiert Path Traversal
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_safe_path_blocks_traversal(workspace):
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
server._safe_path("../secret.txt")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 3. list_files liefert leeren Hinweis
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_list_files_empty(workspace):
|
||||||
|
result = server.list_files()
|
||||||
|
|
||||||
|
assert result == "No files found in the project directory."
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 4. list_files findet Dateien rekursiv
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_list_files_recursive(workspace):
|
||||||
|
src = workspace / "src"
|
||||||
|
src.mkdir()
|
||||||
|
|
||||||
|
(src / "main.py").write_text("print('hello')")
|
||||||
|
|
||||||
|
result = server.list_files()
|
||||||
|
|
||||||
|
assert "src/main.py" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 5. read_file liest Datei korrekt
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_read_file_success(workspace):
|
||||||
|
file = workspace / "hello.txt"
|
||||||
|
file.write_text("Hello World")
|
||||||
|
|
||||||
|
result = server.read_file("hello.txt")
|
||||||
|
|
||||||
|
assert result == "Hello World"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 6. read_file erkennt fehlende Datei
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_read_file_missing(workspace):
|
||||||
|
result = server.read_file("missing.txt")
|
||||||
|
|
||||||
|
assert "does not exist" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 7. write_new_file erstellt Datei
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_write_new_file_success(workspace):
|
||||||
|
result = server.write_new_file("new.txt", "content")
|
||||||
|
|
||||||
|
assert "OK:" in result
|
||||||
|
assert (workspace / "new.txt").exists()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 8. write_new_file verhindert Überschreiben
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_write_new_file_existing(workspace):
|
||||||
|
file = workspace / "exists.txt"
|
||||||
|
file.write_text("old")
|
||||||
|
|
||||||
|
result = server.write_new_file("exists.txt", "new")
|
||||||
|
|
||||||
|
assert "already exists" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 9. create_new_directory erstellt Verzeichnis
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_create_new_directory_success(workspace):
|
||||||
|
result = server.create_new_directory("mydir")
|
||||||
|
|
||||||
|
assert "OK:" in result
|
||||||
|
assert (workspace / "mydir").is_dir()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 10. search_files findet Inhalte
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_search_files_content_match(workspace):
|
||||||
|
file = workspace / "notes.txt"
|
||||||
|
file.write_text("Python MCP Server")
|
||||||
|
|
||||||
|
result = server.search_files("mcp")
|
||||||
|
|
||||||
|
assert "[content]" in result
|
||||||
|
|
||||||
|
|
||||||
|
# =========================================================
|
||||||
|
# EDGE CASE TESTS (11–20)
|
||||||
|
# =========================================================
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 11. Mehrfaches Traversal blockieren
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_safe_path_double_traversal(workspace):
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
server._safe_path("../../../../etc/passwd")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 12. Symlink Escape verhindern
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_safe_path_symlink_escape(workspace):
|
||||||
|
outside = workspace.parent / "outside"
|
||||||
|
outside.mkdir()
|
||||||
|
|
||||||
|
target = outside / "evil.txt"
|
||||||
|
target.write_text("bad")
|
||||||
|
|
||||||
|
link = workspace / "link"
|
||||||
|
link.symlink_to(outside)
|
||||||
|
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
server._safe_path("link/evil.txt")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 13. Dateien ohne Extension blockieren
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_write_file_without_extension(workspace):
|
||||||
|
result = server.write_new_file("README", "test")
|
||||||
|
|
||||||
|
assert "can only write" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 14. Hidden Files blockieren
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_write_hidden_file(workspace):
|
||||||
|
result = server.write_new_file(".env", "SECRET=123")
|
||||||
|
|
||||||
|
assert "can only write" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 15. Binary Files korrekt behandeln
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_read_binary_file(workspace):
|
||||||
|
binary = workspace / "data.bin"
|
||||||
|
binary.write_bytes(b"\xFF\xFE\xFD")
|
||||||
|
|
||||||
|
result = server.read_file("data.bin")
|
||||||
|
|
||||||
|
assert "not a text file" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 16. Sehr große Zeilen durchsuchen
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_search_huge_line(workspace):
|
||||||
|
huge_text = "A" * 1_000_000 + "needle"
|
||||||
|
|
||||||
|
file = workspace / "huge.txt"
|
||||||
|
file.write_text(huge_text)
|
||||||
|
|
||||||
|
result = server.search_files("needle")
|
||||||
|
|
||||||
|
assert "[content]" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 17. Leere Dateien lesen
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_read_empty_file(workspace):
|
||||||
|
file = workspace / "empty.txt"
|
||||||
|
file.write_text("")
|
||||||
|
|
||||||
|
result = server.read_file("empty.txt")
|
||||||
|
|
||||||
|
assert result == ""
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 18. Sonderzeichen im Query
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_search_special_characters(workspace):
|
||||||
|
file = workspace / "test.txt"
|
||||||
|
file.write_text("hello [world] (test)")
|
||||||
|
|
||||||
|
result = server.search_files("[world]")
|
||||||
|
|
||||||
|
assert "[content]" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 19. Unicode-Dateinamen unterstützen
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_write_unicode_filename(workspace):
|
||||||
|
filename = "🔥_überraschung.txt"
|
||||||
|
|
||||||
|
result = server.write_new_file(filename, "unicode")
|
||||||
|
|
||||||
|
assert "OK:" in result
|
||||||
|
assert (workspace / filename).exists()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 20. Tiefe Verzeichnisstrukturen
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_list_files_deep_nesting(workspace):
|
||||||
|
current = workspace
|
||||||
|
|
||||||
|
for i in range(50):
|
||||||
|
current = current / f"dir_{i}"
|
||||||
|
current.mkdir()
|
||||||
|
|
||||||
|
file = current / "deep.txt"
|
||||||
|
file.write_text("deep")
|
||||||
|
|
||||||
|
result = server.list_files()
|
||||||
|
|
||||||
|
assert "deep.txt" in result
|
||||||
@ -0,0 +1,297 @@
|
|||||||
|
import pytest
|
||||||
|
from unittest.mock import Mock, patch
|
||||||
|
from backend.agent.servers import mcp_server_web_search as server
|
||||||
|
|
||||||
|
|
||||||
|
# =========================================================
|
||||||
|
# BASIC TESTS (1–10)
|
||||||
|
# =========================================================
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 1. _validate_url erlaubt HTTPS
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_validate_url_https():
|
||||||
|
url = "https://example.com"
|
||||||
|
|
||||||
|
result = server._validate_url(url)
|
||||||
|
|
||||||
|
assert result == url
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 2. _validate_url erlaubt HTTP
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_validate_url_http():
|
||||||
|
url = "http://example.com"
|
||||||
|
|
||||||
|
result = server._validate_url(url)
|
||||||
|
|
||||||
|
assert result == url
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 3. _validate_url blockiert localhost
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_validate_url_localhost():
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
server._validate_url("http://localhost/admin")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 4. _validate_url blockiert 127.0.0.1
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_validate_url_loopback():
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
server._validate_url("http://127.0.0.1")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 5. _validate_url blockiert private IP
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_validate_url_private_ip():
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
server._validate_url("http://192.168.1.10")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 6. web_search liefert Suchergebnisse
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
@patch("ddgs.DDGS")
|
||||||
|
def test_web_search_success(mock_ddgs):
|
||||||
|
mock_instance = Mock()
|
||||||
|
|
||||||
|
mock_instance.text.return_value = [
|
||||||
|
{
|
||||||
|
"title": "Example",
|
||||||
|
"href": "https://example.com",
|
||||||
|
"body": "Example snippet"
|
||||||
|
}
|
||||||
|
]
|
||||||
|
|
||||||
|
mock_ddgs.return_value = mock_instance
|
||||||
|
|
||||||
|
result = server.web_search("example")
|
||||||
|
|
||||||
|
assert "Title: Example" in result
|
||||||
|
assert "https://example.com" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 7. web_search ohne Ergebnisse
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
@patch("ddgs.DDGS")
|
||||||
|
def test_web_search_no_results(mock_ddgs):
|
||||||
|
mock_instance = Mock()
|
||||||
|
mock_instance.text.return_value = []
|
||||||
|
|
||||||
|
mock_ddgs.return_value = mock_instance
|
||||||
|
|
||||||
|
result = server.web_search("nothing")
|
||||||
|
|
||||||
|
assert "No results found" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 8. fetch_page lädt HTML
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
@patch("requests.get")
|
||||||
|
def test_fetch_page_success(mock_get):
|
||||||
|
response = Mock()
|
||||||
|
|
||||||
|
response.status_code = 200
|
||||||
|
response.text = """
|
||||||
|
<html>
|
||||||
|
<body>
|
||||||
|
<h1>Hello World</h1>
|
||||||
|
</body>
|
||||||
|
</html>
|
||||||
|
"""
|
||||||
|
|
||||||
|
mock_get.return_value = response
|
||||||
|
|
||||||
|
result = server.fetch_page("https://example.com")
|
||||||
|
|
||||||
|
assert "Hello World" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 9. fetch_page entfernt script Tags
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
@patch("requests.get")
|
||||||
|
def test_fetch_page_removes_script(mock_get):
|
||||||
|
response = Mock()
|
||||||
|
|
||||||
|
response.status_code = 200
|
||||||
|
response.text = """
|
||||||
|
<html>
|
||||||
|
<script>alert('xss')</script>
|
||||||
|
<body>Hello</body>
|
||||||
|
</html>
|
||||||
|
"""
|
||||||
|
|
||||||
|
mock_get.return_value = response
|
||||||
|
|
||||||
|
result = server.fetch_page("https://example.com")
|
||||||
|
|
||||||
|
assert "alert" not in result
|
||||||
|
assert "Hello" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 10. fetch_page erkennt HTTP Fehler
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
@patch("requests.get")
|
||||||
|
def test_fetch_page_http_error(mock_get):
|
||||||
|
response = Mock()
|
||||||
|
|
||||||
|
response.status_code = 404
|
||||||
|
response.text = "Not Found"
|
||||||
|
|
||||||
|
mock_get.return_value = response
|
||||||
|
|
||||||
|
result = server.fetch_page("https://example.com")
|
||||||
|
|
||||||
|
assert "HTTP error 404" in result
|
||||||
|
|
||||||
|
|
||||||
|
# =========================================================
|
||||||
|
# EDGE CASE TESTS (11–20)
|
||||||
|
# =========================================================
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 11. Blockiere file:// SSRF
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_validate_url_blocks_file_scheme():
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
server._validate_url("file:///etc/passwd")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 12. Blockiere ftp://
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_validate_url_blocks_ftp():
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
server._validate_url("ftp://example.com")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 13. Blockiere AWS Metadata Endpoint
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_validate_url_blocks_metadata_ip():
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
server._validate_url("http://169.254.169.254")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 14. Blockiere internes Docker Netzwerk
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_validate_url_blocks_docker_network():
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
server._validate_url("http://172.20.0.5")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 15. Sehr lange URL
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
def test_validate_url_very_long():
|
||||||
|
long_url = "https://example.com/" + ("a" * 5000)
|
||||||
|
|
||||||
|
result = server._validate_url(long_url)
|
||||||
|
|
||||||
|
assert result == long_url
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 16. fetch_page behandelt Timeout
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
@patch("requests.get")
|
||||||
|
def test_fetch_page_timeout(mock_get):
|
||||||
|
import requests
|
||||||
|
|
||||||
|
mock_get.side_effect = requests.Timeout("timeout")
|
||||||
|
|
||||||
|
result = server.fetch_page("https://example.com")
|
||||||
|
|
||||||
|
assert "Error fetching page" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 17. fetch_page behandelt Connection Error
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
@patch("requests.get")
|
||||||
|
def test_fetch_page_connection_error(mock_get):
|
||||||
|
import requests
|
||||||
|
|
||||||
|
mock_get.side_effect = requests.ConnectionError("connection failed")
|
||||||
|
|
||||||
|
result = server.fetch_page("https://example.com")
|
||||||
|
|
||||||
|
assert "Error fetching page" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 18. fetch_page truncatet große Seiten
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
@patch("requests.get")
|
||||||
|
def test_fetch_page_truncates_large_content(mock_get):
|
||||||
|
response = Mock()
|
||||||
|
|
||||||
|
response.status_code = 200
|
||||||
|
response.text = "<html><body>" + ("A" * 10000) + "</body></html>"
|
||||||
|
|
||||||
|
mock_get.return_value = response
|
||||||
|
|
||||||
|
result = server.fetch_page("https://example.com")
|
||||||
|
|
||||||
|
assert "[... truncated ...]" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 19. fetch_page bei leerem Body
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
@patch("requests.get")
|
||||||
|
def test_fetch_page_empty_content(mock_get):
|
||||||
|
response = Mock()
|
||||||
|
|
||||||
|
response.status_code = 200
|
||||||
|
response.text = "<html></html>"
|
||||||
|
|
||||||
|
mock_get.return_value = response
|
||||||
|
|
||||||
|
result = server.fetch_page("https://example.com")
|
||||||
|
|
||||||
|
assert "no text content found" in result.lower()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
# 20. web_search behandelt Exception sauber
|
||||||
|
# ---------------------------------------------------------
|
||||||
|
|
||||||
|
@patch("ddgs.DDGS")
|
||||||
|
def test_web_search_exception(mock_ddgs):
|
||||||
|
mock_ddgs.side_effect = Exception("DDGS failed")
|
||||||
|
|
||||||
|
result = server.web_search("test")
|
||||||
|
|
||||||
|
assert "Search error" in result
|
||||||
Loading…
x
Reference in New Issue
Block a user