searxng-use-cli/tests/test_cli_e2e.py

"""End-to-end tests for scripts/search.py.

Covers three layers of the CLI:
  * ``_run_single_query`` — full orchestration (search → dedup → sort →
    limit → domain-filter → fetch), stubbed at the ``search_multi`` /
    ``fetch_top_results`` boundary so no network is touched.
  * ``_emit_error``       — error output + exit (json→stdout, others→stderr).
  * ``main()``            — real CLI via subprocess (--version / --help /
    no-args / --cache-stats).

``_run_single_query`` tests build a ``SimpleNamespace`` args object that
matches the argparse.Namespace shape produced by ``main()``. Cache-related
tests use the ``isolated_cache`` fixture from conftest.py so each test gets
a fresh SQLite cache under a temp dir. ``main()`` tests spawn a real
subprocess with cwd pinned to the project root.
"""
import json
import os
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace

import pytest

from search import _run_single_query, _emit_error, _build_params
import cache as cache_module

PROJECT_ROOT = Path(__file__).resolve().parent.parent


def _make_args(**overrides):
    """Build a minimal args object matching the argparse.Namespace shape
    that ``_run_single_query`` reads. All fields _run_single_query touches
    are present with sensible defaults; tests override only what they need.
    """
    base = dict(
        query="test", format="json", method="GET", timeout=15, retry=0,
        serial=False, no_dedup=False, sort_by="none", max_results=None,
        include_domain=None, exclude_domain=None, fetch=0, fetch_timeout=10,
        fetch_retries=3, max_size=None, cache_ttl=0, snippet_len=0,
        categories=None, language=None, pageno=1, time_range="year",
        safesearch=0, engines="google,bing",
    )
    base.update(overrides)
    return SimpleNamespace(**base)


# ===== _run_single_query =====

def test_run_single_query_success(monkeypatch):
    """Successful search returns (results, None)."""
    args = _make_args()
    fake = {"results": [{"url": "https://a.com/1", "title": "A"}]}
    monkeypatch.setattr("search.search_multi", lambda *a, **kw: fake)

    results, err, _ = _run_single_query("test", args, ["https://x.example.com"], {}, 0)
    assert err is None
    assert results is not None
    assert results["results"][0]["url"] == "https://a.com/1"


def test_run_single_query_failure_returns_error(monkeypatch):
    """When search_multi raises, returns (None, error_str)."""
    args = _make_args()

    def boom(*a, **kw):
        raise RuntimeError("network down")
    monkeypatch.setattr("search.search_multi", boom)

    results, err, _ = _run_single_query("test", args, ["https://x.example.com"], {}, 0)
    assert results is None
    assert "network down" in err


def test_run_single_query_cache_hit_skips_network(isolated_cache, monkeypatch):
    """Cache hit returns the cached result without calling search_multi.

    Pre-populates the cache with the exact params _run_single_query builds,
    then asserts search_multi is never reached (a call would raise).
    """
    args = _make_args(cache_ttl=5)
    cached = {"results": [{"url": "https://a.com/1", "title": "cached"}]}

    params = _build_params("test", args)
    cache_module.put(params, cached, 300)

    def should_not_call(*a, **kw):
        raise AssertionError("search_multi must not be called on cache hit")
    monkeypatch.setattr("search.search_multi", should_not_call)

    results, err, _ = _run_single_query("test", args, ["https://x.example.com"], {}, 300)
    assert err is None
    assert results is not None
    assert results["results"][0]["title"] == "cached"


def test_run_single_query_cache_miss_stores_result(isolated_cache, monkeypatch):
    """Cache miss calls search_multi and stores the result for next time."""
    args = _make_args(cache_ttl=5)
    fake = {"results": [{"url": "https://a.com/1", "title": "A"}]}
    monkeypatch.setattr("search.search_multi", lambda *a, **kw: fake)

    assert cache_module.stats()["entries"] == 0

    results, err, _ = _run_single_query("test", args, ["https://x.example.com"], {}, 300)
    assert err is None

    assert cache_module.stats()["entries"] >= 1


def test_run_single_query_no_dedup_keeps_duplicates(monkeypatch):
    """no_dedup=True skips cross-engine URL deduplication.

    Two results with the same URL but different engines are both kept.
    """
    args = _make_args(no_dedup=True)
    fake = {"results": [
        {"url": "https://a.com/1", "engine": "google"},
        {"url": "https://a.com/1", "engine": "bing"},
    ]}
    monkeypatch.setattr("search.search_multi", lambda *a, **kw: fake)

    results, err, _ = _run_single_query("test", args, ["https://x.example.com"], {}, 0)
    assert err is None
    assert len(results["results"]) == 2


def test_run_single_query_max_results_truncation(monkeypatch):
    """--max-results truncates the list (applied AFTER dedup+sort)."""
    args = _make_args(max_results=2)
    fake = {"results": [{"url": f"https://a.com/{i}"} for i in range(5)]}
    monkeypatch.setattr("search.search_multi", lambda *a, **kw: fake)

    results, err, _ = _run_single_query("test", args, ["https://x.example.com"], {}, 0)
    assert err is None
    assert len(results["results"]) == 2


def test_run_single_query_include_domain_filter(monkeypatch):
    """--include-domain keeps only results whose domain matches."""
    args = _make_args(include_domain="a.com")
    fake = {"results": [
        {"url": "https://a.com/1"},
        {"url": "https://b.com/2"},
        {"url": "https://a.com/3"},
    ]}
    monkeypatch.setattr("search.search_multi", lambda *a, **kw: fake)

    results, err, _ = _run_single_query("test", args, ["https://x.example.com"], {}, 0)
    assert err is None
    urls = [r["url"] for r in results["results"]]
    assert urls == ["https://a.com/1", "https://a.com/3"]


def test_run_single_query_exclude_domain_filter(monkeypatch):
    """--exclude-domain drops results whose domain matches."""
    args = _make_args(exclude_domain="b.com")
    fake = {"results": [
        {"url": "https://a.com/1"},
        {"url": "https://b.com/2"},
        {"url": "https://c.com/3"},
    ]}
    monkeypatch.setattr("search.search_multi", lambda *a, **kw: fake)

    results, err, _ = _run_single_query("test", args, ["https://x.example.com"], {}, 0)
    assert err is None
    urls = [r["url"] for r in results["results"]]
    assert urls == ["https://a.com/1", "https://c.com/3"]


def test_run_single_query_fetch_attaches_fetched(monkeypatch):
    """--fetch N attaches a 'fetched' list (and fetched_source) to results."""
    args = _make_args(fetch=2)
    fake = {"results": [
        {"url": "https://a.com/1"},
        {"url": "https://b.com/2"},
    ]}
    monkeypatch.setattr("search.search_multi", lambda *a, **kw: fake)
    fake_fetched = [
        {"url": "https://a.com/1", "status": "ok", "text": "page A"},
        {"url": "https://b.com/2", "status": "ok", "text": "page B"},
    ]
    monkeypatch.setattr("search.fetch_top_results", lambda *a, **kw: fake_fetched)

    results, err, _ = _run_single_query("test", args, ["https://x.example.com"], {}, 0)
    assert err is None
    assert results["fetched"] == fake_fetched
    assert results["fetched_source"] == "json"


# ===== _emit_error =====

def test_emit_error_json_to_stdout(capsys):
    """In json mode the error is printed as JSON to stdout and exits 1."""
    args = _make_args(format="json")
    with pytest.raises(SystemExit) as exc_info:
        _emit_error("boom", args)
    assert exc_info.value.code == 1
    captured = capsys.readouterr()
    data = json.loads(captured.out)
    assert data["error"] == "boom"
    assert data["exit_code"] == 1


def test_emit_error_brief_keeps_stdout_clean(capsys):
    """In non-json mode stdout stays clean (error routed via logger to stderr)."""
    args = _make_args(format="brief")
    with pytest.raises(SystemExit) as exc_info:
        _emit_error("boom", args)
    assert exc_info.value.code == 1
    captured = capsys.readouterr()
    assert captured.out == ""


def test_emit_error_includes_query_when_provided(capsys):
    """JSON error includes the 'query' field when a query is supplied."""
    args = _make_args(format="json")
    with pytest.raises(SystemExit):
        _emit_error("not found", args, query="hello world")
    captured = capsys.readouterr()
    data = json.loads(captured.out)
    assert data["query"] == "hello world"
    assert data["error"] == "not found"


def test_emit_error_omits_query_when_absent(capsys):
    """JSON error omits the 'query' field entirely when query is None."""
    args = _make_args(format="json")
    with pytest.raises(SystemExit):
        _emit_error("boom", args, query=None)
    captured = capsys.readouterr()
    data = json.loads(captured.out)
    assert "query" not in data


# ===== main() via real subprocess =====

def _run_cli(*args, env=None):
    """Run scripts/search.py as a real subprocess at the project root.

    Uses ``sys.executable`` so the same interpreter that runs pytest runs
    the CLI. cwd is pinned to PROJECT_ROOT so the script's config-file
    auto-discovery (./searxng.toml) behaves deterministically.
    """
    full_env = {**os.environ, **(env or {})}
    cmd = [sys.executable, str(PROJECT_ROOT / "scripts" / "search.py")] + list(args)
    return subprocess.run(
        cmd, cwd=str(PROJECT_ROOT),
        capture_output=True, text=True, timeout=30,
        env=full_env,
    )


def test_cli_version_prints_version():
    """`--version` exits 0 and prints the version string."""
    r = _run_cli("--version")
    assert r.returncode == 0
    assert "1.7.0" in r.stdout
    assert "searxng-cli" in r.stdout


def test_cli_help_exits_zero():
    """`--help` exits 0 and prints usage text on stdout."""
    r = _run_cli("--help")
    assert r.returncode == 0
    assert "Search via a user-supplied SearXNG instance" in r.stdout
    assert "--query" in r.stdout


def test_cli_no_args_exits_nonzero():
    """No arguments → parser.error → non-zero exit.

    argparse uses exit code 2 for parser.error (not 1); we assert non-zero
    plus the message mentions --query.
    """
    r = _run_cli()
    assert r.returncode != 0
    assert "--query" in r.stderr or "required" in r.stderr.lower()


def test_cli_cache_stats_outputs_json(tmp_path):
    """`--cache-stats` prints cache statistics as JSON and exits 0.

    An -i instance is supplied because instance resolution runs before the
    cache-stats branch in main(); the instance is never contacted.
    SEARXNG_CACHE_DIR is redirected to a temp dir for isolation.
    """
    r = _run_cli(
        "--cache-stats", "-i", "https://example.com",
        env={"SEARXNG_CACHE_DIR": str(tmp_path)},
    )
    assert r.returncode == 0
    data = json.loads(r.stdout)
    assert "entries" in data
    assert "path" in data