mirror of
https://github.com/MadsLorentzen/ai-job-search.git
synced 2026-11-06 06:57:50 +00:00
fix(fetch-page): pass --compressed only when curl supports it (#759)
Fixes #705. Probes curl --version once per process and leaves --compressed out when the Features line lists no decoder (libz, brotli, zstd), so the System32 curl on Windows no longer rejects every fetch. Same design as #713 by @DevarshP610, which was opened first.
This commit is contained in:
1 parent
87d1857139
commit
256237903e
2 files changed
+61
-1
No files matched your search
@@ -349,6 +349,16 @@ class Fetching(FetchPageFixture):
|
||||
self.assertEqual(self.out.read_text(encoding="utf-8"), PAGE)
|
||||
self.assertIn("Visible title", out)
|
||||
|
||||
def test_fetch_succeeds_when_curl_lacks_compression(self):
|
||||
orig = fetch_page._curl_supports_compressed
|
||||
fetch_page._curl_supports_compressed = lambda: False
|
||||
self.addCleanup(setattr, fetch_page, "_curl_supports_compressed", orig)
|
||||
code, out, err = self.run_main(f"{self.base}/ok.html", "--out", str(self.out), "--text")
|
||||
self.assertEqual(code, 0, err)
|
||||
self.assertIn("HTTP 200 size=", out)
|
||||
self.assertEqual(self.out.read_text(encoding="utf-8"), PAGE)
|
||||
self.assertIn("Visible title", out)
|
||||
|
||||
|
||||
class CurlArgvShape(unittest.TestCase):
|
||||
"""The grant is `Bash(python tools/fetch_page.py:*)`, so what curl is handed
|
||||
@@ -374,6 +384,38 @@ class CurlArgvShape(unittest.TestCase):
|
||||
import robots_check
|
||||
self.assertEqual(fetch_page.BROWSER, robots_check.BROWSER)
|
||||
|
||||
def test_curl_compressed_flag_requires_decoder_support(self):
|
||||
orig = fetch_page._curl_supports_compressed
|
||||
try:
|
||||
fetch_page._curl_supports_compressed = lambda: False
|
||||
argv_no_decoder = fetch_page.build_curl_argv("https://example.com", Path("out.html"), Path("headers.txt"), 45)
|
||||
self.assertNotIn("--compressed", argv_no_decoder)
|
||||
|
||||
fetch_page._curl_supports_compressed = lambda: True
|
||||
argv_with_decoder = fetch_page.build_curl_argv("https://example.com", Path("out.html"), Path("headers.txt"), 45)
|
||||
self.assertIn("--compressed", argv_with_decoder)
|
||||
finally:
|
||||
fetch_page._curl_supports_compressed = orig
|
||||
|
||||
def test_curl_features_line_controls_compressed_support(self):
|
||||
from unittest.mock import Mock, patch
|
||||
cases = {
|
||||
"Features: AsynchDNS HTTPS-proxy IPv6": False,
|
||||
"Features: AsynchDNS HTTPS-proxy IPv6 libz brotli zstd": True,
|
||||
"Features: libz": True,
|
||||
"Features: brotli": True,
|
||||
"Features: zstd": True,
|
||||
}
|
||||
for features, expected in cases.items():
|
||||
with self.subTest(features=features):
|
||||
with patch.object(fetch_page.subprocess, "run", return_value=Mock(returncode=0, stdout=features)):
|
||||
fetch_page._curl_supports_compressed.cache_clear()
|
||||
self.assertEqual(fetch_page._curl_supports_compressed(), expected)
|
||||
with patch.object(fetch_page.subprocess, "run", side_effect=OSError("not found")):
|
||||
fetch_page._curl_supports_compressed.cache_clear()
|
||||
self.assertFalse(fetch_page._curl_supports_compressed())
|
||||
fetch_page._curl_supports_compressed.cache_clear()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+19
-1
@@ -39,6 +39,7 @@ Exit codes: 0 fetched; 1 refused by robots policy (go to escalation step 4);
|
||||
"""
|
||||
|
||||
import argparse
|
||||
from functools import lru_cache
|
||||
import gzip
|
||||
import hashlib
|
||||
import html
|
||||
@@ -68,14 +69,31 @@ MAX_REDIRECTS = 5
|
||||
MAX_BYTES = 10 * 1024 * 1024
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def _curl_supports_compressed() -> bool:
|
||||
"""Return whether the bare curl used for fetching can decode content."""
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
["curl", "--version"],
|
||||
capture_output=True, text=True, encoding="utf-8", errors="replace",
|
||||
timeout=5,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return False
|
||||
return proc.returncode == 0 and bool(
|
||||
re.search(r"^Features:.*\b(?:libz|brotli|zstd)\b", proc.stdout, re.IGNORECASE | re.MULTILINE)
|
||||
)
|
||||
|
||||
|
||||
def build_curl_argv(url: str, out: Path, headers: Path, timeout: float) -> list:
|
||||
"""One GET without automatic redirects, so the next hop can be gated."""
|
||||
compressed = ["--compressed"] if _curl_supports_compressed() else []
|
||||
return [
|
||||
"curl", "-sS",
|
||||
"--max-redirs", "0",
|
||||
"--max-time", str(timeout),
|
||||
"--max-filesize", str(MAX_BYTES),
|
||||
"--compressed",
|
||||
*compressed,
|
||||
"-A", BROWSER,
|
||||
"-H", f"Accept: {ACCEPT}",
|
||||
"-H", f"Accept-Language: {ACCEPT_LANGUAGE}",
|
||||
|
||||
Reference in new issue
Block a user