search/test_crawler.py

#!/usr/bin/env python3
"""
Local test suite for the crawler (no database required).

Only the pure, DB-free parts are tested: the HostScheduler (per-host
delays, block backoff, Retry-After), the transient-error classification
and the queue-claim decision logic (blocked hosts are skipped, another
domain is picked) against an in-memory queue substitute.
"""

import os
import sys
import time
import unittest
from email import utils as email_utils
from unittest import mock

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))


def _load_env_file():
    """
    Load .env like the systemd EnvironmentFile does, so the tests also
    run outside systemd. Must happen before crawler.py is imported,
    which reads the JONASSEARCH_* variables at import time.
    """
    env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")

    try:
        with open(env_path) as handle:
            for line in handle:
                line = line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                key, _, value = line.partition("=")
                os.environ.setdefault(key.strip(), value.strip())
    except OSError:
        pass


_load_env_file()

import crawler  # noqa: E402


class StubRobots:
    """robots.txt stub: constant delay, no network."""

    def __init__(self, delay=3.0):
        self.delay_value = delay

    def delay(self, url):
        return self.delay_value


class FakeState:
    """In-memory crawl_queue substitute for claim/defer/requeue."""

    def __init__(self, rows=None):
        self.rows = rows if rows is not None else []

    def add(self, **kwargs):
        row = {
            "id": len(self.rows) + 1,
            "url": "",
            "source": None,
            "status": "pending",
            "priority": 0,
            "attempts": 0,
        }
        row.update(kwargs)
        self.rows.append(row)
        return row["id"]


class FakeCursor:
    """Just enough SQL surface for the crawler's queue functions."""

    def __init__(self, state):
        self._state = state
        self._rows = []

    def execute(self, query, params=None):
        query = " ".join(query.split())
        params = params or ()

        if "SELECT id, url, source_url" in query:
            batch = params[-1]
            exclude = set(params[0]) if len(params) == 2 else set()
            candidates = [
                row for row in self._state.rows
                if row["status"] == "pending" and row["id"] not in exclude
            ]
            candidates.sort(key=lambda row: (-row["priority"], row["id"]))
            self._rows = [
                (row["id"], row["url"], row["source"])
                for row in candidates[:batch]
            ]
        elif "SET status = 'processing'" in query:
            (queue_id,) = params
            self._rows = []
            for row in self._state.rows:
                if row["id"] == queue_id and row["status"] == "pending":
                    row["status"] = "processing"
                    row["attempts"] += 1
                    self._rows = [(row["id"], row["url"], row["source"])]
                    break
        elif "GREATEST(attempts - 1, 0)" in query:
            (queue_id,) = params
            for row in self._state.rows:
                if row["id"] == queue_id:
                    row["status"] = "pending"
                    row["attempts"] = max(row["attempts"] - 1, 0)
        elif "WHEN attempts >=" in query:
            max_attempts, queue_id = params
            self._rows = []
            for row in self._state.rows:
                if row["id"] == queue_id:
                    row["status"] = (
                        "failed" if row["attempts"] >= max_attempts
                        else "pending"
                    )
                    self._rows = [(row["status"],)]
        else:
            raise AssertionError("unexpected query: %s" % query[:120])

    def fetchone(self):
        return self._rows[0] if self._rows else None

    def fetchall(self):
        return self._rows

    def __enter__(self):
        return self

    def __exit__(self, *_exc):
        return False


class FakeConnection:
    def __init__(self, state):
        self.state = state

    def cursor(self):
        return FakeCursor(self.state)

    def commit(self):
        pass

    def rollback(self):
        pass


class HostSchedulerTest(unittest.TestCase):

    def test_first_request_is_ready(self):
        scheduler = crawler.HostScheduler()
        self.assertEqual(
            scheduler.wait_time("https://a.example/x", 3.0), 0.0
        )

    def test_delay_is_honored_after_request(self):
        scheduler = crawler.HostScheduler()
        scheduler.wait("https://a.example/x", 0.0)
        remaining = scheduler.wait_time("https://a.example/y", 3.0)
        self.assertGreater(remaining, 2.5)
        self.assertLessEqual(remaining, 3.0)

    def test_short_wait_is_slept_inline(self):
        scheduler = crawler.HostScheduler()
        scheduler.wait("https://a.example/x", 0.0)
        with mock.patch("time.sleep") as sleep:
            self.assertTrue(scheduler.wait("https://a.example/y", 3.0))
        sleep.assert_called_once()

    def test_long_wait_is_deferred_without_touching_schedule(self):
        scheduler = crawler.HostScheduler()
        scheduler.wait("https://a.example/x", 0.0)
        recorded = scheduler.last_request["a.example"]
        with mock.patch("time.sleep") as sleep:
            self.assertFalse(scheduler.wait("https://a.example/y", 60.0))
        sleep.assert_not_called()
        self.assertEqual(scheduler.last_request["a.example"], recorded)

    def test_block_backoff_doubles(self):
        scheduler = crawler.HostScheduler()
        url = "https://a.example/x"
        scheduler.note_http_error(url, 429)
        first = scheduler.wait_time("https://a.example/y", 3.0)
        scheduler.note_http_error(url, 429)
        second = scheduler.wait_time("https://a.example/y", 3.0)
        self.assertGreater(first, crawler.HOST_BLOCK_RULES[429][0] * 0.9)
        self.assertGreater(second, first * 1.5)

    def test_retry_after_is_honored(self):
        scheduler = crawler.HostScheduler()
        scheduler.note_http_error(
            "https://a.example/x", 429, retry_after=1200
        )
        remaining = scheduler.wait_time("https://a.example/y", 3.0)
        self.assertGreater(remaining, 1000)
        self.assertLessEqual(remaining, crawler.HOST_BLOCK_HARD_CAP)

    def test_block_never_exceeds_hard_cap(self):
        scheduler = crawler.HostScheduler()
        url = "https://a.example/x"
        for _ in range(40):
            scheduler.note_http_error(url, 429)
        remaining = scheduler.wait_time(url, 3.0)
        self.assertLessEqual(remaining, crawler.HOST_BLOCK_HARD_CAP + 1)

    def test_success_resets_block_counter(self):
        scheduler = crawler.HostScheduler()
        url = "https://a.example/x"
        scheduler.note_http_error(url, 429)
        scheduler.note_http_error(url, 429)
        scheduler.note_success(url)
        scheduler.note_http_error(url, 429)
        remaining = scheduler.wait_time("https://a.example/y", 3.0)
        self.assertLessEqual(
            remaining, crawler.HOST_BLOCK_RULES[429][0] + 1
        )

    def test_non_block_status_is_ignored(self):
        scheduler = crawler.HostScheduler()
        scheduler.note_http_error("https://a.example/x", 404)
        self.assertEqual(
            scheduler.wait_time("https://a.example/y", 3.0), 0.0
        )

    def test_network_error_backs_off(self):
        scheduler = crawler.HostScheduler()
        scheduler.note_network_error("https://a.example/x")
        remaining = scheduler.wait_time("https://a.example/y", 3.0)
        self.assertGreater(remaining, 20)

    def test_unrelated_host_is_not_blocked(self):
        scheduler = crawler.HostScheduler()
        scheduler.note_http_error("https://a.example/x", 429)
        self.assertEqual(
            scheduler.wait_time("https://b.example/y", 3.0), 0.0
        )


class ErrorClassificationTest(unittest.TestCase):

    def test_network_errors_are_transient(self):
        self.assertTrue(crawler.is_transient_error("network:timeout"))
        self.assertTrue(crawler.is_transient_error("network:[Errno -2] x"))

    def test_rate_limit_and_server_errors_are_transient(self):
        for status in (408, 425, 429, 500, 502, 503, 504):
            self.assertTrue(crawler.is_transient_error("http_%d" % status))

    def test_terminal_errors_are_not_transient(self):
        for error in (
            "http_404", "http_410", "robots", "too_large",
            "not_html", "error:boom", "host_busy", None, "",
        ):
            self.assertFalse(crawler.is_transient_error(error))

    def test_retry_after_seconds_digit(self):
        self.assertEqual(
            crawler._retry_after_seconds({"Retry-After": "120"}), 120.0
        )

    def test_retry_after_http_date(self):
        headers = {
            "Retry-After": email_utils.formatdate(
                time.time() + 120, usegmt=True
            )
        }
        value = crawler._retry_after_seconds(headers)
        self.assertGreater(value, 100)
        self.assertLess(value, 140)

    def test_retry_after_missing_or_garbage(self):
        self.assertIsNone(crawler._retry_after_seconds({}))
        self.assertIsNone(
            crawler._retry_after_seconds({"Retry-After": "soon"})
        )
        self.assertIsNone(crawler._retry_after_seconds(None))


class QueueClaimTest(unittest.TestCase):

    def _state(self):
        state = FakeState()
        state.add(url="https://blocked.example/a", priority=500)
        state.add(url="https://other.example/b", priority=100)
        state.add(url="https://blocked.example/c", priority=400)
        return state

    def _claim(self, state, scheduler=None):
        scheduler = scheduler or crawler.HostScheduler()
        return crawler.claim_next_queue_item(
            FakeConnection(state), scheduler, StubRobots()
        )

    def test_highest_priority_ready_host_wins(self):
        item, wait_hint = self._claim(self._state())
        self.assertEqual(item[1], "https://blocked.example/a")
        self.assertIsNone(wait_hint)

    def test_blocked_host_is_skipped_for_other_domain(self):
        scheduler = crawler.HostScheduler()
        scheduler.blocked_until["blocked.example"] = time.monotonic() + 900
        item, wait_hint = self._claim(self._state(), scheduler)
        self.assertEqual(item[1], "https://other.example/b")
        self.assertGreater(wait_hint, 800)

    def test_rate_limited_host_is_skipped(self):
        scheduler = crawler.HostScheduler()
        scheduler.wait("https://blocked.example/a", 0.0)
        item, _wait_hint = self._claim(self._state(), scheduler)
        self.assertEqual(item[1], "https://other.example/b")

    def test_only_blocked_hosts_left_reports_wait_hint(self):
        state = FakeState()
        state.add(url="https://blocked.example/a", priority=500)
        scheduler = crawler.HostScheduler()
        scheduler.blocked_until["blocked.example"] = time.monotonic() + 900
        item, wait_hint = self._claim(state, scheduler)
        self.assertIsNone(item)
        self.assertGreater(wait_hint, 800)

    def test_empty_queue(self):
        item, wait_hint = self._claim(FakeState())
        self.assertIsNone(item)
        self.assertIsNone(wait_hint)

    def test_claim_marks_processing_and_counts_attempt(self):
        state = self._state()
        self._claim(state)
        self.assertEqual(state.rows[0]["status"], "processing")
        self.assertEqual(state.rows[0]["attempts"], 1)
        self.assertEqual(state.rows[1]["status"], "pending")

    def test_claim_finds_ready_host_behind_blocked_wall(self):
        state = FakeState()
        for index in range(30):
            state.add(
                url="https://blocked.example/%d" % index,
                priority=300 - index,
            )
        state.add(url="https://ready.example/x", priority=10)
        scheduler = crawler.HostScheduler()
        scheduler.blocked_until["blocked.example"] = time.monotonic() + 900
        item, _wait_hint = self._claim(state, scheduler)
        self.assertEqual(item[1], "https://ready.example/x")

    def test_defer_does_not_burn_an_attempt(self):
        state = FakeState()
        queue_id = state.add(
            url="https://slow.example/a",
            status="processing",
            attempts=1,
        )
        crawler.defer_queue_item(FakeConnection(state), queue_id)
        row = state.rows[0]
        self.assertEqual(row["status"], "pending")
        self.assertEqual(row["attempts"], 0)

    def test_requeue_retries_below_attempt_cap(self):
        state = FakeState()
        queue_id = state.add(
            url="https://x.example/a",
            status="processing",
            attempts=1,
        )
        status = crawler.requeue_queue_item(FakeConnection(state), queue_id)
        self.assertEqual(status, "pending")
        self.assertEqual(state.rows[0]["status"], "pending")

    def test_requeue_fails_after_attempt_cap(self):
        state = FakeState()
        queue_id = state.add(
            url="https://x.example/a",
            status="processing",
            attempts=crawler.QUEUE_MAX_ATTEMPTS,
        )
        status = crawler.requeue_queue_item(FakeConnection(state), queue_id)
        self.assertEqual(status, "failed")
        self.assertEqual(state.rows[0]["status"], "failed")


class HostFilterTest(unittest.TestCase):

    def test_localhost_and_private_suffixes_are_blocked(self):
        for host in (
            "localhost", "sub.localhost", "localhost.localdomain",
            "myserver.local", "app.internal",
        ):
            self.assertTrue(
                crawler.is_non_crawlable_host(host), host
            )

    def test_ip_literals_are_blocked(self):
        for host in (
            "127.0.0.1", "10.0.0.5", "192.168.1.10", "172.16.0.1",
            "8.8.8.8", "::1", "fe80::1", "2001:db8::1",
            "::ffff:127.0.0.1",
        ):
            self.assertTrue(
                crawler.is_non_crawlable_host(host), host
            )

    def test_single_label_and_empty_names_are_blocked(self):
        for host in ("mymachine", "", "   ", "."):
            self.assertTrue(
                crawler.is_non_crawlable_host(host), repr(host)
            )

    def test_real_domains_are_allowed(self):
        for host in (
            "example.com", "www.example.com",
            "sub.domain.example.co.uk", "xn--bcher-kva.example",
            "example.com.",
        ):
            self.assertFalse(
                crawler.is_non_crawlable_host(host), host
            )

    def test_normalize_url_rejects_non_public_targets(self):
        for url in (
            "http://localhost/x",
            "http://127.0.0.1/x",
            "http://[::1]:8080/x",
            "http://10.1.2.3/a?b=c",
            "http://intranet.local/x",
            "http://host/x",
        ):
            self.assertIsNone(crawler.normalize_url(url), url)

    def test_normalize_url_keeps_real_domains(self):
        self.assertEqual(
            crawler.normalize_url("http://www.example.com/x?y=1"),
            "http://www.example.com/x?y=1",
        )

    def test_is_non_crawlable_url_handles_malformed_urls(self):
        self.assertTrue(crawler.is_non_crawlable_url("http://localhost/x"))
        self.assertTrue(crawler.is_non_crawlable_url("http://[::1]/"))
        self.assertFalse(crawler.is_non_crawlable_url("https://example.com/"))

    def test_malformed_url_is_treated_as_non_crawlable(self):
        # urlparse raises ValueError on malformed IPv6 brackets; the
        # filter must not crash on it.
        self.assertTrue(
            crawler.is_non_crawlable_url("http://[::1")
        )


if __name__ == "__main__":
    unittest.main()