diff --git a/tests/spiders/test_links.py b/tests/spiders/test_links.py
new file mode 100644
index 0000000..2413992
--- /dev/null
+++ b/tests/spiders/test_links.py
@@ -0,0 +1,243 @@
+"""Tests for `LinkExtractor`."""
+
+import re
+
+import pytest
+
+from scrapling.engines.toolbelt.custom import Response
+from scrapling.spiders.links import IGNORED_EXTENSIONS, LinkExtractor
+
+
+def _make_response(html: str, url: str = "https://example.com/page") -> Response:
+ """Build a minimal Response wrapping the given HTML."""
+ return Response(
+ url=url,
+ content=html,
+ status=200,
+ reason="OK",
+ cookies={},
+ headers={},
+ request_headers={},
+ )
+
+
+HTML_BASIC = """
+
+ post 1
+ post 2
+ external
+ about
+ mail
+ js
+ pdf
+ area
+
+
+"""
+
+
+class TestExtractBasic:
+ def test_default_extracts_a_and_area_with_href(self):
+ resp = _make_response(HTML_BASIC)
+ urls = LinkExtractor().extract(resp)
+ # mailto/javascript filtered (non-http scheme), .pdf filtered (deny_extensions)
+ # link[rel=stylesheet] not in default tags
+ assert "https://example.com/posts/1" in urls
+ assert "https://example.com/posts/2" in urls
+ assert "https://example.com/about" in urls
+ assert "https://example.com/area-link" in urls
+ assert "https://other.com/page" in urls
+ assert all(not u.startswith("mailto:") for u in urls)
+ assert all(not u.startswith("javascript:") for u in urls)
+ assert not any(u.endswith(".pdf") for u in urls)
+ assert not any(u.endswith(".css") for u in urls)
+
+ def test_relative_urls_become_absolute_via_urljoin(self):
+ resp = _make_response('x', url="https://example.com/sub/")
+ assert LinkExtractor().extract(resp) == ["https://example.com/sub/foo/bar"]
+
+ def test_empty_allow_means_match_all(self):
+ resp = _make_response('xy')
+ out = LinkExtractor().extract(resp)
+ assert len(out) == 2
+
+
+class TestAllowDeny:
+ def test_allow_regex_filters_in(self):
+ resp = _make_response(HTML_BASIC)
+ urls = LinkExtractor(allow=r"/posts/").extract(resp)
+ assert urls == ["https://example.com/posts/1", "https://example.com/posts/2"]
+
+ def test_deny_regex_filters_out(self):
+ resp = _make_response(HTML_BASIC)
+ urls = LinkExtractor(deny=r"/posts/").extract(resp)
+ assert "https://example.com/posts/1" not in urls
+ assert "https://example.com/about" in urls
+
+ def test_deny_overrides_allow(self):
+ resp = _make_response(HTML_BASIC)
+ urls = LinkExtractor(allow=r"/posts/", deny=r"/posts/2").extract(resp)
+ assert urls == ["https://example.com/posts/1"]
+
+ def test_compiled_pattern_accepted(self):
+ resp = _make_response(HTML_BASIC)
+ pat = re.compile(r"/posts/\d+$")
+ urls = LinkExtractor(allow=pat).extract(resp)
+ assert len(urls) == 2
+
+ def test_iterable_of_patterns(self):
+ resp = _make_response(HTML_BASIC)
+ urls = LinkExtractor(allow=[r"/posts/", r"/about"]).extract(resp)
+ assert "https://example.com/posts/1" in urls
+ assert "https://example.com/about" in urls
+
+
+class TestDomains:
+ def test_allow_domains_keeps_only_matching(self):
+ resp = _make_response(HTML_BASIC)
+ urls = LinkExtractor(allow_domains="example.com").extract(resp)
+ assert all("example.com" in u for u in urls)
+ assert "https://other.com/page" not in urls
+
+ def test_allow_domains_matches_subdomains(self):
+ html = 'ab'
+ resp = _make_response(html)
+ urls = LinkExtractor(allow_domains="example.com").extract(resp)
+ assert urls == ["https://api.example.com/x"]
+
+ def test_deny_domains_filters_out(self):
+ resp = _make_response(HTML_BASIC)
+ urls = LinkExtractor(deny_domains="other.com").extract(resp)
+ assert "https://other.com/page" not in urls
+ assert "https://example.com/posts/1" in urls
+
+
+class TestRestrict:
+ def test_restrict_css_scopes_extraction(self):
+ html = """
+
+
+ m
+
+ """
+ resp = _make_response(html)
+ urls = LinkExtractor(restrict_css="main").extract(resp)
+ assert urls == ["https://example.com/main-link"]
+
+ def test_restrict_xpath_scopes_extraction(self):
+ html = """
+
+
+
+
+ """
+ resp = _make_response(html)
+ urls = LinkExtractor(restrict_xpath='//div[@id="content"]').extract(resp)
+ assert urls == ["https://example.com/c"]
+
+
+class TestTagsAttrs:
+ def test_custom_tags_and_attrs_for_stylesheets(self):
+ html = 'p'
+ resp = _make_response(html)
+ # Override deny_extensions to allow .css through, and pick up
+ urls = LinkExtractor(tags=("link",), attrs=("href",), deny_extensions=()).extract(resp)
+ assert urls == ["https://example.com/style.css"]
+
+
+class TestCanonicalization:
+ def test_query_params_sorted(self):
+ resp = _make_response('x')
+ urls = LinkExtractor().extract(resp)
+ assert urls == ["https://example.com/x?a=1&b=2"]
+
+ def test_fragment_dropped_by_default(self):
+ resp = _make_response('x')
+ urls = LinkExtractor().extract(resp)
+ assert urls == ["https://example.com/x"]
+
+ def test_keep_fragment_preserves_it(self):
+ resp = _make_response('x')
+ urls = LinkExtractor(keep_fragment=True).extract(resp)
+ assert urls == ["https://example.com/x#section"]
+
+ def test_canonicalize_off_leaves_url_unchanged(self):
+ resp = _make_response('x')
+ urls = LinkExtractor(canonicalize=False).extract(resp)
+ assert urls == ["https://example.com/x?b=2&a=1#f"]
+
+
+class TestDedup:
+ def test_unique_drops_duplicates(self):
+ html = 'abc'
+ resp = _make_response(html)
+ urls = LinkExtractor().extract(resp)
+ # canonicalize collapses /x and /x? together
+ assert urls == ["https://example.com/x"]
+
+
+class TestExtensions:
+ def test_default_deny_extensions_drops_pdf_zip_images(self):
+ html = 'pdfzippngok'
+ resp = _make_response(html)
+ urls = LinkExtractor().extract(resp)
+ assert urls == ["https://example.com/d"]
+
+ def test_custom_deny_extensions_overrides_default(self):
+ html = 'pdfzip'
+ resp = _make_response(html)
+ urls = LinkExtractor(deny_extensions={"zip"}).extract(resp)
+ # .pdf now allowed because we replaced the set
+ assert urls == ["https://example.com/a.pdf"]
+
+ def test_empty_deny_extensions_allows_everything(self):
+ html = 'pdf'
+ resp = _make_response(html)
+ urls = LinkExtractor(deny_extensions=()).extract(resp)
+ assert urls == ["https://example.com/a.pdf"]
+
+
+class TestStrip:
+ def test_strip_removes_whitespace(self):
+ resp = _make_response('x')
+ urls = LinkExtractor().extract(resp)
+ assert urls == ["https://example.com/spaced"]
+
+
+class TestMatches:
+ def test_matches_honors_allow(self):
+ ex = LinkExtractor(allow=r"/posts/")
+ assert ex.matches("https://example.com/posts/1") is True
+ assert ex.matches("https://example.com/about") is False
+
+ def test_matches_honors_deny(self):
+ ex = LinkExtractor(deny=r"/admin")
+ assert ex.matches("https://example.com/admin/x") is False
+ assert ex.matches("https://example.com/posts/1") is True
+
+ def test_matches_honors_allow_domains(self):
+ ex = LinkExtractor(allow_domains="example.com")
+ assert ex.matches("https://api.example.com/x") is True
+ assert ex.matches("https://other.com/x") is False
+
+ def test_matches_honors_deny_extensions(self):
+ ex = LinkExtractor()
+ assert ex.matches("https://example.com/file.pdf") is False
+ assert ex.matches("https://example.com/page") is True
+
+ def test_matches_rejects_non_http_schemes(self):
+ ex = LinkExtractor()
+ assert ex.matches("mailto:x@example.com") is False
+ assert ex.matches("javascript:void(0)") is False
+ assert ex.matches("ftp://example.com/x") is False
+
+ def test_matches_canonicalizes_before_checking(self):
+ ex = LinkExtractor(allow=r"a=1&b=2$")
+ # the URL has params in the wrong order; canonicalize sorts them
+ assert ex.matches("https://example.com/x?b=2&a=1") is True
+
+
+class TestIgnoredExtensions:
+ def test_constant_includes_common_binary_types(self):
+ for ext in ("pdf", "zip", "png", "mp4", "exe"):
+ assert ext in IGNORED_EXTENSIONS
diff --git a/tests/spiders/test_sitemap.py b/tests/spiders/test_sitemap.py
new file mode 100644
index 0000000..cd5739b
--- /dev/null
+++ b/tests/spiders/test_sitemap.py
@@ -0,0 +1,392 @@
+"""Tests for `SitemapParser` and `SitemapSpider`."""
+
+import gzip
+import pickle
+
+import pytest
+
+from scrapling.engines.toolbelt.custom import Response
+from scrapling.spiders.links import LinkExtractor
+from scrapling.spiders.request import Request
+from scrapling.spiders.sitemap import SitemapParser, SitemapResult, SitemapSpider, SitemapUrl
+from scrapling.spiders.templates import CrawlRule
+from scrapling.core._types import Any, AsyncGenerator, Dict, Union
+
+
+URLSET_XML = b"""
+
+
+ https://example.com/posts/1
+ 2026-01-15
+ daily
+ 0.8
+
+
+ https://example.com/posts/2
+ 2026-02-20
+
+
+ https://example.com/about
+
+
+"""
+
+URLSET_WITH_ALTERNATES = b"""
+
+
+ https://example.com/en/page
+
+
+
+
+"""
+
+INDEX_XML = b"""
+
+ https://example.com/posts-sitemap.xml
+ https://example.com/products-sitemap.xml
+ https://example.com/skip-sitemap.xml
+
+"""
+
+# Sitemap without the standard namespace (some sites do this)
+URLSET_NO_NS = b"""
+
+ https://example.com/x
+
+"""
+
+
+class TestSitemapParserUrlset:
+ def test_parse_urlset_with_full_metadata(self):
+ result = SitemapParser().parse(URLSET_XML)
+ assert len(result.urls) == 3
+ assert result.sitemaps == []
+ first = result.urls[0]
+ assert first.loc == "https://example.com/posts/1"
+ assert first.lastmod == "2026-01-15"
+ assert first.changefreq == "daily"
+ assert first.priority == 0.8
+
+ def test_parse_handles_partial_metadata(self):
+ result = SitemapParser().parse(URLSET_XML)
+ second = result.urls[1]
+ assert second.lastmod == "2026-02-20"
+ assert second.changefreq is None
+ assert second.priority is None
+
+ def test_parse_handles_no_namespace(self):
+ result = SitemapParser().parse(URLSET_NO_NS)
+ assert len(result.urls) == 1
+ assert result.urls[0].loc == "https://example.com/x"
+
+
+class TestSitemapParserAlternates:
+ def test_alternate_links_off_by_default(self):
+ result = SitemapParser().parse(URLSET_WITH_ALTERNATES)
+ assert result.urls[0].alternates == []
+
+ def test_alternate_links_on_collects_them(self):
+ result = SitemapParser(alternate_links=True).parse(URLSET_WITH_ALTERNATES)
+ assert result.urls[0].alternates == [
+ "https://example.com/fr/page",
+ "https://example.com/de/page",
+ ]
+
+
+class TestSitemapParserIndex:
+ def test_parse_sitemapindex_returns_child_sitemaps(self):
+ result = SitemapParser().parse(INDEX_XML)
+ assert result.urls == []
+ assert result.sitemaps == [
+ "https://example.com/posts-sitemap.xml",
+ "https://example.com/products-sitemap.xml",
+ "https://example.com/skip-sitemap.xml",
+ ]
+
+
+class TestSitemapParserDecompression:
+ def test_gz_body_via_magic_bytes(self):
+ compressed = gzip.compress(URLSET_XML)
+ result = SitemapParser().parse(compressed)
+ assert len(result.urls) == 3
+
+ def test_gz_body_via_content_type_hint(self):
+ compressed = gzip.compress(URLSET_XML)
+ result = SitemapParser().parse(compressed, content_type="application/x-gzip")
+ assert len(result.urls) == 3
+
+ def test_corrupt_gz_logged_not_raised(self):
+ # Body starts with gzip magic but is not valid gzip
+ body = b"\x1f\x8b" + b"junk data"
+ result = SitemapParser().parse(body)
+ assert result == SitemapResult()
+
+
+class TestSitemapParserMalformed:
+ def test_invalid_xml_returns_empty_result(self):
+ result = SitemapParser().parse(b"")
+ assert result == SitemapResult()
+
+
+class TestFromRobotsTxt:
+ def test_extracts_sitemap_directives(self):
+ body = """
+ User-agent: *
+ Disallow: /admin
+ Sitemap: https://example.com/sitemap.xml
+ Sitemap: https://example.com/posts-sitemap.xml
+ """
+ urls = SitemapParser.from_robots_txt(body)
+ assert urls == [
+ "https://example.com/sitemap.xml",
+ "https://example.com/posts-sitemap.xml",
+ ]
+
+ def test_ignores_comments_and_blank_lines(self):
+ body = """
+ # This is a comment
+ Sitemap: https://example.com/sitemap.xml # inline comment
+ # Sitemap: https://commented.example.com/sitemap.xml
+ """
+ urls = SitemapParser.from_robots_txt(body)
+ assert urls == ["https://example.com/sitemap.xml"]
+
+ def test_directive_match_is_case_insensitive(self):
+ body = "SITEMAP: https://example.com/sitemap.xml\nsitemap: https://example.com/other.xml"
+ urls = SitemapParser.from_robots_txt(body)
+ assert urls == [
+ "https://example.com/sitemap.xml",
+ "https://example.com/other.xml",
+ ]
+
+ def test_returns_empty_when_no_directives(self):
+ body = "User-agent: *\nDisallow: /admin"
+ urls = SitemapParser.from_robots_txt(body)
+ assert urls == []
+
+
+def _make_response(body: bytes, url: str = "https://example.com/sitemap.xml", headers: dict | None = None) -> Response:
+ resp = Response(
+ url=url,
+ content=body,
+ status=200,
+ reason="OK",
+ cookies={},
+ headers=headers or {},
+ request_headers={},
+ )
+ resp.request = Request(url, sid="default")
+ return resp
+
+
+async def _collect(agen: AsyncGenerator) -> list:
+ return [item async for item in agen]
+
+
+class TestSitemapSpiderFlow:
+ @pytest.mark.asyncio
+ async def test_urlset_dispatched_through_rules(self):
+ class S(SitemapSpider):
+ name = "s"
+ sitemap_urls = ["https://example.com/sitemap.xml"]
+
+ def rules(self):
+ return [CrawlRule(LinkExtractor(allow=r"/posts/"), callback=self.parse_post)]
+
+ async def parse_post(self, response):
+ yield {"post": response.url}
+
+ spider = S()
+ out = await _collect(spider._parse_sitemap(_make_response(URLSET_XML)))
+ post_reqs = [r for r in out if "/posts/" in r.url]
+ about_reqs = [r for r in out if "/about" in r.url]
+ # Two posts dispatched to parse_post; about falls through (callback inherited from None → None)
+ assert len(post_reqs) == 2
+ assert all(r.callback == spider.parse_post for r in post_reqs)
+ assert len(about_reqs) == 1
+ assert about_reqs[0].callback is None
+
+ @pytest.mark.asyncio
+ async def test_no_rules_means_all_urls_fall_through(self):
+ class S(SitemapSpider):
+ name = "s"
+ sitemap_urls = ["https://example.com/sitemap.xml"]
+
+ spider = S()
+ out = await _collect(spider._parse_sitemap(_make_response(URLSET_XML)))
+ assert len(out) == 3
+ assert all(r.callback is None for r in out)
+
+ @pytest.mark.asyncio
+ async def test_sitemapindex_descends_into_children(self):
+ class S(SitemapSpider):
+ name = "s"
+ sitemap_urls = ["https://example.com/sitemap.xml"]
+
+ spider = S()
+ out = await _collect(spider._parse_sitemap(_make_response(INDEX_XML)))
+ # No urls in this index, just three child sitemap fetches
+ assert len(out) == 3
+ assert all(r.callback == spider._parse_sitemap for r in out)
+ assert {r.url for r in out} == {
+ "https://example.com/posts-sitemap.xml",
+ "https://example.com/products-sitemap.xml",
+ "https://example.com/skip-sitemap.xml",
+ }
+
+ @pytest.mark.asyncio
+ async def test_sitemap_follow_filters_child_sitemaps(self):
+ class S(SitemapSpider):
+ name = "s"
+ sitemap_urls = ["https://example.com/sitemap.xml"]
+ sitemap_follow = LinkExtractor(allow=r"posts-sitemap")
+
+ spider = S()
+ out = await _collect(spider._parse_sitemap(_make_response(INDEX_XML)))
+ assert {r.url for r in out} == {"https://example.com/posts-sitemap.xml"}
+
+ @pytest.mark.asyncio
+ async def test_alternate_links_dispatched_when_enabled(self):
+ class S(SitemapSpider):
+ name = "s"
+ sitemap_urls = ["https://example.com/sitemap.xml"]
+ sitemap_alternate_links = True
+
+ spider = S()
+ out = await _collect(spider._parse_sitemap(_make_response(URLSET_WITH_ALTERNATES)))
+ urls = {r.url for r in out}
+ assert urls == {
+ "https://example.com/en/page",
+ "https://example.com/fr/page",
+ "https://example.com/de/page",
+ }
+
+ @pytest.mark.asyncio
+ async def test_gzipped_sitemap_handled_via_magic_bytes(self):
+ class S(SitemapSpider):
+ name = "s"
+ sitemap_urls = ["https://example.com/sitemap.xml.gz"]
+
+ spider = S()
+ body = gzip.compress(URLSET_XML)
+ out = await _collect(spider._parse_sitemap(_make_response(body, url="https://example.com/sitemap.xml.gz")))
+ assert len(out) == 3
+
+
+class TestSitemapSpiderStartRequests:
+ @pytest.mark.asyncio
+ async def test_start_requests_uses_sitemap_urls(self):
+ class S(SitemapSpider):
+ name = "s"
+ sitemap_urls = ["https://a.com/s.xml", "https://b.com/s.xml"]
+
+ spider = S()
+ out = [req async for req in spider.start_requests()]
+ assert {r.url for r in out} == {"https://a.com/s.xml", "https://b.com/s.xml"}
+ assert all(r.callback == spider._parse_sitemap for r in out)
+
+ @pytest.mark.asyncio
+ async def test_start_requests_falls_back_to_robots_txt(self):
+ class S(SitemapSpider):
+ name = "s"
+ allowed_domains = {"example.com"}
+
+ spider = S()
+ out = [req async for req in spider.start_requests()]
+ assert len(out) == 1
+ assert out[0].url == "https://example.com/robots.txt"
+ assert out[0].callback == spider._parse_robots
+
+ @pytest.mark.asyncio
+ async def test_start_requests_uses_start_urls_if_no_sitemap_urls(self):
+ class S(SitemapSpider):
+ name = "s"
+ start_urls = ["https://example.com/seed"]
+
+ spider = S()
+ out = [req async for req in spider.start_requests()]
+ assert len(out) == 1
+ assert out[0].url == "https://example.com/seed"
+ # Should NOT have _parse_sitemap as callback (start_urls path treats them as regular pages)
+ assert out[0].callback is None
+
+ @pytest.mark.asyncio
+ async def test_start_requests_raises_when_nothing_configured(self):
+ class S(SitemapSpider):
+ name = "s"
+
+ spider = S()
+ with pytest.raises(RuntimeError, match="needs `sitemap_urls`"):
+ [req async for req in spider.start_requests()]
+
+
+class TestParseRobots:
+ @pytest.mark.asyncio
+ async def test_parse_robots_yields_sitemap_requests(self):
+ class S(SitemapSpider):
+ name = "s"
+ allowed_domains = {"example.com"}
+
+ spider = S()
+ body = b"User-agent: *\nSitemap: https://example.com/sitemap.xml\n"
+ resp = _make_response(body, url="https://example.com/robots.txt")
+ out = await _collect(spider._parse_robots(resp))
+ assert len(out) == 1
+ assert out[0].url == "https://example.com/sitemap.xml"
+ assert out[0].callback == spider._parse_sitemap
+
+ @pytest.mark.asyncio
+ async def test_parse_robots_with_no_directives_warns(self):
+ # Spider's logger has propagate=False, so we attach our own handler to it.
+ import logging
+
+ class S(SitemapSpider):
+ name = "s"
+ allowed_domains = {"example.com"}
+
+ spider = S()
+ records: list[logging.LogRecord] = []
+
+ class _Capture(logging.Handler):
+ def emit(self, record: logging.LogRecord) -> None:
+ records.append(record)
+
+ spider.logger.addHandler(_Capture())
+
+ body = b"User-agent: *\nDisallow: /\n"
+ resp = _make_response(body, url="https://example.com/robots.txt")
+ out = await _collect(spider._parse_robots(resp))
+ assert out == []
+ assert any("No Sitemap:" in r.getMessage() for r in records if r.levelno == logging.WARNING)
+
+
+class TestSitemapSpiderPickle:
+ @pytest.mark.asyncio
+ async def test_pickle_request_with_bound_method_callback_via_rules(self):
+ class S(SitemapSpider):
+ name = "s"
+ sitemap_urls = ["https://example.com/sitemap.xml"]
+
+ def rules(self):
+ return [CrawlRule(LinkExtractor(allow=r"/posts/"), callback=self.parse_post)]
+
+ async def parse_post(self, response):
+ yield {"post": response.url}
+
+ spider = S()
+ out = await _collect(spider._parse_sitemap(_make_response(URLSET_XML)))
+ post_req = next(r for r in out if "/posts/" in r.url)
+ state = post_req.__getstate__()
+ assert state["_callback_name"] == "parse_post"
+ # Round-trip
+ pickled = pickle.dumps(post_req)
+ restored = pickle.loads(pickled)
+ fresh = S()
+ restored._restore_callback(fresh)
+ assert restored.callback == fresh.parse_post
diff --git a/tests/spiders/test_templates.py b/tests/spiders/test_templates.py
new file mode 100644
index 0000000..893e05a
--- /dev/null
+++ b/tests/spiders/test_templates.py
@@ -0,0 +1,230 @@
+"""Tests for `CrawlSpider` and `CrawlRule`."""
+
+import pickle
+
+import pytest
+
+from scrapling.engines.toolbelt.custom import Response
+from scrapling.spiders.links import LinkExtractor
+from scrapling.spiders.request import Request
+from scrapling.spiders.templates import CrawlRule, CrawlSpider
+from scrapling.core._types import Any, AsyncGenerator, Dict, Union
+
+
+HTML = """
+
+ post 1
+ post 2
+ next page
+ about
+
+"""
+
+
+def _make_response(url: str = "https://example.com/") -> Response:
+ """Build a Response with a Request attached so `response.follow()` works."""
+ resp = Response(
+ url=url,
+ content=HTML,
+ status=200,
+ reason="OK",
+ cookies={},
+ headers={},
+ request_headers={},
+ )
+ resp.request = Request(url)
+ return resp
+
+
+class _TestSpider(CrawlSpider):
+ name = "test"
+ start_urls = ["https://example.com/"]
+
+ async def parse_post(self, response: Response) -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
+ yield {"post": response.url}
+
+ async def parse_page(self, response: Response) -> AsyncGenerator[Union[Dict[str, Any], Request, None], None]:
+ yield {"page": response.url}
+
+
+async def _collect(agen: AsyncGenerator) -> list:
+ return [item async for item in agen]
+
+
+class TestCrawlSpider:
+ @pytest.mark.asyncio
+ async def test_empty_rules_yields_nothing(self):
+ class S(CrawlSpider):
+ name = "s"
+ start_urls = ["https://example.com/"]
+
+ spider = S()
+ out = await _collect(spider.parse(_make_response()))
+ assert out == []
+
+ @pytest.mark.asyncio
+ async def test_single_rule_yields_matching_links(self):
+ class S(CrawlSpider):
+ name = "s"
+ start_urls = ["https://example.com/"]
+
+ def rules(self):
+ return [CrawlRule(LinkExtractor(allow=r"/posts/"))]
+
+ spider = S()
+ out = await _collect(spider.parse(_make_response()))
+ urls = [r.url for r in out]
+ assert urls == ["https://example.com/posts/1", "https://example.com/posts/2"]
+
+ @pytest.mark.asyncio
+ async def test_multiple_rules_all_applied(self):
+ class S(CrawlSpider):
+ name = "s"
+ start_urls = ["https://example.com/"]
+
+ def rules(self):
+ return [
+ CrawlRule(LinkExtractor(allow=r"/posts/")),
+ CrawlRule(LinkExtractor(allow=r"/page/")),
+ ]
+
+ spider = S()
+ out = await _collect(spider.parse(_make_response()))
+ urls = [r.url for r in out]
+ assert "https://example.com/posts/1" in urls
+ assert "https://example.com/posts/2" in urls
+ assert "https://example.com/page/2/" in urls
+
+ @pytest.mark.asyncio
+ async def test_rule_with_callback_bound_method(self):
+ spider = _TestSpider()
+ # rules() defaults to []; override at instance level
+ spider.rules = lambda: [ # type: ignore[method-assign]
+ CrawlRule(LinkExtractor(allow=r"/posts/"), callback=spider.parse_post)
+ ]
+ out = await _collect(spider.parse(_make_response()))
+ assert all(r.callback == spider.parse_post for r in out)
+
+ @pytest.mark.asyncio
+ async def test_rule_with_no_callback_leaves_request_callback_none(self):
+ # When CrawlRule.callback is None, response.follow() inherits the original
+ # request's callback. The original request was created with callback=None,
+ # so the resulting request's callback should also be None (engine then
+ # falls back to spider.parse).
+ class S(CrawlSpider):
+ name = "s"
+ start_urls = ["https://example.com/"]
+
+ def rules(self):
+ return [CrawlRule(LinkExtractor(allow=r"/posts/"))]
+
+ spider = S()
+ out = await _collect(spider.parse(_make_response()))
+ assert all(r.callback is None for r in out)
+
+ @pytest.mark.asyncio
+ async def test_process_request_invoked(self):
+ spider = _TestSpider()
+
+ def add_priority(req: Request, response: Response) -> Request:
+ req.priority = 99
+ return req
+
+ spider.rules = lambda: [ # type: ignore[method-assign]
+ CrawlRule(LinkExtractor(allow=r"/posts/"), process_request=add_priority)
+ ]
+ out = await _collect(spider.parse(_make_response()))
+ assert all(r.priority == 99 for r in out)
+
+ @pytest.mark.asyncio
+ async def test_process_request_can_replace_request(self):
+ spider = _TestSpider()
+ replacement = Request("https://replaced.example.com/")
+
+ def replace(req: Request, response: Response) -> Request:
+ return replacement
+
+ spider.rules = lambda: [ # type: ignore[method-assign]
+ CrawlRule(LinkExtractor(allow=r"/posts/"), process_request=replace)
+ ]
+ out = await _collect(spider.parse(_make_response()))
+ assert all(r is replacement for r in out)
+
+ @pytest.mark.asyncio
+ async def test_user_can_compose_super_parse(self):
+ """Override parse() to add custom yields plus call super().parse() for rules."""
+
+ class S(CrawlSpider):
+ name = "s"
+ start_urls = ["https://example.com/"]
+
+ def rules(self):
+ return [CrawlRule(LinkExtractor(allow=r"/posts/"))]
+
+ async def parse(self, response):
+ yield {"custom": "item"}
+ async for req in super().parse(response):
+ yield req
+
+ spider = S()
+ out = await _collect(spider.parse(_make_response()))
+ assert out[0] == {"custom": "item"}
+ assert all(isinstance(x, Request) for x in out[1:])
+ assert len(out) == 3 # 1 dict + 2 requests
+
+ @pytest.mark.asyncio
+ async def test_referer_set_on_followed_requests(self):
+ # `response.follow()` sets the referer header; verify it survives the rule path.
+ class S(CrawlSpider):
+ name = "s"
+ start_urls = ["https://example.com/"]
+
+ def rules(self):
+ return [CrawlRule(LinkExtractor(allow=r"/posts/"))]
+
+ spider = S()
+ out = await _collect(spider.parse(_make_response()))
+ for req in out:
+ assert req._session_kwargs["headers"]["referer"] == "https://example.com/"
+
+
+class TestCrawlSpiderPickle:
+ """Verify Request produced by CrawlSpider survives pickle round-trip with bound-method callbacks."""
+
+ @pytest.mark.asyncio
+ async def test_pickle_request_with_bound_method_callback(self):
+ spider = _TestSpider()
+ spider.rules = lambda: [ # type: ignore[method-assign]
+ CrawlRule(LinkExtractor(allow=r"/posts/"), callback=spider.parse_post)
+ ]
+ out = await _collect(spider.parse(_make_response()))
+ req = out[0]
+
+ # __getstate__ should convert the bound method into a name-string
+ state = req.__getstate__()
+ assert state["callback"] is None
+ assert state["_callback_name"] == "parse_post"
+
+ # Round-trip via pickle (bound methods aren't directly picklable; the
+ # state machinery handles the conversion)
+ pickled = pickle.dumps(req)
+ restored = pickle.loads(pickled)
+ assert restored._callback_name == "parse_post"
+
+ # Then _restore_callback on a fresh spider instance brings the method back
+ fresh_spider = _TestSpider()
+ restored._restore_callback(fresh_spider)
+ assert restored.callback == fresh_spider.parse_post
+
+
+class TestCrawlRule:
+ def test_default_callback_is_none(self):
+ rule = CrawlRule(LinkExtractor())
+ assert rule.callback is None
+ assert rule.follow is None
+ assert rule.process_request is None
+
+ def test_callback_accepts_callable(self):
+ spider = _TestSpider()
+ rule = CrawlRule(LinkExtractor(), callback=spider.parse_post)
+ assert rule.callback == spider.parse_post
