スクレイパーのテスト:フィクスチャ、リプレイ、回帰テストスイート

ほとんどのスクレイパーは同じ方法でテストされる。誰かがライブサイトに対して実行し、出力を見て、正しそうだと判断する。それはうまくいかなくなる日までは機能する。サイトがクラス名を変更し、スクレイパーが空のフィールドを返しても、下流のレポートがおかしくなるまで誰も気づかない。

難しいのは、スクレイパーが自分でコントロールできないものに依存している点だ。ライブサイトを呼び出すテストは遅く、不安定で、警告なしに答えを変えるので、チームはそれを実行しなくなる。実際のページに一切触れないテストは永遠にパスし続け、そのあいだに本番は壊れている。

この問題から抜け出す方法は、問題を二つに分けることだ。コードは決して変化しない記録済みページに対してテストし、サイトの方は変化したときに教えてくれる小さな日次チェックで別途テストする。このガイドでは、そのまま応用できるコードとともに両方を構築する。記録されたフィクスチャ、クローラー全体をオフラインで実行するリプレイセッション、パーサー向けのゴールデンファイルテスト、そしてライブサイト向けのカナリアだ。

壊れうる3つのこと

スクレイパーが不正なデータを出すとき、原因はほぼ常に次の3つのいずれかであり、それぞれ異なる種類のテストが必要になる。

何が壊れたか	例	それを捕まえるテスト
あなたのパーサー	リファクタリングで “1,099.00” が 1.099 になる	記録済みページに対するゴールデンファイルテスト
あなたのクロールロジック	ページネーションが1ページ目で止まる。503が空ページとして扱われる	クローラーをオフラインで実行するリプレイテスト
サイト側	クラス名が変わる、フィールドが移動する、ブロックページが出る	ライブサイトに対する日次カナリア

最初の2つはあなたのコードのバグであり、デプロイ前に、ネットワークなしで数秒で動くテストで捕まえるべきものだ。3つ目はそもそもバグではない。世界が変化しているのであり、ユニットテストではなく監視が必要になる。

フェッチとパースを分離しておく

テストは構造から始まる。ページをパースする関数が同時にフェッチも行っていると、すべてのテストがネットワークに対処しなければならなくなる。パースは純粋な関数のままにしておく。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() はセッションを自分で生成するのではなく、引数として受け取る。この一点の選択こそが、これをテスト可能にしている。本番では 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、ブロックページ。それぞれが恒久的なテストケースになる。本番で生のレスポンスをアーカイブしているなら、archiving scraped pages in WARC のガイドで説明している通り、そのアーカイブは実際の失敗から得られるフィクスチャの供給源になる。

正しいクロールなら必ず満たすべきことを検証する

ゴールデンファイルは正確な出力をチェックする。検証(validation)は、サイトのコンテンツがどう変化しても成り立つべきルールをチェックする。すべてのレコードに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 drift in scraped data のガイドを参照してほしい。

テストスイート

フィクスチャと検証が揃えば、テストは短くなる。標準ライブラリだけを使う。

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 を付けて実行してゴールデンファイルを書き出し、それから手作業でレビューする。これらはパーサーが生成すべきものの仕様となる。それ以降、パーサーの出力に何か変化があれば、誰かがそれを見て、修正なのかバグなのかを判断するまでテストは失敗し続ける。
• リプレイテスト は実際の crawl() をオフラインで実行する。一方はページネーションが最後まで追従されることをチェックし、もう一方は2ページ目の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

5つのテストが、ネットワークなしで一瞬で終わる。この速さこそが重要だ。この速さのテストならコミットのたびに実行できる。その周りのパイプライン、特にフィクスチャテストにプロキシの認証情報が一切不要な理由については、running scrapers in CI/CD のガイドで扱っている。

カナリア: あなたのコードではなくサイトをテストする

フィクスチャはサイトをあった時点の状態で凍結する。サイトがいつ変化したかを知るには、スケジュールに沿ってライブサイトに対して実際のクロールを実行し、結果を検証する。それが上述の canary() 関数であり、すべて問題なければ空のリストを、何かおかしければ具体的な問題のリストを返す。

小さな開始URLと現実的な最小レコード数を指定して、1日1回、または大きなクロールの前に実行する。失敗したときは次の手順を踏む。

1. 壊れたページを新しいフィクスチャとして記録する。
2. ゴールデンファイルテストまたはリプレイテストがそのフィクスチャで失敗するのを確認する。 これにより、テストが問題を再現していることが確認できる。
3. **パーサーを修正し、**新しいゴールデン出力をレビューし、フィクスチャと修正とゴールデンファイルをまとめてコミットする。

サイトの変更は恒久的な回帰テストになるので、同じ壊れ方が気づかれずに再発することは二度とない。カナリアからフィクスチャへ、そして修正へというこのループこそが、壊れやすいスクレイパーを保守されたスクレイパーへと変えるものだ。

カナリアは、パースに関する問題以外の失敗についても教えてくれる。ステータス200を返すブロックページ、地域リダイレクト、コンテンツの代わりに表示される同意ウォールなどだ。成功したように見えて実際には誤ったコンテンツを含むレスポンスがどれほど一般的かについては、silent failure rate の調査を参照してほしい。パイプライン全体の継続的な監視については、how to monitor a web scraping pipeline を参照してほしい。

実践的なルール

• ユニットテストからネットワークを呼び出さない。 ReplaySession がそうしているように、フィクスチャがない場合はエラーにする。
• フィクスチャは小さく、本物にする。 ケースごとに1ページで十分だ。HTMLを削らないこと。削りたくなるノイズこそが、しばしばパーサーを壊すものだ。
• ゴールデンファイルをコードと同様にレビューする。 レビューされていないゴールデンファイルは、パーサーが昨日と同じ動作をすることしか証明しない。
• ケースに名前をつける。日付ではなく。 “売り切れ商品” と名付けたフィクスチャは1年後でも役に立つが、“火曜日のページ” ではそうはいかない。
• フィクスチャは意図的に更新する。 サイトが変わったときは、古いレイアウトが絶対に戻ってこないのでない限り、古いフィクスチャを置き換えるのではなく新しいフィクスチャを追加する。
• フィクスチャに秘密情報を入れない。 記録されたページにはセッショントークンや個人データが含まれうる。コミットする前にクッキーを取り除き、本文を確認する。

よくある質問

ライブサイトに触れずにWebスクレイパーをテストするにはどうすればよいか。

実際のレスポンスを一度フィクスチャとして記録し、HTTPセッションをその記録を返すものに置き換える。パーサーとクローラーは本番とまったく同じように動作するが、オフラインで、ミリ秒単位で、しかも毎回同じ答えが得られる。

ゴールデンファイルテストとは何か。

関数の出力を、保存されレビュー済みの期待される出力のコピーと比較するテストだ。スクレイパーの場合、ゴールデンファイルは記録済みページから生成されるべきレコードのリストになる。出力に何らかの変化があれば、誰かがそれを受け入れるか修正するまでテストは失敗する。

ウェブサイトの変更がスクレイパーを壊したことをどうやって知るか。

カナリアを実行する。スケジュールに沿ってライブサイトに対して行う小さな実際のクロールで、正しい結果なら必ず満たすべき検証ルールを伴う。それが問題を報告したら、新しいページをフィクスチャとして記録し、パーサーを修正し、そのフィクスチャを回帰テストとして保持する。

スクレイパーのテストにプロキシを使うべきか。

ユニットテストとリプレイテストはネットワークを一切使うべきではないので、プロキシも認証情報も不要だ。カナリアはライブサイトに対して実行されるので、本番と同じプロキシ設定を使うべきであり、そうすることで本番が見ているものと同じものを見ることができる。

まとめ

コードとサイトを分けてテストする。記録済みフィクスチャとリプレイセッションにより、パーサーとクローラーをオフラインで、素早く、決定論的に、コミットのたびにテストできる。検証ルールを伴う日次カナリアは、サイトがいつ変化したかを教えてくれる。カナリアが見つけた不具合をすべてフィクスチャに変えていけば、サイトが変化するたびにスクレイパーは信頼性を失うのではなく、増していく。