362 lines
13 KiB
Python
362 lines
13 KiB
Python
"""Tests for SearchManager — no real network calls, all I/O mocked."""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, patch
|
|
|
|
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")
|