대부분의 스크레이퍼는 같은 방식으로 테스트된다. 누군가 실제 사이트에 대해 실행해보고, 출력 결과를 보고, 맞는 것 같다고 판단한다. 이 방식은 문제가 생기기 전까지는 통한다. 사이트가 클래스 이름을 바꾸면 스크레이퍼는 빈 필드를 반환하고, 다운스트림 리포트가 잘못 나올 때까지 아무도 알아차리지 못한다.
어려운 점은 스크레이퍼가 자신이 통제할 수 없는 대상에 의존한다는 것이다. 실제 사이트를 호출하는 테스트는 느리고 불안정하며 경고 없이 결과가 바뀌기 때문에, 팀은 결국 그 테스트를 더 이상 실행하지 않게 된다. 반대로 실제 페이지를 전혀 건드리지 않는 테스트는 영원히 통과하지만, 운영 환경은 망가진다.
해결책은 문제를 둘로 나누는 것이다. 코드는 절대 바뀌지 않는 녹화된 페이지에 대해 테스트하고, 사이트는 변경 여부를 알려주는 작은 일일 점검으로 별도로 테스트한다. 이 가이드는 두 가지를 모두 구축하며, 적용 가능한 코드를 함께 제공한다: 녹화된 픽스처, 크롤러 전체를 오프라인으로 실행하는 재생 세션, 파서를 위한 골든 파일 테스트, 그리고 실제 사이트를 위한 카나리아.
깨질 수 있는 세 가지
스크레이퍼가 잘못된 데이터를 생성할 때, 원인은 거의 항상 다음 세 가지 중 하나이며, 각각 다른 종류의 테스트가 필요하다:
| 무엇이 깨졌는가 | 예시 | 이를 잡아내는 테스트 |
|---|---|---|
| 파서 | 리팩터링으로 “1,099.00”이 1.099로 바뀜 | 녹화된 페이지에 대한 골든 파일 테스트 |
| 크롤링 로직 | 페이지네이션이 첫 페이지 이후 멈춤; 503을 빈 페이지로 처리함 | 크롤러를 오프라인으로 실행하는 재생 테스트 |
| 사이트 | 클래스 이름이 바뀜, 필드 위치가 이동함, 차단 페이지가 나타남 | 실제 사이트에 대한 일일 카나리아 |
앞의 두 가지는 코드의 버그이며 배포 전에, 네트워크 없이 몇 초 안에 실행되는 테스트에서 잡아내야 한다. 세 번째는 전혀 버그가 아니다. 세상이 변한 것이며, 유닛 테스트가 아니라 모니터링이 필요하다.
가져오기와 파싱을 분리하라
테스트는 구조에서 시작된다. 페이지를 파싱하는 함수가 동시에 페이지를 가져오기까지 한다면, 모든 테스트가 네트워크를 다뤄야 한다. 파싱은 순수 함수로 유지하라: 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, failedcrawl()은 세션을 직접 생성하는 대신 인자로 받는다. 이 하나의 선택이 테스트 가능성을 만든다. 운영 환경에서는 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로 스크레이핑한 페이지 아카이빙하기 가이드에서 설명하듯, 그 아카이브들이 실제 실패 사례로부터 나온 픽스처의 좋은 원천이 된다.
올바른 크롤링이라면 반드시 만족해야 할 조건을 검증하라
골든 파일은 정확한 출력을 확인한다. 검증(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()가 세 군데에서 실행된다: 테스트 스위트, 일일 카나리아, 그리고 이상적으로는 데이터가 기록되기 전 모든 운영 실행에서. 스크레이핑된 데이터의 스키마 드리프트 가이드는 시간에 따른 필드 수준 검증을 더 깊이 다룬다.
테스트 스위트
픽스처와 검증이 갖춰지면 테스트는 짧아진다. 표준 라이브러리만 사용한다:
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()을 오프라인으로 실행한다. 하나는 페이지네이션이 끝까지 따라가는지 확인하고, 다른 하나는 두 번째 페이지의 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네트워크 없이 1초도 안 되는 시간에 다섯 개의 테스트가 실행된다. 이 속도가 핵심이다. 이 정도로 빠른 테스트는 모든 커밋마다 실행할 수 있다. CI/CD에서 스크레이퍼 실행하기 가이드는 이를 둘러싼 파이프라인을 다루며, 픽스처 테스트가 왜 프록시 자격 증명을 전혀 필요로 하지 않는지도 설명한다.
카나리아: 코드가 아니라 사이트를 테스트하기
픽스처는 사이트를 그 당시 모습 그대로 고정시킨다. 사이트가 언제 바뀌었는지 알려면, 실제 크롤링을 실제 사이트에 대해 일정에 따라 실행하고 결과를 검증해야 한다. 이것이 위의 canary() 함수다. 모든 것이 정상이면 빈 리스트를 반환하고, 그렇지 않으면 구체적인 문제 목록을 반환한다.
하루에 한 번, 혹은 큰 크롤링 전마다, 작은 시작 URL과 현실적인 최소 레코드 수로 실행하라. 실패했을 때:
- 문제가 생긴 페이지를 새 픽스처로 녹화한다.
- 골든 파일 또는 재생 테스트가 그 픽스처에서 실패하는 것을 확인한다. 이는 테스트가 문제를 재현한다는 것을 확인해준다.
- 파서를 수정하고, 새 골든 출력을 검토한 다음, 픽스처와 수정 사항과 골든 파일을 함께 커밋한다.
사이트 변경은 영구적인 회귀 테스트가 되어, 같은 문제가 다시는 눈에 띄지 않게 재발할 수 없다. 카나리아에서 픽스처로, 그리고 수정으로 이어지는 이 순환이 취약한 스크레이퍼를 유지보수되는 스크레이퍼로 바꾼다.
카나리아는 파싱과 전혀 관련 없는 실패도 알려준다: 상태 코드 200을 반환하는 차단 페이지, 지역 리디렉션, 콘텐츠 대신 나타나는 동의 화면 같은 것들이다. 침묵하는 실패율에 대한 우리의 연구는 성공한 것처럼 보이지만 잘못된 콘텐츠를 담고 있는 응답이 얼마나 흔한지 보여준다. 전체 파이프라인의 지속적인 모니터링에 대해서는 웹 스크레이핑 파이프라인을 모니터링하는 방법을 참고하라.
실용적인 규칙
- 유닛 테스트에서는 절대 네트워크를 호출하지 말라.
ReplaySession이 하듯이, 픽스처가 없으면 오류가 나도록 만들라. - 픽스처는 작고 실제적으로 유지하라. 케이스당 페이지 하나면 충분하다. HTML을 조금도 잘라내지 말라. 잘라내고 싶은 노이즈가 바로 파서를 깨뜨리는 원인일 때가 많다.
- 골든 파일을 코드처럼 검토하라. 검토되지 않은 골든 파일은 파서가 어제 한 일을 오늘도 똑같이 한다는 것만 증명할 뿐이다.
- 날짜가 아니라 케이스에 이름을 붙여라. “품절 상품”이라는 이름의 픽스처는 1년 후에도 유용하지만, “화요일 페이지”라는 이름은 그렇지 않다.
- 픽스처는 의도적으로 갱신하라. 사이트가 바뀌면, 이전 레이아웃이 절대 돌아오지 않을 것이 확실한 경우가 아니라면, 기존 픽스처를 교체하지 말고 새 픽스처를 추가하라.
- 픽스처에서 비밀 정보를 제거하라. 녹화된 페이지에는 세션 토큰이나 개인 정보가 담길 수 있다. 커밋하기 전에 쿠키를 제거하고 본문을 확인하라.
자주 묻는 질문
실제 사이트에 접속하지 않고 웹 스크레이퍼를 테스트하려면 어떻게 하나요?
실제 응답을 한 번 픽스처로 녹화한 다음, HTTP 세션을 그 녹화 내용을 제공하는 것으로 교체한다. 파서와 크롤러는 운영 환경과 완전히 동일하게 실행되지만, 오프라인에서 밀리초 단위로, 매번 같은 결과를 내며 실행된다.
골든 파일 테스트란 무엇인가요?
함수의 출력을 저장되고 검토된 기대 출력 사본과 비교하는 테스트다. 스크레이퍼의 경우, 골든 파일은 녹화된 페이지가 만들어내야 할 레코드 목록이다. 출력에 변화가 생기면 누군가 그것을 수용하거나 고칠 때까지 테스트가 실패한다.
웹사이트 변경이 스크레이퍼를 망가뜨렸는지 어떻게 알 수 있나요?
카나리아를 실행하라: 실제 사이트에 대한 작은 실제 크롤링을 일정에 따라 실행하고, 올바른 결과라면 반드시 만족해야 할 검증 규칙을 적용하라. 문제가 보고되면, 새 페이지를 픽스처로 녹화하고, 파서를 수정하고, 그 픽스처를 회귀 테스트로 유지하라.
스크레이퍼 테스트에 프록시를 사용해야 하나요?
유닛 테스트와 재생 테스트는 네트워크를 전혀 사용하지 않아야 하므로, 프록시나 자격 증명이 필요 없다. 카나리아는 실제 사이트에 대해 실행되며, 운영 환경이 보는 것과 동일한 것을 보기 위해 운영 환경과 같은 프록시 설정을 사용해야 한다.
결론
코드와 사이트를 따로 테스트하라. 녹화된 픽스처와 재생 세션을 사용하면 파서와 크롤러를 오프라인에서, 빠르고 결정론적으로, 모든 커밋마다 테스트할 수 있다. 검증 규칙을 갖춘 일일 카나리아는 사이트가 언제 바뀌었는지 알려준다. 카나리아가 찾아낸 모든 문제를 픽스처로 바꾸면, 스크레이퍼는 사이트가 바뀔 때마다 점점 더 신뢰할 수 있게 된다. 약해지는 것이 아니라.