"""Tests for the --verify instance health-check feature. Covers: _probe_config_endpoint (reachable/disabled/error), verify_instances (reachability, JSON support, POST probe, latency, auth status, error classification), _print_verify_report (JSON and table output). """ import json import urllib.error from unittest.mock import patch, MagicMock from search import _probe_config_endpoint, verify_instances, _print_verify_report def _mock_urlopen(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 # ----- _probe_config_endpoint ----- def test_probe_config_reachable(): """A working /config endpoint returns engines and categories.""" payload = json.dumps({ "engines": [{"name": "google"}, {"name": "bing"}, {"name": "wikipedia"}], "categories": {"general": [], "news": [], "images": []}, }) with patch("urllib.request.urlopen", return_value=_mock_urlopen(payload.encode())): r = _probe_config_endpoint("https://s.example.com", timeout=10) assert r["reachable"] is True assert r["engines"] == ["google", "bing", "wikipedia"] assert set(r["categories"]) == {"general", "news", "images"} assert r["error"] is None def test_probe_config_categories_as_list(): """Some instances return categories as a list instead of a dict.""" payload = json.dumps({ "engines": [{"name": "google"}], "categories": ["general", "news"], }) with patch("urllib.request.urlopen", return_value=_mock_urlopen(payload.encode())): r = _probe_config_endpoint("https://s.example.com", timeout=10) assert r["reachable"] is True assert r["categories"] == ["general", "news"] def test_probe_config_http_error(): """/config returns 403 → reachable=False with HTTP code in error.""" err = urllib.error.HTTPError("url", 403, "Forbidden", {}, None) with patch("urllib.request.urlopen", side_effect=err): r = _probe_config_endpoint("https://s.example.com", timeout=10) assert r["reachable"] is False assert "403" in r["error"] assert r["engines"] == [] def test_probe_config_connection_error(): """Connection error → reachable=False with error message.""" err = urllib.error.URLError("connection refused") with patch("urllib.request.urlopen", side_effect=err): r = _probe_config_endpoint("https://s.example.com", timeout=10) assert r["reachable"] is False assert r["engines"] == [] assert "connection refused" in r["error"] # ----- verify_instances ----- def _mock_json_search_response(results=None): """Mock a successful JSON search response.""" payload = json.dumps({"results": results or [{"title": "t", "url": "https://x.com"}]}) return _mock_urlopen(payload.encode(), content_type="application/json") def _mock_html_search_response(): """Mock an HTML search response (JSON disabled).""" return _mock_urlopen(b"not json", content_type="text/html") def test_verify_reachable_json_supported(): """Instance returns JSON → reachable + json_supported + post probe.""" with patch("urllib.request.urlopen", return_value=_mock_json_search_response()): # /config also needs to respond config_payload = json.dumps({"engines": [{"name": "google"}], "categories": {}}) config_resp = _mock_urlopen(config_payload.encode()) with patch("urllib.request.urlopen", return_value=_mock_json_search_response()): report = verify_instances(["https://s.example.com"]) r = report[0] assert r["url"] == "https://s.example.com" assert r["reachable"] is True assert r["json_supported"] is True def test_verify_html_only_instance(): """Instance returns HTML (no JSON) → reachable but json_supported=False.""" with patch("urllib.request.urlopen", return_value=_mock_html_search_response()): report = verify_instances(["https://s.example.com"]) r = report[0] assert r["reachable"] is True assert r["json_supported"] is False def test_verify_auth_rejected(): """401/403 with auth → auth_status='rejected'.""" err = urllib.error.HTTPError("url", 403, "Forbidden", {}, None) auth_headers = {"Authorization": "Bearer token123"} with patch("urllib.request.urlopen", side_effect=err): report = verify_instances(["https://s.example.com"], auth_headers=auth_headers) r = report[0] assert r["auth_status"] == "rejected" def test_verify_no_auth_returns_na(): """Without auth headers, auth_status is 'n/a'.""" err = urllib.error.URLError("timeout") with patch("urllib.request.urlopen", side_effect=err): report = verify_instances(["https://s.example.com"]) r = report[0] assert r["auth_status"] == "n/a" def test_verify_connection_error(): """Connection error → reachable=False, latency=None (no response received).""" err = urllib.error.URLError("connection refused") with patch("urllib.request.urlopen", side_effect=err): report = verify_instances(["https://s.example.com"]) r = report[0] assert r["reachable"] is False assert r["json_supported"] is False assert r["latency"] is None # no response → no latency assert "connection refused" in r["error"] def test_verify_preserves_input_order(): """Multiple instances: report preserves the original input order.""" urls = ["https://a.example.com", "https://b.example.com", "https://c.example.com"] with patch("urllib.request.urlopen", return_value=_mock_json_search_response()): report = verify_instances(urls) assert [r["url"] for r in report] == urls def test_verify_has_latency(): """Latency is a positive float for reachable instances.""" with patch("urllib.request.urlopen", return_value=_mock_json_search_response()): report = verify_instances(["https://s.example.com"]) assert report[0]["latency"] is not None assert report[0]["latency"] >= 0 def test_verify_has_result_count(): """Result count from the test query is recorded.""" with patch("urllib.request.urlopen", return_value=_mock_json_search_response( [{"title": "a"}, {"title": "b"}, {"title": "c"}])): report = verify_instances(["https://s.example.com"]) assert report[0]["result_count"] == 3 # ----- _print_verify_report ----- def test_print_report_json(capsys): """JSON output is valid JSON with the full report.""" report = [ {"url": "https://a.com", "reachable": True, "json_supported": True, "post_supported": True, "latency": 0.5, "result_count": 10, "engines": ["google"], "auth_status": "n/a", "error": None, "config_endpoint": True}, ] _print_verify_report(report, as_json=True) out, _ = capsys.readouterr() data = json.loads(out) assert data[0]["url"] == "https://a.com" assert data[0]["reachable"] is True def test_print_report_table(capsys): """Table output includes the header and a summary line.""" report = [ {"url": "https://a.com", "reachable": True, "json_supported": True, "post_supported": True, "latency": 0.5, "result_count": 10, "engines": ["google", "bing"], "auth_status": "n/a", "error": None, "config_endpoint": True}, {"url": "https://b.com", "reachable": False, "json_supported": False, "post_supported": None, "latency": None, "result_count": None, "engines": [], "auth_status": "n/a", "error": "timeout", "config_endpoint": None}, ] _print_verify_report(report, as_json=False) out, _ = capsys.readouterr() assert "URL" in out assert "REACH" in out assert "https://a.com" in out assert "https://b.com" in out assert "1/2 instances reachable" in out def test_print_report_empty(capsys): """Empty report produces a table with 0/0 summary.""" _print_verify_report([], as_json=False) out, _ = capsys.readouterr() assert "0/0 instances reachable" in out