VacatureRadar/tests/unit/test_fetcher.py

import httpx
import pytest
from django.test import override_settings

from apps.sources.services.fetcher import (
    ContentRejectedError,
    FetchError,
    FetchTimeoutError,
    PolicyBlockedError,
    RateLimitedError,
    fetch_url,
)
from apps.sources.services.url_security import ValidatedUrl

PUBLIC_DNS = [(None, None, None, None, ("93.184.216.34", 443))]


def _patch_dns(monkeypatch):
    monkeypatch.setattr(
        "apps.sources.services.url_security.socket.getaddrinfo",
        lambda *args, **kwargs: PUBLIC_DNS,
    )
    # validate_public_url captured the default resolver at definition time, so patch call site too.
    original = __import__(
        "apps.sources.services.url_security", fromlist=["validate_public_url"]
    ).validate_public_url

    def validate(url, **kwargs):
        return original(url, resolver=lambda *a, **k: PUBLIC_DNS, **kwargs)

    monkeypatch.setattr("apps.sources.services.fetcher.validate_public_url", validate)


def test_fetcher_success_redirect_304_and_hash(monkeypatch):
    _patch_dns(monkeypatch)

    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/start":
            return httpx.Response(302, headers={"location": "/job"})
        return httpx.Response(
            200,
            headers={"content-type": "text/html; charset=utf-8"},
            content=b"Vacature",
        )

    client = httpx.Client(transport=httpx.MockTransport(handler))
    document = fetch_url("https://example.org/start", client=client)
    assert document.final_url == "https://example.org/job"
    assert document.text == "Vacature"
    assert len(document.sha256) == 64
    client.close()

    client = httpx.Client(
        transport=httpx.MockTransport(lambda request: httpx.Response(304, headers={"etag": "x"}))
    )
    unchanged = fetch_url("https://example.org/job", client=client)
    assert unchanged.status_code == 304
    assert unchanged.content == b""
    client.close()


def test_fetcher_respects_rate_limit_and_retry_after(monkeypatch):
    _patch_dns(monkeypatch)
    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda request: httpx.Response(
                429, headers={"retry-after": "45"}, content=b"too many requests"
            )
        )
    )
    with pytest.raises(RateLimitedError) as exc:
        fetch_url("https://example.org/too-fast", client=client)
    assert exc.value.retry_after_seconds == 45
    client.close()


def test_fetcher_rejects_dns_rebinding(monkeypatch):
    _patch_dns(monkeypatch)
    calls = 0

    def validate(url: str, **kwargs):
        nonlocal calls
        calls += 1
        address = "93.184.216.34" if calls == 1 else "198.51.100.12"
        return ValidatedUrl(url=url, hostname="example.org", port=443, addresses=(address,))

    monkeypatch.setattr("apps.sources.services.fetcher.validate_public_url", validate)

    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda request: httpx.Response(
                200,
                headers={"content-type": "text/html; charset=utf-8"},
                content=b"Vacature",
            )
        )
    )
    with pytest.raises(FetchError, match="DNS-rebindcontrole"):
        fetch_url("https://example.org/start", client=client)
    client.close()


class _PeerStream:
    def __init__(self, address: str) -> None:
        self.address = address

    def get_extra_info(self, key: str):
        return (self.address, 443) if key == "server_addr" else None


def test_fetcher_accepts_public_cdn_rotation_when_connected_peer_is_public(monkeypatch):
    calls = 0

    def validate(url: str, **kwargs):
        nonlocal calls
        calls += 1
        address = "150.171.109.35" if calls == 1 else "150.171.109.36"
        return ValidatedUrl(url=url, hostname="cdn.example.org", port=443, addresses=(address,))

    monkeypatch.setattr("apps.sources.services.fetcher.validate_public_url", validate)

    def handler(request: httpx.Request) -> httpx.Response:
        status_code = 302 if request.url.path == "/start" else 200
        headers = {"location": "/jobs"} if status_code == 302 else {"content-type": "text/html"}
        return httpx.Response(
            status_code,
            headers=headers,
            content=b"" if status_code == 302 else b"Vacature",
            extensions={"network_stream": _PeerStream("150.171.109.36")},
        )

    client = httpx.Client(transport=httpx.MockTransport(handler))

    document = fetch_url("https://cdn.example.org/start", client=client)

    assert document.status_code == 200
    assert document.final_url == "https://cdn.example.org/jobs"
    client.close()


def test_fetcher_rejects_private_connected_peer(monkeypatch):
    _patch_dns(monkeypatch)
    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda request: httpx.Response(
                200,
                headers={"content-type": "text/html"},
                content=b"Vacature",
                extensions={"network_stream": _PeerStream("10.0.0.5")},
            )
        )
    )

    with pytest.raises(FetchError, match="Niet-publiek verbonden IP-adres"):
        fetch_url("https://example.org/jobs", client=client)
    client.close()


def test_fetcher_rejects_policy_content_size_and_http(monkeypatch):
    _patch_dns(monkeypatch)
    with pytest.raises(PolicyBlockedError):
        fetch_url("https://linkedin.com/jobs", client=httpx.Client())

    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda request: httpx.Response(200, headers={"content-type": "image/png"}, content=b"x")
        )
    )
    with pytest.raises(ContentRejectedError):
        fetch_url("https://example.org/image", client=client)
    client.close()

    with override_settings(FETCHER_MAX_BYTES=2):
        client = httpx.Client(
            transport=httpx.MockTransport(
                lambda request: httpx.Response(
                    200, headers={"content-type": "text/plain"}, content=b"long"
                )
            )
        )
        with pytest.raises(ContentRejectedError):
            fetch_url("https://example.org/large", client=client)
        client.close()

    client = httpx.Client(
        transport=httpx.MockTransport(lambda request: httpx.Response(500, content=b"error"))
    )
    with pytest.raises(FetchError, match="HTTP 500"):
        fetch_url("https://example.org/error", client=client)
    client.close()

    client = httpx.Client(
        transport=httpx.MockTransport(
            lambda request: (_ for _ in ()).throw(httpx.TimeoutException("timeout"))
        )
    )
    with pytest.raises(FetchTimeoutError):
        fetch_url("https://example.org/slow", client=client)
    client.close()


def test_fetcher_rejects_redirect_without_location(monkeypatch):
    _patch_dns(monkeypatch)
    client = httpx.Client(transport=httpx.MockTransport(lambda request: httpx.Response(302)))
    with pytest.raises(FetchError, match="Location"):
        fetch_url("https://example.org/start", client=client)
    client.close()