How to use test_pull method in Airtest

Best Python code snippet using Airtest

test_pipeline.py

Source:test_pipeline.py

...64        self.assertEqual(self.l, [0, 2, 4, 6, 8])65    def test_run_parallel(self):66        self.pl.run_parallel()67        self.assertEqual(self.l, [0, 2, 4, 6, 8])68    def test_pull(self):69        pl = pipeline.Pipeline((_produce(), _work()))70        self.assertEqual(list(pl.pull()), [0, 2, 4, 6, 8])71    def test_pull_chain(self):72        pl = pipeline.Pipeline((_produce(), _work()))73        pl2 = pipeline.Pipeline((pl.pull(), _work()))74        self.assertEqual(list(pl2.pull()), [0, 4, 8, 12, 16])75class ParallelStageTest(unittest.TestCase):76    def setUp(self):77        self.l = []78        self.pl = pipeline.Pipeline((79            _produce(), (_work(), _work()), _consume(self.l)80        ))81    def test_run_sequential(self):82        self.pl.run_sequential()83        self.assertEqual(self.l, [0, 2, 4, 6, 8])84    def test_run_parallel(self):85        self.pl.run_parallel()86        # Order possibly not preserved; use set equality.87        self.assertEqual(set(self.l), set([0, 2, 4, 6, 8]))88    def test_pull(self):89        pl = pipeline.Pipeline((_produce(), (_work(), _work())))90        self.assertEqual(list(pl.pull()), [0, 2, 4, 6, 8])91class ExceptionTest(unittest.TestCase):92    def setUp(self):93        self.l = []94        self.pl = pipeline.Pipeline((_produce(), _exc_work(),95                                     _consume(self.l)))96    def test_run_sequential(self):97        self.assertRaises(TestException, self.pl.run_sequential)98    def test_run_parallel(self):99        self.assertRaises(TestException, self.pl.run_parallel)100    def test_pull(self):101        pl = pipeline.Pipeline((_produce(), _exc_work()))102        pull = pl.pull()103        for i in range(3):104            pull.next()105        self.assertRaises(TestException, pull.next)106class ParallelExceptionTest(unittest.TestCase):107    def setUp(self):108        self.l = []109        self.pl = pipeline.Pipeline((110            _produce(), (_exc_work(), _exc_work()), _consume(self.l)111        ))112    def test_run_parallel(self):113        self.assertRaises(TestException, self.pl.run_parallel)114class ConstrainedThreadedPipelineTest(unittest.TestCase):115    def test_constrained(self):116        l = []117        # Do a "significant" amount of work...118        pl = pipeline.Pipeline((_produce(1000), _work(), _consume(l)))119        # ... with only a single queue slot.120        pl.run_parallel(1)121        self.assertEqual(l, [i * 2 for i in range(1000)])122    def test_constrained_exception(self):123        # Raise an exception in a constrained pipeline.124        l = []125        pl = pipeline.Pipeline((_produce(1000), _exc_work(), _consume(l)))126        self.assertRaises(TestException, pl.run_parallel, 1)127    def test_constrained_parallel(self):128        l = []129        pl = pipeline.Pipeline((130            _produce(1000), (_work(), _work()), _consume(l)131        ))132        pl.run_parallel(1)133        self.assertEqual(set(l), set(i * 2 for i in range(1000)))134class BubbleTest(unittest.TestCase):135    def setUp(self):136        self.l = []137        self.pl = pipeline.Pipeline((_produce(), _bub_work(),138                                     _consume(self.l)))139    def test_run_sequential(self):140        self.pl.run_sequential()141        self.assertEqual(self.l, [0, 2, 4, 8])142    def test_run_parallel(self):143        self.pl.run_parallel()144        self.assertEqual(self.l, [0, 2, 4, 8])145    def test_pull(self):146        pl = pipeline.Pipeline((_produce(), _bub_work()))147        self.assertEqual(list(pl.pull()), [0, 2, 4, 8])148class MultiMessageTest(unittest.TestCase):149    def setUp(self):150        self.l = []151        self.pl = pipeline.Pipeline((152            _produce(), _multi_work(), _consume(self.l)153        ))154    def test_run_sequential(self):155        self.pl.run_sequential()156        self.assertEqual(self.l, [0, 0, 1, -1, 2, -2, 3, -3, 4, -4])157    def test_run_parallel(self):158        self.pl.run_parallel()159        self.assertEqual(self.l, [0, 0, 1, -1, 2, -2, 3, -3, 4, -4])160    def test_pull(self):161        pl = pipeline.Pipeline((_produce(), _multi_work()))162        self.assertEqual(list(pl.pull()), [0, 0, 1, -1, 2, -2, 3, -3, 4, -4])163class StageDecoratorTest(unittest.TestCase):164    def test_stage_decorator(self):165        @pipeline.stage166        def add(n, i):167            return i + n168        pl = pipeline.Pipeline([169            iter([1, 2, 3]),170            add(2)171        ])172        self.assertEqual(list(pl.pull()), [3, 4, 5])173    def test_mutator_stage_decorator(self):174        @pipeline.mutator_stage...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.