299 lines
8.1 KiB
Python
299 lines
8.1 KiB
Python
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("backend.agent.servers.mcp_server_web_search.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("backend.agent.servers.mcp_server_web_search.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")
|
||
|
||
# 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
|
||
|
||
|
||
# ---------------------------------------------------------
|
||
# 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 "HTTP-Fehler" in result or "Error" 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("backend.agent.servers.mcp_server_web_search.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 |