419 lines
15 KiB
Python
419 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Local test suite for the search engine (no database required).
|
|
|
|
Only the pure, DB-free parts are tested: parameter normalization,
|
|
SQL assembly and the query-parsing helpers. Database-backed behaviour
|
|
is covered by the statement timeouts in production.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
import unittest
|
|
from datetime import timedelta, timezone
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
|
|
def _load_env_file():
|
|
"""
|
|
Load ../.env like the systemd EnvironmentFile does, so the tests
|
|
also run outside systemd. Must happen before engine.py is imported,
|
|
which reads the JONASSEARCH_* variables at import time.
|
|
"""
|
|
env_path = os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)),
|
|
os.pardir,
|
|
".env",
|
|
)
|
|
|
|
try:
|
|
with open(env_path) as handle:
|
|
for line in handle:
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
os.environ.setdefault(key.strip(), value.strip())
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
_load_env_file()
|
|
|
|
import engine # noqa: E402
|
|
|
|
|
|
class ParseRankWeightsTest(unittest.TestCase):
|
|
|
|
def test_defaults_map_to_pg_order(self):
|
|
# ts_rank_cd erwartet [D, C, B, A]
|
|
self.assertEqual(
|
|
engine.parse_rank_weights(None),
|
|
[0.1, 0.3, 0.6, 1.0],
|
|
)
|
|
|
|
def test_custom_spec(self):
|
|
self.assertEqual(
|
|
engine.parse_rank_weights("A=1.0,B=0.5,C=0.25,D=0.05"),
|
|
[0.05, 0.25, 0.5, 1.0],
|
|
)
|
|
|
|
def test_partial_spec_keeps_defaults(self):
|
|
self.assertEqual(
|
|
engine.parse_rank_weights("B=0.9"),
|
|
[0.1, 0.3, 0.9, 1.0],
|
|
)
|
|
|
|
def test_invalid_key(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.parse_rank_weights("E=1.0")
|
|
|
|
def test_invalid_value(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.parse_rank_weights("A=xyz")
|
|
|
|
def test_missing_equals(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.parse_rank_weights("A 1.0")
|
|
|
|
|
|
class NormalizeSitesTest(unittest.TestCase):
|
|
|
|
def test_plain(self):
|
|
self.assertEqual(engine.normalize_sites(["example.com"]),
|
|
["example.com"])
|
|
|
|
def test_url_and_www(self):
|
|
self.assertEqual(
|
|
engine.normalize_sites(["https://www.Example.com/pfad?x=1"]),
|
|
["example.com"],
|
|
)
|
|
|
|
def test_wildcard(self):
|
|
self.assertEqual(engine.normalize_sites("*.example.com"),
|
|
["example.com"])
|
|
|
|
def test_dedup_and_multiple(self):
|
|
self.assertEqual(
|
|
engine.normalize_sites(["example.com", "EXAMPLE.com", "de.example.org"]),
|
|
["example.com", "de.example.org"],
|
|
)
|
|
|
|
def test_empty(self):
|
|
self.assertEqual(engine.normalize_sites(None), [])
|
|
self.assertEqual(engine.normalize_sites(["", " "]), [])
|
|
|
|
def test_invalid_raises(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.normalize_sites(["not a domain!"])
|
|
|
|
|
|
class NormalizeTldsTest(unittest.TestCase):
|
|
|
|
def test_plain_gets_leading_dot(self):
|
|
# Der Crawler speichert die TLD mit führendem Punkt ('.de').
|
|
self.assertEqual(engine.normalize_tlds(["de"]), [".de"])
|
|
|
|
def test_dot_and_case(self):
|
|
self.assertEqual(engine.normalize_tlds([".DE", "at"]), [".de", ".at"])
|
|
|
|
def test_invalid_raises(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.normalize_tlds(["de/)"])
|
|
|
|
def test_empty(self):
|
|
self.assertEqual(engine.normalize_tlds(None), [])
|
|
|
|
|
|
class NormalizeSortTest(unittest.TestCase):
|
|
|
|
def test_defaults(self):
|
|
self.assertEqual(engine.normalize_sort(None), "relevance")
|
|
self.assertEqual(engine.normalize_sort("relevance"), "relevance")
|
|
self.assertEqual(engine.normalize_sort("freshness"), "freshness")
|
|
|
|
def test_aliases(self):
|
|
self.assertEqual(engine.normalize_sort("date"), "freshness")
|
|
self.assertEqual(engine.normalize_sort("score"), "relevance")
|
|
|
|
def test_invalid(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.normalize_sort("banana")
|
|
|
|
|
|
class ParseTimeBoundTest(unittest.TestCase):
|
|
|
|
def test_empty_is_none(self):
|
|
self.assertIsNone(engine.parse_time_bound(None))
|
|
self.assertIsNone(engine.parse_time_bound(""))
|
|
|
|
def test_relative_days(self):
|
|
before = engine.parse_time_bound("7d", "after")
|
|
now = datetime_now()
|
|
delta = now - before
|
|
self.assertGreater(delta.total_seconds(), 7 * 86400 - 5)
|
|
self.assertLess(delta.total_seconds(), 7 * 86400 + 5)
|
|
|
|
def test_relative_hours_and_weeks(self):
|
|
before = engine.parse_time_bound("24h", "after")
|
|
self.assertAlmostEqual(
|
|
(datetime_now() - before).total_seconds(), 86400, delta=5
|
|
)
|
|
before = engine.parse_time_bound("2w", "after")
|
|
self.assertAlmostEqual(
|
|
(datetime_now() - before).total_seconds(), 14 * 86400, delta=5
|
|
)
|
|
|
|
def test_relative_zero_raises(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.parse_time_bound("0d")
|
|
|
|
def test_iso_date_becomes_utc(self):
|
|
dt = engine.parse_time_bound("2026-01-31", "after")
|
|
self.assertEqual(dt.tzinfo, timezone.utc)
|
|
self.assertEqual((dt.year, dt.month, dt.day), (2026, 1, 31))
|
|
|
|
def test_iso_datetime_with_z(self):
|
|
dt = engine.parse_time_bound("2026-01-31T12:00:00Z", "before")
|
|
self.assertEqual(dt.utcoffset(), timedelta(0))
|
|
self.assertEqual(dt.hour, 12)
|
|
|
|
def test_naive_datetime_becomes_utc(self):
|
|
dt = engine.parse_time_bound("2026-01-31 12:00:00", "after")
|
|
self.assertEqual(dt.tzinfo, timezone.utc)
|
|
|
|
def test_invalid_raises(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.parse_time_bound("gestern")
|
|
|
|
|
|
def datetime_now():
|
|
return engine.datetime.now(timezone.utc)
|
|
|
|
|
|
class PrefixLexemeTest(unittest.TestCase):
|
|
|
|
def test_last_word(self):
|
|
self.assertEqual(engine.prefix_lexeme("hallo welt"), "welt:*")
|
|
|
|
def test_lowercased(self):
|
|
self.assertEqual(engine.prefix_lexeme("SuchMASCHINE"), "suchmaschine:*")
|
|
|
|
def test_skips_short_tokens(self):
|
|
self.assertEqual(engine.prefix_lexeme("a b suchbegriff"),
|
|
"suchbegriff:*")
|
|
|
|
def test_punctuation_only(self):
|
|
self.assertIsNone(engine.prefix_lexeme("!!! ..."))
|
|
|
|
def test_umlauts(self):
|
|
self.assertEqual(engine.prefix_lexeme("größen"), "größen:*")
|
|
|
|
def test_empty(self):
|
|
self.assertIsNone(engine.prefix_lexeme(""))
|
|
self.assertIsNone(engine.prefix_lexeme(None))
|
|
|
|
|
|
class FilterClausesTest(unittest.TestCase):
|
|
|
|
def test_defaults_filter_errors_and_non_html(self):
|
|
sql, params = engine._filter_clauses([], [], None, None,
|
|
False, False)
|
|
self.assertIn("p.http_status = 200", sql)
|
|
self.assertIn("text/html%", sql)
|
|
self.assertEqual(params, {})
|
|
|
|
def test_include_errors(self):
|
|
sql, _ = engine._filter_clauses([], [], None, None, True, False)
|
|
self.assertNotIn("http_status", sql)
|
|
|
|
def test_sites_clause(self):
|
|
sql, params = engine._filter_clauses(
|
|
engine.normalize_sites(["example.com"]), [], None, None,
|
|
False, False,
|
|
)
|
|
self.assertIn("p.domain = ANY(%(sites)s)", sql)
|
|
self.assertIn("p.domain LIKE ANY(%(site_pats)s)", sql)
|
|
self.assertEqual(params["sites"], ["example.com"])
|
|
self.assertEqual(params["site_pats"], ["%.example.com"])
|
|
|
|
def test_time_bounds(self):
|
|
after = engine.datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
before = engine.datetime(2026, 2, 1, tzinfo=timezone.utc)
|
|
sql, params = engine._filter_clauses(
|
|
[], [], after, before, False, False,
|
|
)
|
|
self.assertIn("COALESCE(p.crawled_at, p.discovered_at) >= %(after)s",
|
|
sql)
|
|
self.assertIn("COALESCE(p.crawled_at, p.discovered_at) < %(before)s",
|
|
sql)
|
|
self.assertEqual(params["after"], after)
|
|
self.assertEqual(params["before"], before)
|
|
|
|
|
|
class SearchSqlTest(unittest.TestCase):
|
|
|
|
def test_relevance_ordering(self):
|
|
sql = engine._search_sql("search_vector", "relevance", "")
|
|
self.assertIn(
|
|
"ts_rank_cd(%(weights)s::real[], p.search_vector, q.tsq,",
|
|
sql,
|
|
)
|
|
self.assertIn("%(norm)s::integer", sql)
|
|
self.assertIn("(r.text_rank * f.fresh) DESC", sql)
|
|
self.assertIn("websearch_to_tsquery(%(ts_config)s, %(query)s)", sql)
|
|
self.assertIn("LIMIT %(limit)s OFFSET %(offset)s", sql)
|
|
# content darf im Ranking-Query nicht gelesen werden (TOAST-Schonung)
|
|
self.assertIsNone(re.search(r"\bp\.content\b", sql),
|
|
"content-Spalte im Ranking-Query gelesen")
|
|
|
|
def test_capped_fallback_sql(self):
|
|
sql = engine._search_sql("search_vector", "relevance", "",
|
|
capped=True)
|
|
self.assertIn("LIMIT %(rank_cap)s", sql)
|
|
self.assertIn("p.search_vector AS vector", sql)
|
|
self.assertIn("ts_rank_cd(%(weights)s::real[], p.vector,", sql)
|
|
self.assertIn("LIMIT %(limit)s OFFSET %(offset)s", sql)
|
|
# auch der gekappte Query darf kein content lesen
|
|
self.assertIsNone(re.search(r"\bp\.content\b", sql))
|
|
|
|
def test_freshness_ordering(self):
|
|
sql = engine._search_sql("search_vector", "freshness", " AND x")
|
|
self.assertIn(
|
|
"COALESCE(p.crawled_at, p.discovered_at) DESC NULLS LAST", sql
|
|
)
|
|
self.assertIn(" AND x", sql)
|
|
|
|
def test_vector_column_is_interpolated(self):
|
|
sql = engine._search_sql("my_vector", "relevance", "")
|
|
self.assertIn("p.my_vector", sql)
|
|
self.assertNotIn("p.search_vector", sql)
|
|
|
|
|
|
class CountSqlTest(unittest.TestCase):
|
|
|
|
def test_count_uses_filters(self):
|
|
sql = engine._count_sql("search_vector", " AND p.http_status = 200")
|
|
self.assertIn("count(*) AS n", sql)
|
|
self.assertIn("p.search_vector @@ q.tsq AND p.http_status = 200", sql)
|
|
|
|
|
|
class ImageSearchSqlTest(unittest.TestCase):
|
|
|
|
def test_image_sql_uses_pictures_and_per_page_limit(self):
|
|
sql = engine._image_search_sql("search_vector", "relevance", "")
|
|
self.assertIn("unnest(p.pictures)", sql)
|
|
self.assertIn("LIMIT %(images_per_source)s", sql)
|
|
self.assertIn("i.image_url", sql)
|
|
self.assertIn("p.url AS page_url", sql)
|
|
self.assertIn("(r.text_rank * f.fresh) DESC", sql)
|
|
self.assertIn("LIMIT %(limit)s OFFSET %(offset)s", sql)
|
|
# Ranking-Query darf content nicht lesen (TOAST-Schonung).
|
|
self.assertIsNone(re.search(r"\bp\.content\b", sql))
|
|
|
|
def test_image_sql_capped_fallback(self):
|
|
sql = engine._image_search_sql(
|
|
"search_vector", "relevance", "", capped=True,
|
|
)
|
|
self.assertIn("LIMIT %(rank_cap)s", sql)
|
|
self.assertIn("p.pictures", sql)
|
|
self.assertIn("LIMIT %(limit)s OFFSET %(offset)s", sql)
|
|
|
|
def test_image_sql_freshness_order(self):
|
|
sql = engine._image_search_sql("search_vector", "freshness", "")
|
|
self.assertIn(
|
|
"COALESCE(p.crawled_at, p.discovered_at) DESC NULLS LAST",
|
|
sql,
|
|
)
|
|
|
|
def test_image_sql_uses_filters(self):
|
|
sql = engine._image_search_sql(
|
|
"search_vector", "relevance", " AND p.http_status = 200",
|
|
)
|
|
self.assertIn(" AND p.http_status = 200", sql)
|
|
|
|
def test_image_count_is_capped_per_source_page(self):
|
|
sql = engine._image_count_sql("search_vector", "")
|
|
self.assertIn("cardinality(p.pictures)", sql)
|
|
self.assertIn("%(images_per_source)s", sql)
|
|
self.assertIn("sum(", sql.lower())
|
|
|
|
|
|
class SetStatementTimeoutTest(unittest.TestCase):
|
|
|
|
def test_casts_to_int(self):
|
|
cur = mock.Mock()
|
|
engine._set_statement_timeout(cur, "4000")
|
|
cur.execute.assert_called_once_with("SET statement_timeout = 4000")
|
|
|
|
|
|
class SearchValidationTest(unittest.TestCase):
|
|
|
|
def test_empty_query_raises(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.search(" ")
|
|
|
|
def test_bad_page_raises(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.search("test", page="x")
|
|
|
|
def test_per_page_capped(self):
|
|
with mock.patch.object(engine, "db_connect") as db:
|
|
conn = mock.MagicMock()
|
|
db.return_value = conn
|
|
# 1. fetchone: geparste tsquery, 2. fetchone: Trefferzahl
|
|
conn.cursor.return_value.__enter__.return_value\
|
|
.fetchone.side_effect = [{"tsq": "test"}, {"n": 0}]
|
|
conn.cursor.return_value.__enter__.return_value\
|
|
.fetchall.return_value = []
|
|
with mock.patch.object(engine, "resolve_vector_column",
|
|
return_value="search_vector"):
|
|
data = engine.search("test", per_page=10000)
|
|
self.assertEqual(data["per_page"], engine.MAX_PER_PAGE)
|
|
|
|
def test_query_error_for_bad_filters(self):
|
|
with mock.patch.object(engine, "db_connect") as db:
|
|
db.return_value = mock.MagicMock()
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.search("test", sites=["ungültig!"])
|
|
|
|
|
|
class ImageSearchValidationTest(unittest.TestCase):
|
|
|
|
def test_empty_query_raises(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.search_images(" ")
|
|
|
|
def test_bad_page_raises(self):
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.search_images("test", page="x")
|
|
|
|
def test_query_error_for_bad_filters(self):
|
|
with mock.patch.object(engine, "db_connect") as db:
|
|
db.return_value = mock.MagicMock()
|
|
with self.assertRaises(engine.QueryError):
|
|
engine.search_images("test", sites=["ungültig!"])
|
|
|
|
def test_per_page_capped(self):
|
|
with mock.patch.object(engine, "db_connect") as db:
|
|
conn = mock.MagicMock()
|
|
db.return_value = conn
|
|
cur = conn.cursor.return_value.__enter__.return_value
|
|
cur.fetchone.side_effect = [{"tsq": "test"}, {"n": 0}]
|
|
cur.fetchall.return_value = []
|
|
with mock.patch.object(engine, "resolve_vector_column",
|
|
return_value="search_vector"):
|
|
data = engine.search_images("test", per_page=10000)
|
|
self.assertEqual(
|
|
data["per_page"], engine.MAX_IMAGE_RESULTS_PER_PAGE
|
|
)
|
|
self.assertFalse(data["has_more"])
|
|
self.assertEqual(data["results"], [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|