Add a test for pipelines using asyncio.

This commit is contained in:
Andrey Rakhmatullin 2019-09-10 14:57:07 +05:00
parent 1f9cef787d
commit bfdd552a32
1 changed files with 15 additions and 0 deletions

View File

@ -1,3 +1,5 @@
import asyncio
from twisted.internet import defer
from twisted.internet.defer import Deferred
from twisted.trial import unittest
@ -33,6 +35,13 @@ class AsyncDefPipeline:
return item
class AsyncDefAsyncioPipeline:
async def process_item(self, item, spider):
await asyncio.sleep(0.2)
item['pipeline_passed'] = True
return item
class ItemSpider(Spider):
name = 'itemspider'
@ -82,3 +91,9 @@ class PipelineTestCase(unittest.TestCase):
crawler = self._create_crawler(AsyncDefPipeline)
yield crawler.crawl(mockserver=self.mockserver)
self.assertEqual(len(self.items), 1)
@defer.inlineCallbacks
def test_asyncdef_asyncio_pipeline(self):
crawler = self._create_crawler(AsyncDefAsyncioPipeline)
yield crawler.crawl(mockserver=self.mockserver)
self.assertEqual(len(self.items), 1)