Attempt to un-hang tests

Author: janbuchar

tests/unit/crawlers/_basic/test_basic_crawler.py

@@ -4,6 +4,7 @@
 import asyncio
 import json
 import logging
+import multiprocessing
 import os
 import re
 import sys
@@ -1868,6 +1869,18 @@ def _process_run_crawlers(crawler_inputs: list[_CrawlerInput], storage_dir: str)
     ]
 
 
+def _run_in_subprocess(func: Any, *args: Any, **kwargs: Any) -> Any:
+    """Spawn a child process via ``ProcessPoolExecutor``, run *func*, and return the result.
+
+    Uses the ``spawn`` start method to avoid inheriting the parent's tokio runtime thread state
+    (created by ``pyo3-async-runtimes``) via ``fork``, which causes the child to hang on exit on
+    Linux.  See pyo3-async-runtimes#40 / #64.
+    """
+    ctx = multiprocessing.get_context('spawn')
+    with ProcessPoolExecutor(max_workers=1, mp_context=ctx) as executor:
+        return executor.submit(func, *args, **kwargs).result()
+
+
 async def test_crawler_state_persistence(tmp_path: Path) -> None:
     """Test that crawler statistics and state persist and are loaded correctly.
 
@@ -1877,37 +1890,34 @@ async def test_crawler_state_persistence(tmp_path: Path) -> None:
         storage_client=FileSystemStorageClient(), configuration=Configuration(storage_dir=str(tmp_path))
     )
 
-    with ProcessPoolExecutor() as executor:
-        # Crawl 2 requests in the first run and automatically persist the state.
-        first_run_state = executor.submit(
-            _process_run_crawlers,
-            crawler_inputs=[_CrawlerInput(requests=['https://a.placeholder.com', 'https://b.placeholder.com'])],
-            storage_dir=str(tmp_path),
-        ).result()[0]
-        # Expected state after first crawler run
-        assert first_run_state.requests_finished == 2
-        state = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_0')
-        assert state.get('urls') == ['https://a.placeholder.com', 'https://b.placeholder.com']
-
-    # Do not reuse the executor to simulate a fresh process to avoid modified class attributes.
-    with ProcessPoolExecutor() as executor:
-        # Crawl 1 additional requests in the second run, but use previously automatically persisted state.
-        second_run_state = executor.submit(
-            _process_run_crawlers,
-            crawler_inputs=[_CrawlerInput(requests=['https://c.placeholder.com'])],
-            storage_dir=str(tmp_path),
-        ).result()[0]
-
-        # Expected state after second crawler run
-        # 2 requests from first run and 1 request from second run.
-        assert second_run_state.requests_finished == 3
-
-        state = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_0')
-        assert state.get('urls') == [
-            'https://a.placeholder.com',
-            'https://b.placeholder.com',
-            'https://c.placeholder.com',
-        ]
+    # Crawl 2 requests in the first run and automatically persist the state.
+    first_run_state = _run_in_subprocess(
+        _process_run_crawlers,
+        crawler_inputs=[_CrawlerInput(requests=['https://a.placeholder.com', 'https://b.placeholder.com'])],
+        storage_dir=str(tmp_path),
+    )[0]
+    # Expected state after first crawler run
+    assert first_run_state.requests_finished == 2
+    state = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_0')
+    assert state.get('urls') == ['https://a.placeholder.com', 'https://b.placeholder.com']
+
+    # Crawl 1 additional request in the second run, but use previously automatically persisted state.
+    second_run_state = _run_in_subprocess(
+        _process_run_crawlers,
+        crawler_inputs=[_CrawlerInput(requests=['https://c.placeholder.com'])],
+        storage_dir=str(tmp_path),
+    )[0]
+
+    # Expected state after second crawler run
+    # 2 requests from first run and 1 request from second run.
+    assert second_run_state.requests_finished == 3
+
+    state = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_0')
+    assert state.get('urls') == [
+        'https://a.placeholder.com',
+        'https://b.placeholder.com',
+        'https://c.placeholder.com',
+    ]
 
     assert first_run_state.crawler_started_at == second_run_state.crawler_started_at
     assert first_run_state.crawler_finished_at
@@ -1926,41 +1936,39 @@ async def test_crawler_state_persistence_2_crawlers_with_migration(tmp_path: Pat
         storage_client=FileSystemStorageClient(), configuration=Configuration(storage_dir=str(tmp_path))
     )
 
-    with ProcessPoolExecutor() as executor:
-        # Run 2 crawler, each crawl 1 request in and automatically persist the state.
-        first_run_states = executor.submit(
-            _process_run_crawlers,
-            crawler_inputs=[
-                _CrawlerInput(requests=['https://a.placeholder.com']),
-                _CrawlerInput(requests=['https://c.placeholder.com']),
-            ],
-            storage_dir=str(tmp_path),
-        ).result()
-        # Expected state after first crawler run
-        assert first_run_states[0].requests_finished == 1
-        assert first_run_states[1].requests_finished == 1
-        state_0 = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_0')
-        assert state_0.get('urls') == ['https://a.placeholder.com']
-        state_1 = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_1')
-        assert state_1.get('urls') == ['https://c.placeholder.com']
-
-    with ProcessPoolExecutor() as executor:
-        # Run 2 crawler, each crawl 1 request in and automatically persist the state.
-        second_run_states = executor.submit(
-            _process_run_crawlers,
-            crawler_inputs=[
-                _CrawlerInput(requests=['https://b.placeholder.com']),
-                _CrawlerInput(requests=['https://d.placeholder.com']),
-            ],
-            storage_dir=str(tmp_path),
-        ).result()
-        # Expected state after first crawler run
-        assert second_run_states[0].requests_finished == 2
-        assert second_run_states[1].requests_finished == 2
-        state_0 = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_0')
-        assert state_0.get('urls') == ['https://a.placeholder.com', 'https://b.placeholder.com']
-        state_1 = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_1')
-        assert state_1.get('urls') == ['https://c.placeholder.com', 'https://d.placeholder.com']
+    # Run 2 crawlers, each crawl 1 request and automatically persist the state.
+    first_run_states = _run_in_subprocess(
+        _process_run_crawlers,
+        crawler_inputs=[
+            _CrawlerInput(requests=['https://a.placeholder.com']),
+            _CrawlerInput(requests=['https://c.placeholder.com']),
+        ],
+        storage_dir=str(tmp_path),
+    )
+    # Expected state after first crawler run
+    assert first_run_states[0].requests_finished == 1
+    assert first_run_states[1].requests_finished == 1
+    state_0 = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_0')
+    assert state_0.get('urls') == ['https://a.placeholder.com']
+    state_1 = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_1')
+    assert state_1.get('urls') == ['https://c.placeholder.com']
+
+    # Run 2 crawlers again, each crawl 1 more request.
+    second_run_states = _run_in_subprocess(
+        _process_run_crawlers,
+        crawler_inputs=[
+            _CrawlerInput(requests=['https://b.placeholder.com']),
+            _CrawlerInput(requests=['https://d.placeholder.com']),
+        ],
+        storage_dir=str(tmp_path),
+    )
+    # Expected state after second crawler run
+    assert second_run_states[0].requests_finished == 2
+    assert second_run_states[1].requests_finished == 2
+    state_0 = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_0')
+    assert state_0.get('urls') == ['https://a.placeholder.com', 'https://b.placeholder.com']
+    state_1 = await state_kvs.get_value(f'{BasicCrawler._CRAWLEE_STATE_KEY}_1')
+    assert state_1.get('urls') == ['https://c.placeholder.com', 'https://d.placeholder.com']
 
 
 async def test_crawler_intermediate_statistics() -> None: