Repository: apify/crawlee-python
File: tests/unit/basic_crawler/test_basic_crawler.py (Python)

Code before:
from crawlee.statistics import FinalStatistics
from crawlee.storages import Dataset, KeyValueStore, RequestList, RequestQueue

if TYPE_CHECKING:
    from collections.abc import AsyncGenerator, Sequence

    import respx
    from yarl import URL

    from crawlee._types import JsonSerializable


@pytest.fixture
async def mock_event_manager() -> AsyncGenerator[EventManager, None]:
    async with EventManager(persist_state_interval=timedelta(milliseconds=50)) as event_manager:
        with patch('crawlee.service_container.get_event_manager', return_value=event_manager):
            yield event_manager


@pytest.fixture
async def key_value_store() -> AsyncGenerator[KeyValueStore, None]:
    kvs = await KeyValueStore.open()
    yield kvs
    await kvs.drop()
    kvs._clear_cache()
    kvs._drop_persist_state_event()


async def test_processes_requests() -> None:
    crawler = BasicCrawler(request_provider=RequestList(['http://a.com/', 'http://b.com/', 'http://c.com/']))
    calls = list[str]()

    @crawler.router.default_handler
    async def handler(context: BasicCrawlingContext) -> None:
        calls.append(context.request.url)

    await crawler.run()

    assert calls == ['http://a.com/', 'http://b.com/', 'http://c.com/']


async def test_processes_requests_from_run_args() -> None:
    crawler = BasicCrawler(request_provider=RequestList())
    calls = list[str]()

    @crawler.router.default_handler
    async def handler(context: BasicCrawlingContext) -> None:
        calls.append(context.request.url)

    await crawler.run(['http://a.com/', 'http://b.com/', 'http://c.com/'])


Review comment (suggestion):
Might be better to use a mock event manager and just discard it at the end of the test. That way the `KeyValueStore` doesn't have to deal with unregistering the listeners at all, and I think that's a good thing.

Code after:
from crawlee.storages import Dataset, KeyValueStore, RequestList, RequestQueue

if TYPE_CHECKING:
    from collections.abc import AsyncGenerator, Sequence

    import respx
    from yarl import URL

    from crawlee._types import JsonSerializable


@pytest.fixture
async def key_value_store() -> AsyncGenerator[KeyValueStore, None]:
    kvs = await KeyValueStore.open()
    yield kvs
    await kvs.drop()


async def test_processes_requests() -> None:
    crawler = BasicCrawler(request_provider=RequestList(['http://a.com/', 'http://b.com/', 'http://c.com/']))
    calls = list[str]()

    @crawler.router.default_handler
    async def handler(context: BasicCrawlingContext) -> None:
        calls.append(context.request.url)

    await crawler.run()

    assert calls == ['http://a.com/', 'http://b.com/', 'http://c.com/']


async def test_processes_requests_from_run_args() -> None:
    crawler = BasicCrawler(request_provider=RequestList())
    calls = list[str]()

    @crawler.router.default_handler
    async def handler(context: BasicCrawlingContext) -> None:
        calls.append(context.request.url)

    await crawler.run(['http://a.com/', 'http://b.com/', 'http://c.com/'])

    assert calls == ['http://a.com/', 'http://b.com/', 'http://c.com/']


async def test_allows_multiple_run_calls() -> None:
    crawler = BasicCrawler(request_provider=RequestList())
    calls = list[str]()

    @crawler.router.default_handler
    async def handler(context: BasicCrawlingContext) -> None:
        calls.append(context.request.url)