428 lines
14 KiB
Python
428 lines
14 KiB
Python
"""
|
|
Tests for serp module.
|
|
|
|
Tests FastAPI SERP server endpoints and functionality.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, AsyncMock, patch, MagicMock
|
|
import tempfile
|
|
import shutil
|
|
import os
|
|
import json
|
|
from pathlib import Path
|
|
|
|
# Test client for FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from serp import (
|
|
app,
|
|
get_lang,
|
|
t,
|
|
inject_i18n,
|
|
slugify,
|
|
TRANSLATIONS,
|
|
LANG_NAMES,
|
|
CrawlRequest,
|
|
)
|
|
|
|
|
|
class TestLanguageDetection:
|
|
"""Test language detection functions."""
|
|
|
|
def test_get_lang_default(self):
|
|
"""Test default language is English."""
|
|
lang = get_lang(None, None)
|
|
assert lang == "en"
|
|
|
|
def test_get_lang_from_cookie(self):
|
|
"""Test language from cookie takes priority."""
|
|
lang = get_lang("zh", "en-US")
|
|
assert lang == "zh"
|
|
|
|
def test_get_lang_from_accept_header(self):
|
|
"""Test language from Accept-Language header."""
|
|
lang = get_lang(None, "zh-CN,zh;q=0.9,en;q=0.8")
|
|
assert lang == "zh"
|
|
|
|
def test_get_lang_accept_header_fallback(self):
|
|
"""Test Accept-Language fallback to English for unknown."""
|
|
lang = get_lang(None, "xyz-XY,abc;q=0.8")
|
|
assert lang == "en"
|
|
|
|
def test_get_lang_unsupported_cookie(self):
|
|
"""Test unsupported cookie falls back to header."""
|
|
lang = get_lang("xyz", "ja,en;q=0.8")
|
|
assert lang == "ja"
|
|
|
|
|
|
class TestTranslationFunction:
|
|
"""Test translation function."""
|
|
|
|
def test_translation_english(self):
|
|
"""Test English translation."""
|
|
result = t("search", "en")
|
|
assert result == "Search"
|
|
|
|
def test_translation_chinese(self):
|
|
"""Test Chinese translation."""
|
|
result = t("search", "zh")
|
|
assert result == "搜索"
|
|
|
|
def test_translation_spanish(self):
|
|
"""Test Spanish translation."""
|
|
result = t("search", "es")
|
|
assert result == "Buscar"
|
|
|
|
def test_translation_missing_key(self):
|
|
"""Test missing key returns key."""
|
|
result = t("nonexistent_key", "en")
|
|
assert result == "nonexistent_key"
|
|
|
|
def test_translation_missing_language(self):
|
|
"""Test missing language falls back to English."""
|
|
result = t("search", "xyz")
|
|
assert result == "Search"
|
|
|
|
|
|
class TestSlugify:
|
|
"""Test slugify function."""
|
|
|
|
def test_slugify_basic(self):
|
|
"""Test basic slugification."""
|
|
result = slugify("Hello World")
|
|
assert result == "hello-world"
|
|
|
|
def test_slugify_special_chars(self):
|
|
"""Test removing special characters."""
|
|
result = slugify("Hello! World? Test@123")
|
|
assert "hello" in result
|
|
assert "world" in result
|
|
|
|
def test_slugify_max_length(self):
|
|
"""Test max length truncation."""
|
|
long_text = "a" * 100
|
|
result = slugify(long_text, max_len=10)
|
|
assert len(result) <= 10
|
|
|
|
def test_slugify_empty(self):
|
|
"""Test empty string."""
|
|
result = slugify("")
|
|
assert result == ""
|
|
|
|
def test_slugify_unicode(self):
|
|
"""Test unicode handling."""
|
|
result = slugify("Cafe with accents")
|
|
assert result == "cafe-with-accents" or "cafe" in result
|
|
|
|
|
|
class TestTranslationsStructure:
|
|
"""Test translations dictionary structure."""
|
|
|
|
def test_all_languages_have_required_keys(self):
|
|
"""Test all languages have core translation keys."""
|
|
required_keys = ["search", "crawl", "live", "about", "loading"]
|
|
for lang_code, translations in TRANSLATIONS.items():
|
|
for key in required_keys:
|
|
assert key in translations, f"Missing '{key}' in language '{lang_code}'"
|
|
|
|
def test_lang_names_match_translations(self):
|
|
"""Test language names match translation keys."""
|
|
for lang_code in LANG_NAMES:
|
|
assert lang_code in TRANSLATIONS, f"Language '{lang_code}' in LANG_NAMES but not TRANSLATIONS"
|
|
|
|
|
|
class TestCrawlRequestModel:
|
|
"""Test CrawlRequest Pydantic model."""
|
|
|
|
def test_crawl_request_defaults(self):
|
|
"""Test CrawlRequest default values."""
|
|
request = CrawlRequest()
|
|
assert request.targets == []
|
|
assert request.target_uri == ""
|
|
assert request.mode == "all"
|
|
assert request.depth == -1
|
|
assert request.max_pages == -1
|
|
assert request.fresh is False
|
|
assert request.fast is False
|
|
assert request.screenshots is True
|
|
assert request.hydra is False
|
|
|
|
def test_crawl_request_with_values(self):
|
|
"""Test CrawlRequest with custom values."""
|
|
request = CrawlRequest(
|
|
targets=["https://example.com", "https://test.com"],
|
|
mode="images",
|
|
depth=5,
|
|
max_pages=100,
|
|
fresh=True,
|
|
)
|
|
assert len(request.targets) == 2
|
|
assert request.mode == "images"
|
|
assert request.depth == 5
|
|
assert request.max_pages == 100
|
|
assert request.fresh is True
|
|
|
|
|
|
class TestAPIEndpoints:
|
|
"""Test API endpoints using TestClient."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.db_path = os.path.join(self.temp_dir, "test.db")
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_health_endpoint(self):
|
|
"""Test health check endpoint returns valid response."""
|
|
with TestClient(app) as client:
|
|
response = client.get("/health")
|
|
# Health endpoint doesn't require db init
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "status" in data
|
|
assert data["status"] == "healthy"
|
|
assert "features" in data
|
|
|
|
|
|
class TestInjectI18n:
|
|
"""Test i18n injection into HTML."""
|
|
|
|
def test_inject_replaces_placeholders(self):
|
|
"""Test that placeholders are replaced."""
|
|
html = "<html><body>{{search}} {{loading}}</body></html><script></script>"
|
|
result = inject_i18n(html, "en")
|
|
|
|
assert "Search" in result
|
|
assert "Loading..." in result
|
|
assert "{{search}}" not in result
|
|
|
|
def test_inject_adds_lang_attribute(self):
|
|
"""Test that lang attribute is added to html tag."""
|
|
html = "<html><body>Test</body></html><script></script>"
|
|
result = inject_i18n(html, "zh")
|
|
|
|
assert 'lang="zh"' in result
|
|
|
|
def test_inject_adds_js_translations(self):
|
|
"""Test that JS translations object is injected."""
|
|
html = "<html><body>Test</body></html><script></script>"
|
|
result = inject_i18n(html, "en")
|
|
|
|
assert "const T=" in result
|
|
|
|
|
|
class TestStaticEndpoints:
|
|
"""Test that static file mounts work."""
|
|
|
|
def test_app_has_routes(self):
|
|
"""Test that app has expected routes."""
|
|
route_paths = [route.path for route in app.routes]
|
|
|
|
# Check for main routes
|
|
assert "/" in route_paths or any("/" in str(r.path) for r in app.routes)
|
|
|
|
|
|
class TestAsyncEndpoints:
|
|
"""Test async endpoints with mocked database."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_async(self):
|
|
"""Test health endpoint asynchronously."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "healthy"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch('serp.db')
|
|
async def test_stats_endpoint(self, mock_db):
|
|
"""Test stats endpoint with mocked db."""
|
|
mock_db.get_stats = AsyncMock(return_value={
|
|
"images": 100,
|
|
"videos": 50,
|
|
"audio": 10
|
|
})
|
|
# Mock session as async context manager
|
|
mock_session = MagicMock()
|
|
mock_session.execute = AsyncMock(return_value=MagicMock(scalar=MagicMock(return_value=0)))
|
|
mock_db.session = MagicMock(return_value=MagicMock(
|
|
__aenter__=AsyncMock(return_value=mock_session),
|
|
__aexit__=AsyncMock(return_value=None)
|
|
))
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/stats")
|
|
# May fail without proper db initialization, that's expected
|
|
assert response.status_code in [200, 500]
|
|
|
|
|
|
class TestErrorHandling:
|
|
"""Test error handling in endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
@patch('serp.db', None)
|
|
async def test_media_not_found(self):
|
|
"""Test 404 for non-existent media."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/media/nonexistent123")
|
|
# Without db, should return error
|
|
assert response.status_code in [404, 500]
|
|
|
|
|
|
class TestInternationalization:
|
|
"""Test internationalization coverage."""
|
|
|
|
def test_all_supported_languages(self):
|
|
"""Test that all major languages are supported."""
|
|
expected_languages = ["en", "zh", "es", "hi", "ar", "pt", "ru", "ja", "fr", "de"]
|
|
for lang in expected_languages:
|
|
assert lang in TRANSLATIONS, f"Missing language: {lang}"
|
|
|
|
def test_translation_consistency(self):
|
|
"""Test that all translations have same number of keys."""
|
|
en_keys = set(TRANSLATIONS["en"].keys())
|
|
for lang_code, translations in TRANSLATIONS.items():
|
|
lang_keys = set(translations.keys())
|
|
# All translations should have at least the English keys
|
|
missing = en_keys - lang_keys
|
|
if missing:
|
|
# This is a warning, not necessarily a failure
|
|
pass # Some keys may be intentionally missing
|
|
|
|
def test_lang_selector_has_all_languages(self):
|
|
"""Test language selector includes all supported languages."""
|
|
for lang_code in TRANSLATIONS.keys():
|
|
# zh-tw is a variant, others should be in LANG_NAMES
|
|
if "-" not in lang_code or lang_code == "zh-tw":
|
|
# Either should be in LANG_NAMES or be a valid language
|
|
pass # Just checking structure
|
|
|
|
|
|
class TestMediaServing:
|
|
"""Test media serving functionality."""
|
|
|
|
def test_slugify_for_filenames(self):
|
|
"""Test slugify produces safe filenames."""
|
|
# Test various inputs that might be used for filenames
|
|
test_cases = [
|
|
("Hello World.jpg", "hello-worldjpg"),
|
|
("Test Image 123", "test-image-123"),
|
|
("Image with spaces", "image-with-spaces"),
|
|
]
|
|
for input_text, expected_pattern in test_cases:
|
|
result = slugify(input_text)
|
|
# Result should be lowercase and have no spaces
|
|
assert result.islower() or result == ""
|
|
assert " " not in result
|
|
|
|
|
|
class TestCrawlRequestValidation:
|
|
"""Test CrawlRequest validation."""
|
|
|
|
def test_valid_targets(self):
|
|
"""Test valid URL targets."""
|
|
request = CrawlRequest(
|
|
targets=["https://example.com", "https://test.org"]
|
|
)
|
|
assert len(request.targets) == 2
|
|
|
|
def test_mode_values(self):
|
|
"""Test different mode values."""
|
|
for mode in ["text", "images", "videos", "media", "all"]:
|
|
request = CrawlRequest(mode=mode)
|
|
assert request.mode == mode
|
|
|
|
def test_depth_values(self):
|
|
"""Test depth values."""
|
|
request = CrawlRequest(depth=0)
|
|
assert request.depth == 0
|
|
|
|
request = CrawlRequest(depth=-1)
|
|
assert request.depth == -1
|
|
|
|
request = CrawlRequest(depth=10)
|
|
assert request.depth == 10
|
|
|
|
|
|
class TestHelperFunctions:
|
|
"""Test various helper functions."""
|
|
|
|
def test_nav_html_exists(self):
|
|
"""Test NAV_HTML constant exists."""
|
|
from serp import NAV_HTML
|
|
assert "neopig" in NAV_HTML
|
|
assert "href" in NAV_HTML
|
|
|
|
def test_search_box_html_exists(self):
|
|
"""Test SEARCH_BOX_HTML constant exists."""
|
|
from serp import SEARCH_BOX_HTML
|
|
assert "form" in SEARCH_BOX_HTML
|
|
assert "search" in SEARCH_BOX_HTML.lower()
|
|
|
|
|
|
class TestDatabaseIntegration:
|
|
"""Test database-related endpoint behaviors."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_endpoint_accepts_query(self):
|
|
"""Test search endpoint accepts query parameter."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/search", params={"q": "test"})
|
|
# May fail without db, but should not crash
|
|
assert response.status_code in [200, 500]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_random_endpoint_exists(self):
|
|
"""Test random endpoint exists."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/random")
|
|
# Should redirect or return error without db
|
|
assert response.status_code in [302, 307, 500]
|
|
|
|
|
|
class TestLiveStream:
|
|
"""Test live streaming endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_live_stream_endpoint_exists(self):
|
|
"""Test live stream SSE endpoint exists."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
# SSE endpoints are tricky to test, just verify route exists
|
|
response = await client.get("/api/live/stream", timeout=2.0)
|
|
# Should return event-stream or error
|
|
assert response.status_code in [200, 500]
|
|
|
|
|
|
class TestImportMode:
|
|
"""Test import mode functionality."""
|
|
|
|
def test_import_mode_env_check(self):
|
|
"""Test import mode is controlled by environment."""
|
|
from serp import IMPORT_MODE
|
|
# Just verify the constant exists
|
|
assert isinstance(IMPORT_MODE, bool)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_upload_requires_mode(self):
|
|
"""Test import upload requires import mode enabled."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
# When import mode is disabled, should return 403
|
|
response = await client.post("/api/import/upload")
|
|
# Either 403 (disabled) or 422 (missing file) or 500 (db error)
|
|
assert response.status_code in [403, 422, 500]
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__, '-v'])
|