Initial checkin of atest package of tools.
Change-Id: I2b0768f5a8b30fcbb4c8dc828e76d5809fb20155
diff --git a/android.py b/android.py
index ca7eaf4..1fcf06c 100644
--- a/android.py
+++ b/android.py
@@ -1,3 +1,4 @@
+# python3.4
# Copyright (C) 2009 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
@@ -12,62 +13,133 @@
# License for the specific language governing permissions and limitations under
# the License.
-__author__ = 'Damon Kohler <damonkohler@gmail.com>'
+"""
+JSON RPC interface to android scripting engine.
+"""
-import collections
-import json
+__author__ = 'Keith Dart <dart@google.com>'
+__oldauthor__ = 'Damon Kohler <damonkohler@gmail.com>'
+
import os
+import json
import socket
-import sys
-PORT = os.environ.get('AP_PORT')
-HOST = os.environ.get('AP_HOST')
-HANDSHAKE = os.environ.get('AP_HANDSHAKE')
-Result = collections.namedtuple('Result', 'id,result,error')
+HOST = os.environ.get('AP_HOST', None)
+PORT = os.environ.get('AP_PORT', 9999)
+
+
+class SL4AException(Exception):
+ pass
+
+class SL4AAPIError(SL4AException):
+ pass
+
+class SL4AProtocolError(SL4AException):
+ pass
+
+
+def IDCounter():
+ i = 0
+ while True:
+ yield i
+ i += 1
class Android(object):
+ COUNTER = IDCounter()
- def __init__(self, cmd='initiate', uid=-1, port=None, addr=None):
- self.PORT = os.environ.get('AP_PORT')
- if port is not None:
- self.PORT = port
- if addr is None:
- addr = HOST, self.PORT
- self.conn = socket.create_connection(addr)
- self.client = self.conn.makefile()
- self.id = 0
- handshake = {'cmd':cmd,'uid':uid}
- self.client.write(json.dumps(handshake)+'\n')
- self.client.flush()
- resp = self.client.readline()
- #print(resp)
- result = json.loads(resp)
- self.uid = uid
- if result['status']:
- self.uid = result['uid']
- #self._authenticate(HANDSHAKE)
+ def __init__(self, cmd='initiate', uid=-1, port=PORT, addr=HOST):
+ conn = socket.create_connection((addr, port))
+ self.uid = None
+ self.client = conn.makefile(mode="brw")
+ handshake = {'cmd':cmd, 'uid':uid}
+ self.client.write(json.dumps(handshake).encode("utf8")+b'\n')
+ self.client.flush()
+ resp = self.client.readline()
+ if not resp:
+ raise SL4AProtocolError("No response from handshake.")
+ result = json.loads(str(resp, encoding="utf8"))
+ if result['status']:
+ self.uid = result['uid']
+ else:
+ self.uid = -1
- def _rpc(self, method, *args):
- data = {'id': self.id,
- 'method': method,
- 'params': args}
- request = json.dumps(data)
- #print(request)
- self.client.write(request+'\n')
- self.client.flush()
- response = self.client.readline()
- self.id += 1
- print("Response: "+str(response))
- result = json.loads(response)
- if result['error'] is not None:
- print(result['error'])
- # namedtuple doesn't work with unicode keys.
- return Result(id=result['id'], result=result['result'],
- error=result['error'], )
+ def _rpc(self, method, *args):
+ apiid = next(Android.COUNTER)
+ data = {'id': apiid,
+ 'method': method,
+ 'params': args}
+ request = json.dumps(data)
+ self.client.write(request.encode("utf8")+b'\n')
+ self.client.flush()
+ response = self.client.readline()
+ result = json.loads(str(response, encoding="utf8"))
+ if result['error']:
+ raise SL4AAPIError(result['error'])
+ if result['id'] != apiid:
+ raise SL4AProtocolError("Mismatched API id")
+ return result['result']
- def __getattr__(self, name):
- def rpc_call(*args):
- #print(args)
- return self._rpc(name, *args)
- return rpc_call
+ def __getattr__(self, name):
+ def rpc_call(*args):
+ return self._rpc(name, *args)
+ return rpc_call
+
+
+def start_forwarding(port, localport=PORT):
+ os.system("adb forward tcp:{} tcp:{}".format(localport, port))
+
+
+def android(argv):
+ import getopt
+
+ def _usage():
+ print("""Usage: android -p <port> [-l <localport>] [-u <uid>] <apicall> [<apiargs>...]
+ Example: android.py -p 56241 makeToast hello""")
+
+ localport = PORT
+ port = 0
+ uid = -1
+ cmd = "initiate"
+ try:
+ opts, args = getopt.getopt(argv[1:], "l:p:u:", ["localport=", "port=", "uid="])
+ except getopt.GetoptError:
+ _usage()
+ return
+ for opt, optarg in opts:
+ if opt in ("-p", "--port"):
+ try:
+ port = int(optarg)
+ except ValueError:
+ _usage()
+ return
+ if opt in ("-u", "--uid"):
+ try:
+ uid = int(optarg)
+ except ValueError:
+ _usage()
+ return
+ cmd = "continue"
+
+ if not args:
+ _usage()
+ return
+
+ if not port:
+ _usage()
+ return
+
+ if cmd == "initiate":
+ start_forwarding(port, localport)
+
+ a = Android(cmd=cmd, uid=uid, port=localport)
+ print ("UID:", a.uid)
+ rpc = getattr(a, args[0])
+ rpc(*args[1:])
+
+
+# Run as top-level script for handy test utility
+if __name__ == "__main__":
+ import sys
+ android(sys.argv)
+
