KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _setup_stream_server method in Airtest

Best Python code snippet using Airtest

javacap.py

Source:javacap.py

...17    def __init__(self, adb):18        super(Javacap, self).__init__(adb)19        self.frame_gen = None20    @on_method_ready('install_or_upgrade')21    def _setup_stream_server(self):22        """23        Setup stream server24        Returns:25            adb shell process, non-blocking stream reader and local port26        """27        localport, deviceport = self.adb.setup_forward("localabstract:javacap_{}".format)28        deviceport = deviceport[len("localabstract:"):]29        # setup agent proc30        apkpath = self.adb.path_app(self.APP_PKG)31        cmds = ["CLASSPATH=" + apkpath, 'exec', 'app_process', '/system/bin', self.SCREENCAP_SERVICE,32                "--scale", "100", "--socket", "%s" % deviceport, "-lazy", "2>&1"]33        proc = self.adb.start_shell(cmds)34        # check proc output35        nbsp = NonBlockingStreamReader(proc.stdout, print_output=True, name="javacap_sever")36        while True:37            line = nbsp.readline(timeout=5.0)38            if line is None:39                raise RuntimeError("javacap server setup timeout")40            if b"Capture server listening on" in line:41                break42            if b"Address already in use" in line:43                raise RuntimeError("javacap server setup error: %s" % line)44        # reg_cleanup(proc.kill)45        return proc, nbsp, localport46    def get_frames(self):47        """48        Get the screen frames49        Returns:50            None51        """52        proc, nbsp, localport = self._setup_stream_server()53        s = SafeSocket()54        s.connect((self.adb.host, localport))55        t = s.recv(24)56        # javacap header57        LOGGING.debug(struct.unpack("<2B5I2B", t))58        LOGGING.info(struct.unpack("<2B5I2B", t))59        stopping = False60        while not stopping:61            s.send(b"1")62            # recv frame header, count frame_size63            if self.RECVTIMEOUT is not None:64                header = s.recv_with_timeout(4, self.RECVTIMEOUT)65            else:66                header = s.recv(4)...

Javecap.py

Source:Javecap.py

...20        self.frame_gen = None21        self.proc = None22        self.nbsp = None23        # start server24        self._setup_stream_server()25        self.adb.set_forward('localabstract:%s' % self.JAC_LOCAL_NAME)26        self.JAC_PORT = self.adb.get_forward_port(self.JAC_LOCAL_NAME)27        logger.info('javacap init, port:{} name:{}', self.JAC_PORT, self.JAC_LOCAL_NAME)28    def _setup_stream_server(self):29        apkpath = self.adb.path_app(self.APP_PKG)30        cmds = ["CLASSPATH=" + apkpath, 'exec', 'app_process', '/system/bin', self.SCREENCAP_SERVICE,31                "--scale", "100", "--socket", "%s" % self.JAC_LOCAL_NAME, "-lazy", "2>&1"]32        proc = self.adb.start_shell(cmds)33        nbsp = NonBlockingStreamReader(proc.stdout, print_output=True, name="javacap_sever")34        while True:35            line = nbsp.readline(timeout=5.0)36            if line is None:37                raise RuntimeError("javacap server setup timeout")38            if b"Capture server listening on" in line:39                break40            if b"Address already in use" in line:41                raise RuntimeError("javacap server setup error: %s" % line)42        reg_cleanup(proc.kill)43        return proc, nbsp44    def get_frames(self):45        proc, nbsp = self._setup_stream_server()46        s = SafeSocket()47        s.connect((self.adb.host, self.JAC_PORT))48        t = s.recv(24)49        # javacap header50        logger.debug(struct.unpack("<2B5I2B", t))51        stopping = False52        while not stopping:53            s.send(b"1")54            # recv frame header, count frame_size55            if self.RECVTIMEOUT is not None:56                header = s.recv_with_timeout(4, self.RECVTIMEOUT)57            else:58                header = s.recv(4)59            if header is None:...

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.