"""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=.""" # 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)