Metona Team
0468dd4e9d
Multi-instance failover, exponential-backoff retry, SQLite cache, batch mode, domain filter, cross-engine dedup, result sorting, CSV export, structured logging, enhanced Markdown conversion, 155 pytest tests, Gitea Actions CI
497 lines
15 KiB
Python
497 lines
15 KiB
Python
"""Tests for scripts/search.py — pure-logic functions (no network).
|
|
|
|
Covers: parse_instances (URL normalization, lists, whitespace),
|
|
_normalize_csv, filter_results_by_domain (include/exclude, www, case,
|
|
precedence), _read_queries_file (comments, blanks, missing file),
|
|
_cfg_int (str/int/absent/bad), _build_params (param construction,
|
|
time_range=none exclusion), and _merge_headers.
|
|
"""
|
|
import pytest
|
|
from search import (
|
|
_build_params,
|
|
_cfg_int,
|
|
_format_results,
|
|
_merge_headers,
|
|
_normalize_csv,
|
|
_read_queries_file,
|
|
deduplicate_results,
|
|
filter_results_by_domain,
|
|
load_config,
|
|
parse_instances,
|
|
sort_results,
|
|
)
|
|
|
|
|
|
# ----- parse_instances -----
|
|
|
|
def test_parse_instances_single():
|
|
assert parse_instances("https://example.com") == ["https://example.com"]
|
|
|
|
|
|
def test_parse_instances_adds_https_prefix():
|
|
assert parse_instances("example.com") == ["https://example.com"]
|
|
|
|
|
|
def test_parse_instances_strips_trailing_slash():
|
|
assert parse_instances("https://example.com/") == ["https://example.com"]
|
|
|
|
|
|
def test_parse_instances_multiple_comma():
|
|
assert parse_instances("a.com,b.com") == ["https://a.com", "https://b.com"]
|
|
|
|
|
|
def test_parse_instances_handles_whitespace():
|
|
assert parse_instances(" a.com , b.com ") == ["https://a.com", "https://b.com"]
|
|
|
|
|
|
def test_parse_instances_empty_string():
|
|
assert parse_instances("") == []
|
|
|
|
|
|
def test_parse_instances_preserves_http():
|
|
assert parse_instances("http://localhost:8080") == ["http://localhost:8080"]
|
|
|
|
|
|
def test_parse_instances_skips_empty_entries():
|
|
assert parse_instances("a.com,,b.com,") == ["https://a.com", "https://b.com"]
|
|
|
|
|
|
# ----- _normalize_csv -----
|
|
|
|
def test_normalize_csv_strips_spaces():
|
|
assert _normalize_csv("google, bing, brave") == "google,bing,brave"
|
|
|
|
|
|
def test_normalize_csv_drops_empty_parts():
|
|
assert _normalize_csv("google,,bing,") == "google,bing"
|
|
|
|
|
|
def test_normalize_csv_empty_input():
|
|
assert _normalize_csv("") == ""
|
|
|
|
|
|
# ----- filter_results_by_domain -----
|
|
|
|
def _results(*urls):
|
|
return {"results": [{"url": u, "title": u} for u in urls]}
|
|
|
|
|
|
def test_filter_no_args_returns_unchanged():
|
|
r = _results("https://a.com/1", "https://b.com/2")
|
|
out = filter_results_by_domain(dict(r))
|
|
assert len(out["results"]) == 2
|
|
|
|
|
|
def test_filter_include_allowlist():
|
|
r = _results("https://a.com/1", "https://b.com/2", "https://c.com/3")
|
|
out = filter_results_by_domain(dict(r), include_domains=["a.com"])
|
|
assert len(out["results"]) == 1
|
|
assert out["results"][0]["url"] == "https://a.com/1"
|
|
|
|
|
|
def test_filter_include_www_normalized():
|
|
"""www. prefix is stripped for matching, so 'a.com' matches 'www.a.com'."""
|
|
r = _results("https://www.a.com/1", "https://b.com/2")
|
|
out = filter_results_by_domain(dict(r), include_domains=["a.com"])
|
|
assert len(out["results"]) == 1
|
|
assert out["results"][0]["url"] == "https://www.a.com/1"
|
|
|
|
|
|
def test_filter_exclude_blocklist():
|
|
r = _results("https://a.com/1", "https://b.com/2")
|
|
out = filter_results_by_domain(dict(r), exclude_domains=["b.com"])
|
|
assert len(out["results"]) == 1
|
|
assert out["results"][0]["url"] == "https://a.com/1"
|
|
|
|
|
|
def test_filter_exclude_overrides_include():
|
|
"""When a domain is in BOTH lists, exclude wins (result dropped).
|
|
|
|
Rationale: include filters first (allowlist), then exclude filters the
|
|
survivors (blocklist). A domain listed in both is kept by include then
|
|
removed by exclude — exclude is the more explicit "do not want" intent.
|
|
"""
|
|
r = _results("https://a.com/1")
|
|
out = filter_results_by_domain(dict(r), include_domains=["a.com"],
|
|
exclude_domains=["a.com"])
|
|
assert len(out["results"]) == 0
|
|
|
|
|
|
def test_filter_case_insensitive():
|
|
r = _results("https://A.COM/1")
|
|
out = filter_results_by_domain(dict(r), include_domains=["a.com"])
|
|
assert len(out["results"]) == 1
|
|
|
|
|
|
def test_filter_empty_results_list():
|
|
out = filter_results_by_domain({"results": []}, include_domains=["a.com"])
|
|
assert out["results"] == []
|
|
|
|
|
|
def test_filter_no_results_key():
|
|
"""Missing 'results' key should not raise."""
|
|
out = filter_results_by_domain({}, include_domains=["a.com"])
|
|
assert out == {}
|
|
|
|
|
|
def test_filter_multiple_include():
|
|
r = _results("https://a.com/1", "https://b.com/2", "https://c.com/3")
|
|
out = filter_results_by_domain(dict(r), include_domains=["a.com", "c.com"])
|
|
assert len(out["results"]) == 2
|
|
|
|
|
|
# ----- _read_queries_file -----
|
|
|
|
def test_read_queries_file_basic(tmp_path):
|
|
f = tmp_path / "queries.txt"
|
|
f.write_text("query one\n# comment\n\nquery two\n", encoding="utf-8")
|
|
assert _read_queries_file(str(f)) == ["query one", "query two"]
|
|
|
|
|
|
def test_read_queries_file_missing_raises():
|
|
with pytest.raises(RuntimeError):
|
|
_read_queries_file("nonexistent_file.txt")
|
|
|
|
|
|
def test_read_queries_file_all_comments(tmp_path):
|
|
f = tmp_path / "empty.txt"
|
|
f.write_text("# only comments\n# another\n", encoding="utf-8")
|
|
assert _read_queries_file(str(f)) == []
|
|
|
|
|
|
def test_read_queries_file_strips_whitespace(tmp_path):
|
|
f = tmp_path / "q.txt"
|
|
f.write_text(" spaced query \n", encoding="utf-8")
|
|
assert _read_queries_file(str(f)) == ["spaced query"]
|
|
|
|
|
|
# ----- _cfg_int -----
|
|
|
|
def test_cfg_int_present_int_value():
|
|
assert _cfg_int({"timeout": 15}, "timeout", 30) == 15
|
|
|
|
|
|
def test_cfg_int_present_str_value():
|
|
"""TOML may carry the value as a string; _cfg_int must coerce."""
|
|
assert _cfg_int({"timeout": "15"}, "timeout", 30) == 15
|
|
|
|
|
|
def test_cfg_int_absent_returns_default():
|
|
assert _cfg_int({}, "timeout", 30) == 30
|
|
|
|
|
|
def test_cfg_int_bad_value_returns_default():
|
|
assert _cfg_int({"timeout": "abc"}, "timeout", 30) == 30
|
|
|
|
|
|
def test_cfg_int_none_default():
|
|
"""Defaults may be None (e.g. --retry); absent key must return None."""
|
|
assert _cfg_int({}, "max_retries", None) is None
|
|
|
|
|
|
# ----- _build_params -----
|
|
|
|
class _Args:
|
|
"""Minimal argparse.Namespace stand-in for _build_params tests."""
|
|
|
|
def __init__(self, **overrides):
|
|
self.categories = None
|
|
self.language = None
|
|
self.pageno = 1
|
|
self.time_range = "year"
|
|
self.safesearch = 0
|
|
self.engines = "google,bing"
|
|
for k, v in overrides.items():
|
|
setattr(self, k, v)
|
|
|
|
|
|
def test_build_params_minimal():
|
|
p = _build_params("hello", _Args())
|
|
assert p["q"] == "hello"
|
|
assert p["format"] == "json"
|
|
assert "categories" not in p
|
|
assert "language" not in p
|
|
assert p["engines"] == "google,bing"
|
|
|
|
|
|
def test_build_params_normalizes_categories():
|
|
p = _build_params("x", _Args(categories="general, news"))
|
|
assert p["categories"] == "general,news"
|
|
|
|
|
|
def test_build_params_time_range_none_excluded():
|
|
p = _build_params("x", _Args(time_range="none"))
|
|
assert "time_range" not in p
|
|
|
|
|
|
def test_build_params_pageno_as_string():
|
|
p = _build_params("x", _Args(pageno=3))
|
|
assert p["pageno"] == "3"
|
|
|
|
|
|
def test_build_params_safesearch_as_string():
|
|
p = _build_params("x", _Args(safesearch=1))
|
|
assert p["safesearch"] == "1"
|
|
|
|
|
|
# ----- _merge_headers -----
|
|
|
|
def test_merge_headers_basic():
|
|
assert _merge_headers({"a": 1}, {"b": 2}) == {"a": 1, "b": 2}
|
|
|
|
|
|
def test_merge_headers_later_overrides():
|
|
assert _merge_headers({"a": 1}, {"a": 2}) == {"a": 2}
|
|
|
|
|
|
def test_merge_headers_skips_none_dicts():
|
|
assert _merge_headers(None, {"a": 1}, None) == {"a": 1}
|
|
|
|
|
|
def test_merge_headers_all_none():
|
|
assert _merge_headers(None, None) == {}
|
|
|
|
|
|
# ----- deduplicate_results -----
|
|
|
|
def test_dedup_removes_exact_duplicate_url():
|
|
r = {"results": [
|
|
{"url": "https://a.com/1", "engine": "google", "score": 1.0},
|
|
{"url": "https://a.com/1", "engine": "bing", "score": 0.5},
|
|
]}
|
|
out = deduplicate_results(r)
|
|
assert len(out["results"]) == 1
|
|
assert out["results"][0]["engine"] == "google" # first kept
|
|
|
|
|
|
def test_dedup_strips_tracking_params():
|
|
"""utm_*, gclid, fbclid, etc. are stripped before comparison."""
|
|
r = {"results": [
|
|
{"url": "https://a.com/page?utm_source=x&id=1"},
|
|
{"url": "https://a.com/page?id=1&utm_medium=y"},
|
|
]}
|
|
out = deduplicate_results(r)
|
|
assert len(out["results"]) == 1
|
|
|
|
|
|
def test_dedup_strips_fragment():
|
|
r = {"results": [
|
|
{"url": "https://a.com/page#section1"},
|
|
{"url": "https://a.com/page#section2"},
|
|
]}
|
|
out = deduplicate_results(r)
|
|
assert len(out["results"]) == 1
|
|
|
|
|
|
def test_dedup_normalizes_scheme_host_case():
|
|
r = {"results": [
|
|
{"url": "HTTPS://Example.COM/path"},
|
|
{"url": "https://example.com/path"},
|
|
]}
|
|
out = deduplicate_results(r)
|
|
assert len(out["results"]) == 1
|
|
|
|
|
|
def test_dedup_normalizes_param_order():
|
|
r = {"results": [
|
|
{"url": "https://a.com/p?a=1&b=2"},
|
|
{"url": "https://a.com/p?b=2&a=1"},
|
|
]}
|
|
out = deduplicate_results(r)
|
|
assert len(out["results"]) == 1
|
|
|
|
|
|
def test_dedup_keeps_different_urls():
|
|
r = {"results": [
|
|
{"url": "https://a.com/1"},
|
|
{"url": "https://a.com/2"},
|
|
]}
|
|
out = deduplicate_results(r)
|
|
assert len(out["results"]) == 2
|
|
|
|
|
|
def test_dedup_keeps_url_less_results():
|
|
"""Results without a URL are never deduped (kept as-is)."""
|
|
r = {"results": [
|
|
{"title": "no url"},
|
|
{"title": "also no url"},
|
|
]}
|
|
out = deduplicate_results(r)
|
|
assert len(out["results"]) == 2
|
|
|
|
|
|
def test_dedup_empty_results():
|
|
out = deduplicate_results({"results": []})
|
|
assert out["results"] == []
|
|
|
|
|
|
def test_dedup_no_results_key():
|
|
out = deduplicate_results({})
|
|
assert out == {}
|
|
|
|
|
|
# ----- sort_results -----
|
|
|
|
def test_sort_by_score_descending():
|
|
r = {"results": [
|
|
{"url": "https://a.com", "score": 0.5},
|
|
{"url": "https://b.com", "score": 2.0},
|
|
{"url": "https://c.com", "score": 1.0},
|
|
]}
|
|
out = sort_results(r, "score")
|
|
assert [x["url"] for x in out["results"]] == [
|
|
"https://b.com", "https://c.com", "https://a.com"
|
|
]
|
|
|
|
|
|
def test_sort_by_score_none_at_end():
|
|
"""Entries without score keep relative order at the end."""
|
|
r = {"results": [
|
|
{"url": "https://a.com", "score": 1.0},
|
|
{"url": "https://b.com"}, # no score
|
|
{"url": "https://c.com", "score": 3.0},
|
|
{"url": "https://d.com"}, # no score
|
|
]}
|
|
out = sort_results(r, "score")
|
|
assert out["results"][0]["url"] == "https://c.com"
|
|
assert out["results"][1]["url"] == "https://a.com"
|
|
# no-score entries keep relative order: b before d
|
|
assert out["results"][2]["url"] == "https://b.com"
|
|
assert out["results"][3]["url"] == "https://d.com"
|
|
|
|
|
|
def test_sort_by_date_descending():
|
|
r = {"results": [
|
|
{"url": "https://a.com", "published_date": "2024-01-01"},
|
|
{"url": "https://b.com", "published_date": "2024-06-15"},
|
|
{"url": "https://c.com", "published_date": "2024-03-10"},
|
|
]}
|
|
out = sort_results(r, "date")
|
|
assert [x["url"] for x in out["results"]] == [
|
|
"https://b.com", "https://c.com", "https://a.com"
|
|
]
|
|
|
|
|
|
def test_sort_by_date_none_at_end():
|
|
r = {"results": [
|
|
{"url": "https://a.com", "published_date": "2024-01-01"},
|
|
{"url": "https://b.com"}, # no date
|
|
]}
|
|
out = sort_results(r, "date")
|
|
assert out["results"][0]["url"] == "https://a.com"
|
|
assert out["results"][1]["url"] == "https://b.com"
|
|
|
|
|
|
def test_sort_by_engine_ascending():
|
|
r = {"results": [
|
|
{"url": "https://a.com", "engine": "duckduckgo"},
|
|
{"url": "https://b.com", "engine": "bing"},
|
|
{"url": "https://c.com", "engine": "google"},
|
|
]}
|
|
out = sort_results(r, "engine")
|
|
assert [x["engine"] for x in out["results"]] == ["bing", "duckduckgo", "google"]
|
|
|
|
|
|
def test_sort_by_none_preserves_order():
|
|
r = {"results": [
|
|
{"url": "https://a.com", "score": 0.5},
|
|
{"url": "https://b.com", "score": 2.0},
|
|
]}
|
|
out = sort_results(r, "none")
|
|
assert [x["url"] for x in out["results"]] == ["https://a.com", "https://b.com"]
|
|
|
|
|
|
def test_sort_empty_results():
|
|
out = sort_results({"results": []}, "score")
|
|
assert out["results"] == []
|
|
|
|
|
|
def test_sort_no_results_key():
|
|
out = sort_results({}, "score")
|
|
assert out == {}
|
|
|
|
|
|
# ----- _format_results (csv) -----
|
|
|
|
def test_format_csv_basic():
|
|
results = {"results": [
|
|
{"title": "T1", "url": "https://a.com", "engine": "google", "score": 1.0,
|
|
"published_date": "2024-01-01", "content": "snippet one"},
|
|
]}
|
|
args = _Args(format="csv", snippet_len=0, fetch=0)
|
|
out = _format_results(results, args)
|
|
assert "title,url,engine,score,published_date,content" in out
|
|
assert "T1" in out
|
|
assert "https://a.com" in out
|
|
assert "google" in out
|
|
assert "1.0" in out
|
|
assert "snippet one" in out
|
|
|
|
|
|
def test_format_csv_multiple_rows():
|
|
results = {"results": [
|
|
{"title": "A", "url": "https://a.com", "engine": "google", "score": 2.0,
|
|
"published_date": "", "content": "ca"},
|
|
{"title": "B", "url": "https://b.com", "engine": "bing", "score": 1.0,
|
|
"published_date": "2024-06-01", "content": "cb"},
|
|
]}
|
|
args = _Args(format="csv", snippet_len=0, fetch=0)
|
|
out = _format_results(results, args)
|
|
lines = out.strip().split("\n")
|
|
assert len(lines) == 3 # header + 2 rows
|
|
assert lines[0].startswith("title,url")
|
|
|
|
|
|
def test_format_csv_empty_results():
|
|
results = {"results": []}
|
|
args = _Args(format="csv", snippet_len=0, fetch=0)
|
|
out = _format_results(results, args)
|
|
assert "title,url,engine,score,published_date,content" in out
|
|
|
|
|
|
def test_format_csv_missing_fields():
|
|
"""Results with missing fields → empty string in CSV, no crash."""
|
|
results = {"results": [
|
|
{"title": "Only Title"}, # no url, engine, score, etc.
|
|
]}
|
|
args = _Args(format="csv", snippet_len=0, fetch=0)
|
|
out = _format_results(results, args)
|
|
assert "Only Title" in out
|
|
|
|
|
|
def test_format_csv_comma_in_content_escaped():
|
|
"""Commas in content are properly quoted by the csv module."""
|
|
results = {"results": [
|
|
{"title": "T", "url": "https://a.com", "engine": "g", "score": 1.0,
|
|
"published_date": "", "content": "has, comma"},
|
|
]}
|
|
args = _Args(format="csv", snippet_len=0, fetch=0)
|
|
out = _format_results(results, args)
|
|
assert '"has, comma"' in out
|
|
|
|
|
|
# ----- load_config (--config FILE) -----
|
|
|
|
def test_load_config_explicit_path(tmp_path):
|
|
f = tmp_path / "test.toml"
|
|
f.write_text(
|
|
'[searxng]\ninstance = "https://x.example.com"\ntimeout = 20\nformat = "brief"\n',
|
|
encoding="utf-8")
|
|
cfg = load_config(str(f))
|
|
assert cfg["instance"] == "https://x.example.com"
|
|
assert cfg["timeout"] == 20
|
|
assert cfg["format"] == "brief"
|
|
|
|
|
|
def test_load_config_nonexistent_returns_empty():
|
|
cfg = load_config("/nonexistent/path/config.toml")
|
|
assert cfg == {}
|
|
|
|
|
|
def test_load_config_top_level_table(tmp_path):
|
|
"""Config without [searxng] section — top-level keys used directly."""
|
|
f = tmp_path / "flat.toml"
|
|
f.write_text('instance = "https://flat.example.com"\n', encoding="utf-8")
|
|
cfg = load_config(str(f))
|
|
assert cfg["instance"] == "https://flat.example.com"
|