How to use _emulatorCommand method in fMBT

Best Python code snippet using fMBT_python

fmbtandroid.py

Source:fmbtandroid.py Github

copy

Full Screen

...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",...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run fMBT automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful