from __future__ import annotations from typing import Any import httpx import pytest from resume_agent import web def _mock_client( monkeypatch: pytest.MonkeyPatch, handler: Any, captured: dict[str, Any] | None = None, ) -> None: original_client = httpx.Client transport = httpx.MockTransport(handler) def client_factory(**kwargs: Any) -> httpx.Client: if captured is not None: captured.update(kwargs) return original_client(transport=transport, **kwargs) monkeypatch.setattr(web.httpx, "Client", client_factory) monkeypatch.setattr(web, "_validate_public_url", lambda _url: None) def test_job_fetch_retries_transient_connection_timeout( monkeypatch: pytest.MonkeyPatch, ) -> None: attempts = 0 def handler(request: httpx.Request) -> httpx.Response: nonlocal attempts attempts += 1 if attempts == 1: raise httpx.ConnectTimeout("timed out", request=request) return httpx.Response( 200, headers={"content-type": "text/html; charset=utf-8"}, text="

Backend Engineer

", ) _mock_client(monkeypatch, handler) result = web.fetch_job_page( "https://jobs.example/role", timeout_seconds=120, retries=1, retry_backoff_seconds=0, ) assert result == "Backend Engineer" assert attempts == 2 def test_job_fetch_reports_timeout_as_job_page_error( monkeypatch: pytest.MonkeyPatch, ) -> None: attempts = 0 def handler(request: httpx.Request) -> httpx.Response: nonlocal attempts attempts += 1 raise httpx.ConnectTimeout("timed out", request=request) _mock_client(monkeypatch, handler) with pytest.raises(web.JobPageError, match="after 3 attempts"): web.fetch_job_page( "https://jobs.example/role", timeout_seconds=120, retries=2, retry_backoff_seconds=0, ) assert attempts == 3 def test_job_fetch_timeout_is_configurable( monkeypatch: pytest.MonkeyPatch, ) -> None: captured: dict[str, Any] = {} def handler(_request: httpx.Request) -> httpx.Response: return httpx.Response( 200, headers={"content-type": "text/plain"}, text="Complete job description", ) _mock_client(monkeypatch, handler, captured) monkeypatch.setenv("RESUME_AGENT_JOB_FETCH_TIMEOUT_SECONDS", "240") monkeypatch.setenv("RESUME_AGENT_JOB_FETCH_RETRIES", "0") assert web.fetch_job_page("https://jobs.example/role") == "Complete job description" timeout = captured["timeout"] assert timeout.connect == 60 assert timeout.read == 240 def test_job_fetch_rejects_invalid_timeout_configuration( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("RESUME_AGENT_JOB_FETCH_TIMEOUT_SECONDS", "not-a-number") with pytest.raises(web.JobPageError, match="must be a number"): web.fetch_job_page("https://jobs.example/role")