wehub-resource-sync
247153575d
Tests / tests (map[TOXENV:py310], macos-latest, 3.10) (push) Has been cancelled
Tests / tests (map[TOXENV:py311], macos-latest, 3.11) (push) Has been cancelled
Tests / tests (map[TOXENV:py312], macos-latest, 3.12) (push) Has been cancelled
Tests / tests (map[TOXENV:py313], macos-latest, 3.13) (push) Has been cancelled
283 lines
10 KiB
Python
283 lines
10 KiB
Python
"""Tests for force-stop checkpoint preservation in CrawlerEngine.
|
|
|
|
Regression tests for the bug where force-stop (second Ctrl+C) called
|
|
cancel_scope.cancel() BEFORE saving the checkpoint, causing:
|
|
1. _save_checkpoint() to be aborted by anyio's Cancelled exception
|
|
2. self.paused never set to True
|
|
3. The finally block to DELETE the previous checkpoint (cleanup runs on non-paused exit)
|
|
|
|
Total progress loss: user's checkpoint from a long crawl is irrecoverably deleted.
|
|
"""
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import anyio
|
|
import pytest
|
|
|
|
from scrapling.spiders.engine import CrawlerEngine
|
|
from scrapling.spiders.request import Request
|
|
from scrapling.spiders.session import SessionManager
|
|
from scrapling.spiders.checkpoint import CheckpointManager, CheckpointData
|
|
from scrapling.core._types import Any, Dict, Set, AsyncGenerator
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mock helpers (minimal, matching test_engine.py conventions)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class MockResponse:
|
|
def __init__(self, status=200, body=b"ok", url="https://example.com"):
|
|
self.status = status
|
|
self.body = body
|
|
self.url = url
|
|
self.request: Any = None
|
|
self.meta: Dict[str, Any] = {}
|
|
|
|
def __str__(self):
|
|
return self.url
|
|
|
|
|
|
class MockSession:
|
|
def __init__(self, delay: float = 0.0):
|
|
self._is_alive = False
|
|
self._delay = delay
|
|
|
|
async def __aenter__(self):
|
|
self._is_alive = True
|
|
return self
|
|
|
|
async def __aexit__(self, *args):
|
|
self._is_alive = False
|
|
|
|
async def fetch(self, url: str, **kwargs):
|
|
if self._delay:
|
|
await anyio.sleep(self._delay)
|
|
resp = MockResponse(url=url)
|
|
return resp
|
|
|
|
|
|
class _LogCounterStub:
|
|
def get_counts(self):
|
|
return {"debug": 0, "info": 0, "warning": 0, "error": 0, "critical": 0}
|
|
|
|
|
|
class SlowSpider:
|
|
"""Spider with slow-responding requests to simulate in-flight tasks during force-stop."""
|
|
|
|
def __init__(self, num_urls: int = 10):
|
|
self.concurrent_requests = 4
|
|
self.concurrent_requests_per_domain = 0
|
|
self.download_delay = 0.0
|
|
self.max_blocked_retries = 3
|
|
self.allowed_domains = set()
|
|
self.fp_include_kwargs = False
|
|
self.fp_include_headers = False
|
|
self.fp_keep_fragments = False
|
|
self.robots_txt_obey = False
|
|
self.development_mode = False
|
|
self.development_cache_dir = None
|
|
self.start_urls = []
|
|
self.name = "slow_spider"
|
|
self._log_counter = _LogCounterStub()
|
|
self._num_urls = num_urls
|
|
self.on_start_calls = []
|
|
self.on_close_calls = 0
|
|
|
|
async def parse(self, response) -> AsyncGenerator[Dict[str, Any] | Request | None, None]:
|
|
yield {"url": str(response)}
|
|
|
|
async def on_start(self, resuming=False):
|
|
self.on_start_calls.append({"resuming": resuming})
|
|
|
|
async def on_close(self):
|
|
self.on_close_calls += 1
|
|
|
|
async def on_error(self, request, error):
|
|
pass
|
|
|
|
async def on_scraped_item(self, item):
|
|
return item
|
|
|
|
async def is_blocked(self, response):
|
|
return False
|
|
|
|
async def retry_blocked_request(self, request, response):
|
|
return request
|
|
|
|
async def start_requests(self) -> AsyncGenerator[Request, None]:
|
|
for i in range(self._num_urls):
|
|
yield Request(f"https://example.com/page/{i}", sid="default")
|
|
|
|
|
|
def _make_engine(spider=None, session=None, crawldir=None, interval=300.0):
|
|
spider = spider or SlowSpider()
|
|
sm = SessionManager()
|
|
sm.add("default", session or MockSession())
|
|
return CrawlerEngine(spider, sm, crawldir=crawldir, interval=interval)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestForceStopCheckpointPreservation:
|
|
"""Verify checkpoint is saved BEFORE cancel_scope.cancel() on force-stop."""
|
|
|
|
@pytest.mark.anyio
|
|
async def test_force_stop_saves_checkpoint_before_cancel(self):
|
|
"""Core regression test: force-stop must save checkpoint, not delete it."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
spider = SlowSpider(num_urls=20)
|
|
# Use a slow session so tasks are in-flight when we force-stop
|
|
session = MockSession(delay=0.5)
|
|
engine = _make_engine(spider, session, crawldir=tmpdir, interval=0)
|
|
|
|
checkpoint_path = Path(tmpdir) / "checkpoint.pkl"
|
|
|
|
async def force_stop_after_delay():
|
|
"""Simulate two rapid Ctrl+C presses."""
|
|
# Wait for some tasks to start
|
|
await anyio.sleep(0.1)
|
|
engine.request_pause() # First Ctrl+C
|
|
await anyio.sleep(0.05)
|
|
engine.request_pause() # Second Ctrl+C (force stop)
|
|
|
|
async with anyio.create_task_group() as tg:
|
|
tg.start_soon(force_stop_after_delay)
|
|
await engine.crawl()
|
|
|
|
# The checkpoint file MUST exist after force-stop
|
|
assert checkpoint_path.exists(), (
|
|
"Checkpoint file was not saved (or was deleted) after force-stop. "
|
|
"This means the cancel_scope.cancel() ran before _save_checkpoint()."
|
|
)
|
|
# Engine must report as paused
|
|
assert engine.paused is True
|
|
|
|
@pytest.mark.anyio
|
|
async def test_graceful_pause_still_saves_checkpoint(self):
|
|
"""Single Ctrl+C (graceful pause) should save checkpoint as before."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
spider = SlowSpider(num_urls=5)
|
|
session = MockSession(delay=0.3)
|
|
engine = _make_engine(spider, session, crawldir=tmpdir, interval=0)
|
|
|
|
checkpoint_path = Path(tmpdir) / "checkpoint.pkl"
|
|
|
|
async def pause_after_delay():
|
|
await anyio.sleep(0.1)
|
|
engine.request_pause()
|
|
|
|
async with anyio.create_task_group() as tg:
|
|
tg.start_soon(pause_after_delay)
|
|
await engine.crawl()
|
|
|
|
assert checkpoint_path.exists(), "Checkpoint not saved on graceful pause"
|
|
assert engine.paused is True
|
|
|
|
@pytest.mark.anyio
|
|
async def test_force_stop_checkpoint_is_loadable(self):
|
|
"""Checkpoint saved during force-stop must be valid and loadable."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
spider = SlowSpider(num_urls=15)
|
|
session = MockSession(delay=0.4)
|
|
engine = _make_engine(spider, session, crawldir=tmpdir, interval=0)
|
|
|
|
async def force_stop():
|
|
await anyio.sleep(0.1)
|
|
engine.request_pause()
|
|
await anyio.sleep(0.05)
|
|
engine.request_pause()
|
|
|
|
async with anyio.create_task_group() as tg:
|
|
tg.start_soon(force_stop)
|
|
await engine.crawl()
|
|
|
|
# Load the checkpoint and verify it's valid
|
|
manager = CheckpointManager(tmpdir)
|
|
data = await manager.load()
|
|
assert data is not None, "Checkpoint data could not be loaded"
|
|
assert isinstance(data, CheckpointData)
|
|
# seen set should have some entries (requests were enqueued)
|
|
assert len(data.seen) > 0
|
|
|
|
@pytest.mark.anyio
|
|
async def test_normal_completion_cleans_up_checkpoint(self):
|
|
"""Normal completion (no pause) should still clean up checkpoint files."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
spider = SlowSpider(num_urls=2)
|
|
session = MockSession(delay=0.0)
|
|
engine = _make_engine(spider, session, crawldir=tmpdir, interval=0)
|
|
|
|
await engine.crawl()
|
|
|
|
checkpoint_path = Path(tmpdir) / "checkpoint.pkl"
|
|
# No pause → checkpoint should be cleaned up
|
|
assert not checkpoint_path.exists()
|
|
assert engine.paused is False
|
|
|
|
@pytest.mark.anyio
|
|
async def test_force_stop_without_checkpoint_system(self):
|
|
"""Force-stop without crawldir should not crash."""
|
|
spider = SlowSpider(num_urls=10)
|
|
session = MockSession(delay=0.3)
|
|
engine = _make_engine(spider, session, crawldir=None)
|
|
|
|
async def force_stop():
|
|
await anyio.sleep(0.1)
|
|
engine.request_pause()
|
|
await anyio.sleep(0.05)
|
|
engine.request_pause()
|
|
|
|
async with anyio.create_task_group() as tg:
|
|
tg.start_soon(force_stop)
|
|
await engine.crawl()
|
|
|
|
# Should not crash and should not be marked as paused
|
|
# (no checkpoint system = no pause state)
|
|
assert engine.paused is False
|
|
|
|
@pytest.mark.anyio
|
|
async def test_force_stop_preserves_existing_checkpoint(self):
|
|
"""If a checkpoint already exists, force-stop must not delete it."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# First run: do a graceful pause to create a checkpoint
|
|
spider1 = SlowSpider(num_urls=10)
|
|
session1 = MockSession(delay=0.2)
|
|
engine1 = _make_engine(spider1, session1, crawldir=tmpdir, interval=0)
|
|
|
|
async def pause1():
|
|
await anyio.sleep(0.1)
|
|
engine1.request_pause()
|
|
|
|
async with anyio.create_task_group() as tg:
|
|
tg.start_soon(pause1)
|
|
await engine1.crawl()
|
|
|
|
checkpoint_path = Path(tmpdir) / "checkpoint.pkl"
|
|
assert checkpoint_path.exists(), "First run should create checkpoint"
|
|
first_checkpoint_size = checkpoint_path.stat().st_size
|
|
|
|
# Second run: force-stop (the fix ensures checkpoint is updated, not deleted)
|
|
spider2 = SlowSpider(num_urls=10)
|
|
session2 = MockSession(delay=0.3)
|
|
engine2 = _make_engine(spider2, session2, crawldir=tmpdir, interval=0)
|
|
|
|
async def force_stop2():
|
|
await anyio.sleep(0.1)
|
|
engine2.request_pause()
|
|
await anyio.sleep(0.05)
|
|
engine2.request_pause()
|
|
|
|
async with anyio.create_task_group() as tg:
|
|
tg.start_soon(force_stop2)
|
|
await engine2.crawl()
|
|
|
|
# Checkpoint must still exist (updated, not deleted)
|
|
assert checkpoint_path.exists(), (
|
|
"Force-stop deleted the checkpoint instead of preserving it"
|
|
)
|