This repository has been archived by the owner. It is now read-only.
Permalink
Browse files

Instead of monkeyrunner, create our own client which uses the protocol

Seems to be some socket issues with the upstream monkeyrunner, sigh...
  • Loading branch information...
wlach committed Mar 22, 2012
1 parent 1f4e4ee commit 8d034212bc38c1fc80b9fe792c0b06919c74141a
Showing with 67 additions and 13 deletions.
  1. +66 −12 bin/devicecontroller.py
  2. +1 −1 bin/runtest.py
@@ -6,24 +6,78 @@
# part of the tools provided with the Android SDK (see:
# http://developer.android.com/guide/developing/tools/monkeyrunner_concepts.html)
import select
import socket
import subprocess
import sys
import time
from com.android.monkeyrunner import MonkeyRunner, MonkeyDevice
class Device(object):
import mozdevice
def __init__(self):
self.device = MonkeyRunner.waitForConnection()
self.width = int(self.device.getProperty('display.width'))
self.height = int(self.device.getProperty('display.height'))
class MonkeyConnection(object):
def __init__(self, port=9999):
# kill any existing instances of the monkey process
droid = mozdevice.DroidADB()
droid.killProcess('app_process')
subprocess.check_call(["adb", "forward", "tcp:%s" % port, "tcp:%s" % port])
p = subprocess.Popen(["adb", "shell", "monkey", "--port", str(port)])
connected = False
tries = 0
while not connected and tries < 20:
time.sleep(0.5)
try:
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._sock.connect(('localhost', port))
self.dimensions = self._get_display_dimensions()
connected = True
except Exception, e:
print "Can't connect to localhost:%s, retrying..." % port
tries+=1
if not connected:
raise Exception("Could not open monkey connection!")
print "Connected!"
time.sleep(1)
def _send_cmd(self, cmd):
select.select([], [self._sock], [])
sent = self._sock.send("%s\r\n" % cmd)
ret = None
while not ret:
select.select([self._sock], [], [])
ret = self._sock.recv(1024)
if not ret.startswith("OK"):
raise Exception("Exception running command '%s'. Got: %s" % (cmd, ret))
if len(ret) > 3:
return ret[3:]
return ""
def _get_display_dimensions(self):
width = int(self._send_cmd("getvar display.width"))
height = int(self._send_cmd("getvar display.height"))
return [width, height]
def drag(self, touch_start, touch_end, duration=1.0, num_steps=5):
self._send_cmd("touch down %s %s" % touch_start)
delta = ((touch_end[0] - touch_start[0]) / num_steps,
(touch_end[1] - touch_start[1]) / num_steps)
for i in range(num_steps):
current = (touch_start[0] + delta[0]*i,
touch_start[1] + delta[1]*i)
self._send_cmd("touch move %s %s" % current)
time.sleep(duration/num_steps)
self._send_cmd("touch up %s %s" % touch_end)
def scroll_down(self):
x = int(self.width / 2)
ybottom = self.height - 100
ytop = 200
self.device.drag((x,ybottom), (x, ytop), 0.1, 5)
x = int(self.dimensions[0] / 2)
ybottom = self.dimensions[1] - 100
ytop = 120
self.drag((x,ybottom), (x,ytop), 0.1, 5)
device = Device()
connection = MonkeyConnection()
print "READY"
sys.stdout.flush()
@@ -47,7 +101,7 @@ def scroll_down(self):
print "QUIT: %s" % time.time()
break
elif cmd == "scroll_down":
device.scroll_down()
connection.scroll_down()
elif cmd == "sleep":
sleeptime = 1
if len(params) > 0:
@@ -74,7 +74,7 @@ def __init__(self, capture_metadata, capture_file, controller):
# open up a monkeyrunner process on startup, wait for it to give a
# line so we don't have to wait
# for it later (note that not all tests use monkeyrunner)
args = ['monkeyrunner', os.path.join(BINDIR, 'devicecontroller.py')]
args = ['python', os.path.join(BINDIR, 'devicecontroller.py')]
self.monkey_proc = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
line = self.monkey_proc.stdout.readline()

0 comments on commit 8d03421

Please sign in to comment.