130 lines
3.7 KiB
Python
130 lines
3.7 KiB
Python
"""Proxy parsing and runtime-status tests for scraper."""
|
|
|
|
import inspect
|
|
|
|
import pytest
|
|
|
|
import scraper
|
|
|
|
|
|
def _clear_proxy_env(monkeypatch):
|
|
monkeypatch.delenv("HTTPS_PROXY", raising=False)
|
|
monkeypatch.delenv("https_proxy", raising=False)
|
|
|
|
|
|
def _proxy_url_parser():
|
|
parser = getattr(scraper, "_get_proxy_url", None)
|
|
if parser is None:
|
|
pytest.skip("scraper does not expose a proxy URL parser")
|
|
return parser
|
|
|
|
|
|
async def _get_network_status(mock_pool, monkeypatch):
|
|
monkeypatch.setattr(
|
|
scraper,
|
|
"_proxy_enabled_effective",
|
|
None,
|
|
raising=False,
|
|
)
|
|
status_fn = getattr(scraper, "get_network_status")
|
|
signature = inspect.signature(status_fn)
|
|
if "pool" in signature.parameters:
|
|
result = status_fn(mock_pool)
|
|
if inspect.isawaitable(result):
|
|
return await result
|
|
return result
|
|
|
|
effective_proxy_fn = getattr(scraper, "_get_effective_proxy_url", None)
|
|
if effective_proxy_fn is not None:
|
|
async def fake_get_pool():
|
|
return mock_pool
|
|
|
|
monkeypatch.setattr(scraper, "get_pool", fake_get_pool, raising=False)
|
|
result = effective_proxy_fn()
|
|
if inspect.isawaitable(result):
|
|
await result
|
|
|
|
result = status_fn()
|
|
if inspect.isawaitable(result):
|
|
return await result
|
|
return result
|
|
|
|
|
|
def _block_public_ip_fetch(monkeypatch):
|
|
async def fail_fetch(*_args, **_kwargs):
|
|
raise AssertionError("network status tests must not fetch public IPs")
|
|
|
|
monkeypatch.setattr(scraper, "_fetch_public_ip", fail_fetch, raising=False)
|
|
|
|
|
|
def test_proxy_url_parser_accepts_full_proxy_url(monkeypatch):
|
|
_clear_proxy_env(monkeypatch)
|
|
monkeypatch.setenv("HTTPS_PROXY", "http://user:pass@proxy.example:8080")
|
|
|
|
assert _proxy_url_parser()() == "http://user:pass@proxy.example:8080"
|
|
|
|
|
|
def test_proxy_url_parser_builds_url_from_colon_format(monkeypatch):
|
|
_clear_proxy_env(monkeypatch)
|
|
monkeypatch.setenv("HTTPS_PROXY", "proxy.example:8080:user@example:p a/s")
|
|
|
|
assert (
|
|
_proxy_url_parser()()
|
|
== "http://user%40example:p%20a%2Fs@proxy.example:8080"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("proxy_value", ["", "proxy.example:8080:user"])
|
|
def test_proxy_url_parser_rejects_missing_or_malformed_proxy(
|
|
monkeypatch, proxy_value
|
|
):
|
|
_clear_proxy_env(monkeypatch)
|
|
monkeypatch.setenv("HTTPS_PROXY", proxy_value)
|
|
|
|
assert _proxy_url_parser()() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_network_status_reports_proxy_available_when_env_is_usable(
|
|
mock_pool, monkeypatch
|
|
):
|
|
_clear_proxy_env(monkeypatch)
|
|
_block_public_ip_fetch(monkeypatch)
|
|
monkeypatch.setenv("HTTPS_PROXY", "proxy.example:8080:user:pass")
|
|
mock_pool.fetchval.return_value = None
|
|
|
|
status = await _get_network_status(mock_pool, monkeypatch)
|
|
|
|
assert status["proxy_available"] is True
|
|
assert status["proxy_enabled"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_network_status_respects_disabled_proxy_setting(
|
|
mock_pool, monkeypatch
|
|
):
|
|
_clear_proxy_env(monkeypatch)
|
|
_block_public_ip_fetch(monkeypatch)
|
|
monkeypatch.setenv("HTTPS_PROXY", "proxy.example:8080:user:pass")
|
|
mock_pool.fetchval.return_value = "false"
|
|
|
|
status = await _get_network_status(mock_pool, monkeypatch)
|
|
|
|
assert status["proxy_available"] is True
|
|
assert status["proxy_enabled"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_network_status_reports_unavailable_for_malformed_proxy(
|
|
mock_pool, monkeypatch
|
|
):
|
|
_clear_proxy_env(monkeypatch)
|
|
_block_public_ip_fetch(monkeypatch)
|
|
monkeypatch.setenv("HTTPS_PROXY", "proxy.example:8080:user")
|
|
mock_pool.fetchval.return_value = None
|
|
|
|
status = await _get_network_status(mock_pool, monkeypatch)
|
|
|
|
assert status["proxy_available"] is False
|
|
assert status["proxy_enabled"] is False
|