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, _format_results
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.

    v2.0.0: added referer, no_fallback, fetch_report, request_delay.
    """
    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",
        # v2.0.0 new fields
        referer=None, no_fallback=False, fetch_report=False,
        request_delay=0.3,
    )
    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.

    ``encoding="utf-8"`` is explicit because Windows Chinese systems default
    to GBK for subprocess text decoding, which crashes on non-ASCII chars
    in --help output (e.g. box-drawing chars, em-dashes). The script itself
    already forces UTF-8 stdout via ``force_utf8_stdout()``, so this just
    matches the decoding side.
    """
    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, encoding="utf-8", timeout=30,
        env=full_env,
    )


def test_cli_version_prints_version():
    """`--version` exits 0 and prints the version string."""
    from _config import VERSION
    r = _run_cli("--version")
    assert r.returncode == 0
    assert VERSION 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


# ===== --dump-schema (v1.8.0) =====

def test_cli_dump_schema_outputs_valid_json():
    """`--dump-schema` prints valid JSON with schema_version/title/properties.

    The schema dump is a non-search operation: it exits 0 and does NOT
    require --query or -i (it short-circuits before instance resolution).
    """
    r = _run_cli("--dump-schema")
    assert r.returncode == 0
    data = json.loads(r.stdout)
    assert data["schema_version"] == "1.0"
    assert data["title"] == "SearXNG CLI Search Result"
    assert "properties" in data
    assert isinstance(data["properties"], dict)


def test_cli_dump_schema_does_not_require_query():
    """`--dump-schema` exits 0 without --query (non-search operation)."""
    r = _run_cli("--dump-schema")
    assert r.returncode == 0
    # The --query-required check must not fire for --dump-schema.
    assert "required" not in r.stderr.lower()


# ===== --stream mutual exclusion (v1.8.0) =====

def test_cli_stream_with_queries_file_rejected(tmp_path):
    """--stream + --queries-file is rejected with E_INPUT on stdout.

    Batch mode emits a JSON array, not JSON Lines; the combination is
    explicitly rejected so AI agents don't silently get the wrong format.
    """
    qf = tmp_path / "queries.txt"
    qf.write_text("query\n", encoding="utf-8")
    r = _run_cli("--stream", "--queries-file", str(qf),
                 "--format", "json", "-i", "https://x.example.com")
    assert r.returncode == 1
    data = json.loads(r.stdout)
    assert data["error_code"] == "E_INPUT"


def test_cli_stream_with_csv_format_rejected():
    """--stream + --format csv is rejected with E_INPUT.

    csv is a non-json format, so _emit_error routes the error to stderr
    (as a ``[E_INPUT]`` prefixed log line) rather than stdout JSON.
    """
    r = _run_cli("--stream", "--format", "csv", "-q", "test",
                 "-i", "https://x.example.com")
    assert r.returncode == 1
    assert "E_INPUT" in r.stderr


# ===== schema_version in single-query JSON (v1.8.0) =====

def test_format_results_json_includes_schema_version():
    """Single-query JSON output includes schema_version: '1.0'."""
    results = {"results": [{"title": "T", "url": "https://example.com"}]}
    args = SimpleNamespace(format="json")
    out = _format_results(results, args)
    parsed = json.loads(out)
    assert parsed["schema_version"] == "1.0"


# ===== _fallback field must not leak to JSON output (v1.8.0) =====

def test_run_single_query_html_fallback_pops_underscore_fallback(monkeypatch):
    """HTML-fallback's internal _fallback field is removed before output.

    _run_single_query pops _fallback at the end so it never appears in
    the JSON payload; fetched_source (the public field) is set separately
    on the --fetch path.
    """
    args = _make_args(format="json")
    fake = {"results": [{"url": "https://a.com/1", "title": "A"}],
            "_fallback": "html"}
    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
    # _fallback must be popped so it doesn't leak into JSON output
    assert "_fallback" not in results
    # And the serialized JSON also omits it
    out = _format_results(results, args)
    parsed = json.loads(out)
    assert "_fallback" not in parsed


def test_run_single_query_fetch_sets_fetched_source_from_html_fallback(monkeypatch):
    """--fetch path sets fetched_source from _fallback (or 'json').

    When the search came via HTML fallback (_fallback='html'), the
    fetched_source field reflects that origin, while _fallback itself is
    cleaned up and never reaches JSON output.
    """
    args = _make_args(fetch=1)
    fake = {"results": [{"url": "https://a.com/1"}], "_fallback": "html"}
    monkeypatch.setattr("search.search_multi", lambda *a, **kw: fake)
    monkeypatch.setattr("search.fetch_top_results", lambda *a, **kw: [
        {"url": "https://a.com/1", "status": "ok", "text": "page A",
         "text_length": 6, "truncated": False},
    ])

    results, err, _ = _run_single_query("test", args,
                                        ["https://x.example.com"], {}, 0)
    assert err is None
    # fetched_source reflects the HTML fallback origin
    assert results["fetched_source"] == "html"
    # _fallback still cleaned up
    assert "_fallback" not in results