import json
import re
import pytest
import config
from config import CrawlSettings, EntityConfig
from fetcher import NotFound
from spiders.base import Spider
class FakeFetcher:
"""Serves list pages for a fixed set of IDs the way Wowhead does: rows are capped per page,
and a filter bound above `filter_limit` is silently dropped."""
def __init__(self, ids, cap=4, filter_limit=1000):
self.ids, self.cap, self.filter_limit, self.urls = sorted(ids), cap, filter_limit, []
self.settings = CrawlSettings(concurrency=2)
def get(self, url):
self.urls.append(url)
m = re.search(r";(\d+):(\d+)$", url)
if m:
lo, hi = (b if b <= self.filter_limit else None for b in map(int, m.groups()))
rows = [{"id": i, "name": f"thing {i}"} for i in self.ids
if (lo is None or i >= lo) and (hi is None or i <= hi)]
rows.sort(key=lambda r: r["id"] % 7) # Wowhead sorts by popularity, not id
data = json.dumps(rows[:self.cap])
return f"new Listview({{template: 'thing', id: 'things', _truncated: {int(len(rows) > self.cap)}, data:{data}}});"
entity_id = int(url.rsplit("=", 1)[1])
if entity_id == 13:
raise NotFound(url)
return f"
thing {entity_id}
"
class ThingSpider(Spider):
entity = EntityConfig(name="thing", list_path="things", id_filter=1, max_id=100_000)
listview_id = "things"
csv_fields = ["id", "name", "url"]
def list_row(self, raw, categories):
return {"id": raw["id"], "name": raw["name"]}
def parse_page(self, html, row):
return {"id": int(row["id"]), "title": html[4:-5]}
@pytest.fixture(autouse=True)
def tmp_output(tmp_path, monkeypatch):
monkeypatch.setattr(config, "OUTPUT_DIR", tmp_path)
monkeypatch.setattr("spiders.base.OUTPUT_DIR", tmp_path)
return tmp_path
IDS = [1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610, 987]
def test_list_bisects_until_nothing_is_truncated():
fetcher = FakeFetcher(IDS)
path = ThingSpider(fetcher).crawl_list()
lines = path.read_text(encoding="utf-8").splitlines()
assert lines[0] == "id,name,url"
assert [int(line.split(",")[0]) for line in lines[1:]] == IDS
assert lines[1].endswith("/forever/thing=1")
def test_list_skips_ranges_past_the_filter_limit():
fetcher = FakeFetcher(IDS)
ThingSpider(fetcher).crawl_list()
# Ranges starting above 1000 are answered with "id <= hi" (everything) and skipped, not split.
assert len(fetcher.urls) < 40
def test_list_page_total(tmp_output):
from parsers.common import list_page
html = "new Listview({id: 'x', note: \"5,442 quests found (1,000 displayed)\", _truncated: 1, data:[]});"
assert list_page(html, "x") == ([], True, 5442)
assert list_page("new Listview({id: 'x', data:[{\"id\": 1}]});", "x") == ([{"id": 1}], False, 1)
def test_details_resume_and_record_missing():
fetcher = FakeFetcher(IDS)
spider = ThingSpider(fetcher)
spider.crawl_list()
path = spider.crawl_details(limit=5)
assert len(path.read_text(encoding="utf-8").splitlines()) == 5
fetcher.urls.clear()
spider.crawl_details()
records = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
assert sorted(r["id"] for r in records) == IDS
assert len(fetcher.urls) == len(IDS) - 5 # only the remainder was fetched
assert next(r for r in records if r["id"] == 13)["not_found"] is True
assert next(r for r in records if r["id"] == 8)["title"] == "thing 8"
def test_find_missing_refetches_only_skipped_rows():
fetcher = FakeFetcher(IDS)
spider = ThingSpider(fetcher)
spider.crawl_list()
spider.crawl_details(limit=8) # IDs 1..34
path = spider.entity.detail_jsonl()
kept = [line for line in path.read_text(encoding="utf-8").splitlines()
if json.loads(line)["id"] not in (3, 13)] # pretend these were skipped while blocked
path.write_text("\n".join(kept) + "\n", encoding="utf-8")
assert spider.find_missing(dry_run=True) == [3, 13]
fetcher.urls.clear()
assert spider.find_missing() == [3, 13]
assert len(fetcher.urls) == 2 # IDs past 34 are left for `detail`
assert spider.find_missing() == []