109 lines
3.0 KiB
Python
109 lines
3.0 KiB
Python
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="<html><body><h1>Backend Engineer</h1></body></html>",
|
|
)
|
|
|
|
_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")
|