test: add tests accordingly

2026-05-11 02:33:43 +03:00
parent f7da15771b
commit 8a8b2d14ba
3 changed files with 865 additions and 0 deletions
@@ -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 = """
+<html><body>
+  <a href="/posts/1">post 1</a>
+  <a href="/posts/2">post 2</a>
+  <a href="https://other.com/page">external</a>
+  <a href="/about">about</a>
+  <a href="mailto:x@example.com">mail</a>
+  <a href="javascript:alert(1)">js</a>
+  <a href="/file.pdf">pdf</a>
+  <area href="/area-link">area</area>
+  <link rel="stylesheet" href="/style.css">
+</body></html>
+"""
+
+
+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('<a href="foo/bar">x</a>', 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('<a href="/anything">x</a><a href="/else">y</a>')
+        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 = '<a href="https://api.example.com/x">a</a><a href="https://other.com/y">b</a>'
+        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 = """
+        <html><body>
+            <nav><a href="/nav-link">n</a></nav>
+            <main><a href="/main-link">m</a></main>
+        </body></html>
+        """
+        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 = """
+        <html><body>
+            <div id="header"><a href="/h">h</a></div>
+            <div id="content"><a href="/c">c</a></div>
+        </body></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 = '<link rel="stylesheet" href="/style.css"><a href="/page">p</a>'
+        resp = _make_response(html)
+        # Override deny_extensions to allow .css through, and pick up <link href>
+        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('<a href="/x?b=2&a=1">x</a>')
+        urls = LinkExtractor().extract(resp)
+        assert urls == ["https://example.com/x?a=1&b=2"]
+
+    def test_fragment_dropped_by_default(self):
+        resp = _make_response('<a href="/x#section">x</a>')
+        urls = LinkExtractor().extract(resp)
+        assert urls == ["https://example.com/x"]
+
+    def test_keep_fragment_preserves_it(self):
+        resp = _make_response('<a href="/x#section">x</a>')
+        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('<a href="/x?b=2&a=1#f">x</a>')
+        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 = '<a href="/x">a</a><a href="/x">b</a><a href="/x?">c</a>'
+        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 = '<a href="/a.pdf">pdf</a><a href="/b.zip">zip</a><a href="/c.png">png</a><a href="/d">ok</a>'
+        resp = _make_response(html)
+        urls = LinkExtractor().extract(resp)
+        assert urls == ["https://example.com/d"]
+
+    def test_custom_deny_extensions_overrides_default(self):
+        html = '<a href="/a.pdf">pdf</a><a href="/b.zip">zip</a>'
+        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 = '<a href="/a.pdf">pdf</a>'
+        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('<a href="  /spaced  ">x</a>')
+        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
@@ -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"""<?xml version="1.0" encoding="UTF-8"?>
+<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
+  <url>
+    <loc>https://example.com/posts/1</loc>
+    <lastmod>2026-01-15</lastmod>
+    <changefreq>daily</changefreq>
+    <priority>0.8</priority>
+  </url>
+  <url>
+    <loc>https://example.com/posts/2</loc>
+    <lastmod>2026-02-20</lastmod>
+  </url>
+  <url>
+    <loc>https://example.com/about</loc>
+  </url>
+</urlset>
+"""
+
+URLSET_WITH_ALTERNATES = b"""<?xml version="1.0" encoding="UTF-8"?>
+<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9"
+        xmlns:xhtml="http://www.w3.org/1999/xhtml">
+  <url>
+    <loc>https://example.com/en/page</loc>
+    <xhtml:link rel="alternate" hreflang="fr" href="https://example.com/fr/page"/>
+    <xhtml:link rel="alternate" hreflang="de" href="https://example.com/de/page"/>
+  </url>
+</urlset>
+"""
+
+INDEX_XML = b"""<?xml version="1.0" encoding="UTF-8"?>
+<sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
+  <sitemap><loc>https://example.com/posts-sitemap.xml</loc></sitemap>
+  <sitemap><loc>https://example.com/products-sitemap.xml</loc></sitemap>
+  <sitemap><loc>https://example.com/skip-sitemap.xml</loc></sitemap>
+</sitemapindex>
+"""
+
+# Sitemap without the standard namespace (some sites do this)
+URLSET_NO_NS = b"""<?xml version="1.0"?>
+<urlset>
+  <url><loc>https://example.com/x</loc></url>
+</urlset>
+"""
+
+
+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"<not valid xml")
+        assert result == SitemapResult()
+
+    def test_unknown_root_returns_empty_result(self):
+        result = SitemapParser().parse(b"<?xml version='1.0'?><foo><bar/></foo>")
+        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
@@ -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 = """
+<html><body>
+  <a href="/posts/1">post 1</a>
+  <a href="/posts/2">post 2</a>
+  <a href="/page/2/">next page</a>
+  <a href="/about">about</a>
+</body></html>
+"""
+
+
+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