diff --git a/tests/e2e/actor_source_base/server.py b/tests/e2e/actor_source_base/server.py index 04d46b97..84c1d39a 100644 --- a/tests/e2e/actor_source_base/server.py +++ b/tests/e2e/actor_source_base/server.py @@ -17,6 +17,9 @@ /redirect-target - Page the redirect lands on /slow - Answers only after 10 minutes, to keep a request in flight while a run is interrupted /delayed/ - Answers after 5 seconds with a link to /delayed/, so a request is always in flight + /files/. - A binary file of bytes, with the content type of (bin, pdf or png) + /start-urls.txt - A plain-text list of URLs, for the `requestsFromUrl` request list source. The URLs point to + example.com, since the URL extraction only matches hosts with a top-level domain. The homepage includes both direct product links (for Scrapy spiders that look for /products/ links on the start page) and category links (for testing crawl depth with Crawlee crawlers). @@ -42,16 +45,31 @@ '3': {'name': 'Widget C', 'price': '$39.99', 'description': 'A premium widget for professionals'}, } +_FILE_CONTENT_TYPES = { + 'bin': 'application/octet-stream', + 'pdf': 'application/pdf', + 'png': 'image/png', +} -async def _send_html(send: Send, html: str, status: int = 200) -> None: + +def file_body(size: int) -> bytes: + """Return the body of the `/files/.` route. Tests import it to compare the downloaded bytes.""" + return bytes(i % 256 for i in range(size)) + + +async def _send_body(send: Send, body: bytes, content_type: str, status: int = 200) -> None: await send( { 'type': 'http.response.start', 'status': status, - 'headers': [[b'content-type', b'text/html; charset=utf-8']], + 'headers': [[b'content-type', content_type.encode()]], } ) - await send({'type': 'http.response.body', 'body': html.encode()}) + await send({'type': 'http.response.body', 'body': body}) + + +async def _send_html(send: Send, html: str, status: int = 200) -> None: + await _send_body(send, html.encode(), 'text/html; charset=utf-8', status) async def _send_redirect(send: Send, location: str) -> None: @@ -149,6 +167,19 @@ async def app(scope: dict[str, Any], _receive: Receive, send: Send) -> None: f'Next' f'', ) + elif path.startswith('/files/'): + size, _, extension = path.split('/')[-1].partition('.') + content_type = _FILE_CONTENT_TYPES.get(extension) + if not size.isdigit() or content_type is None: + await _send_html(send, 'Not Found', 404) + return + await _send_body(send, file_body(int(size)), content_type) + elif path == '/start-urls.txt': + await _send_body( + send, + b'Products to scrape:\nhttps://example.com/products/2\nhttps://example.com/products/3\n', + 'text/plain; charset=utf-8', + ) elif path.startswith('/deep/'): try: n = int(path.split('/')[-1]) diff --git a/tests/e2e/test_actor_events.py b/tests/e2e/test_actor_events.py index da1a9378..7954f554 100644 --- a/tests/e2e/test_actor_events.py +++ b/tests/e2e/test_actor_events.py @@ -1,11 +1,15 @@ from __future__ import annotations import asyncio +from datetime import timedelta from typing import TYPE_CHECKING +from .._utils import poll_until_condition from apify import Actor if TYPE_CHECKING: + from apify_client import ApifyClientAsync + from .conftest import MakeActorFunction, RunActorFunction @@ -167,3 +171,85 @@ async def handler(context: ParselCrawlingContext) -> None: assert run_log is not None assert 'error occurred during closing of asynchronous generator' not in run_log assert 'asynchronous generator is already running' not in run_log + + +async def test_aborting_event_on_graceful_abort( + make_actor: MakeActorFunction, + apify_client_async: ApifyClientAsync, +) -> None: + """A graceful abort from the API reaches the `ABORTING` listener before the platform stops the run.""" + + async def main() -> None: + from crawlee.events._types import Event + + async def on_aborting() -> None: + await Actor.set_value('ABORTING_RECEIVED', {'received': True}) + + async with Actor: + Actor.on(Event.ABORTING, on_aborting) + Actor.log.info('Waiting for the abort') + await asyncio.sleep(300) + + actor = await make_actor(label='actor-aborting-event', main_func=main) + run = await actor.start() + run_client = apify_client_async.run(run.id) + + # Abort the run only once the listener is registered. The container startup time is highly variable, so poll the + # log with a growing interval. + log_client = run_client.log() + log = await poll_until_condition( + log_client.get, + lambda log: bool(log and 'Waiting for the abort' in log), + timeout=300, + poll_interval=2, + backoff_factor=1.2, + ) + assert log is not None + assert 'Waiting for the abort' in log, f'The run did not start waiting in time:\n{log}' + + await run_client.abort(gracefully=True) + run_result = await run_client.wait_for_finish(wait_duration=timedelta(seconds=600)) + assert run_result is not None + assert run_result.status == 'ABORTED' + + record = await run_client.key_value_store().get_record('ABORTING_RECEIVED') + assert record is not None, 'The ABORTING listener was not called.' + assert record['value'] == {'received': True} + + +async def test_migrating_listeners_run_before_reboot( + make_actor: MakeActorFunction, + run_actor: RunActorFunction, +) -> None: + """`Actor.reboot()` calls the `MIGRATING` listeners and the `PERSIST_STATE` listeners with `is_migrating=True`.""" + + async def main() -> None: + from crawlee.events._types import Event, EventPersistStateData + + async def on_migrating() -> None: + await Actor.set_value('MIGRATING_RECEIVED', {'received': True}) + + async def on_persist_state(data: EventPersistStateData) -> None: + if data.is_migrating: + await Actor.set_value('PERSIST_STATE_MIGRATING', {'received': True}) + + async with Actor: + run = await Actor.apify_client.run(Actor.configuration.actor_run_id or '').get() + assert run is not None, 'The Actor could not read its own run.' + + if run.stats.reboot_count == 0: + Actor.on(Event.MIGRATING, on_migrating) + Actor.on(Event.PERSIST_STATE, on_persist_state) + await Actor.reboot() + + actor = await make_actor(label='actor-migrating-event', main_func=main) + run_result = await run_actor(actor) + + assert run_result.status == 'SUCCEEDED' + assert run_result.stats.reboot_count == 1 + + kvs = actor.last_run().key_value_store() + for key in ('MIGRATING_RECEIVED', 'PERSIST_STATE_MIGRATING'): + record = await kvs.get_record(key) + assert record is not None, f'The listener did not write {key}.' + assert record['value'] == {'received': True} diff --git a/tests/e2e/test_actor_key_value_store.py b/tests/e2e/test_actor_key_value_store.py index 47edf688..eceac62e 100644 --- a/tests/e2e/test_actor_key_value_store.py +++ b/tests/e2e/test_actor_key_value_store.py @@ -140,3 +140,36 @@ async def main() -> None: run_result = await run_actor(actor) assert run_result.status == 'SUCCEEDED' + + +async def test_use_state_survives_reboot( + make_actor: MakeActorFunction, + run_actor: RunActorFunction, +) -> None: + """The state from `Actor.use_state()` is persisted before a reboot and loaded again in the next container.""" + + async def main() -> None: + async with Actor: + run = await Actor.apify_client.run(Actor.configuration.actor_run_id or '').get() + assert run is not None, 'The Actor could not read its own run.' + + state = await Actor.use_state({'containers': 0}) + containers = state['containers'] + assert isinstance(containers, int), f'state={dict(state)}' + state['containers'] = containers + 1 + + if run.stats.reboot_count == 0: + state['before_reboot'] = ['a', 'b'] + await Actor.reboot() + else: + assert state.get('before_reboot') == ['a', 'b'], f'state={dict(state)}' + + actor = await make_actor(label='use-state-reboot', main_func=main) + run_result = await run_actor(actor) + + assert run_result.status == 'SUCCEEDED' + assert run_result.stats.reboot_count == 1 + + state_record = await actor.last_run().key_value_store().get_record('APIFY_GLOBAL_STATE') + assert state_record is not None + assert state_record['value'] == {'containers': 2, 'before_reboot': ['a', 'b']} diff --git a/tests/e2e/test_actor_request_list.py b/tests/e2e/test_actor_request_list.py new file mode 100644 index 00000000..a0b0cbad --- /dev/null +++ b/tests/e2e/test_actor_request_list.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from .._utils import poll_until_condition +from apify import Actor + +if TYPE_CHECKING: + from apify_client import ApifyClientAsync + + from .conftest import MakeActorFunction, RunActorFunction + +_EXPECTED_REQUESTS = [ + ('https://example.com/products/1', 'direct'), + ('https://example.com/products/2', 'remote'), + ('https://example.com/products/3', 'remote'), +] + + +async def test_request_list_from_actor_input( + make_actor: MakeActorFunction, + run_actor: RunActorFunction, + apify_client_async: ApifyClientAsync, +) -> None: + """Direct URLs and a `requestsFromUrl` list from the Actor input reach the crawler with their user data.""" + + async def main() -> None: + from crawlee.crawlers import BasicCrawler, BasicCrawlingContext + + from apify.request_loaders import ApifyRequestList + + async with Actor: + actor_input = await Actor.get_input() or {} + request_list = await ApifyRequestList.open( + request_list_sources_input=actor_input.get('requestListSources', []), + ) + crawler = BasicCrawler(request_manager=await request_list.to_tandem()) + + # The handler only records the requests. It doesn't fetch them, so the URLs don't have to resolve. + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + await context.push_data({'url': context.request.url, 'source': context.request.user_data.get('source')}) + + await crawler.run() + + actor = await make_actor(label='request-list', main_func=main) + run_result = await run_actor( + actor, + run_input={ + 'requestListSources': [ + {'url': 'https://example.com/products/1', 'userData': {'source': 'direct'}}, + {'requestsFromUrl': 'http://localhost:8080/start-urls.txt', 'userData': {'source': 'remote'}}, + ], + }, + ) + assert run_result.status == 'SUCCEEDED' + + dataset = actor.last_run().dataset() + items = await poll_until_condition( + dataset.list_items, + lambda page: len(page.items) >= len(_EXPECTED_REQUESTS), + timeout=30, + ) + assert sorted((item['url'], item['source']) for item in items.items) == _EXPECTED_REQUESTS + + # The tandem moves every request from the list into the default RQ and marks it as handled there. + assert run_result.default_request_queue_id is not None + rq_client = apify_client_async.request_queue(run_result.default_request_queue_id) + requests = await rq_client.list_requests() + handled_urls = sorted(request.url for request in requests.items if request.handled_at is not None) + assert handled_urls == [url for url, _ in _EXPECTED_REQUESTS] diff --git a/tests/e2e/test_crawlee/actor_source/main_delayed_chain_crawler.py b/tests/e2e/test_crawlee/actor_source/main_delayed_chain_crawler.py new file mode 100644 index 00000000..87366100 --- /dev/null +++ b/tests/e2e/test_crawlee/actor_source/main_delayed_chain_crawler.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from crawlee import ConcurrencySettings +from crawlee.crawlers import ParselCrawler, ParselCrawlingContext + +from apify import Actor + +if TYPE_CHECKING: + from crawlee.crawlers import BasicCrawlingContext + +IN_FLIGHT_LOG_MARKER = 'Sending the request to the server' +"""Logged before every navigation, so a test knows which request is in flight.""" + + +async def main() -> None: + async with Actor: + # One request at a time, so the test knows what is in flight when it interrupts the run. + crawler = ParselCrawler(concurrency_settings=ConcurrencySettings(max_concurrency=1, desired_concurrency=1)) + + @crawler.pre_navigation_hook + async def mark_in_flight(context: BasicCrawlingContext) -> None: + Actor.log.info(f'{IN_FLIGHT_LOG_MARKER}: {context.request.url}') + + @crawler.router.default_handler + async def handler(context: ParselCrawlingContext) -> None: + await context.push_data({'url': context.request.url}) + await context.enqueue_links() + + await crawler.run(['http://localhost:8080/delayed/1']) diff --git a/tests/e2e/test_crawlee/actor_source/main_file_download_crawler.py b/tests/e2e/test_crawlee/actor_source/main_file_download_crawler.py new file mode 100644 index 00000000..d8ac9eca --- /dev/null +++ b/tests/e2e/test_crawlee/actor_source/main_file_download_crawler.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from crawlee.crawlers import FileDownloadCrawler, FileDownloadCrawlingContext +from crawlee.storages import RequestQueue + +from apify import Actor + +_BUFFERED_URLS = ['http://localhost:8080/files/1024.pdf', 'http://localhost:8080/files/2048.png'] +_STREAMED_URLS = ['http://localhost:8080/files/5242880.bin'] + + +async def _store_file(context: FileDownloadCrawlingContext, content: bytes, chunk_count: int) -> None: + key = context.request.url.rsplit('/', 1)[-1] + kvs = await context.get_key_value_store() + await kvs.set_value(key, content, content_type=context.http_response.headers.get('content-type')) + await context.push_data({'url': context.request.url, 'key': key, 'chunk_count': chunk_count}) + + +async def main() -> None: + async with Actor: + buffered_crawler = FileDownloadCrawler() + + @buffered_crawler.router.default_handler + async def buffered_handler(context: FileDownloadCrawlingContext) -> None: + content = await context.http_response.read() + await _store_file(context, content, chunk_count=1) + + await buffered_crawler.run(_BUFFERED_URLS) + + # The default RQ already holds the buffered crawl, so the streamed crawl gets a queue of its own. + streamed_crawler = FileDownloadCrawler(stream=True, request_manager=await RequestQueue.open(alias='streamed')) + + @streamed_crawler.router.default_handler + async def streamed_handler(context: FileDownloadCrawlingContext) -> None: + chunks = [chunk async for chunk in context.http_response.read_stream()] + await _store_file(context, b''.join(chunks), chunk_count=len(chunks)) + + await streamed_crawler.run(_STREAMED_URLS) diff --git a/tests/e2e/test_crawlee/actor_source/main_reboot_crawler.py b/tests/e2e/test_crawlee/actor_source/main_reboot_crawler.py new file mode 100644 index 00000000..09d9d841 --- /dev/null +++ b/tests/e2e/test_crawlee/actor_source/main_reboot_crawler.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from crawlee import ConcurrencySettings +from crawlee.crawlers import ParselCrawler, ParselCrawlingContext + +from apify import Actor + +REBOOT_URL = 'http://localhost:8080/about' +"""The page whose handler reboots the first container, so part of the site is crawled before and part after.""" + + +async def main() -> None: + async with Actor: + run = await Actor.apify_client.run(Actor.configuration.actor_run_id or '').get() + assert run is not None, 'The Actor could not read its own run.' + is_first_container = run.stats.reboot_count == 0 + + # One request at a time, so no other request is in progress when the handler reboots the run. + crawler = ParselCrawler( + max_crawl_depth=2, + concurrency_settings=ConcurrencySettings(max_concurrency=1, desired_concurrency=1), + ) + + @crawler.router.default_handler + async def handler(context: ParselCrawlingContext) -> None: + if context.request.url == REBOOT_URL and is_first_container: + request_manager = await crawler.get_request_manager() + await request_manager.reclaim_request(context.request) + await Actor.reboot() + + await context.enqueue_links() + + if '/products/' in context.request.url: + await context.push_data( + { + 'url': context.request.url, + 'name': context.selector.css('h1::text').get('').strip(), + } + ) + + await crawler.run(['http://localhost:8080/']) + + await Actor.set_value( + 'CRAWLER_RESULT', + {'requests_finished': crawler.statistics.state.requests_finished}, + ) diff --git a/tests/e2e/test_crawlee/test_file_download_crawler.py b/tests/e2e/test_crawlee/test_file_download_crawler.py new file mode 100644 index 00000000..0cc1660d --- /dev/null +++ b/tests/e2e/test_crawlee/test_file_download_crawler.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from ..._utils import poll_until_condition +from ..actor_source_base.server import file_body +from .conftest import read_actor_source + +if TYPE_CHECKING: + from ..conftest import MakeActorFunction, RunActorFunction + +# The key, size and content type of each file the Actor downloads from the `/files/.` route. +_EXPECTED_FILES = { + '1024.pdf': (1024, 'application/pdf'), + '2048.png': (2048, 'image/png'), + '5242880.bin': (5242880, 'application/octet-stream'), +} + + +async def test_file_download_crawler(make_actor: MakeActorFunction, run_actor: RunActorFunction) -> None: + """Buffered and streamed downloads store the exact file bytes and content type in the KVS.""" + actor = await make_actor( + label='crawl-file-download', + source_files={ + 'actor/main.py': read_actor_source('main_file_download_crawler.py'), + }, + ) + run_result = await run_actor(actor) + assert run_result.status == 'SUCCEEDED' + + dataset = actor.last_run().dataset() + items = await poll_until_condition( + dataset.list_items, + lambda page: len(page.items) >= len(_EXPECTED_FILES), + timeout=30, + ) + items_by_key = {item['key']: item for item in items.items} + assert set(items_by_key) == set(_EXPECTED_FILES) + + # A 5 MiB body that arrives in one chunk would mean the streamed crawler buffered it. + assert items_by_key['5242880.bin']['chunk_count'] > 1 + + kvs = actor.last_run().key_value_store() + for key, (size, content_type) in _EXPECTED_FILES.items(): + record = await kvs.get_record_as_bytes(key) + assert record is not None, f'Missing KVS record: {key}' + assert record['content_type'].split(';')[0] == content_type, f'{key}: {record["content_type"]}' + assert record['value'] == file_body(size), f'{key}: the stored bytes differ from the downloaded file' diff --git a/tests/e2e/test_crawlee/test_graceful_abort_crawler.py b/tests/e2e/test_crawlee/test_graceful_abort_crawler.py new file mode 100644 index 00000000..38158183 --- /dev/null +++ b/tests/e2e/test_crawlee/test_graceful_abort_crawler.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +from collections import Counter +from datetime import timedelta +from typing import TYPE_CHECKING + +from ..._utils import poll_until_condition +from .actor_source.main_delayed_chain_crawler import IN_FLIGHT_LOG_MARKER +from .conftest import read_actor_source + +if TYPE_CHECKING: + from apify_client import ApifyClientAsync + + from ..conftest import MakeActorFunction + + +async def test_graceful_abort_crawler(make_actor: MakeActorFunction, apify_client_async: ApifyClientAsync) -> None: + """A graceful abort persists the crawler statistics, and every handled request has its item pushed exactly once.""" + actor = await make_actor( + label='crawl-graceful-abort', + source_files={ + 'actor/main.py': read_actor_source('main_delayed_chain_crawler.py'), + }, + ) + run = await actor.start() + run_client = apify_client_async.run(run.id) + + # The crawler handles one request at a time, so once it starts the third page, the first two are handled. The + # container startup time is highly variable, so poll the log with a growing interval. + third_page_marker = f'{IN_FLIGHT_LOG_MARKER}: http://localhost:8080/delayed/3' + log_client = run_client.log() + log = await poll_until_condition( + log_client.get, + lambda log: bool(log and third_page_marker in log), + timeout=300, + poll_interval=2, + backoff_factor=1.2, + ) + assert log is not None + assert third_page_marker in log, f'The run did not log {third_page_marker!r} in time:\n{log}' + + await run_client.abort(gracefully=True) + run_result = await run_client.wait_for_finish(wait_duration=timedelta(seconds=600)) + assert run_result is not None + assert run_result.status == 'ABORTED' + + assert run.default_request_queue_id is not None + requests = await apify_client_async.request_queue(run.default_request_queue_id).list_requests() + handled_urls = [request.url for request in requests.items if request.handled_at is not None] + pending_urls = [request.url for request in requests.items if request.handled_at is None] + assert len(handled_urls) >= 2, f'Fewer than two requests were marked as handled: {requests.items}' + assert pending_urls, f'No request stayed pending: {requests.items}' + + items = await poll_until_condition( + run_client.dataset().list_items, + lambda page: len(page.items) >= len(handled_urls), + timeout=30, + ) + assert Counter(item['url'] for item in items.items) == Counter(handled_urls) + + # The platform sends `persistState` with the abort, and the crawler persists its statistics in the default KVS. + kvs_client = run_client.key_value_store() + keys = await kvs_client.list_keys() + statistics_keys = [item.key for item in keys.items if 'CRAWLER_STATISTICS' in item.key] + assert len(statistics_keys) == 1, f'Expected one crawler statistics record: {[item.key for item in keys.items]}' + statistics_record = await kvs_client.get_record(statistics_keys[0]) + assert statistics_record is not None + requests_finished = statistics_record['value']['requestsFinished'] + assert 2 <= requests_finished <= len(handled_urls), f'requestsFinished={requests_finished}' diff --git a/tests/e2e/test_crawlee/test_ppe_crawler.py b/tests/e2e/test_crawlee/test_ppe_crawler.py new file mode 100644 index 00000000..e9c9b726 --- /dev/null +++ b/tests/e2e/test_crawlee/test_ppe_crawler.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +from decimal import Decimal +from typing import TYPE_CHECKING + +from ..._utils import poll_until_condition +from .conftest import read_actor_source + +if TYPE_CHECKING: + from apify_client import ApifyClientAsync + + from ..conftest import MakeActorFunction, RunActorFunction + + +async def test_ppe_crawler_stops_pushing_once_over_budget( + make_actor: MakeActorFunction, + run_actor: RunActorFunction, + apify_client_async: ApifyClientAsync, +) -> None: + """A pay-per-event crawler charges the synthetic dataset-item event and stops pushing once over budget.""" + actor = await make_actor( + label='crawl-ppe', + source_files={ + 'actor/main.py': read_actor_source('main_parsel_crawler.py'), + }, + ) + + await actor.update( + pricing_infos=[ + { + 'pricingModel': 'PAY_PER_EVENT', + 'apifyMarginPercentage': 0.0, + 'createdAt': '2024-01-01T00:00:00.000Z', + 'startedAt': '2024-01-01T00:00:00.000Z', + 'pricingPerEvent': { + 'actorChargeEvents': { + 'apify-default-dataset-item': { + 'eventTitle': 'Default dataset item', + 'eventPriceUsd': 0.05, + 'eventDescription': 'One item written to the default dataset', + }, + }, + }, + }, + ] + ) + + # The crawler scrapes three products. At $0.05 an item, a $0.075 budget covers one of them. The second push + # overcharges by one item, so the run is over budget and the third push is dropped and never charged. + run = await run_actor(actor, max_total_charge_usd=Decimal('0.075')) + + # The overcharge makes the platform auto-abort the run, which races the Actor's clean exit. + assert run.status in ('SUCCEEDED', 'ABORTED') + + expected_charges = {'apify-default-dataset-item': 2} + + # The platform derives the charges from dataset writes asynchronously, so they can lag behind the run reaching + # a terminal status. Refetch until they propagate. + charged_run = await poll_until_condition( + apify_client_async.run(run.id).get, + lambda refetched: refetched is not None and refetched.charged_event_counts == expected_charges, + timeout=120, + poll_interval=1, + ) + + assert charged_run is not None + assert charged_run.charged_event_counts == expected_charges + + # `total` counts the whole dataset, unlike `count`, which only covers the items returned in this page. + items = await actor.last_run().dataset().list_items() + assert items.total == 2 diff --git a/tests/e2e/test_crawlee/test_reboot_crawler.py b/tests/e2e/test_crawlee/test_reboot_crawler.py new file mode 100644 index 00000000..d8d148d5 --- /dev/null +++ b/tests/e2e/test_crawlee/test_reboot_crawler.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +from collections import Counter +from typing import TYPE_CHECKING + +from ..._utils import poll_until_condition +from .conftest import read_actor_source + +if TYPE_CHECKING: + from ..conftest import MakeActorFunction, RunActorFunction + +# With `max_crawl_depth=2`, the crawler reaches the homepage, 2 categories, the about page, 3 products, `/deep/1` +# and `/deep/2`. +_EXPECTED_PAGE_COUNT = 9 + +_EXPECTED_PRODUCTS = {'Widget A', 'Widget B', 'Widget C'} + + +async def test_reboot_crawler(make_actor: MakeActorFunction, run_actor: RunActorFunction) -> None: + """A crawl split by a reboot visits every page once and keeps its statistics from before the reboot.""" + actor = await make_actor( + label='crawl-reboot', + source_files={ + 'actor/main.py': read_actor_source('main_reboot_crawler.py'), + }, + ) + run_result = await run_actor(actor) + assert run_result.status == 'SUCCEEDED' + assert run_result.stats.reboot_count == 1 + + dataset = actor.last_run().dataset() + items = await poll_until_condition( + dataset.list_items, + lambda page: len(page.items) >= len(_EXPECTED_PRODUCTS), + timeout=30, + ) + assert Counter(item['name'] for item in items.items) == Counter(_EXPECTED_PRODUCTS) + + result = await actor.last_run().key_value_store().get_record('CRAWLER_RESULT') + assert result is not None + assert result['value']['requests_finished'] == _EXPECTED_PAGE_COUNT