How to use setup_client_backend method in Airtest

Best Python code snippet using Airtest

MiniTouch.py

Source:MiniTouch.py

...137        self.serial = serial138        self.WIDTH = width139        self.HEIGHT = height140        #self.setup_server()141        self.setup_client_backend()142    def raw_cmd(self, *args):143        try:144            cmds = ['adb'] + ['-s'] + [self.serial] + list(args)145            p = subprocess.Popen(cmds, stdout=subprocess.PIPE, stderr=subprocess.PIPE)146            c = p.communicate()[0]147            return c148        except Exception, e:149            pass150    def __transform_xy(self, x, y):151        """152        Transform coordinates (x, y) according to the device display153        Args:154            x: coordinate x155            y: coordinate y156        Returns:157            transformed coordinates (x, y)158        """159        if not (self.display_info and self.display_info['max_x'] and self.display_info['max_y']):160            return x, y161        width, height = self.wx_x,self.wx_y162        if width > height and self.display_info['orientation'] in [1, 3]:163            width, height = height, width164        nx = x * self.max_x / width165        ny = y * self.max_y / height166        return nx, ny167    def get_std_encoding(self,stream):168        """169        Get encoding of the stream170        Args:171            stream: stream172        Returns:173            encoding or file system encoding174        """175        return getattr(stream, "encoding", None) or sys.getfilesystemencoding()176    def setup_server(self):177        if self.server_proc:178            self.server_proc.kill()179            self.server_proc = None180        cmd = ('adb -s '+ self.serial +' forward tcp:'+ self.port+' localabstract:minitouch')181        out = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)182        cmds = ('adb -s '+self.serial +" shell /data/local/tmp/minitouch -n " + self.port+" 2>&1")183        p = subprocess.Popen(cmds, stdin=subprocess.PIPE,stdout=subprocess.PIPE, stderr=subprocess.PIPE)184        nbsp = NonBlockingStreamReader(p.stdout, name="minitouch_server")185        while True:186            line = nbsp.readline(timeout=5.0)187            if line is None:188                raise RuntimeError("minitouch setup timeout")189            line = line.decode(self.get_std_encoding(sys.stdout))190            # è¯å«åºsetupæåçlogï¼å¹¶å¹éåºmax_x, max_y191            m = re.match("Type \w touch device .+ \((\d+)x(\d+) with \d+ contacts\) detected on .+ \(.+\)", line)192            if m:193                self.max_x, self.max_y = int(m.group(1)), int(m.group(2))194                break195            else:196                self.max_x = 32768197                self.max_y = 32768198        # nbsp.kill() # ä¿çï¼ä¸æäºï¼åé¢è¿ä¼ç»§ç»è¯»åå¹¶pirnt199        if p.poll() is not None:200            # server setup error, may be already setup by others201            # subprocess exit immediately202            raise RuntimeError("minitouch server quit immediately")203        self.server_proc = p204        # reg_cleanup(self.server_proc.kill)205        return p206    def touch(self, tuple_xy, duration=0.01):207        x, y = tuple_xy208        self.handle("d 0 %d %d 50\nc\n" % (x, y))209        time.sleep(duration)210        #self.handle("u 0 \nc\n")211    def touch_end(self):212        self.handle("u 0 \nc\n")213    def move_start(self,tuple_to_xy):214        tox, toy = tuple_to_xy215        self.handle(b"m 0 %d %d 50\nc\n" % (tox, toy))216    def swipe(self, tuple_from_xy, tuple_to_xy, duration=0.8, steps=5):217        """218        Perform swipe event219        minitouch protocol example::220            d 0 0 0 50221            c222            m 0 20 0 50223            c224            m 0 40 0 50225            c226            m 0 60 0 50227            c228            m 0 80 0 50229            c230            m 0 100 0 50231            c232            u 0233            c234        Args:235            tuple_from_xy: start point236            tuple_to_xy: end point237            duration: time interval for swipe duration, default is 0.8238            steps: size of swipe step, default is 5239        Returns:240            None241        """242        from_x, from_y = tuple_from_xy243        to_x, to_y = tuple_to_xy244        from_x, from_y = self.__transform_xy(from_x, from_y)245        to_x, to_y = self.__transform_xy(to_x, to_y)246        interval = float(duration) / (steps + 1)247        self.handle(b"d 0 %d %d 50\nc\n" % (from_x, from_y))248        time.sleep(interval)249        for i in range(1, steps):250            self.handle(b"m 0 %d %d 50\nc\n" % (251                from_x + (to_x - from_x) * i / steps,252                from_y + (to_y - from_y) * i / steps,253            ))254            time.sleep(interval)255        for i in range(10):256            self.handle(b"m 0 %d %d 50\nc\n" % (to_x, to_y))257        time.sleep(interval)258        self.handle(b"u 0\nc\n")259    def pinch(self, center=None, percent=0.5, duration=0.5, steps=5, in_or_out='in'):260        """261        Perform pinch action262        minitouch protocol example::263            d 0 0 100 50264            d 1 100 0 50265            c266            m 0 10 90 50267            m 1 90 10 50268            c269            m 0 20 80 50270            m 1 80 20 50271            c272            m 0 20 80 50273            m 1 80 20 50274            c275            m 0 30 70 50276            m 1 70 30 50277            c278            m 0 40 60 50279            m 1 60 40 50280            c281            m 0 50 50 50282            m 1 50 50 50283            c284            u 0285            u 1286            c287        """288        w, h = self.display_info['width'], self.display_info['height']289        if isinstance(center, (list, tuple)):290            x0, y0 = center291        elif center is None:292            x0, y0 = w / 2, h / 2293        else:294            raise RuntimeError("center should be None or list/tuple, not %s" % repr(center))295        x1, y1 = x0 - w * percent / 2, y0 - h * percent / 2296        x2, y2 = x0 + w * percent / 2, y0 + h * percent / 2297        cmds = []298        if in_or_out == 'in':299            cmds.append(b"d 0 %d %d 50\nd 1 %d %d 50\nc\n" % (x1, y1, x2, y2))300            for i in range(1, steps):301                cmds.append(b"m 0 %d %d 50\nm 1 %d %d 50\nc\n" % (302                    x1+(x0-x1)*i/steps, y1+(y0-y1)*i/steps,303                    x2+(x0-x2)*i/steps, y2+(y0-y2)*i/steps304                ))305            cmds.append(b"m 0 %d %d 50\nm 1 %d %d 50\nc\n" % (x0, y0, x0, y0))306            cmds.append(b"u 0\nu 1\nc\n")307        elif in_or_out == 'out':308            cmds.append(b"d 0 %d %d 50\nd 1 %d %d 50\nc\n" % (x0, y0, x0, y0))309            for i in range(1, steps):310                cmds.append(b"m 0 %d %d 50\nm 1 %d %d 50\nc\n" % (311                    x0+(x1-x0)*i/steps, y0+(y1-y0)*i/steps,312                    x0+(x2-x0)*i/steps, y0+(y2-y0)*i/steps313                ))314            cmds.append(b"m 0 %d %d 50\nm 1 %d %d 50\nc\n" % (x1, y1, x2, y2))315            cmds.append(b"u 0\nu 1\nc\n")316        else:317            raise RuntimeError("center should be 'in' or 'out', not %s" % repr(in_or_out))318        interval = float(duration)/(steps+1)319        for i, c in enumerate(cmds):320            self.handle(c)321            time.sleep(interval)322    def operate(self, args):323        if args["type"] == "down":324            x, y = args["x"], args["y"]325            # support py 3326            cmd = "d 0 %d %d 50 \nc\n" % (x, y)327        elif args["type"] == "move":328            x, y = args["x"], args["y"]329            # support py 3330            cmd = "m 0 %d %d 50 \nc\n" % (x, y)331        elif args["type"] == "up":332            # support py 3333            cmd = "u 0 \nc\n"334        elif args["type"] == "up1":335            # support py 3336            cmd = "u 1 \nc\n"337        else:338            raise RuntimeError("invalid operate args: %s" % args)339        self.handle(cmd)340    def safe_send(self, data):341        """342        Send data to client343        Args:344            data: data to send345        Raises:346            Exception: when data cannot be sent347        Returns:348            None349        """350        try:351            self.client.send(data)352        except Exception as err:353            # raise MinitouchError(err)354            # raise err355           print "send some commands error!"356    def _backend_worker(self):357        """358        Backend worker queue thread359        Returns:360            None361        """362        while not self.backend_stop_event.isSet():363            cmd = self.backend_queue.get()364            if cmd is None:365                break366            self.safe_send(cmd)367    def minitouch_thread(self,serial):368        try:369            cmd = ('adb -s '+serial + ' forward tcp:'+ self.port+' localabstract:minitouch')370            out = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)371            cmds = ("adb -s "+serial + " shell /data/local/tmp/minitouch")372            p = subprocess.Popen(cmds, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)373            time.sleep(2)374            self.minitch_flag = True375        except Exception,e:376            import traceback377            traceback.print_exc()378            self.minitch_flag = False379    def setup_client_backend(self):380        self.backend_queue = queue.Queue()381        self.backend_stop_event = threading.Event()382        self.setup_client()383        t = threading.Thread(target=self._backend_worker, name="minitouch")384        t.daemon = True385        t.start()386        self.backend_thread = t387        self.handle = self.backend_queue.put388    def setup_client1(self):389        s = socket.socket()  # åå»º socket å¯¹è±¡390        host = '127.0.0.1'  # è·åæ¬å°ä¸»æºå391        s.connect((host, self.port))392        s.settimeout(2)393        header = ""...

base_touch.py

Source:base_touch.py

...31        """32        self.install()33        self.setup_server()34        if self.backend:35            self.setup_client_backend()36        else:37            self.setup_client()38    def uninstall(self):39        """40        Uninstall airtest touch41        Returns:42            None43        """44        raise NotImplemented45    def install(self):46        """47        Install airtest touch48        Returns:49            None50        """51        raise NotImplemented52    def setup_server(self):53        """54        Setip touch server and adb forward55        Returns:56            server process57        """58        raise NotImplemented59    def safe_send(self, data):60        """61        Send data to client62        Args:63            data: data to send64        Raises:65            Exception: when data cannot be sent66        Returns:67            None68        """69        if isinstance(data, six.text_type):70            data = data.encode('utf-8')71        try:72            self.client.send(data)73        except Exception as err:74            # raise MinitouchError(err)75            raise err76    def _backend_worker(self):77        """78        Backend worker queue thread79        Returns:80            None81        """82        while not self.backend_stop_event.isSet():83            cmd = self.backend_queue.get()84            if cmd is None:85                break86            self.safe_send(cmd)87    def setup_client_backend(self):88        """89        Setup backend client thread as daemon90        Returns:91            None92        """93        self.backend_queue = queue.Queue()94        self.backend_stop_event = threading.Event()95        self.setup_client()96        t = threading.Thread(target=self._backend_worker, name="airtouch")97        # t.daemon = True98        t.start()99        self.backend_thread = t100        self.handle = self.backend_queue.put101    def setup_client(self):...

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.