searxng-use-cli/tests/test_stream_progress.py

"""Tests for --stream (JSON Lines output) and --progress (progress events).

Covers: stream output format (result/done/error events), progress event
emission (start/cache_hit/cache_store/done/error/fetch_*), progress
disabled by default, stream only works with --format json.
"""
import json
import sys
import urllib.error
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest

import common
from common import emit_progress, set_progress_enabled
import search as search_mod
from search import _run_single_query, main


# ----- emit_progress: default disabled -----

def test_progress_disabled_by_default(capsys):
    """Without --progress, emit_progress is a no-op."""
    set_progress_enabled(False)
    emit_progress("start", query="test", instances=1)
    out, err = capsys.readouterr()
    assert out == ""
    assert err == ""


def test_progress_enabled_emits_json(capsys):
    """With --progress, emit_progress outputs JSON Lines to stderr."""
    set_progress_enabled(True)
    try:
        emit_progress("start", query="test", instances=2)
        _, err = capsys.readouterr()
        data = json.loads(err.strip())
        assert data["event"] == "start"
        assert data["query"] == "test"
        assert data["instances"] == 2
    finally:
        set_progress_enabled(False)


def test_progress_multiple_events(capsys):
    """Multiple events produce multiple JSON Lines."""
    set_progress_enabled(True)
    try:
        emit_progress("start", query="q", instances=1)
        emit_progress("cache_hit", query="q", ttl=30)
        emit_progress("done", results=5, query="q")
        _, err = capsys.readouterr()
        lines = [l for l in err.strip().split("\n") if l]
        assert len(lines) == 3
        events = [json.loads(l)["event"] for l in lines]
        assert events == ["start", "cache_hit", "done"]
    finally:
        set_progress_enabled(False)


def test_progress_event_with_error_code(capsys):
    """Error events include error_code field."""
    set_progress_enabled(True)
    try:
        emit_progress("error", error="timeout", error_code="E_NETWORK",
                      query="test")
        _, err = capsys.readouterr()
        data = json.loads(err.strip())
        assert data["error_code"] == "E_NETWORK"
    finally:
        set_progress_enabled(False)


# ----- _run_single_query: progress events -----

def _make_args(**overrides):
    """Construct a minimal args object for _run_single_query."""
    defaults = 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",
    )
    defaults.update(overrides)
    return SimpleNamespace(**defaults)


def test_run_single_query_emits_start_and_done(capsys):
    """_run_single_query emits start and done events when progress is enabled."""
    set_progress_enabled(True)
    try:
        mock_results = {"results": [{"title": "t", "url": "https://x.com"}]}
        with patch.object(search_mod, "search_multi", return_value=mock_results):
            _run_single_query("test", _make_args(),
                              ["https://x.example.com"], {}, 0)
        _, err = capsys.readouterr()
        lines = [json.loads(l) for l in err.strip().split("\n") if l]
        events = [e["event"] for e in lines]
        assert "start" in events
        assert "done" in events
        done_event = next(e for e in lines if e["event"] == "done")
        assert done_event["results"] == 1
    finally:
        set_progress_enabled(False)


def test_run_single_query_emits_cache_hit(capsys, isolated_cache):
    """Cache hit emits cache_hit event."""
    set_progress_enabled(True)
    try:
        mock_results = {"results": [{"title": "t", "url": "https://x.com"}]}
        with patch.object(search_mod, "search_multi", return_value=mock_results):
            # First call: cache miss, stores result
            _run_single_query("test", _make_args(cache_ttl=30),
                              ["https://x.example.com"], {}, 1800)
            capsys.readouterr()  # clear
            # Second call: cache hit
            _run_single_query("test", _make_args(cache_ttl=30),
                              ["https://x.example.com"], {}, 1800)
        _, err = capsys.readouterr()
        lines = [json.loads(l) for l in err.strip().split("\n") if l]
        events = [e["event"] for e in lines]
        assert "cache_hit" in events
    finally:
        set_progress_enabled(False)


def test_run_single_query_emits_error_on_failure(capsys):
    """Search failure emits error event with error_code."""
    set_progress_enabled(True)
    try:
        import urllib.error
        err = urllib.error.HTTPError("url", 403, "Forbidden", {}, None)
        with patch.object(search_mod, "search_multi", side_effect=err):
            _run_single_query("test", _make_args(),
                              ["https://x.example.com"], {}, 0)
        _, err_out = capsys.readouterr()
        lines = [json.loads(l) for l in err_out.strip().split("\n") if l]
        error_events = [e for e in lines if e["event"] == "error"]
        assert len(error_events) == 1
        assert error_events[0]["error_code"] == "E_AUTH"
    finally:
        set_progress_enabled(False)


# ----- --stream: JSON Lines output -----

def test_stream_outputs_json_lines(capsys):
    """--stream outputs each result as a JSON Line + a done event."""
    mock_results = {"results": [
        {"title": "first", "url": "https://a.com"},
        {"title": "second", "url": "https://b.com"},
    ]}
    with patch.object(search_mod, "search_multi", return_value=mock_results):
        with pytest.raises(SystemExit) as exc_info:
            main.__wrapped__ if hasattr(main, "__wrapped__") else None
            # Call main with --stream
            with patch.object(sys, "argv", ["search.py", "-q", "test",
                                            "-i", "https://x.example.com",
                                            "--stream", "--format", "json"]):
                # Mock the early argparse for logging
                with patch.object(search_mod, "setup_logging"):
                    main()
        assert exc_info.value.code == 0
    out, _ = capsys.readouterr()
    lines = [json.loads(l) for l in out.strip().split("\n") if l]
    # Should have 2 result events + 1 done event
    result_events = [e for e in lines if e["type"] == "result"]
    done_events = [e for e in lines if e["type"] == "done"]
    assert len(result_events) == 2
    assert len(done_events) == 1
    assert done_events[0]["count"] == 2
    assert result_events[0]["result"]["title"] == "first"


def test_stream_empty_results_exit_2(capsys):
    """--stream with empty results exits with code 2 and emits done with count=0."""
    mock_results = {"results": []}
    with patch.object(search_mod, "search_multi", return_value=mock_results):
        with pytest.raises(SystemExit) as exc_info:
            with patch.object(sys, "argv", ["search.py", "-q", "test",
                                            "-i", "https://x.example.com",
                                            "--stream", "--format", "json"]):
                with patch.object(search_mod, "setup_logging"):
                    main()
    assert exc_info.value.code == 2
    out, _ = capsys.readouterr()
    lines = [json.loads(l) for l in out.strip().split("\n") if l]
    done_events = [e for e in lines if e["type"] == "done"]
    assert len(done_events) == 1
    assert done_events[0]["count"] == 0


def test_stream_result_event_shape():
    """Each result event has type=result and result=<result dict>."""
    # Unit test the stream output logic directly
    results = [{"title": "t", "url": "https://x.com"}]
    lines = []
    for r in results:
        lines.append(json.dumps({"type": "result", "result": r}))
    lines.append(json.dumps({"type": "done", "count": len(results)}))
    parsed = [json.loads(l) for l in lines]
    assert parsed[0]["type"] == "result"
    assert parsed[0]["result"]["url"] == "https://x.com"
    assert parsed[-1]["type"] == "done"
    assert parsed[-1]["count"] == 1


# ----- stream done/error: schema_version & recovery_hint (v1.8.0) -----

def test_stream_done_includes_schema_version(capsys):
    """stream done event includes the schema_version field."""
    mock_results = {"results": [{"title": "t", "url": "https://x.com"}]}
    with patch.object(search_mod, "search_multi", return_value=mock_results):
        with pytest.raises(SystemExit) as exc_info:
            with patch.object(sys, "argv", ["search.py", "-q", "test",
                                            "-i", "https://x.example.com",
                                            "--stream", "--format", "json",
                                            "--retry", "0"]):
                with patch.object(search_mod, "setup_logging"):
                    main()
    assert exc_info.value.code == 0
    out, _ = capsys.readouterr()
    lines = [json.loads(l) for l in out.strip().split("\n") if l]
    done_events = [e for e in lines if e["type"] == "done"]
    assert len(done_events) == 1
    assert done_events[0]["schema_version"] == "1.0"


def test_stream_error_includes_recovery_hint(capsys):
    """stream error event includes recovery_hint for classified errors.

    HTTP 403 -> E_AUTH, which has a recovery_hint in RECOVERY_HINTS.
    --retry 0 avoids backoff sleeps on the (mocked) 403.
    """
    err = urllib.error.HTTPError("url", 403, "Forbidden", {}, None)
    with patch.object(search_mod, "search_multi", side_effect=err):
        with pytest.raises(SystemExit) as exc_info:
            with patch.object(sys, "argv", ["search.py", "-q", "test",
                                            "-i", "https://x.example.com",
                                            "--stream", "--format", "json",
                                            "--retry", "0"]):
                with patch.object(search_mod, "setup_logging"):
                    main()
    assert exc_info.value.code == 1
    out, _ = capsys.readouterr()
    lines = [json.loads(l) for l in out.strip().split("\n") if l]
    error_events = [e for e in lines if e["type"] == "error"]
    assert len(error_events) == 1
    assert error_events[0]["error_code"] == "E_AUTH"
    assert "recovery_hint" in error_events[0]


# ----- instance_try / instance_ok / instance_fail progress events (v1.8.0) -----
#
# These events are emitted inside search_multi (not _run_single_query), so
# we must let the real search_multi run and only mock urllib.request.urlopen.

def _mock_urlopen_resp(data: bytes, content_type="application/json"):
    """Build a MagicMock that quacks like an urlopen context manager."""
    resp = MagicMock()
    resp.read.return_value = data
    resp.headers = {"Content-Type": content_type}
    resp.__enter__.return_value = resp
    resp.__exit__.return_value = None
    return resp


def _url_router(routing):
    """Build a side_effect that dispatches urlopen by request URL substring."""
    def _side_effect(req, *args, **kwargs):
        url = getattr(req, "full_url", str(req))
        for key, resp in routing.items():
            if key in url:
                if isinstance(resp, BaseException):
                    raise resp
                return resp
        raise AssertionError(f"unexpected urlopen for {url!r}")
    return _side_effect


def test_progress_emits_instance_try_ok_fail(capsys):
    """instance_try/instance_ok/instance_fail events are emitted to stderr.

    Uses serial mode (parallel=False) so the events come out in a
    deterministic order: try a -> fail a -> try b -> ok b.
    """
    set_progress_enabled(True)
    try:
        payload = json.dumps(
            {"results": [{"title": "b", "url": "https://b.com"}]}
        ).encode()
        router = _url_router({
            "a.example.com": urllib.error.URLError("down"),
            "b.example.com": _mock_urlopen_resp(payload),
        })
        with patch("urllib.request.urlopen", side_effect=router):
            search_mod.search_multi(
                ["https://a.example.com", "https://b.example.com"],
                {"q": "test", "format": "json"},
                parallel=False, retry_per=0,
            )
        _, err = capsys.readouterr()
        lines = [json.loads(l) for l in err.strip().split("\n") if l]
        events = [e["event"] for e in lines]
        assert "instance_try" in events
        assert "instance_fail" in events
        assert "instance_ok" in events
        # instance_fail carries error_code
        fail_events = [e for e in lines if e["event"] == "instance_fail"]
        assert len(fail_events) >= 1
        assert "error_code" in fail_events[0]
        # instance_ok carries results count
        ok_events = [e for e in lines if e["event"] == "instance_ok"]
        assert len(ok_events) >= 1
        assert "results" in ok_events[0]
    finally:
        set_progress_enabled(False)