测试爬虫：Fixtures、重放与回归套件

网站一变化，爬虫就会失效，而直接访问真实网站的测试也会随之失效。如何用录制的 fixtures、重放和每日金丝雀测试来测试爬虫。

大多数爬虫的测试方式都是一样的：有人针对真实网站运行一遍，看看输出结果,觉得看起来没问题就算了。这种方式在出问题之前都管用。网站改了一个类名,爬虫返回空字段,直到下游的报告出错才有人注意到。

难点在于，爬虫依赖的是你无法控制的东西。调用真实网站的测试又慢又不稳定，答案还会在没有任何预警的情况下发生变化，于是团队干脆不再运行这些测试。而那些从不接触真实页面的测试则会永远通过，哪怕生产环境早已出了问题。

解决办法是把问题一分为二。针对永不改变的录制页面测试你的代码，再用一个简短的每日检查单独测试网站本身，告诉你网站何时发生了变化。本指南将搭建这两部分内容，并附上可供你直接套用的代码：录制固件、离线运行整个爬虫的回放会话、针对解析器的黄金文件测试，以及针对真实网站的金丝雀检测。

三种可能出错的情况

当爬虫产出错误数据时，原因几乎总是以下三种之一，而每一种都需要不同类型的测试：

出了什么问题	示例	能捕获它的测试
你的解析器	重构把 “1,099.00” 变成了 1.099	针对录制页面的黄金文件测试
你的爬取逻辑	翻页在第一页后就停止；503 被当作空页面处理	离线运行爬虫的回放测试
网站本身	类名变了，字段位置变了，出现了拦截页面	针对真实网站的每日金丝雀检测

前两种是你代码中的 bug，应该在部署前，在无需网络、几秒内就能运行完的测试中被发现。第三种根本不是 bug；而是外部世界发生了变化，需要的是监控而不是单元测试。

把抓取和解析分开

测试从结构开始。如果解析页面的函数同时也负责抓取页面，那么每个测试都得处理网络问题。要让解析保持为纯函数：输入 HTML，输出记录。爬虫本身则变成一个很薄的循环，负责抓取页面并把它们交给解析器。

下面是我们要测试的爬虫，用于一个带分页的列表网站：

from urllib.parse import urljoin

from lxml import html as lxml_html


def parse_listing(page_html):
    """Products on one listing page. Pure function: HTML in, records out."""
    doc = lxml_html.fromstring(page_html)
    items = []
    for card in doc.cssselect("li.product"):
        price = card.cssselect(".price")[0].text_content().strip()
        items.append({
            "sku": card.get("data-sku"),
            "name": card.cssselect("h2")[0].text_content().strip(),
            "price": float(price.replace("$", "").replace(",", "")),
            "in_stock": not card.cssselect(".sold-out"),
        })
    return items


def next_page(page_html, base_url):
    links = lxml_html.fromstring(page_html).cssselect("a[rel=next]")
    return urljoin(base_url, links[0].get("href")) if links else None


def crawl(session, start_url, max_pages=50):
    """Follow pagination; return the records and the URLs that failed."""
    records, failed, url = [], [], start_url
    for _ in range(max_pages):
        response = session.get(url, timeout=30)
        if response.status_code != 200:
            failed.append((url, response.status_code))
            break
        records.extend(parse_listing(response.text))
        url = next_page(response.text, response.url)
        if url is None:
            break
    return records, failed

crawl() 把 session 作为参数传入,而不是自己创建一个。正是这一个选择让它变得可测试：生产环境中你传入一个 requests.Session，测试中你传入一个提供录制页面的替代对象。

一次性录制固件

固件是一份真实响应，保存到磁盘上并附带足够的元数据以便精确重放：URL、重定向后的最终 URL、状态码和编码。请以原始字节而非解码后的文本形式保存响应体，这样编码问题会被重现出来，而不是被掩盖掉。

import hashlib
import json
from pathlib import Path


def _name(url):
    return hashlib.sha256(url.encode()).hexdigest()[:16]


def record(session, url, directory):
    """Fetch a page once and save everything a test needs to replay it."""
    response = session.get(url, timeout=30)
    directory = Path(directory)
    directory.mkdir(parents=True, exist_ok=True)
    name = _name(url)
    (directory / f"{name}.html").write_bytes(response.content)
    (directory / f"{name}.json").write_text(json.dumps({
        "url": url,
        "final_url": response.url,
        "status": response.status_code,
        "encoding": response.encoding,
        "headers": {"Content-Type": response.headers.get("Content-Type", "")},
    }, indent=2))
    return name


class Replayed:
    def __init__(self, meta, body):
        self.url = meta["final_url"]
        self.status_code = meta["status"]
        self.headers = meta["headers"]
        self.content = body
        self.text = body.decode(meta["encoding"] or "utf-8", errors="replace")


class ReplaySession:
    """Stands in for requests.Session: serves recorded pages, never the network."""

    def __init__(self, directory):
        self.directory = Path(directory)
        self.requested = []

    def get(self, url, **kwargs):
        self.requested.append(url)
        name = _name(url)
        meta_path = self.directory / f"{name}.json"
        if not meta_path.exists():
            raise LookupError(f"no fixture recorded for {url}")
        meta = json.loads(meta_path.read_text())
        return Replayed(meta, (self.directory / f"{name}.html").read_bytes())

ReplaySession 具有和 requests.Session 相同的 get() 方法，所以 crawl() 根本分辨不出区别。它对任何未被录制的 URL 都会抛出错误，这一点很重要：如果测试在固件缺失时悄悄回退到真实网络，那它在你本地机器上会很快，但在 CI 中就会变慢且不稳定。

录制的不应只是正常页面。那些在生产中会弄坏爬虫的情况才最值得保留：分页的最后一页、空的搜索结果、售罄的商品、503 响应、拦截页面。每一种都要变成一个永久的测试用例。如果你在生产环境中归档原始响应,正如我们关于以 WARC 格式归档抓取页面的指南中所描述的那样，那么这些归档就是来自真实故障的现成固件来源。

验证任何正确爬取都必须满足的条件

黄金文件检查的是精确的输出。验证检查的则是无论网站内容如何变化都必须成立的规则：每条记录都有 ID，ID 唯一，名称不为空，价格是合理的数字，并且本次爬取返回的记录数大致符合预期。

def validate(records, min_records=1):
    """Checks that hold for any correct crawl, whatever the site shows today."""
    problems = []
    if len(records) < min_records:
        problems.append(f"only {len(records)} records, expected at least {min_records}")
    seen = set()
    for i, r in enumerate(records):
        if not r.get("sku"):
            problems.append(f"record {i}: missing sku")
        elif r["sku"] in seen:
            problems.append(f"record {i}: duplicate sku {r['sku']}")
        seen.add(r.get("sku"))
        if not r.get("name"):
            problems.append(f"record {i}: empty name")
        if not isinstance(r.get("price"), float) or not 0 < r["price"] < 100_000:
            problems.append(f"record {i}: implausible price {r.get('price')!r}")
    return problems


def canary(session, start_url, crawl, min_records):
    """Run the real crawl against the live site and list anything wrong.

    Schedule it daily: an empty list means the scraper still works today.
    """
    records, failed = crawl(session, start_url)
    return validate(records, min_records) + [f"fetch failed: {url} ({status})" for url, status in failed]

同一个 validate() 在三个地方运行：在测试套件中、在每日金丝雀检测中，以及理想情况下，在数据写入之前的每次生产运行中。我们关于抓取数据中的 schema 漂移的指南对字段级别随时间推移的检查做了更深入的探讨。

测试套件

有了固件和验证逻辑之后，测试本身会很简短。它们只用标准库：

import json
import os
import unittest
from pathlib import Path

from fixtures import ReplaySession
from shop import crawl, parse_listing
from validate import canary, validate

FIXTURES = Path(__file__).parent / "fixtures"
GOLDEN = Path(__file__).parent / "golden"
UPDATE = os.environ.get("UPDATE_GOLDEN") == "1"


class ParserTests(unittest.TestCase):
    def test_every_fixture_matches_its_golden_file(self):
        for page in sorted(FIXTURES.glob("*.html")):
            meta = json.loads(page.with_suffix(".json").read_text())
            if meta["status"] != 200:
                continue
            with self.subTest(url=meta["url"]):
                parsed = parse_listing(page.read_text())
                golden = GOLDEN / f"{page.stem}.json"
                if UPDATE:
                    golden.write_text(json.dumps(parsed, indent=2))
                self.assertEqual(parsed, json.loads(golden.read_text()))


class CrawlTests(unittest.TestCase):
    def test_crawl_follows_pagination_offline(self):
        session = ReplaySession(FIXTURES)
        records, failed = crawl(session, "https://shop.example/laptops?page=1")
        self.assertEqual(len(session.requested), 2)
        self.assertEqual(failed, [])
        self.assertEqual(validate(records, min_records=4), [])

    def test_crawl_reports_a_failed_page(self):
        session = ReplaySession(FIXTURES)
        records, failed = crawl(session, "https://shop.example/monitors?page=1")
        self.assertEqual(failed, [("https://shop.example/monitors?page=2", 503)])
        self.assertEqual(len(records), 2)


class ValidationTests(unittest.TestCase):
    def test_catches_broken_records(self):
        bad = [{"sku": "A1", "name": "", "price": 0.0}, {"sku": "A1", "name": "x", "price": "12"}]
        self.assertEqual(validate(bad), [
            "record 0: empty name",
            "record 0: implausible price 0.0",
            "record 1: duplicate sku A1",
            "record 1: implausible price '12'",
        ])

    def test_canary_flags_a_partial_crawl(self):
        problems = canary(ReplaySession(FIXTURES), "https://shop.example/monitors?page=1", crawl, min_records=4)
        self.assertEqual(problems, [
            "only 2 records, expected at least 4",
            "fetch failed: https://shop.example/monitors?page=2 (503)",
        ])


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

黄金文件测试会解析每一个录制的页面，并把结果与一份存储好的 JSON 文件进行比较。第一次运行时，用 UPDATE_GOLDEN=1 来写入黄金文件，然后手动检查它们：它们就是解析器应该产出什么的规格说明。此后，解析器输出的任何变化都会导致测试失败，直到有人查看并判断这是一次修复还是一个 bug。
回放测试离线运行真实的 crawl()。其中一个检查分页是否被追踪到底；另一个检查第二页上的 503 是否被正确报告，而不是被悄悄当成结果的结束。
验证测试检查这些规则是否真的能捕获到损坏的记录，以及金丝雀检测是否能报告出一次不完整的爬取。

对从一个小型测试商店录制的固件运行该测试套件：

test_crawl_follows_pagination_offline (test_shop.CrawlTests) ... ok
test_crawl_reports_a_failed_page (test_shop.CrawlTests) ... ok
test_every_fixture_matches_its_golden_file (test_shop.ParserTests) ... ok
test_canary_flags_a_partial_crawl (test_shop.ValidationTests) ... ok
test_catches_broken_records (test_shop.ValidationTests) ... ok

----------------------------------------------------------------------
Ran 5 tests in 0.019s

OK