How to use close_pipe method in Airtest

Best Python code snippet using Airtest

test_DALSA.py

Source:test_DALSA.py

...15        self.assertTrue(self.dev.setup_pipe())16        result, data = self.dev.grab_pipe()17        self.assertTrue(result)18        self.assertEqual(len(data), 2048)19        self.assertTrue(self.dev.close_pipe())20    def test_speed(self):21        self.assertTrue(self.dev.setup_pipe())22        result, data = self.dev.grab_pipe()23        forc_res = self.dev.open_port()24        forc_res = self.dev.start_scan()25        avg_default = numpy.average(data)26        time_start = time.time()27        for i in range(255):28            result = self.dev.set_gain(100)29            result, data = self.dev.grab_pipe()30        time_stop = time.time()31        time_diff = time_stop - time_start32        log.info("time diff: %s" % time_diff)33        self.assertTrue(self.dev.close_pipe())34    def test_gain(self):35        # Make sure no light is reaching the detector when running this test36        # Yes, this is the correct order - see the barbecue application for37        # details. An alternative if speed is not required is the open write38        # close pattern in the test below39        self.assertTrue(self.dev.setup_pipe())40        result, data = self.dev.grab_pipe()41        forc_res = self.dev.open_port()42        forc_res = self.dev.start_scan()43        # Set to default gain44        result = self.dev.set_gain(187)45        result, data = self.dev.grab_pipe()46        47        avg_default = numpy.average(data)48        time.sleep(0.5)49        result = self.dev.set_gain(100)50        result, data = self.dev.grab_pipe()51        avg_low_gain = numpy.average(data)52        gain_diff_threshold = 100053        gain_diff = abs(avg_low_gain - avg_default)54        self.assertGreater(gain_diff, gain_diff_threshold)55        # Now set it back to default 187, and make sure the difference56        # is minimal57        result = self.dev.set_gain(187)58        result, data = self.dev.grab_pipe()59        new_default = numpy.average(data)60        new_diff = abs(new_default - avg_default)61        self.assertLess(new_diff, 10)62        forc_res = self.dev.close_port()63        self.assertTrue(self.dev.close_pipe())64    def test_openwriteclose_gain(self):65        self.assertTrue(self.dev.setup_pipe())66        result, data = self.dev.grab_pipe()67        self.assertTrue(result)68        self.assertEqual(len(data), 2048)69        self.assertTrue(self.dev.open_write_close("gain 187"))70        result, data = self.dev.grab_pipe()71        avg_default = numpy.average(data)72        self.assertTrue(self.dev.open_write_close("gain 100"))73        result, data = self.dev.grab_pipe()74        avg_low_gain = numpy.average(data)75        gain_diff_threshold = 100076        gain_diff = abs(avg_low_gain - avg_default)77        self.assertGreater(gain_diff, gain_diff_threshold)78        self.assertTrue(self.dev.close_pipe())79    def test_offset(self):80        self.assertTrue(self.dev.setup_pipe())81        result, data = self.dev.grab_pipe()82        self.assertTrue(result)83        self.assertEqual(len(data), 2048)84        self.assertTrue(self.dev.open_write_close("offset 0"))85        result, data = self.dev.grab_pipe()86        avg_default = numpy.average(data)87        self.assertTrue(self.dev.open_write_close("offset 255"))88        result, data = self.dev.grab_pipe()89        avg_high_offset = numpy.average(data)90        self.assertGreater(avg_default, avg_high_offset)91        self.assertTrue(self.dev.close_pipe())92        93class TestBasler4K(unittest.TestCase):94    def setUp(self):95        self.dev = BaslerSprint4K()96    def test_pipe_cycle(self):97        self.assertTrue(self.dev.setup_pipe())98        result, data = self.dev.grab_pipe()99        self.assertTrue(result)100        self.assertEqual(len(data), 4096)101        self.assertTrue(self.dev.close_pipe())102if __name__ == "__main__":...

snippet.py

Source:snippet.py

...14        15        # optional: close pipe after 20 seconds16        self.ioloop.add_timeout(time.time()+(20), self.close_pipe)17        18    def close_pipe(self):19        self.ioloop.remove_handler(self.pipe)20        # add link to end user to load more output21        self.write('''22            <script>23                (function(){24                    document.write("</pre>");25                    document.write("<a href='javascript:location.href=location.href'>load more...</a>");26                    document.body.scrollTop = document.body.scrollHeight;27                })()28            </script>        29        ''')30        self.finish()31        self.pipe.close()32        33        try:34            self.process.kill()35        except:36            pass37        38    def on_read(self, fd, events): 39        buffer = self.pipe.read(256) 40        try: 41            assert buffer 42            self.write(buffer) 43            self.flush() 44        except: 45            self.close_pipe()46            47    def _spawn_process(self, commandline):48        self.process = subprocess.Popen(49            commandline, 50            shell=True, 51            stdout=subprocess.PIPE, 52            stdin=subprocess.PIPE, 53            stderr=subprocess.PIPE)54            55        return self.process56        57    def _get_pipe(self, commandline):58        return self._spawn_process(commandline).stdout59            ...

1064355.py

Source:1064355.py

...14        15        # optional: close pipe after 20 seconds16        self.ioloop.add_timeout(time.time()+(20), self.close_pipe)17        18    def close_pipe(self):19        self.ioloop.remove_handler(self.pipe)20        # add link to end user to load more output21        self.write('''22            <script>23                (function(){24                    document.write("</pre>");25                    document.write("<a href='javascript:location.href=location.href'>load more...</a>");26                    document.body.scrollTop = document.body.scrollHeight;27                })()28            </script>        29        ''')30        self.finish()31        self.pipe.close()32        33        try:34            self.process.kill()35        except:36            pass37        38    def on_read(self, fd, events): 39        buffer = self.pipe.read(256) 40        try: 41            assert buffer 42            self.write(buffer) 43            self.flush() 44        except: 45            self.close_pipe()46            47    def _spawn_process(self, commandline):48        self.process = subprocess.Popen(49            commandline, 50            shell=True, 51            stdout=subprocess.PIPE, 52            stdin=subprocess.PIPE, 53            stderr=subprocess.PIPE)54            55        return self.process56        57    def _get_pipe(self, commandline):58        return self._spawn_process(commandline).stdout59            ...

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.