Publish curated VacatureRadar source snapshot

tests/security/test_robots.py

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+import httpx
+import pytest
+from django.utils import timezone
+
+from apps.sources.models import Source, SourceRobotsCache
+from apps.sources.services.robots import assess_robots
+from apps.sources.services.url_security import ValidatedUrl
+
+
+def _allow_test_dns(monkeypatch):
+    monkeypatch.setattr(
+        "apps.sources.services.robots.validate_public_url",
+        lambda url, **kwargs: ValidatedUrl(
+            url=url,
+            hostname="jobs.example.org",
+            port=443,
+            addresses=("93.184.216.34",),
+        ),
+    )
+
+
+@pytest.mark.security
+def test_robots_uses_user_agent_matching_and_allows_default(db, monkeypatch):
+    _allow_test_dns(monkeypatch)
+    source = Source.objects.create(
+        name="Example jobs",
+        source_type=Source.Type.EMPLOYER,
+        base_url="https://jobs.example.org/vacatures/",
+        domain="jobs.example.org",
+        status=Source.Status.ACTIVE,
+        policy=Source.Policy.ALLOW,
+    )
+    calls = {"count": 0}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls["count"] += 1
+        assert request.url.path == "/robots.txt"
+        body = (
+            "User-agent: *\n"
+            "Disallow: /admin\n"
+            "User-agent: vacatureradar/0.1 (+local-personal-use)\n"
+            "Allow: /admin\n"
+        )
+        return httpx.Response(200, text=body)
+
+    client = httpx.Client(transport=httpx.MockTransport(handler))
+    decision = assess_robots(
+        source.base_url + "admin",
+        source=source,
+        user_agent="VacatureRadar/0.1 (+local-personal-use)",
+        client=client,
+        now=timezone.now(),
+    )
+    client.close()
+    assert decision.allowed
+    assert calls["count"] == 1
+
+
+@pytest.mark.security
+def test_robots_cache_refreshes_after_ttl(db, settings, monkeypatch):
+    _allow_test_dns(monkeypatch)
+    source = Source.objects.create(
+        name="Example jobs",
+        source_type=Source.Type.EMPLOYER,
+        base_url="https://jobs.example.org/vacatures/",
+        domain="jobs.example.org",
+        status=Source.Status.ACTIVE,
+        policy=Source.Policy.ALLOW,
+    )
+    settings.ROBOTS_CACHE_TTL_SECONDS = 0
+
+    calls = {"count": 0}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls["count"] += 1
+        return httpx.Response(200, text="User-agent: *\nAllow: /")
+
+    client = httpx.Client(transport=httpx.MockTransport(handler))
+    assess_robots(
+        source.base_url, source=source, user_agent="test-agent", client=client, now=timezone.now()
+    )
+    assert calls["count"] == 1
+    assess_robots(
+        source.base_url,
+        source=source,
+        user_agent="test-agent",
+        client=client,
+        now=timezone.now(),
+    )
+    assert calls["count"] == 2
+    assert SourceRobotsCache.objects.filter(origin__contains="jobs.example.org").exists()
+    client.close()
+
+
+@pytest.mark.security
+def test_robots_follows_and_revalidates_redirects(db, monkeypatch):
+    validated_urls = []
+
+    def validate(url, **kwargs):
+        validated_urls.append(url)
+        if "127.0.0.1" in url:
+            from apps.sources.services.url_security import UnsafeUrlError
+
+            raise UnsafeUrlError("Privéadres")
+        return ValidatedUrl(
+            url=url,
+            hostname="jobs.example.org",
+            port=443,
+            addresses=("93.184.216.34",),
+        )
+
+    monkeypatch.setattr("apps.sources.services.robots.validate_public_url", validate)
+    source = Source.objects.create(
+        name="Redirect jobs",
+        source_type=Source.Type.ATS,
+        base_url="https://ats.example.org/jobs",
+        domain="ats.example.org",
+        status=Source.Status.ACTIVE,
+        policy=Source.Policy.ALLOW,
+    )
+
+    def safe_handler(request: httpx.Request) -> httpx.Response:
+        if request.url.host == "ats.example.org":
+            return httpx.Response(
+                302,
+                headers={"location": "https://jobs.example.org/robots.txt"},
+                request=request,
+            )
+        return httpx.Response(200, text="User-agent: *\nAllow: /", request=request)
+
+    client = httpx.Client(transport=httpx.MockTransport(safe_handler))
+    assert assess_robots(source.base_url, source=source, client=client).allowed is True
+    client.close()
+    assert validated_urls == [
+        "https://ats.example.org/robots.txt",
+        "https://jobs.example.org/robots.txt",
+    ]
+
+    SourceRobotsCache.objects.all().delete()
+
+    def unsafe_handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            302, headers={"location": "http://127.0.0.1/robots.txt"}, request=request
+        )
+
+    client = httpx.Client(transport=httpx.MockTransport(unsafe_handler))
+    blocked = assess_robots(source.base_url, source=source, client=client)
+    client.close()
+    assert blocked.allowed is False
+    assert "Privéadres" in blocked.reason