Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use _emulatorCommand method in fMBT

Best Python code snippet using fMBT_python
fmbtandroid.py
Source:fmbtandroid.py
...1969            command.extend(adbCommand)1970        else:1971            command.append(adbCommand)1972        return _run(command, expectedExitStatus=expect, timeout=timeout)1973    def _emulatorCommand(self, command):1974        if not self._emulatorSocket:1975            try:1976                emulatorPort = int(re.findall("emulator-([0-9]*)", self._serialNumber)[0])1977            except (IndexError, ValueError):1978                raise FMBTAndroidError("emulator port detection failed")1979            try:1980                self._emulatorSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)1981                self._emulatorSocket.connect(("localhost", emulatorPort))1982            except socket.error, e:1983                raise FMBTAndroidError("connecting to the emulator failed: %s" % (e,))1984        self._emulatorSocket.sendall(command + "\n")1985        data = self._emulatorSocket.recv(4096)1986        try:1987            data = data.splitlines()[-1].strip()1988        except IndexError:1989            raise FMBTAndroidError("no response from the emulator")1990        if data.startswith("OK"):1991            return True, data1992        else:1993            return False, data1994    def _runSetupCmd(self, cmd, expectedExitStatus=0):1995        _adapterLog('setting up connections: "%s"' % (cmd,))1996        try:1997            if expectedExitStatus == None: # execute asynchronously1998                self._runAdb(cmd, expectedExitStatus)1999            else: # command is expected to exit2000                self._runAdb(cmd, expectedExitStatus, timeout=_SHORT_TIMEOUT)2001        except (FMBTAndroidRunError, AndroidDeviceNotFound), e:2002            _adapterLog("connection setup problem: %s" % (e,))2003            return False2004        return True2005    def _detectFeatures(self):2006        # check supported features2007        outputLines = self._runAdb(["shell", "getprop", "ro.build.version.release"],2008                                   timeout=_SHORT_TIMEOUT)[1].splitlines()2009        if len(outputLines) >= 1:2010            self._platformVersion = outputLines[0].strip().split("=")[-1]2011        else:2012            self._platformVersion = "N/A"2013        outputLines = self._runAdb(["shell", "id"],2014                                   timeout=_SHORT_TIMEOUT)[1].splitlines()2015        if len(outputLines) == 1 and "uid=0" in outputLines[0]:2016            self._shellUid0 = True2017        else:2018            self._shellUid0 = False2019        outputLines = self._runAdb(["shell", "su", "root", "id"],2020                                   timeout=_SHORT_TIMEOUT)[1].splitlines()2021        if len(outputLines) == 1 and "uid=0" in outputLines[0]:2022            self._shellSupportsSu = True2023        else:2024            self._shellSupportsSu = False2025        outputLines = self._runAdb(["shell", "tar"],2026                                   expectedExitStatus=EXITSTATUS_ANY,2027                                   timeout=_SHORT_TIMEOUT)[1].splitlines()2028        if len(outputLines) == 1 and "bin" in outputLines[0]:2029            self._shellSupportsTar = False2030        else:2031            self._shellSupportsTar = True2032        outputLines = self._runAdb(["shell", "echo -n foo | toybox base64"],2033                                   expectedExitStatus=EXITSTATUS_ANY,2034                                   timeout=_SHORT_TIMEOUT)[1].splitlines()2035        if len(outputLines) == 1 and "Zm9v" in outputLines[0]:2036            self._shellSupportsToyboxBase64 = True2037        else:2038            self._shellSupportsToyboxBase64 = False2039        outputLines = self._runAdb(["shell", "echo -n foo | uuencode -"],2040                                   expectedExitStatus=EXITSTATUS_ANY,2041                                   timeout=_SHORT_TIMEOUT)[1].splitlines()2042        if len(outputLines) > 0 and "begin" in outputLines[0]:2043            self._shellSupportsUuencode = True2044        else:2045            self._shellSupportsUuencode = False2046    def _resetWindow(self):2047        setupCommands = [["shell", "service" , "call", "window", "1", "i32", "4939"],2048                         ["forward", "tcp:"+str(self._windowPortForward), "tcp:4939"]]2049        for c in setupCommands:2050            self._runSetupCmd(c)2051    def _resetMonkey(self, timeout=12, pollDelay=.25):2052        tryKillingMonkeyOnFailure = 12053        failureCountSinceKill = 02054        endTime = time.time() + timeout2055        if self._shellUid0:2056            monkeyLaunch = ["monkey"]2057        elif self._shellSupportsSu:2058            monkeyLaunch = ["su", "root", "monkey"]2059        else:2060            monkeyLaunch = ["monkey"]2061        if self._monkeyOptions:2062            monkeyLaunch += self._monkeyOptions2063        while time.time() < endTime:2064            monkeyShellCmd = (" ".join(monkeyLaunch + ["--port", "1080"]) +2065                              " >/sdcard/fmbtandroid.monkey.outerr 2>&1")2066            _adapterLog('launching monkey: adb shell "%s"' % (monkeyShellCmd,))2067            self._runAdb(["shell", monkeyShellCmd], expectedExitStatus=None)2068            time.sleep(pollDelay)2069            if not self._runSetupCmd(["forward", "tcp:"+str(self._monkeyPortForward), "tcp:1080"]):2070                time.sleep(pollDelay)2071                failureCountSinceKill += 12072                continue2073            try:2074                self._monkeySocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)2075                self._monkeySocket.connect((self._m_host, self._monkeyPortForward))2076                self._monkeySocket.setblocking(0)2077                self._monkeySocket.settimeout(5.0)2078                _ping = self._monkeyCommand("getvar build.version.release", retry=0)[1]2079                if len(_ping) > 0:2080                    return True2081            except Exception, e:2082                _, monkeyOutput, _ = self._runAdb(["shell", "cat /sdcard/fmbtandroid.monkey.outerr"],2083                                                  expectedExitStatus=EXITSTATUS_ANY,2084                                                  timeout=_SHORT_TIMEOUT)2085                if "/sdcard/fmbtandroid.monkey.outerr: No such file or directory" in monkeyOutput:2086                    msg = 'cannot read/write /sdcard on device %s' % (self._serialNumber,)2087                    _adapterLog(msg)2088                    raise AndroidConnectionError(msg)2089                elif "Error: Unknown option:" in monkeyOutput:2090                    uo = [l for l in monkeyOutput.splitlines() if "Error: Unknown option:" in l][0].split(":")[-1].strip()2091                    _adapterLog('detected an unknown option for monkey: "%s". Disabling it.' % (uo,))2092                    try:2093                        monkeyLaunch.remove(uo)2094                    except ValueError:2095                        pass2096                    continue2097                elif "Error binding to network socket" in monkeyOutput:2098                    _adapterLog('monkey network socket binding failed, killing old monkey')2099                    self.pkill("monkey")2100                    time.sleep(pollDelay)2101                    continue2102                _adapterLog("monkey connection failed, output: %s" % (monkeyOutput,))2103                failureCountSinceKill += 12104            time.sleep(pollDelay)2105            if failureCountSinceKill > 2 and tryKillingMonkeyOnFailure > 0:2106                self.pkill("monkey")2107                tryKillingMonkeyOnFailure -= 12108                failureCountSinceKill = 02109                time.sleep(pollDelay)2110        if self._stopOnError:2111            msg = 'Android monkey error: cannot connect to "adb shell monkey --port 1080" to device %s' % (self._serialNumber)2112            _adapterLog(msg)2113            raise AndroidConnectionError(msg)2114        else:2115            return False2116    def _monkeyCommand(self, command, retry=3):2117        try:2118            self._monkeySocket.sendall(command + "\n")2119            data = self._monkeySocket.recv(4096).strip()2120            if len(data) == 0 and retry > 0:2121                return self._monkeyCommand(command, retry-1)2122            if data == "OK":2123                return True, None2124            elif data.startswith("OK:"):2125                return True, data.split("OK:")[1]2126            _adapterLog("monkeyCommand failing... command: '%s' response: '%s'" % (command, data))2127            return False, None2128        except socket.error:2129            try: self._monkeySocket.close()2130            except: pass2131            if retry > 0:2132                self._resetMonkey()2133                return self._monkeyCommand(command, retry=retry-1)2134            else:2135                raise AndroidConnectionError('Android monkey socket connection lost while sending command "%s"' % (command,))2136    def install(self, filename, lock, reinstall, downgrade,2137                sdcard, algo, key, iv):2138        cmd = ["install"]2139        if lock:2140            cmd.append("-l")2141        if reinstall:2142            cmd.append("-r")2143        if downgrade:2144            cmd.append("-d")2145        if sdcard:2146            cmd.append("-s")2147        if algo != None:2148            cmd.extend(["--algo", algo])2149        if key != None:2150            cmd.extend(["--key", key])2151        if iv != None:2152            cmd.extend(["--iv", iv])2153        cmd.append(filename)2154        status, output, error = self._runAdb(cmd, [0, 1], timeout=_LONG_TIMEOUT)2155        if "Success" in output:2156            return True2157        else:2158            return output + "\n" + error2159    def uninstall(self, apkname, keepData):2160        cmd = ["uninstall"]2161        if keepData:2162            cmd.append("-k")2163        cmd.append(apkname)2164        status, output, error = self._runAdb(2165            cmd, expectedExitStatus=EXITSTATUS_ANY, timeout=_LONG_TIMEOUT)2166        if "Success" in output:2167            return True2168        else:2169            return False2170    def pkill(self, pattern, signal=15, exact=False):2171        """send signal to all processes where process name contains pattern"""2172        _, ps, _ = self._runAdb(["shell", "ps"], timeout=_SHORT_TIMEOUT)2173        if self._shellSupportsSu:2174            shell_kill = ["shell", "su", "root", "kill"]2175        else:2176            shell_kill = ["shell", "kill"]2177        pids = []2178        for line in [l.strip() for l in ps.splitlines()]:2179            fields = line.split()2180            if len(fields) > 7:2181                if exact:2182                    if pattern == fields[7]:2183                        pids.append(fields[1])2184                else:2185                    if pattern in " ".join(fields[7:]):2186                        pids.append(fields[1])2187        if pids:2188            _adapterLog(str(shell_kill + ["-" + str(signal)] + pids))2189            self._runAdb(shell_kill + ["-" + str(signal)] + pids,2190                         expectedExitStatus=EXITSTATUS_ANY,2191                         timeout=_SHORT_TIMEOUT)2192            return True2193        else:2194            return False2195    def recvPlatformVersion(self):2196        return self._platformVersion2197    def reboot(self, reconnect, firstBootAfterFlashing, timeout):2198        if firstBootAfterFlashing:2199            self._runAdb("root", expectedExitStatus=EXITSTATUS_ANY,2200                         timeout=_SHORT_TIMEOUT)2201            time.sleep(2)2202            self._runAdb(["shell", "rm",2203                          "/data/data/com.android.launcher/shared_prefs/com.android.launcher2.prefs.xml"],2204                         expectedExitStatus=EXITSTATUS_ANY,2205                         timeout=_SHORT_TIMEOUT)2206        self._runAdb("reboot", expectedExitStatus=EXITSTATUS_ANY)2207        _adapterLog("rebooting " + self._serialNumber)2208        if reconnect:2209            time.sleep(2)2210            endTime = time.time() + timeout2211            status, _, _ = self._runAdb("wait-for-device", expectedExitStatus=None, timeout=timeout)2212            if status != 0:2213                raise AndroidDeviceNotFound('"timeout -k 1 %s adb wait-for-device" status %s' % (timeout, status))2214            self._detectFeatures()2215            while time.time() < endTime:2216                try:2217                    if self._resetMonkey(timeout=1, pollDelay=1):2218                        break2219                except AndroidConnectionError:2220                    pass2221                time.sleep(1)2222            else:2223                msg = "reboot: reconnecting to " + self._serialNumber + " failed"2224                _adapterLog(msg)2225                raise AndroidConnectionError(msg)2226            self._resetWindow()2227        return True2228    def recvVariable(self, variableName):2229        ok, value = self._monkeyCommand("getvar " + variableName)2230        if ok: return value2231        else:2232            # LOG: getvar variableName failed2233            return None2234    def recvScreenSize(self):2235        _, output, _ = self._runAdb(["shell", "dumpsys", "display"], 0,2236                                    timeout=_SHORT_TIMEOUT)2237        try:2238            # parse default display properties2239            ddName, ddWidth, ddHeight, ddWdpi, ddHdpi = re.findall(2240                r'DisplayDeviceInfo\{[^,]*"([^"]*)"[:,] ([0-9]*) x ([0-9]*),.*, ([0-9.]*) x ([0-9.]*) dpi,.*FLAG_DEFAULT_DISPLAY.*\}',2241                output)[0]2242            ddWidth, ddHeight = int(ddWidth), int(ddHeight)2243        except (IndexError, ValueError), e:2244            _adapterLog('recvScreenSize: cannot read size from "%s"' %2245                        (output,))2246            raise FMBTAndroidError('cannot read screen size from dumpsys')2247        vpWidth, vpHeight = self.recvDefaultViewportSize()2248        if ((vpWidth > vpHeight) and (ddWidth < ddHeight) or2249            (vpWidth < vpHeight) and (ddWidth > ddHeight)):2250            ddWidth, ddHeight = ddHeight, ddWidth2251        return int(ddWidth), int(ddHeight)2252    def recvDefaultViewportSize(self):2253        _, output, _ = self._runAdb(["shell", "dumpsys", "display"], 0,2254                                    timeout=_SHORT_TIMEOUT)2255        try:2256            _, w, h = re.findall("mDefaultViewport(\[0\])?=DisplayViewport\{.*deviceWidth=([0-9]*), deviceHeight=([0-9]*)\}", output)[0]2257            width = int(w)2258            height = int(h)2259        except (IndexError, ValueError), e:2260            _adapterLog('recvScreenSize: cannot read size from "%s"' %2261                        (output,))2262            raise FMBTAndroidError('cannot read screen size from dumpsys')2263        return width, height2264    def recvCurrentDisplayOrientation(self):2265        _, output, _ = self._runAdb(["shell", "dumpsys", "display"], 0,2266                                    timeout=_SHORT_TIMEOUT)2267        s = re.findall("mCurrentOrientation=([0-9])", output)2268        if s:2269            return int(s[0])2270        else:2271            return None2272    def recvDisplayPowered(self):2273        _, output, _ = self._runAdb(["shell", "dumpsys", "power"], 0,2274                                    timeout=_SHORT_TIMEOUT)2275        s = re.findall("Display Power: state=(OFF|ON)", output)2276        if s:2277            return s[0] == "ON"2278        else:2279            return None2280    def recvShowingLockscreen(self):2281        _, output, _ = self._runAdb(["shell", "dumpsys", "window"], 0,2282                                    timeout=_SHORT_TIMEOUT)2283        s = re.findall("mShowingLockscreen=(true|false)", output)2284        if s:2285            if s[0] == "true":2286                return True2287            else:2288                return False2289        else:2290            return None2291    def recvLastAccelerometer(self):2292        _, output, _ = self._runAdb(["shell", "dumpsys", "sensorservice"], 0,2293                                    timeout=_SHORT_TIMEOUT)2294        s = re.findall("3-axis Accelerometer.*last=<([- .0-9]*),([- .0-9]*),([- .0-9]*)>", output)2295        try:2296            rv = tuple([float(d) for d in s[0]])2297        except (IndexError, ValueError):2298            rv = (None, None, None)2299        return rv2300    def sendAcceleration(self, abc):2301        """abc is a tuple of 1, 2 or 3 floats, new accelerometer readings"""2302        try:2303            self._emulatorCommand("sensor set acceleration %s" %2304                                  (":".join([str(value) for value in abc]),))2305        except FMBTAndroidError, e:2306            raise FMBTAndroidError(2307                "accelerometer can be set only on emulator (%s)" % (e,))2308        return True2309    def sendAccelerometerRotation(self, value):2310        if value:2311            sendValue = "i:1"2312        else:2313            sendValue = "i:0"2314        try:2315            self._runAdb(["shell", "content", "insert",2316                          "--uri", "content://settings/system",2317                          "--bind", "name:s:accelerometer_rotation",...
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.