"""Tests for scripts/common.py — auth, charset, retry policy, proxy. Covers: build_auth_headers (Bearer/Basic/precedence), resolve_auth_* priority (CLI > file > env, file parsing, empty-file errors), detect_charset (header/meta/fallback), is_retryable_error (429/5xx yes, 404/403 no, connection errors yes), and apply_proxy (env vars + NO_PROXY preservation). """ import base64 import os import urllib.error import pytest from common import ( RETRYABLE_STATUS, apply_proxy, build_auth_headers, detect_charset, is_retryable_error, resolve_auth_basic, resolve_auth_bearer, ) # ----- build_auth_headers ----- def test_build_auth_bearer(): assert build_auth_headers(bearer_token="tok123") == \ {"Authorization": "Bearer tok123"} def test_build_auth_basic(): h = build_auth_headers(basic_auth="user:pass") assert h["Authorization"].startswith("Basic ") decoded = base64.b64decode(h["Authorization"].split(" ", 1)[1]).decode() assert decoded == "user:pass" def test_build_auth_bearer_wins_over_basic(): """Bearer takes precedence when both are provided.""" h = build_auth_headers(bearer_token="tok", basic_auth="u:p") assert h == {"Authorization": "Bearer tok"} def test_build_auth_none_returns_empty(): assert build_auth_headers() == {} # ----- resolve_auth_basic ----- def test_resolve_auth_basic_cli_wins(): assert resolve_auth_basic(cli_value="cliu:clip") == "cliu:clip" def test_resolve_auth_basic_file(tmp_path): f = tmp_path / "auth.txt" f.write_text("# header\nu:secret\n", encoding="utf-8") assert resolve_auth_basic(file_path=str(f)) == "u:secret" def test_resolve_auth_basic_env(monkeypatch): monkeypatch.setenv("SEARXNG_BASIC_AUTH", "envu:envp") assert resolve_auth_basic() == "envu:envp" def test_resolve_auth_basic_priority_cli_over_file(tmp_path): f = tmp_path / "auth.txt" f.write_text("fileu:filep", encoding="utf-8") assert resolve_auth_basic(cli_value="cliu:clip", file_path=str(f)) == "cliu:clip" def test_resolve_auth_basic_empty_file_raises(tmp_path): f = tmp_path / "empty.txt" f.write_text("# only comment\n", encoding="utf-8") with pytest.raises(RuntimeError): resolve_auth_basic(file_path=str(f)) def test_resolve_auth_basic_missing_file_raises(): with pytest.raises(RuntimeError): resolve_auth_basic(file_path="nonexistent_file.txt") # ----- resolve_auth_bearer ----- def test_resolve_auth_bearer_cli(): assert resolve_auth_bearer(cli_value="tok") == "tok" def test_resolve_auth_bearer_env(monkeypatch): monkeypatch.setenv("SEARXNG_BEARER_TOKEN", "envtok") assert resolve_auth_bearer() == "envtok" def test_resolve_auth_bearer_file(tmp_path): f = tmp_path / "tok.txt" f.write_text("# header\ntoken123\n", encoding="utf-8") assert resolve_auth_bearer(file_path=str(f)) == "token123" # ----- detect_charset ----- def test_detect_charset_from_header(): assert detect_charset(b"hello", "text/html; charset=utf-8") == "utf-8" def test_detect_charset_from_meta_tag(): html = b'hello' assert detect_charset(html, "text/html") == "gbk" def test_detect_charset_fallback_utf8(): assert detect_charset(b"plain", "text/html") == "utf-8" def test_detect_charset_invalid_header_falls_back(): """An invalid charset in the header should not raise; falls back to utf-8.""" assert detect_charset(b"hello", "text/html; charset=nonexistent_encoding") == "utf-8" # ----- is_retryable_error ----- def _http_error(code): return urllib.error.HTTPError("http://x", code, "msg", {}, None) def test_retryable_429(): assert is_retryable_error(_http_error(429)) is True def test_retryable_502(): assert is_retryable_error(_http_error(502)) is True def test_retryable_503(): assert is_retryable_error(_http_error(503)) is True def test_retryable_504(): assert is_retryable_error(_http_error(504)) is True def test_non_retryable_404(): assert is_retryable_error(_http_error(404)) is False def test_retryable_403(): # 403 是可重试的：让 fetch_url 的 UA-fallback 循环有机会切换到浏览器 UA。 # 真正的认证错误会在重试耗尽后由 classify_error 归为 E_AUTH。 assert is_retryable_error(_http_error(403)) is True def test_non_retryable_200(): assert is_retryable_error(_http_error(200)) is False def test_retryable_url_error(): assert is_retryable_error(urllib.error.URLError("refused")) is True def test_retryable_os_error(): assert is_retryable_error(OSError("timeout")) is True def test_retryable_status_set_contents(): # 403 加入可重试集合，让 UA-fallback 在 UA 被屏蔽时有机会切换浏览器 UA assert RETRYABLE_STATUS == frozenset({403, 429, 502, 503, 504}) # ----- apply_proxy ----- def test_apply_proxy_sets_env_vars(monkeypatch): monkeypatch.delenv("HTTP_PROXY", raising=False) monkeypatch.delenv("HTTPS_PROXY", raising=False) monkeypatch.delenv("NO_PROXY", raising=False) apply_proxy("http://corp:8080") assert os.environ["HTTP_PROXY"] == "http://corp:8080" assert os.environ["HTTPS_PROXY"] == "http://corp:8080" assert os.environ["NO_PROXY"] == "localhost,127.0.0.1,::1" def test_apply_proxy_empty_is_noop(monkeypatch): monkeypatch.delenv("HTTP_PROXY", raising=False) apply_proxy("") assert "HTTP_PROXY" not in os.environ def test_apply_proxy_preserves_existing_no_proxy(monkeypatch): """setdefault must NOT overwrite a user-set NO_PROXY.""" monkeypatch.setenv("NO_PROXY", "custom.host") apply_proxy("http://corp:8080") assert os.environ["NO_PROXY"] == "custom.host"