import random import pytest from twisted.internet import defer, reactor from twisted.python.failure import Failure from twisted.trial import unittest from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen from scrapy.utils.defer import ( aiter_errback, deferred_f_from_coro_f, iter_errback, maybe_deferred_to_future, mustbe_deferred, parallel_async, process_chain, process_parallel, ) class TestMustbeDeferred(unittest.TestCase): def test_success_function(self): steps = [] def _append(v): steps.append(v) return steps dfd = mustbe_deferred(_append, 1) dfd.addCallback(self.assertEqual, [1, 2]) # it is [1] with maybeDeferred steps.append(2) # add another value, that should be caught by assertEqual return dfd def test_unfired_deferred(self): steps = [] def _append(v): steps.append(v) dfd = defer.Deferred() reactor.callLater(0, dfd.callback, steps) return dfd dfd = mustbe_deferred(_append, 1) dfd.addCallback(self.assertEqual, [1, 2]) # it is [1] with maybeDeferred steps.append(2) # add another value, that should be caught by assertEqual return dfd def cb1(value, arg1, arg2): return f"(cb1 {value} {arg1} {arg2})" def cb2(value, arg1, arg2): return defer.succeed(f"(cb2 {value} {arg1} {arg2})") def cb3(value, arg1, arg2): return f"(cb3 {value} {arg1} {arg2})" def cb_fail(value, arg1, arg2): return Failure(TypeError()) def eb1(failure, arg1, arg2): return f"(eb1 {failure.value.__class__.__name__} {arg1} {arg2})" class TestDeferUtils(unittest.TestCase): @defer.inlineCallbacks def test_process_chain(self): x = yield process_chain([cb1, cb2, cb3], "res", "v1", "v2") assert x == "(cb3 (cb2 (cb1 res v1 v2) v1 v2) v1 v2)" with pytest.raises(TypeError): yield process_chain([cb1, cb_fail, cb3], "res", "v1", "v2") @defer.inlineCallbacks def test_process_parallel(self): x = yield process_parallel([cb1, cb2, cb3], "res", "v1", "v2") assert x == ["(cb1 res v1 v2)", "(cb2 res v1 v2)", "(cb3 res v1 v2)"] def test_process_parallel_failure(self): d = process_parallel([cb1, cb_fail, cb3], "res", "v1", "v2") self.failUnlessFailure(d, TypeError) return d class TestIterErrback: def test_iter_errback_good(self): def itergood(): 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(): 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(unittest.TestCase): @deferred_f_from_coro_f async def test_aiter_errback_good(self): async def itergood(): 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 @deferred_f_from_coro_f async def test_iter_errback_bad(self): async def iterbad(): 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(unittest.TestCase): @deferred_f_from_coro_f async def test_deferred_f_from_coro_f(self): pass @deferred_f_from_coro_f async def test_deferred_f_from_coro_f_generator(self): yield @pytest.mark.xfail(reason="Checks that the test is actually executed", strict=True) @deferred_f_from_coro_f async def test_deferred_f_from_coro_f_xfail(self): raise RuntimeError("This is expected to be raised") class TestAsyncCooperator(unittest.TestCase): """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, results): if random.random() < 0.4: # simulate async processing dfd = defer.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 @staticmethod def get_async_iterable(length): # simulate a simple callback without delays between results return as_async_generator(range(length)) @staticmethod async def get_async_iterable_with_delays(length): # simulate a callback with delays between some of the results for i in range(length): if random.random() < 0.1: dfd = defer.Deferred() delay = random.random() / 20 reactor.callLater(delay, dfd.callback, None) await maybe_deferred_to_future(dfd) yield i @defer.inlineCallbacks def test_simple(self): for length in [20, 50, 100]: results = [] ait = self.get_async_iterable(length) dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results) yield dl assert list(range(length)) == sorted(results) @defer.inlineCallbacks def test_delays(self): for length in [20, 50, 100]: results = [] ait = self.get_async_iterable_with_delays(length) dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results) yield dl assert list(range(length)) == sorted(results)