Best Python code snippet using fMBT_python
fmbttizen-agent.py
Source:fmbttizen-agent.py
...428 mouse_device.move(x, y)429 return True, None430 except Exception, e:431 return False, str(e)432def sendRelMove(x, y):433 try:434 mouse_device.setXY(0,0)435 mouse_device.move(x, y)436 mouse_device.setXY(0,0)437 return True, None438 except Exception, e:439 return False, str(e)440def sendHwFingerDown(x, y, button):441 try:442 if touch_device:443 touch_device.pressFinger(button, x, y)444 else:445 mouse_device.move(x, y)446 mouse_device.press(button)447 return True, None448 except Exception, e:449 return False, str(e)450def sendHwFingerUp(x, y, button):451 try:452 if touch_device:453 touch_device.releaseFinger(button)454 else:455 mouse_device.move(x, y)456 mouse_device.release(button)457 return True, None458 except Exception, e:459 return False, str(e)460def sendHwKey(keyName, delayBeforePress, delayBeforeRelease):461 keyName = keyName.upper()462 if keyName.startswith("KEY_"):463 keyName = keyName.lstrip("KEY_")464 fd = None465 closeFd = False466 try:467 # Explicit IO device defined for the key?468 inputDevice = deviceToEventFile[hwKeyDevice[keyName]]469 except:470 # Fall back to giving input with keyboard - given that there is one471 if not kbInputDevFd:472 return False, 'No input device for key "%s"' % (keyName,)473 fd = kbInputDevFd474 try: keyCode = _inputKeyNameToCode[keyName]475 except KeyError:476 try: keyCode = fmbtuinput.toKeyCode(keyName)477 except ValueError: return False, 'No keycode for key "%s"' % (keyName,)478 try:479 if not fd:480 fd = os.open(inputDevice, os.O_WRONLY | os.O_NONBLOCK)481 closeFd = True482 except: return False, 'Unable to open input device "%s" for writing' % (inputDevice,)483 if delayBeforePress > 0: time.sleep(delayBeforePress)484 if delayBeforePress >= 0:485 if os.write(fd, struct.pack(_input_event, int(time.time()), 0, _EV_KEY, keyCode, 1)) > 0:486 os.write(fd, struct.pack(_input_event, 0, 0, 0, 0, 0))487 if delayBeforeRelease > 0: time.sleep(delayBeforeRelease)488 if delayBeforeRelease >= 0:489 if os.write(fd, struct.pack(_input_event, int(time.time()), 0, _EV_KEY, keyCode, 0)) > 0:490 os.write(fd, struct.pack(_input_event, 0, 0, 0, 0, 0))491 if closeFd:492 os.close(fd)493 return True, None494def specialCharToXString(c):495 c2s = {'\n': "Return",496 ' ': "space", '!': "exclam", '"': "quotedbl",497 '#': "numbersign", '$': "dollar", '%': "percent",498 '&': "ampersand", "'": "apostrophe",499 '(': "parenleft", ')': "parenright", '*': "asterisk",500 '+': "plus", '-': "minus", '.': "period", '/': "slash",501 ':': "colon", ';': "semicolon", '<': "less", '=': "equal",502 '>': "greater", '?': "question", '@': "at",503 '_': "underscore"}504 return c2s.get(c, c)505def specialCharToUsKeys(c):506 # character -> ([modifier, [modifier...]] keycode)507 c2s = {'\n': ("KEY_ENTER",),508 ' ': ("KEY_SPACE",),509 '`': ("KEY_GRAVE",), '~': ("KEY_LEFTSHIFT", "KEY_GRAVE"),510 '!': ("KEY_LEFTSHIFT", "KEY_1"),511 '@': ("KEY_LEFTSHIFT", "KEY_2"),512 '#': ("KEY_LEFTSHIFT", "KEY_3"),513 '$': ("KEY_LEFTSHIFT", "KEY_4"),514 '%': ("KEY_LEFTSHIFT", "KEY_5"),515 '^': ("KEY_LEFTSHIFT", "KEY_6"),516 '&': ("KEY_LEFTSHIFT", "KEY_7"),517 '*': ("KEY_LEFTSHIFT", "KEY_8"),518 '(': ("KEY_LEFTSHIFT", "KEY_9"),519 ')': ("KEY_LEFTSHIFT", "KEY_0"),520 '-': ("KEY_MINUS",), '_': ("KEY_LEFTSHIFT", "KEY_MINUS"),521 '=': ("KEY_EQUAL",), '+': ("KEY_LEFTSHIFT", "KEY_EQUAL"),522 '\t': ("KEY_TAB",),523 '[': ("KEY_LEFTBRACE",), '{': ("KEY_LEFTSHIFT", "KEY_LEFTBRACE"),524 ']': ("KEY_RIGHTBRACE",), '}': ("KEY_LEFTSHIFT", "KEY_RIGHTBRACE"),525 ';': ("KEY_SEMICOLON",), ':': ("KEY_LEFTSHIFT", "KEY_SEMICOLON"),526 "'": ("KEY_APOSTROPHE",), '"': ("KEY_LEFTSHIFT", "KEY_APOSTROPHE"),527 '\\': ("KEY_BACKSLASH",), '|': ("KEY_LEFTSHIFT", "KEY_BACKSLASH"),528 ',': ("KEY_COMMA",), '<': ("KEY_LEFTSHIFT", "KEY_COMMA"),529 '.': ("KEY_DOT",), '>': ("KEY_LEFTSHIFT", "KEY_DOT"),530 '/': ("KEY_SLASH",), '?': ("KEY_LEFTSHIFT", "KEY_SLASH"),531 }532 return c2s.get(c, c)533mtEvents = {} # slot -> (tracking_id, x, y)534def inputEventSend(inputDevFd, eventType, event, param):535 t = time.time()536 tsec = int(t)537 tusec = int(1000000*(t-tsec))538 os.write(inputDevFd, struct.pack(_input_event,539 tsec, tusec, eventType, event, param))540def mtEventSend(eventType, event, param):541 """multitouch device event"""542 return inputEventSend(mtInputDevFd, eventType, event, param)543def mtGestureStart(x, y):544 mtGestureStart.trackingId += 1545 trackingId = mtGestureStart.trackingId546 for freeSlot in xrange(16):547 if not freeSlot in mtEvents: break548 else: raise ValueError("No free multitouch event slots available")549 mtEvents[freeSlot] = [trackingId, x, y]550 mtEventSend(_EV_ABS, _ABS_MT_SLOT, freeSlot)551 mtEventSend(_EV_ABS, _ABS_MT_TRACKING_ID, trackingId)552 mtEventSend(_EV_ABS, _ABS_MT_POSITION_X, x)553 mtEventSend(_EV_ABS, _ABS_MT_POSITION_Y, y)554 mtEventSend(_EV_ABS, _ABS_X, x)555 mtEventSend(_EV_ABS, _ABS_Y, y)556 mtEventSend(0, 0, 0) # SYNC557 return freeSlot558mtGestureStart.trackingId = 0559def mtGestureMove(slot, x, y):560 if x == mtEvents[slot][1] and y == mtEvents[slot][2]: return561 mtEventSend(_EV_ABS, _ABS_MT_SLOT, slot)562 mtEventSend(_EV_ABS, _ABS_MT_TRACKING_ID, mtEvents[slot][0])563 if x != mtEvents[slot][1] and 0 <= x <= root_width:564 mtEventSend(_EV_ABS, _ABS_MT_POSITION_X, x)565 mtEvents[slot][1] = x566 if y != mtEvents[slot][2] and 0 <= y <= root_height:567 mtEventSend(_EV_ABS, _ABS_MT_POSITION_Y, y)568 mtEvents[slot][2] = y569 if 0 <= x <= root_width:570 mtEventSend(_EV_ABS, _ABS_X, x)571 if 0 <= y <= root_height:572 mtEventSend(_EV_ABS, _ABS_Y, y)573 mtEventSend(0, 0, 0)574def mtGestureEnd(slot):575 mtEventSend(_EV_ABS, _ABS_MT_SLOT, slot)576 mtEventSend(_EV_ABS, _ABS_MT_TRACKING_ID, -1)577 mtEventSend(0, 0, 0) # SYNC578 del mtEvents[slot]579def mtLinearGesture(listOfStartEndPoints, duration, movePoints, sleepBeforeMove=0, sleepAfterMove=0):580 # listOfStartEndPoints: [ [(finger1startX, finger1startY), (finger1endX, finger1endY)],581 # [(finger2startX, finger2startY), (finger2endX, finger2endY)], ...]582 startPoints = [startEnd[0] for startEnd in listOfStartEndPoints]583 xDist = [startEnd[1][0] - startEnd[0][0] for startEnd in listOfStartEndPoints]584 yDist = [startEnd[1][1] - startEnd[0][1] for startEnd in listOfStartEndPoints]585 movePointsF = float(movePoints)586 fingers = []587 for (x, y) in startPoints:588 fingers.append(mtGestureStart(x, y))589 if sleepBeforeMove > 0: time.sleep(sleepBeforeMove)590 if movePoints > 0:591 intermediateSleep = float(duration) / movePoints592 for i in xrange(1, movePoints + 1):593 if intermediateSleep > 0:594 time.sleep(intermediateSleep)595 for fingerIndex, finger in enumerate(fingers):596 mtGestureMove(finger,597 startPoints[fingerIndex][0] + int(xDist[fingerIndex]*i/movePointsF),598 startPoints[fingerIndex][1] + int(yDist[fingerIndex]*i/movePointsF))599 if sleepAfterMove > 0: time.sleep(sleepAfterMove)600 for finger in fingers:601 mtGestureEnd(finger)602 return True, None603def typeCharX(origChar):604 modifiers = []605 c = specialCharToXString(origChar)606 keysym = libX11.XStringToKeysym(c)607 if keysym == NoSymbol:608 return False609 keycode = libX11.XKeysymToKeycode(display, keysym)610 first = (keycode - cMinKeycode.value) * cKeysymsPerKeycode.value611 try:612 if chr(keysyms[first + 1]) == origChar:613 modifiers.append(shiftModifier)614 except ValueError: pass615 for m in modifiers:616 libXtst.XTestFakeKeyEvent(display, m, X_True, X_CurrentTime)617 libXtst.XTestFakeKeyEvent(display, keycode, X_True, X_CurrentTime)618 libXtst.XTestFakeKeyEvent(display, keycode, X_False, X_CurrentTime)619 for m in modifiers[::-1]:620 libXtst.XTestFakeKeyEvent(display, m, X_False, X_CurrentTime)621 return True622def typeCharHw(origChar):623 for c in origChar:624 modifiers = []625 keyCode = None626 c = specialCharToUsKeys(c)627 if isinstance(c, tuple):628 modifiers = c[:-1]629 keyCode = c[-1]630 elif c in string.uppercase:631 modifiers = ["KEY_LEFTSHIFT"]632 keyCode = "KEY_" + c633 elif c in string.lowercase or c in string.digits:634 keyCode = "KEY_" + c.upper()635 else:636 # do not know how to type the character637 pass638 if keyCode:639 for m in modifiers:640 keyboard_device.press(m)641 keyboard_device.tap(keyCode)642 for m in modifiers[::-1]:643 keyboard_device.release(m)644 return True645if g_Xavailable:646 typeChar = typeCharX647else:648 typeChar = typeCharHw649def typeSequence(s, delayBetweenChars=0):650 skipped = []651 for c in s:652 if not typeChar(c):653 skipped.append(c)654 if delayBetweenChars != 0:655 time.sleep(delayBetweenChars)656 if skipped: return False, skipped657 else: return True, skipped658def takeScreenshotOnX():659 image_p = libX11.XGetImage(display, root_window,660 0, 0, root_width, root_height,661 X_AllPlanes, X_ZPixmap)662 image = image_p[0]663 # FMBTRAWX11 image format header:664 # FMBTRAWX11 [width] [height] [color depth] [bits per pixel]<linefeed>665 # Binary data666 rawfmbt_header = "FMBTRAWX11 %d %d %d %d\n" % (667 image.width, image.height, root_depth.value, image.bits_per_pixel)668 rawfmbt_data = ctypes.string_at(image.data, image.height * image.bytes_per_line)669 compressed_image = rawfmbt_header + zlib.compress(rawfmbt_data, 3)670 libX11.XDestroyImage(image_p)671 return True, compressed_image672def westonTakeScreenshotRoot(retry=2):673 if westonTakeScreenshotRoot.ssFilenames == None:674 westonTakeScreenshotRoot.ssFilenames = findWestonScreenshotFilenames()675 if not westonTakeScreenshotRoot.ssFilenames:676 return False, "cannot find weston screenshot directory"677 try:678 for ssFilename in westonTakeScreenshotRoot.ssFilenames:679 if os.access(ssFilename, os.R_OK):680 os.remove(ssFilename)681 keyboard_device.press("KEY_LEFTMETA")682 keyboard_device.tap("s")683 keyboard_device.release("KEY_LEFTMETA")684 time.sleep(0.5)685 # find which screenshot file got created?686 for ssFilename in westonTakeScreenshotRoot.ssFilenames:687 if os.access(ssFilename, os.R_OK):688 break689 else:690 if retry > 0:691 return westonTakeScreenshotRoot(retry-1)692 else:693 return False, "weston did not create any of files %s" % (694 westonTakeScreenshotRoot.ssFilenames,)695 # wait for the screenshot writer to finish696 writerPid = fuser(ssFilename)697 if writerPid != None:698 time.sleep(0.1)699 while fuser(ssFilename, [writerPid]) != None:700 time.sleep(0.1)701 shutil.move(ssFilename, "/tmp/screenshot.png")702 os.chmod("/tmp/screenshot.png", 0666)703 except Exception, e:704 return False, str(e)705 return True, None706westonTakeScreenshotRoot.ssFilenames = None707def takeScreenshotOnWeston():708 if iAmRoot:709 rv, status = westonTakeScreenshotRoot()710 else:711 rv, status = subAgentCommand("root", "tizen", "ss weston-root")712 if rv == False:713 return rv, status714 return True, file("/tmp/screenshot.png").read()715def fuser(filename, usualSuspects=None):716 """Returns the pid of a user of given file, or None"""717 filepath = os.path.realpath(filename)718 if not os.access(filepath, os.R_OK):719 raise ValueError('No such file: "%s"' % (filename,))720 if usualSuspects == None:721 procFds = glob.glob("/proc/[1-9][0-9][0-9]*/fd/*")722 else:723 procFds = []724 for pid in usualSuspects:725 procFds.extend(glob.glob("/proc/%s/fd/*" % (pid,)))726 for symlink in procFds:727 try:728 if os.path.realpath(symlink) == filepath:729 return int(symlink.split('/')[2])730 except OSError:731 pass732def findWestonScreenshotFilenames():733 # find weston cwd734 dirs = []735 for exe in glob.glob("/proc/[1-9][0-9][0-9]*/exe"):736 try:737 if os.path.realpath(exe) == "/usr/bin/weston":738 dirs.append(os.path.realpath(os.path.dirname(exe) + "/cwd"))739 except OSError:740 pass741 return [d + "/wayland-screenshot.png" for d in sorted(set(dirs))]742if g_Xavailable:743 takeScreenshot = takeScreenshotOnX744else:745 takeScreenshot = takeScreenshotOnWeston746def shellSOE(command, asyncStatus, asyncOut, asyncError, usePty):747 if usePty:748 command = '''python -c "import pty; pty.spawn(%s)" ''' % (repr(shlex.split(command)),),749 if (asyncStatus, asyncOut, asyncError) != (None, None, None):750 # prepare for decoupled asynchronous execution751 if asyncStatus == None: asyncStatus = "/dev/null"752 if asyncOut == None: asyncOut = "/dev/null"753 if asyncError == None: asyncError = "/dev/null"754 try:755 stdinFile = file("/dev/null", "r")756 stdoutFile = file(asyncOut, "a+")757 stderrFile = file(asyncError, "a+", 0)758 statusFile = file(asyncStatus, "a+")759 except IOError, e:760 return False, (None, None, e)761 try:762 if os.fork() > 0:763 # parent returns after successful fork, there no764 # direct visibility to async child process beyond this765 # point.766 stdinFile.close()767 stdoutFile.close()768 stderrFile.close()769 statusFile.close()770 return True, (0, None, None) # async parent finishes here771 except OSError, e:772 return False, (None, None, e)773 os.setsid()774 else:775 stdinFile = subprocess.PIPE776 stdoutFile = subprocess.PIPE777 stderrFile = subprocess.PIPE778 try:779 p = subprocess.Popen(command, shell=True,780 stdin=stdinFile,781 stdout=stdoutFile,782 stderr=stderrFile,783 close_fds=True)784 except Exception, e:785 return False, (None, None, e)786 if asyncStatus == None and asyncOut == None and asyncError == None:787 # synchronous execution, read stdout and stderr788 out, err = p.communicate()789 else:790 # asynchronous execution, store status to file791 statusFile.write(str(p.wait()) + "\n")792 statusFile.close()793 out, err = None, None794 sys.exit(0) # async child finishes here795 return True, (p.returncode, out, err)796def waitOutput(nonblockingFd, acceptedOutputs, timeout, pollInterval=0.1):797 start = time.time()798 endTime = start + timeout799 s = ""800 try: s += nonblockingFd.read()801 except IOError: pass802 foundOutputs = [ao for ao in acceptedOutputs if ao in s]803 while len(foundOutputs) == 0 and time.time() < endTime:804 time.sleep(pollInterval)805 try: s += nonblockingFd.read()806 except IOError: pass807 foundOutputs = [ao for ao in acceptedOutputs if ao in s]808 return foundOutputs, s809_subAgents = {}810def openSubAgent(username, password):811 p = subprocess.Popen('''python -c 'import pty; pty.spawn(["su", "-c", "python /tmp/fmbttizen-agent.py --sub-agent", "-", "%s"])' ''' % (username,),812 shell=True,813 stdin=subprocess.PIPE,814 stdout=subprocess.PIPE,815 stderr=subprocess.PIPE)816 # Read in non-blocking mode to ensure agent starts correctly817 fl = fcntl.fcntl(p.stdout.fileno(), fcntl.F_GETFL)818 fcntl.fcntl(p.stdout.fileno(), fcntl.F_SETFL, fl | os.O_NONBLOCK)819 output2 = ""820 seenPrompts, output1 = waitOutput(p.stdout, ["Password:", "FMBTAGENT"], 5.0)821 if "Password:" in seenPrompts:822 p.stdin.write(password + "\r")823 output1 = ""824 seenPrompts, output2 = waitOutput(p.stdout, ["FMBTAGENT"], 5.0)825 if not "FMBTAGENT" in seenPrompts:826 p.terminate()827 return (None, 'fMBT agent with username "%s" does not answer.' % (username,),828 output1 + output2)829 # Agent is alive, continue in blocking mode830 fcntl.fcntl(p.stdout.fileno(), fcntl.F_SETFL, fl)831 return p, "", ""832def subAgentCommand(username, password, cmd):833 if not username in _subAgents:834 process, output, error = openSubAgent(username, password)835 if process == None:836 return None, (-1, output, error)837 else:838 _subAgents[username] = process839 p = _subAgents[username]840 p.stdin.write(cmd + "\r")841 p.stdin.flush()842 answer = p.stdout.readline().rstrip()843 if answer.startswith("FMBTAGENT OK "):844 return True, _decode(answer[len("FMBTAGENT OK "):])845 else:846 return False, _decode(answer[len("FMBTAGENT ERROR "):])847def closeSubAgents():848 for username in _subAgents:849 subAgentCommand(username, None, "quit")850if __name__ == "__main__":851 try:852 origTermAttrs = termios.tcgetattr(sys.stdin.fileno())853 hasTerminal = True854 except termios.error:855 origTermAttrs = None856 hasTerminal = False857 if hasTerminal and not "--keep-echo" in sys.argv and not "--debug" in sys.argv:858 # Disable terminal echo859 newTermAttrs = origTermAttrs860 newTermAttrs[3] = origTermAttrs[3] & ~termios.ECHO861 termios.tcsetattr(sys.stdin.fileno(), termios.TCSANOW, newTermAttrs)862 if "--no-x" in sys.argv:863 debug("X disabled")864 g_Xavailable = False865 platformInfo = {}866 platformInfo["input devices"] = fmbtuinput._g_deviceNames.keys()867 # Send version number, enter main loop868 write_response(True, platformInfo)869 cmd = read_cmd()870 while cmd:871 if cmd.startswith("bl "): # set display backlight time872 if iAmRoot:873 timeout = int(cmd[3:].strip())874 try:875 file("/opt/var/kdb/db/setting/lcd_backlight_normal","wb").write(struct.pack("ii",0x29,timeout))876 write_response(True, None)877 except Exception, e: write_response(False, e)878 else:879 write_response(*subAgentCommand("root", "tizen", cmd))880 elif cmd.startswith("er "): # event recorder881 if iAmRoot:882 cmd, arg = cmd.split(" ", 1)883 if arg.startswith("start "):884 filterOpts = _decode(arg.split()[1])885 if touch_device:886 filterOpts["touchScreen"] = touch_device887 fmbtuinput.startQueueingEvents(filterOpts)888 write_response(True, None)889 elif arg == "stop":890 events = fmbtuinput.stopQueueingEvents()891 write_response(True, None)892 elif arg == "fetch":893 events = fmbtuinput.fetchQueuedEvents()894 write_response(True, events)895 else:896 write_response(*subAgentCommand("root", "tizen", cmd))897 elif cmd.startswith("gd"): # get display status898 try:899 p = subprocess.Popen(['/usr/bin/xset', 'q'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)900 out, err = p.communicate()901 if "Monitor is Off" in out: write_response(True, "Off")902 elif "Monitor is On" in out: write_response(True, "On")903 else: write_response(False, err)904 except Exception, e: write_response(False, e)905 elif cmd.startswith("tm "): # touch move(x, y)906 xs, ys = cmd[3:].strip().split()907 if g_Xavailable:908 libXtst.XTestFakeMotionEvent(display, current_screen, int(xs), int(ys), X_CurrentTime)909 libX11.XFlush(display)910 else:911 if iAmRoot: rv, msg = sendHwMove(int(xs), int(ys))912 else: rv, msg = subAgentCommand("root", "tizen", cmd)913 write_response(True, None)914 elif cmd.startswith("tr "): # relative move(x, y)915 xd, yd = cmd[3:].strip().split()916 if iAmRoot: rv, msg = sendRelMove(int(xd), int(yd))917 else: rv, msg = subAgentCommand("root", "tizen", cmd)918 write_response(True, None)919 elif cmd.startswith("tt "): # touch tap(x, y, button)920 x, y, button = [int(i) for i in cmd[3:].strip().split()]921 if g_Xavailable:922 libXtst.XTestFakeMotionEvent(display, current_screen, x, y, X_CurrentTime)923 libXtst.XTestFakeButtonEvent(display, button, X_True, X_CurrentTime)924 libXtst.XTestFakeButtonEvent(display, button, X_False, X_CurrentTime)925 libX11.XFlush(display)926 rv, msg = True, None927 else:928 if iAmRoot: rv, msg = sendHwTap(x, y, button-1)929 else: rv, msg = subAgentCommand("root", "tizen", cmd)930 write_response(rv, msg)...
fmbttizen.py
Source:fmbttizen.py
...665 def sendMtLinearGesture(self, *args):666 return self._agentCmd("ml %s" % (_encode(args)))[0]667 def sendScreenshotRotation(self, angle):668 return self._agentCmd("sa %s" % (angle,))[0]669 def sendRelMove(self, x, y):670 return self._agentCmd("tr %s %s" % (x, y))[0]671 def sendTap(self, x, y):672 return self._agentCmd("tt %s %s 1" % (x, y))[0]673 def sendTouchDown(self, x, y):674 return self._agentCmd("td %s %s 1" % (x, y))[0]675 def sendTouchMove(self, x, y):676 return self._agentCmd("tm %s %s" % (x, y))[0]677 def sendTouchUp(self, x, y):678 return self._agentCmd("tu %s %s 1" % (x, y))[0]679 def sendType(self, string):680 return self._agentCmd("kt %s" % (_encode(string)))[0]681 def sendDisplayBacklightTime(self, timeout):682 return self._agentCmd("bl %s" % (timeout,))[0]683 def sendRecStart(self, devices=[]):...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!