Add simple tests for process_iterable_helper.

tests/test_utils_middlewares.py

@@ -0,0 +1,90 @@
+import collections
+import sys
+
+from pytest import mark
+from twisted.trial import unittest
+
+from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
+from scrapy.utils.defer import deferred_f_from_coro_f
+from scrapy.utils.middlewares import process_iterable_helper
+
+
+def predicate1(o):
+    return bool(o % 2)
+
+
+def predicate2(o):
+    return o < 10
+
+
+def processor(o):
+    return o * 2
+
+
+class ProcessIterableHelperNormalTest(unittest.TestCase):
+
+    def test_normal_in_predicate(self):
+        iterable1 = iter([1, 2, 3])
+        iterable2 = process_iterable_helper(iterable1, in_predicate=predicate1)
+        self.assertIsInstance(iterable2, collections.abc.Iterable)
+        list2 = list(iterable2)
+        self.assertEqual(list2, [1, 3])
+
+    def test_normal_out_predicate(self):
+        iterable1 = iter([1, 2, 10, 3, 15])
+        iterable2 = process_iterable_helper(iterable1, out_predicate=predicate2)
+        self.assertIsInstance(iterable2, collections.abc.Iterable)
+        list2 = list(iterable2)
+        self.assertEqual(list2, [1, 2, 3])
+
+    def test_normal_processor(self):
+        iterable1 = iter([1, 2, 3])
+        iterable2 = process_iterable_helper(iterable1, processor=processor)
+        self.assertIsInstance(iterable2, collections.abc.Iterable)
+        list2 = list(iterable2)
+        self.assertEqual(list2, [2, 4, 6])
+
+    def test_normal_combined(self):
+        iterable1 = iter([1, 2, 10, 3, 6, 18, 5, 15])
+        iterable2 = process_iterable_helper(iterable1, in_predicate=predicate1,
+                                            out_predicate=predicate2, processor=processor)
+        self.assertIsInstance(iterable2, collections.abc.Iterable)
+        list2 = list(iterable2)
+        self.assertEqual(list2, [2, 6])
+
+
+@mark.skipif(sys.version_info < (3, 6), reason="Async generators require Python 3.6 or higher")
+class ProcessIterableHelperAsyncTest(unittest.TestCase):
+
+    @deferred_f_from_coro_f
+    async def test_async_in_predicate(self):
+        iterable1 = as_async_generator([1, 2, 3])
+        iterable2 = process_iterable_helper(iterable1, in_predicate=predicate1)
+        self.assertIsInstance(iterable2, collections.abc.AsyncIterable)
+        list2 = await collect_asyncgen(iterable2)
+        self.assertEqual(list2, [1, 3])
+
+    @deferred_f_from_coro_f
+    async def test_async_out_predicate(self):
+        iterable1 = as_async_generator([1, 2, 10, 3, 15])
+        iterable2 = process_iterable_helper(iterable1, out_predicate=predicate2)
+        self.assertIsInstance(iterable2, collections.abc.AsyncIterable)
+        list2 = await collect_asyncgen(iterable2)
+        self.assertEqual(list2, [1, 2, 3])
+
+    @deferred_f_from_coro_f
+    async def test_async_processor(self):
+        iterable1 = as_async_generator([1, 2, 3])
+        iterable2 = process_iterable_helper(iterable1, processor=processor)
+        self.assertIsInstance(iterable2, collections.abc.AsyncIterable)
+        list2 = await collect_asyncgen(iterable2)
+        self.assertEqual(list2, [2, 4, 6])
+
+    @deferred_f_from_coro_f
+    async def test_async_combined(self):
+        iterable1 = as_async_generator([1, 2, 10, 3, 6, 18, 5, 15])
+        iterable2 = process_iterable_helper(iterable1, in_predicate=predicate1,
+                                            out_predicate=predicate2, processor=processor)
+        self.assertIsInstance(iterable2, collections.abc.AsyncIterable)
+        list2 = await collect_asyncgen(iterable2)
+        self.assertEqual(list2, [2, 6])