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

How to use _resetWindow method in fMBT

Best Python code snippet using fMBT_python

fmbtandroid.py

Source:fmbtandroid.py

...1016        self.setDisplayToScreenCoords(lambda x, y: (x, y))1017        self._detectFeatures()1018        try:1019            self._resetMonkey()1020            self._resetWindow()1021        finally:1022            # Next _AndroidDeviceConnection instance will use different ports1023            self._w_port = _AndroidDeviceConnection._w_port1024            self._m_port = _AndroidDeviceConnection._m_port1025            _AndroidDeviceConnection._w_port += 1001026            _AndroidDeviceConnection._m_port += 1001027    def __del__(self):1028        try: self._monkeySocket.close()1029        except: pass1030    def target(self):1031        return self._serialNumber1032    def _cat(self, remoteFilename):1033        fd, filename = tempfile.mkstemp("fmbtandroid-cat-")1034        os.close(fd)1035        self._runAdb(["pull", remoteFilename, filename], 0)1036        contents = file(filename).read()1037        os.remove(filename)1038        return contents1039    def _runAdb(self, command, expectedExitStatus=0, timeout=None):1040        if not self._stopOnError:1041            expect = None1042        else:1043            expect = expectedExitStatus1044        if type(command) == list:1045            command = ["adb", "-s", self._serialNumber] + command1046        else:1047            command = ["adb", "-s", self._serialNumber, command]1048        return _run(command, expectedExitStatus=expect, timeout=timeout)1049    def _runSetupCmd(self, cmd, expectedExitStatus = 0):1050        _adapterLog('setting up connections: "%s"' % (cmd,))1051        try:1052            self._runAdb(cmd, expectedExitStatus)1053        except (FMBTAndroidRunError, AndroidDeviceNotFound), e:1054            _adapterLog("connection setup problem: %s" % (e,))1055            return False1056        return True1057    def _detectFeatures(self):1058        # check supported features1059        outputLines = self._runAdb(["shell", "id"])[1].splitlines()1060        if len(outputLines) == 1 and "uid=0" in outputLines[0]:1061            self._shellUid0 = True1062        else:1063            self._shellUid0 = False1064        outputLines = self._runAdb(["shell", "su", "root", "id"])[1].splitlines()1065        if len(outputLines) == 1 and "uid=0" in outputLines[0]:1066            self._shellSupportsSu = True1067        else:1068            self._shellSupportsSu = False1069        outputLines = self._runAdb(["shell", "tar"])[1].splitlines()1070        if len(outputLines) == 1 and "bin" in outputLines[0]:1071            self._shellSupportsTar = False1072        else:1073            self._shellSupportsTar = True1074    def _resetWindow(self):1075        setupCommands = [["shell", "service" , "call", "window", "1", "i32", "4939"],1076                         ["forward", "tcp:"+str(self._w_port), "tcp:4939"]]1077        for c in setupCommands:1078            self._runSetupCmd(c)1079    def _resetMonkey(self, timeout=3, pollDelay=.25):1080        tryKillingMonkeyOnFailure = 11081        failureCountSinceKill = 01082        endTime = time.time() + timeout1083        if self._shellUid0:1084            monkeyLaunch = ["monkey"]1085        elif self._shellSupportsSu:1086            monkeyLaunch = ["su", "root", "monkey"]1087        else:1088            monkeyLaunch = ["monkey"]1089        while time.time() < endTime:1090            if not self._runSetupCmd(["shell"] + monkeyLaunch + ["--port", "1080"], None):1091                time.sleep(pollDelay)1092                failureCountSinceKill += 11093                continue1094            time.sleep(pollDelay)1095            if not self._runSetupCmd(["forward", "tcp:"+str(self._m_port), "tcp:1080"]):1096                time.sleep(pollDelay)1097                failureCountSinceKill += 11098                continue1099            try:1100                self._monkeySocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)1101                self._monkeySocket.connect((self._m_host, self._m_port))1102                self._monkeySocket.setblocking(0)1103                self._monkeySocket.settimeout(1.0)1104                self._platformVersion = self._monkeyCommand("getvar build.version.release", retry=0)[1]1105                if len(self._platformVersion) > 0:1106                    self._monkeySocket.settimeout(5.0)1107                    return True1108            except Exception, e:1109                failureCountSinceKill += 11110            time.sleep(pollDelay)1111            if failureCountSinceKill > 2 and tryKillingMonkeyOnFailure > 0:1112                if self._shellSupportsSu:1113                    self._runSetupCmd(["shell", "su", "root", "pkill", "monkey"])1114                else:1115                    self._runSetupCmd(["shell", "pkill", "monkey"])1116                tryKillingMonkeyOnFailure -= 11117                failureCountSinceKill = 01118                time.sleep(pollDelay)1119        if self._stopOnError:1120            msg = 'Android monkey error: cannot connect to "adb shell monkey --port 1080" to device %s' % (self._serialNumber)1121            _adapterLog(msg)1122            raise AndroidConnectionError(msg)1123        else:1124            return False1125    def _monkeyCommand(self, command, retry=3):1126        try:1127            self._monkeySocket.sendall(command + "\n")1128            data = self._monkeySocket.recv(4096).strip()1129            if len(data) == 0 and retry > 0:1130                return self._monkeyCommand(command, retry-1)1131            if data == "OK":1132                return True, None1133            elif data.startswith("OK:"):1134                return True, data.split("OK:")[1]1135            _adapterLog("monkeyCommand failing... command: '%s' response: '%s'" % (command, data))1136            return False, None1137        except socket.error:1138            try: self._monkeySocket.close()1139            except: pass1140            if retry > 0:1141                self._resetMonkey()1142                return self._monkeyCommand(command, retry=retry-1)1143            else:1144                raise AndroidConnectionError('Android monkey socket connection lost while sending command "%s"' % (command,))1145    def reboot(self, reconnect, firstBootAfterFlashing, timeout):1146        if firstBootAfterFlashing:1147            self._runAdb("root")1148            time.sleep(2)1149            self._runAdb(["shell", "rm", "/data/data/com.android.launcher/shared_prefs/com.android.launcher2.prefs.xml"])1150        self._runAdb("reboot")1151        _adapterLog("rebooting " + self._serialNumber)1152        if reconnect:1153            time.sleep(2)1154            endTime = time.time() + timeout1155            status, _, _ = self._runAdb("wait-for-device", expectedExitStatus=None, timeout=timeout)1156            if status != 0:1157                raise AndroidDeviceNotFound('"timeout %s adb wait-for-device" status %s' % (timeout, status))1158            self._detectFeatures()1159            while time.time() < endTime:1160                try:1161                    if self._resetMonkey(timeout=1, pollDelay=1):1162                        break1163                except AndroidConnectionError:1164                    pass1165                time.sleep(1)1166            else:1167                msg = "reboot: reconnecting to " + self._serialNumber + " failed"1168                _adapterLog(msg)1169                raise AndroidConnectionError(msg)1170            self._resetWindow()1171        return True1172    def recvVariable(self, variableName):1173        ok, value = self._monkeyCommand("getvar " + variableName)1174        if ok: return value1175        else:1176            # LOG: getvar variableName failed1177            return None1178    def recvScreenSize(self):1179        try:1180            height = int(self.recvVariable("display.height"))1181            width = int(self.recvVariable("display.width"))1182        except TypeError:1183            return None, None1184        return width, height1185    def recvTopAppWindow(self):1186        _, output, _ = self._runAdb(["shell", "dumpsys", "window"], 0)1187        if self._platformVersion >= "4.2":1188            s = re.findall("mCurrentFocus=Window\{(#?[0-9A-Fa-f]{8})( [^ ]*)? (?P<winName>[^}]*)\}", output)1189        else:1190            s = re.findall("mCurrentFocus=Window\{(#?[0-9A-Fa-f]{8}) (?P<winName>[^ ]*) [^ ]*\}", output)1191        if s and len(s[-1][-1].strip()) > 1:1192            topWindowName = s[-1][-1]1193            if len(s) > 0:1194                _adapterLog('recvTopAppWindow warning: several mCurrentFocus windows: "%s"'1195                            % ('", "'.join([w[-1] for w in s]),))1196        else: topWindowName = None1197        s = re.findall("mFocusedApp=AppWindowToken.*ActivityRecord\{#?[0-9A-Fa-f]{8}( [^ ]*)? (?P<appName>[^}]*)\}", output)1198        if s and len(s[0][-1].strip()) > 1:1199            topAppName = s[0][-1].strip()1200        else:1201            topAppName = None1202        return topAppName, topWindowName1203    def sendTap(self, xCoord, yCoord):1204        xCoord, yCoord = self._screenToDisplay(xCoord, yCoord)1205        return self._monkeyCommand("tap " + str(xCoord) + " " + str(yCoord))[0]1206    def sendKeyUp(self, key):1207        return self._monkeyCommand("key up " + key)[0]1208    def sendKeyDown(self, key):1209        return self._monkeyCommand("key down " + key)[0]1210    def sendTouchUp(self, xCoord, yCoord):1211        xCoord, yCoord = self._screenToDisplay(xCoord, yCoord)1212        return self._monkeyCommand("touch up " + str(xCoord) + " " + str(yCoord))[0]1213    def sendTouchDown(self, xCoord, yCoord):1214        xCoord, yCoord = self._screenToDisplay(xCoord, yCoord)1215        return self._monkeyCommand("touch down " + str(xCoord) + " " + str(yCoord))[0]1216    def sendTouchMove(self, xCoord, yCoord):1217        xCoord, yCoord = self._screenToDisplay(xCoord, yCoord)1218        return self._monkeyCommand("touch move " + str(xCoord) + " " + str(yCoord))[0]1219    def sendTrackBallMove(self, dx, dy):1220        dx, dy = self._screenToDisplay(dx, dy)1221        return self._monkeyCommand("trackball " + str(dx) + " " + str(dy))[0]1222    def sendPress(self, key):1223        return self._monkeyCommand("press " + key)[0]1224    def sendType(self, text):1225        for lineIndex, line in enumerate(text.split('\n')):1226            if lineIndex > 0: self.sendPress("KEYCODE_ENTER")1227            for wordIndex, word in enumerate(line.split(' ')):1228                if wordIndex > 0: self.sendPress("KEYCODE_SPACE")1229                if len(word) > 0 and not self._monkeyCommand("type " + word)[0]:1230                    _adapterLog('sendType("%s") failed when sending word "%s"' %1231                                (text, word))1232                    return False1233        return True1234    def sendWake(self):1235        return self._monkeyCommand("wake")[0]1236    def recvScreenshot(self, filename, retry=2, retryDelay=1.0):1237        """1238        Capture a screenshot and copy the image file to given path or1239        system temp folder.1240        Returns True on success, otherwise False.1241        """1242        remotefile = '/sdcard/' + os.path.basename(filename)1243        self._runAdb(['shell', 'screencap', '-p', remotefile], 0)1244        status, out, err = self._runAdb(['pull', remotefile, filename], [0, 1])1245        if status != 0:1246            raise FMBTAndroidError("Failed to fetch screenshot from the device: %s. SD card required." % ((out + err).strip(),))1247        status, _, _ = self._runAdb(['shell', 'rm', remotefile], 0)1248        if os.path.getsize(filename) == 0:1249            _adapterLog("received screenshot of size 0")1250            if retry > 0:1251                time.sleep(retryDelay)1252                return self.recvScreenshot(filename, retry-1, retryDelay)1253            else:1254                raise FMBTAndroidError("Screenshot file size 0")1255        return True1256    def setScreenToDisplayCoords(self, screenToDisplayFunction):1257        self._screenToDisplay = screenToDisplayFunction1258    def setDisplayToScreenCoords(self, displayToScreenFunction):1259        self._displayToScreen = displayToScreenFunction1260    def shellSOE(self, shellCommand):1261        fd, filename = tempfile.mkstemp(prefix="fmbtandroid-shellcmd-")1262        remotename = '/sdcard/' + os.path.basename(filename)1263        os.write(fd, shellCommand + "\n")1264        os.close(fd)1265        self._runAdb(["push", filename, remotename], 0)1266        cmd = "source %s >%s.out 2>%s.err; echo $? > %s.status" % ((remotename,)*4)1267        if self._shellSupportsTar:1268            # do everything we can in one command to minimise adb1269            # commands: execute command, record results, package,1270            # print uuencoded package and remove remote temp files1271            cmd += "; cd %s; tar czf - %s.out %s.err %s.status | uuencode %s.tar.gz; rm -f %s*" % (1272                (os.path.dirname(remotename),) + ((os.path.basename(remotename),) * 5))1273            status, output, error = self._runAdb(["shell", cmd], 0)1274            file(filename, "w").write(output)1275            uu.decode(filename, out_file=filename + ".tar.gz")1276            import tarfile1277            tar = tarfile.open(filename + ".tar.gz")1278            basename = os.path.basename(filename)1279            stdout = tar.extractfile(basename + ".out").read()1280            stderr = tar.extractfile(basename + ".err").read()1281            try: exitstatus = int(tar.extractfile(basename + ".status").read())1282            except: exitstatus = None1283            os.remove(filename)1284            os.remove(filename + ".tar.gz")1285        else:1286            # need to pull files one by one, slow.1287            self._runAdb(["shell", cmd], 0)1288            stdout = self._cat(remotename + ".out")1289            stderr = self._cat(remotename + ".err")1290            try: exitstatus = int(self._cat(remotename + ".status"))1291            except: exitstatus = None1292            self._runAdb(["shell", "rm -f "+remotename+"*"])1293        return exitstatus, stdout, stderr1294    def recvViewData(self, retry=3):1295        _dataBufferLen = 4096 * 161296        try:1297            self._windowSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)1298            self._windowSocket.connect( (self._w_host, self._w_port) )1299            self._windowSocket.settimeout(60)1300            # DUMP -1: get foreground window info1301            if self._windowSocket.sendall("DUMP -1\n") == 0:1302                # LOG: readGUI cannot write to window socket1303                raise AndroidConnectionError("writing socket failed")1304            # Read until a "DONE" line or timeout1305            data = ""1306            while True:1307                newData = ''1308                try:1309                    newData = self._windowSocket.recv(_dataBufferLen)1310                except socket.timeout:1311                    data = None1312                    break1313                data += newData1314                if data.splitlines()[-1] == "DONE" or newData == '':1315                    break1316            return data1317        except Exception, msg:1318            _adapterLog("recvViewData: window socket error: %s" % (msg,))1319            if retry > 0:1320                try: self._windowSocket.close()1321                except: pass1322                self._resetWindow()1323                time.sleep(0.5)1324                return self.recvViewData(retry=retry-1)1325            else:1326                msg = "recvViewData: cannot read window socket"1327                _adapterLog(msg)1328                raise AndroidConnectionError(msg)1329        finally:1330            try: self._windowSocket.close()1331            except: pass1332class FMBTAndroidError(Exception): pass1333class FMBTAndroidRunError(FMBTAndroidError): pass1334class AndroidConnectionError(FMBTAndroidError): pass1335class AndroidConnectionLost(AndroidConnectionError): pass1336class AndroidDeviceNotFound(AndroidConnectionError): pass

signal_window.py

Source:signal_window.py

...14        self.index = 015        self._createDto()16    def _createDto(self):17        self.dto = WindowDto(self.windowSize, self.fields)18    def _resetWindow(self):19        self.index = 020        self._createDto()21    def addData(self, data):22        '''23        expects data like this      24        {25            "X": {26                "value":     1,27                "quality":   2  28            },29            "F3": {30                "value":     3,31                "quality":   4  32            }, ...33        }34        35        :param dict data: 36        '''37        #TODO potential bottleneck38        self.dto.addData(data)39        self.index += 140        41        if self.isFull():42            data = self._doWindowFunction(self.dto)43            self.collectedQueue.put(data)44            self._resetWindow()45    def isFull(self):46        return self.index >= self.windowSize47    def _doWindowFunction(self, data):48        pass49    def __repr__(self):  # pragma: no cover50        return "%s: { windowSize = %d, numValue = %d }" % (self.__class__.__name__, self.windowSize, self.index)51    52class RectangularSignalWindow(SignalWindow):53    '''54    Interface for collector function55    '''56    57    def __init__(self, collectedQueue, windowSize, fields):58        super(RectangularSignalWindow, self).__init__(collectedQueue, windowSize, fields)...

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.