mirror of https://github.com/scrapy/scrapy.git
401 lines
12 KiB
Python
401 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import random
|
|
from asyncio import Future
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import pytest
|
|
from twisted.internet.defer import Deferred, inlineCallbacks
|
|
|
|
from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
|
|
from scrapy.utils.defer import (
|
|
aiter_errback,
|
|
deferred_f_from_coro_f,
|
|
deferred_from_coro,
|
|
deferred_to_future,
|
|
iter_errback,
|
|
maybe_deferred_to_future,
|
|
mustbe_deferred,
|
|
parallel_async,
|
|
)
|
|
from tests.utils.decorators import coroutine_test, inline_callbacks_test
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import AsyncGenerator, Awaitable, Callable, Generator
|
|
|
|
|
|
@pytest.mark.requires_reactor # mustbe_deferred() requires a reactor
|
|
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
|
|
class TestMustbeDeferred:
|
|
@inline_callbacks_test
|
|
def test_success_function(self) -> Generator[Deferred[Any], Any, None]:
|
|
steps: list[int] = []
|
|
|
|
def _append(v: int) -> list[int]:
|
|
steps.append(v)
|
|
return steps
|
|
|
|
def _assert(v: list[int]) -> None:
|
|
assert v == [1, 2] # it is [1] with maybeDeferred
|
|
|
|
dfd = mustbe_deferred(_append, 1)
|
|
dfd.addCallback(_assert)
|
|
steps.append(2) # add another value, that should be caught by assertEqual
|
|
yield dfd
|
|
|
|
@inline_callbacks_test
|
|
def test_unfired_deferred(self) -> Generator[Deferred[Any], Any, None]:
|
|
steps: list[int] = []
|
|
|
|
def _append(v: int) -> Deferred[list[int]]:
|
|
from twisted.internet import reactor
|
|
|
|
steps.append(v)
|
|
dfd: Deferred[list[int]] = Deferred()
|
|
reactor.callLater(0, dfd.callback, steps)
|
|
return dfd
|
|
|
|
def _assert(v: list[int]) -> None:
|
|
assert v == [1, 2]
|
|
|
|
dfd = mustbe_deferred(_append, 1)
|
|
dfd.addCallback(_assert)
|
|
steps.append(2) # add another value, that should be caught by assertEqual
|
|
yield dfd
|
|
|
|
|
|
class TestIterErrback:
|
|
def test_iter_errback_good(self):
|
|
def itergood() -> Generator[int, None, None]:
|
|
yield from range(10)
|
|
|
|
errors = []
|
|
out = list(iter_errback(itergood(), errors.append))
|
|
assert out == list(range(10))
|
|
assert not errors
|
|
|
|
def test_iter_errback_bad(self):
|
|
def iterbad() -> Generator[int, None, None]:
|
|
for x in range(10):
|
|
if x == 5:
|
|
1 / 0
|
|
yield x
|
|
|
|
errors = []
|
|
out = list(iter_errback(iterbad(), errors.append))
|
|
assert out == [0, 1, 2, 3, 4]
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0].value, ZeroDivisionError)
|
|
|
|
|
|
class TestAiterErrback:
|
|
@coroutine_test
|
|
async def test_aiter_errback_good(self):
|
|
async def itergood() -> AsyncGenerator[int, None]:
|
|
for x in range(10):
|
|
yield x
|
|
|
|
errors = []
|
|
out = await collect_asyncgen(aiter_errback(itergood(), errors.append))
|
|
assert out == list(range(10))
|
|
assert not errors
|
|
|
|
@coroutine_test
|
|
async def test_iter_errback_bad(self):
|
|
async def iterbad() -> AsyncGenerator[int, None]:
|
|
for x in range(10):
|
|
if x == 5:
|
|
1 / 0
|
|
yield x
|
|
|
|
errors = []
|
|
out = await collect_asyncgen(aiter_errback(iterbad(), errors.append))
|
|
assert out == [0, 1, 2, 3, 4]
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0].value, ZeroDivisionError)
|
|
|
|
|
|
class TestAsyncDefTestsuite:
|
|
@coroutine_test
|
|
async def test_coroutine_test(self):
|
|
pass
|
|
|
|
@pytest.mark.xfail(reason="Checks that the test is actually executed", strict=True)
|
|
@coroutine_test
|
|
async def test_coroutine_test_xfail(self):
|
|
raise RuntimeError("This is expected to be raised")
|
|
|
|
|
|
@pytest.mark.requires_reactor # parallel_async() requires a reactor
|
|
class TestParallelAsync:
|
|
"""This tests _AsyncCooperatorAdapter by testing parallel_async which is its only usage.
|
|
|
|
parallel_async is called with the results of a callback (so an iterable of items, requests and None,
|
|
with arbitrary delays between values), and it uses Scraper._process_spidermw_output as the callable
|
|
(so a callable that returns a Deferred for an item, which will fire after pipelines process it, and
|
|
None for everything else). The concurrent task count is the CONCURRENT_ITEMS setting.
|
|
|
|
We want to test different concurrency values compared to the iterable length.
|
|
We also want to simulate the real usage, with arbitrary delays between getting the values
|
|
from the iterable. We also want to simulate sync and async results from the callable.
|
|
"""
|
|
|
|
CONCURRENT_ITEMS = 50
|
|
|
|
@staticmethod
|
|
def callable(o: int, results: list[int]) -> Deferred[None] | None:
|
|
from twisted.internet import reactor
|
|
|
|
if random.random() < 0.4:
|
|
# simulate async processing
|
|
dfd: Deferred[None] = Deferred()
|
|
dfd.addCallback(lambda _: results.append(o))
|
|
delay = random.random() / 8
|
|
reactor.callLater(delay, dfd.callback, None)
|
|
return dfd
|
|
# simulate trivial sync processing
|
|
results.append(o)
|
|
return None
|
|
|
|
def callable_wrapped(
|
|
self,
|
|
o: int,
|
|
results: list[int],
|
|
parallel_count: list[int],
|
|
max_parallel_count: list[int],
|
|
) -> Deferred[None] | None:
|
|
parallel_count[0] += 1
|
|
max_parallel_count[0] = max(max_parallel_count[0], parallel_count[0])
|
|
dfd = self.callable(o, results)
|
|
|
|
def decrement(_: Any = None) -> None:
|
|
assert parallel_count[0] > 0, parallel_count[0]
|
|
parallel_count[0] -= 1
|
|
|
|
if dfd is not None:
|
|
dfd.addBoth(decrement)
|
|
else:
|
|
decrement()
|
|
return dfd
|
|
|
|
@staticmethod
|
|
def get_async_iterable(length: int) -> AsyncGenerator[int, None]:
|
|
# simulate a simple callback without delays between results
|
|
return as_async_generator(range(length))
|
|
|
|
@staticmethod
|
|
async def get_async_iterable_with_delays(length: int) -> AsyncGenerator[int, None]:
|
|
# simulate a callback with delays between some of the results
|
|
from twisted.internet import reactor
|
|
|
|
for i in range(length):
|
|
if random.random() < 0.1:
|
|
dfd: Deferred[None] = Deferred()
|
|
delay = random.random() / 20
|
|
reactor.callLater(delay, dfd.callback, None)
|
|
await maybe_deferred_to_future(dfd)
|
|
yield i
|
|
|
|
@inline_callbacks_test
|
|
def test_simple(self):
|
|
for length in [20, 50, 100]:
|
|
parallel_count = [0]
|
|
max_parallel_count = [0]
|
|
results = []
|
|
ait = self.get_async_iterable(length)
|
|
dl = parallel_async(
|
|
ait,
|
|
self.CONCURRENT_ITEMS,
|
|
self.callable_wrapped,
|
|
results,
|
|
parallel_count,
|
|
max_parallel_count,
|
|
)
|
|
yield dl
|
|
assert list(range(length)) == sorted(results)
|
|
assert parallel_count[0] == 0
|
|
assert max_parallel_count[0] <= self.CONCURRENT_ITEMS, max_parallel_count[0]
|
|
|
|
@inline_callbacks_test
|
|
def test_delays(self):
|
|
for length in [20, 50, 100]:
|
|
parallel_count = [0]
|
|
max_parallel_count = [0]
|
|
results = []
|
|
ait = self.get_async_iterable_with_delays(length)
|
|
dl = parallel_async(
|
|
ait,
|
|
self.CONCURRENT_ITEMS,
|
|
self.callable_wrapped,
|
|
results,
|
|
parallel_count,
|
|
max_parallel_count,
|
|
)
|
|
yield dl
|
|
assert list(range(length)) == sorted(results)
|
|
assert parallel_count[0] == 0
|
|
assert max_parallel_count[0] <= self.CONCURRENT_ITEMS, max_parallel_count[0]
|
|
|
|
|
|
class TestDeferredFromCoro:
|
|
def test_deferred(self):
|
|
d = Deferred()
|
|
result = deferred_from_coro(d)
|
|
assert isinstance(result, Deferred)
|
|
assert result is d
|
|
|
|
def test_object(self):
|
|
result = deferred_from_coro(42)
|
|
assert result == 42
|
|
|
|
@inline_callbacks_test
|
|
def test_coroutine(self):
|
|
async def coroutine() -> int:
|
|
return 42
|
|
|
|
result = deferred_from_coro(coroutine())
|
|
assert isinstance(result, Deferred)
|
|
coro_result = yield result
|
|
assert coro_result == 42
|
|
|
|
@pytest.mark.only_asyncio
|
|
@inline_callbacks_test
|
|
def test_coroutine_asyncio(self):
|
|
async def coroutine() -> int:
|
|
await asyncio.sleep(0.01)
|
|
return 42
|
|
|
|
result = deferred_from_coro(coroutine())
|
|
assert isinstance(result, Deferred)
|
|
coro_result = yield result
|
|
assert coro_result == 42
|
|
|
|
@pytest.mark.only_asyncio
|
|
@inline_callbacks_test
|
|
def test_future(self):
|
|
future = Future()
|
|
result = deferred_from_coro(future)
|
|
assert isinstance(result, Deferred)
|
|
future.set_result(42)
|
|
future_result = yield result
|
|
assert future_result == 42
|
|
|
|
|
|
class TestDeferredFFromCoroF:
|
|
@inlineCallbacks
|
|
def _assert_result(
|
|
self, c_f: Callable[[], Awaitable[int]]
|
|
) -> Generator[Deferred[Any], Any, None]:
|
|
d_f = deferred_f_from_coro_f(c_f)
|
|
d = d_f()
|
|
assert isinstance(d, Deferred)
|
|
result = yield d
|
|
assert result == 42
|
|
|
|
@inline_callbacks_test
|
|
def test_coroutine(self):
|
|
async def c_f() -> int:
|
|
return 42
|
|
|
|
yield self._assert_result(c_f)
|
|
|
|
@pytest.mark.only_asyncio
|
|
@inline_callbacks_test
|
|
def test_coroutine_asyncio(self):
|
|
async def c_f() -> int:
|
|
await asyncio.sleep(0.01)
|
|
return 42
|
|
|
|
yield self._assert_result(c_f)
|
|
|
|
@pytest.mark.only_asyncio
|
|
@inline_callbacks_test
|
|
def test_future(self):
|
|
def c_f() -> Future[int]:
|
|
f: Future[int] = Future()
|
|
f.set_result(42)
|
|
return f
|
|
|
|
yield self._assert_result(c_f)
|
|
|
|
|
|
@pytest.mark.only_asyncio
|
|
class TestDeferredToFuture:
|
|
@coroutine_test
|
|
async def test_deferred(self):
|
|
d = Deferred()
|
|
result = deferred_to_future(d)
|
|
assert isinstance(result, Future)
|
|
d.callback(42)
|
|
future_result = await result
|
|
assert future_result == 42
|
|
|
|
@coroutine_test
|
|
async def test_wrapped_coroutine(self):
|
|
async def c_f() -> int:
|
|
return 42
|
|
|
|
d = deferred_from_coro(c_f())
|
|
result = deferred_to_future(d)
|
|
assert isinstance(result, Future)
|
|
future_result = await result
|
|
assert future_result == 42
|
|
|
|
@coroutine_test
|
|
async def test_wrapped_coroutine_asyncio(self):
|
|
async def c_f() -> int:
|
|
await asyncio.sleep(0.01)
|
|
return 42
|
|
|
|
d = deferred_from_coro(c_f())
|
|
result = deferred_to_future(d)
|
|
assert isinstance(result, Future)
|
|
future_result = await result
|
|
assert future_result == 42
|
|
|
|
|
|
@pytest.mark.only_asyncio
|
|
class TestMaybeDeferredToFutureAsyncio:
|
|
@coroutine_test
|
|
async def test_deferred(self):
|
|
d = Deferred()
|
|
result = maybe_deferred_to_future(d)
|
|
assert isinstance(result, Future)
|
|
d.callback(42)
|
|
future_result = await result
|
|
assert future_result == 42
|
|
|
|
@coroutine_test
|
|
async def test_wrapped_coroutine(self):
|
|
async def c_f() -> int:
|
|
return 42
|
|
|
|
d = deferred_from_coro(c_f())
|
|
result = maybe_deferred_to_future(d)
|
|
assert isinstance(result, Future)
|
|
future_result = await result
|
|
assert future_result == 42
|
|
|
|
@coroutine_test
|
|
async def test_wrapped_coroutine_asyncio(self):
|
|
async def c_f() -> int:
|
|
await asyncio.sleep(0.01)
|
|
return 42
|
|
|
|
d = deferred_from_coro(c_f())
|
|
result = maybe_deferred_to_future(d)
|
|
assert isinstance(result, Future)
|
|
future_result = await result
|
|
assert future_result == 42
|
|
|
|
|
|
@pytest.mark.only_not_asyncio
|
|
class TestMaybeDeferredToFutureNotAsyncio:
|
|
@coroutine_test
|
|
async def test_deferred(self):
|
|
d = Deferred()
|
|
result = maybe_deferred_to_future(d)
|
|
assert isinstance(result, Deferred)
|
|
assert result is d
|