`tests/test_image_rail_content.py`

54 lines
"""Copyright rail, layer 2: images/scans served from EXTENSIONLESS URLs are refused on
Content-Type or magic bytes before the body is read, and nothing is cached."""
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import pytest

from conrad.crawlers import base

JPEG = b"\xff\xd8\xff\xe0" + b"\x00" * 64
ROUTES = {
    "/picture": ("image/jpeg", JPEG),              # honest image type, no extension
    "/disguised": ("text/html", JPEG),             # lies about its type
    "/scan": ("application/pdf", b"%PDF-1.4 x"),   # newspaper scan
    "/page": ("text/html; charset=utf-8", b"<html>Conrad cartoon caption</html>"),
}


class H(BaseHTTPRequestHandler):
    def do_GET(self):
        ctype, body = ROUTES[self.path]
        self.send_response(200)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *a):
        pass


@pytest.fixture()
def server(monkeypatch, tmp_path):
    monkeypatch.setattr(base.config, "REQUEST_DELAY", 0)
    monkeypatch.setattr(base.config, "CACHE_DIR", tmp_path)
    srv = HTTPServer(("127.0.0.1", 0), H)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    yield f"http://127.0.0.1:{srv.server_port}", tmp_path
    srv.shutdown()


@pytest.mark.parametrize("path", ["/picture", "/disguised", "/scan"])
def test_binary_refused_and_not_cached(server, path):
    url, cache = server
    with pytest.raises(base.Blocked, match="copyright rail"):
        base.Http().get(url + path, check_robots=False)
    assert not any((cache / "http").iterdir()), "refused response must never reach the cache"


def test_html_still_fetched(server):
    url, _ = server
    status, body = base.Http().get(url + "/page", check_robots=False)
    assert status == 200 and "Conrad cartoon caption" in body