mirror of
https://repo.dactyloidae.xyz/Dactyloidae/UXP.git
synced 2026-08-28 18:28:38 +09:00
674 lines
23 KiB
Python
674 lines
23 KiB
Python
# This Source Code Form is subject to the terms of the Mozilla Public
|
|
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
|
|
# You can obtain one at http://mozilla.org/MPL/2.0/.
|
|
|
|
import hashlib
|
|
import mozlog
|
|
import logging
|
|
import os
|
|
import posixpath
|
|
import re
|
|
import struct
|
|
import StringIO
|
|
import zlib
|
|
|
|
from functools import wraps
|
|
|
|
|
|
class DMError(Exception):
|
|
"generic devicemanager exception."
|
|
|
|
def __init__(self, msg='', fatal=False):
|
|
self.msg = msg
|
|
self.fatal = fatal
|
|
|
|
def __str__(self):
|
|
return self.msg
|
|
|
|
|
|
def abstractmethod(method):
|
|
line = method.func_code.co_firstlineno
|
|
filename = method.func_code.co_filename
|
|
|
|
@wraps(method)
|
|
def not_implemented(*args, **kwargs):
|
|
raise NotImplementedError('Abstract method %s at File "%s", line %s '
|
|
'should be implemented by a concrete class' %
|
|
(repr(method), filename, line))
|
|
return not_implemented
|
|
|
|
|
|
class DeviceManager(object):
|
|
"""
|
|
Represents a connection to a device. Once an implementation of this class
|
|
is successfully instantiated, you may do things like list/copy files to
|
|
the device, launch processes on the device, and install or remove
|
|
applications from the device.
|
|
|
|
Never instantiate this class directly! Instead, instantiate an
|
|
implementation of it like DeviceManagerADB or DeviceManagerSUT.
|
|
"""
|
|
|
|
_logcatNeedsRoot = True
|
|
default_timeout = 300
|
|
short_timeout = 30
|
|
|
|
def __init__(self, logLevel=None, deviceRoot=None):
|
|
try:
|
|
self._logger = mozlog.get_default_logger(component="mozdevice")
|
|
if not self._logger: # no global structured logger, fall back to reg logging
|
|
self._logger = mozlog.unstructured.getLogger("mozdevice")
|
|
if logLevel is not None:
|
|
self._logger.setLevel(logLevel)
|
|
except AttributeError:
|
|
# Structured logging doesn't work on Python 2.6
|
|
self._logger = None
|
|
self._logLevel = logLevel
|
|
self._remoteIsWin = None
|
|
self._isDeviceRootSetup = False
|
|
self._deviceRoot = deviceRoot
|
|
|
|
def _log(self, data):
|
|
"""
|
|
This helper function is called by ProcessHandler to log
|
|
the output produced by processes
|
|
"""
|
|
self._logger.debug(data)
|
|
|
|
@property
|
|
def remoteIsWin(self):
|
|
if self._remoteIsWin is None:
|
|
self._remoteIsWin = self.getInfo("os")["os"][0] == "windows"
|
|
return self._remoteIsWin
|
|
|
|
@property
|
|
def logLevel(self):
|
|
return self._logLevel
|
|
|
|
@logLevel.setter
|
|
def logLevel_setter(self, newLogLevel):
|
|
self._logLevel = newLogLevel
|
|
self._logger.setLevel(self._logLevel)
|
|
|
|
@property
|
|
def debug(self):
|
|
self._logger.warning("dm.debug is deprecated. Use logLevel.")
|
|
levels = {logging.DEBUG: 5, logging.INFO: 3, logging.WARNING: 2,
|
|
logging.ERROR: 1, logging.CRITICAL: 0}
|
|
return levels[self.logLevel]
|
|
|
|
@debug.setter
|
|
def debug_setter(self, newDebug):
|
|
self._logger.warning("dm.debug is deprecated. Use logLevel.")
|
|
newDebug = 5 if newDebug > 5 else newDebug # truncate >=5 to 5
|
|
levels = {5: logging.DEBUG, 3: logging.INFO, 2: logging.WARNING,
|
|
1: logging.ERROR, 0: logging.CRITICAL}
|
|
self.logLevel = levels[newDebug]
|
|
|
|
@abstractmethod
|
|
def getInfo(self, directive=None):
|
|
"""
|
|
Returns a dictionary of information strings about the device.
|
|
|
|
:param directive: information you want to get. Options are:
|
|
|
|
- `os` - name of the os
|
|
- `id` - unique id of the device
|
|
- `uptime` - uptime of the device
|
|
- `uptimemillis` - uptime of the device in milliseconds
|
|
(NOT supported on all implementations)
|
|
- `systime` - system time of the device
|
|
- `screen` - screen resolution
|
|
- `memory` - memory stats
|
|
- `memtotal` - total memory available on the device, for example 927208 kB
|
|
- `process` - list of running processes (same as ps)
|
|
- `disk` - total, free, available bytes on disk
|
|
- `power` - power status (charge, battery temp)
|
|
- `temperature` - device temperature
|
|
|
|
If `directive` is `None`, will return all available information
|
|
"""
|
|
|
|
@abstractmethod
|
|
def getCurrentTime(self):
|
|
"""
|
|
Returns device time in milliseconds since the epoch.
|
|
"""
|
|
|
|
def getIP(self, interfaces=['eth0', 'wlan0']):
|
|
"""
|
|
Returns the IP of the device, or None if no connection exists.
|
|
"""
|
|
for interface in interfaces:
|
|
match = re.match(r"%s: ip (\S+)" % interface,
|
|
self.shellCheckOutput(['ifconfig', interface],
|
|
timeout=self.short_timeout))
|
|
if match:
|
|
return match.group(1)
|
|
|
|
def recordLogcat(self):
|
|
"""
|
|
Clears the logcat file making it easier to view specific events.
|
|
"""
|
|
# TODO: spawn this off in a separate thread/process so we can collect all
|
|
# the logcat information
|
|
|
|
# Right now this is just clearing the logcat so we can only see what
|
|
# happens after this call.
|
|
self.shellCheckOutput(['/system/bin/logcat', '-c'], root=self._logcatNeedsRoot,
|
|
timeout=self.short_timeout)
|
|
|
|
def getLogcat(self, filterSpecs=["dalvikvm:I", "ConnectivityService:S",
|
|
"WifiMonitor:S", "WifiStateTracker:S",
|
|
"wpa_supplicant:S", "NetworkStateTracker:S"],
|
|
format="time",
|
|
filterOutRegexps=[]):
|
|
"""
|
|
Returns the contents of the logcat file as a list of
|
|
'\n' terminated strings
|
|
"""
|
|
cmdline = ["/system/bin/logcat", "-v", format, "-d"] + filterSpecs
|
|
output = self.shellCheckOutput(cmdline,
|
|
root=self._logcatNeedsRoot,
|
|
timeout=self.short_timeout)
|
|
lines = output.replace('\r\n', '\n').splitlines(True)
|
|
|
|
for regex in filterOutRegexps:
|
|
lines = [line for line in lines if not re.search(regex, line)]
|
|
|
|
return lines
|
|
|
|
def saveScreenshot(self, filename):
|
|
"""
|
|
Takes a screenshot of what's being display on the device. Uses
|
|
"screencap" on newer (Android 3.0+) devices (and some older ones with
|
|
the functionality backported). This function also works on B2G.
|
|
|
|
Throws an exception on failure. This will always fail on devices
|
|
without the screencap utility.
|
|
"""
|
|
screencap = '/system/bin/screencap'
|
|
if not self.fileExists(screencap):
|
|
raise DMError("Unable to capture screenshot on device: no screencap utility")
|
|
|
|
with open(filename, 'w') as pngfile:
|
|
# newer versions of screencap can write directly to a png, but some
|
|
# older versions can't
|
|
tempScreenshotFile = self.deviceRoot + "/ss-dm.tmp"
|
|
self.shellCheckOutput(["sh", "-c", "%s > %s" %
|
|
(screencap, tempScreenshotFile)],
|
|
root=True)
|
|
buf = self.pullFile(tempScreenshotFile)
|
|
width = int(struct.unpack("I", buf[0:4])[0])
|
|
height = int(struct.unpack("I", buf[4:8])[0])
|
|
with open(filename, 'w') as pngfile:
|
|
pngfile.write(self._writePNG(buf[12:], width, height))
|
|
self.removeFile(tempScreenshotFile)
|
|
|
|
@abstractmethod
|
|
def pushFile(self, localFilename, remoteFilename, retryLimit=1, createDir=True):
|
|
"""
|
|
Copies localname from the host to destname on the device.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def pushDir(self, localDirname, remoteDirname, retryLimit=1, timeout=None):
|
|
"""
|
|
Push local directory from host to remote directory on the device,
|
|
"""
|
|
|
|
@abstractmethod
|
|
def pullFile(self, remoteFilename, offset=None, length=None):
|
|
"""
|
|
Returns contents of remoteFile using the "pull" command.
|
|
|
|
:param remoteFilename: Path to file to pull from remote device.
|
|
:param offset: Offset in bytes from which to begin reading (optional)
|
|
:param length: Number of bytes to read (optional)
|
|
"""
|
|
|
|
@abstractmethod
|
|
def getFile(self, remoteFilename, localFilename):
|
|
"""
|
|
Copy file from remote device to local file on host.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def getDirectory(self, remoteDirname, localDirname, checkDir=True):
|
|
"""
|
|
Copy directory structure from device (remoteDirname) to host (localDirname).
|
|
"""
|
|
|
|
@abstractmethod
|
|
def validateFile(self, remoteFilename, localFilename):
|
|
"""
|
|
Returns True if a file on the remote device has the same md5 hash as a local one.
|
|
"""
|
|
|
|
def validateDir(self, localDirname, remoteDirname):
|
|
"""
|
|
Returns True if remoteDirname on device is same as localDirname on host.
|
|
"""
|
|
|
|
self._logger.info("validating directory: %s to %s" % (localDirname, remoteDirname))
|
|
for root, dirs, files in os.walk(localDirname):
|
|
parts = root.split(localDirname)
|
|
for f in files:
|
|
remoteRoot = remoteDirname + '/' + parts[1]
|
|
remoteRoot = remoteRoot.replace('/', '/')
|
|
if (parts[1] == ""):
|
|
remoteRoot = remoteDirname
|
|
remoteName = remoteRoot + '/' + f
|
|
if (self.validateFile(remoteName, os.path.join(root, f)) is not True):
|
|
return False
|
|
return True
|
|
|
|
@abstractmethod
|
|
def mkDir(self, remoteDirname):
|
|
"""
|
|
Creates a single directory on the device file system.
|
|
"""
|
|
|
|
def mkDirs(self, filename):
|
|
"""
|
|
Make directory structure on the device.
|
|
|
|
WARNING: does not create last part of the path. For example, if asked to
|
|
create `/mnt/sdcard/foo/bar/baz`, it will only create `/mnt/sdcard/foo/bar`
|
|
"""
|
|
filename = posixpath.normpath(filename)
|
|
containing = posixpath.dirname(filename)
|
|
if not self.dirExists(containing):
|
|
parts = filename.split('/')
|
|
name = "/" if not self.remoteIsWin else parts.pop(0)
|
|
for part in parts[:-1]:
|
|
if part != "":
|
|
name = posixpath.join(name, part)
|
|
self.mkDir(name) # mkDir will check previous existence
|
|
|
|
@abstractmethod
|
|
def dirExists(self, dirpath):
|
|
"""
|
|
Returns whether dirpath exists and is a directory on the device file system.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def fileExists(self, filepath):
|
|
"""
|
|
Return whether filepath exists on the device file system,
|
|
regardless of file type.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def listFiles(self, rootdir):
|
|
"""
|
|
Lists files on the device rootdir.
|
|
|
|
Returns array of filenames, ['file1', 'file2', ...]
|
|
"""
|
|
|
|
@abstractmethod
|
|
def removeFile(self, filename):
|
|
"""
|
|
Removes filename from the device.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def removeDir(self, remoteDirname):
|
|
"""
|
|
Does a recursive delete of directory on the device: rm -Rf remoteDirname.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def moveTree(self, source, destination):
|
|
"""
|
|
Does a move of the file or directory on the device.
|
|
|
|
:param source: Path to the original file or directory
|
|
:param destination: Path to the destination file or directory
|
|
"""
|
|
|
|
@abstractmethod
|
|
def copyTree(self, source, destination):
|
|
"""
|
|
Does a copy of the file or directory on the device.
|
|
|
|
:param source: Path to the original file or directory
|
|
:param destination: Path to the destination file or directory
|
|
"""
|
|
|
|
@abstractmethod
|
|
def chmodDir(self, remoteDirname, mask="777"):
|
|
"""
|
|
Recursively changes file permissions in a directory.
|
|
"""
|
|
|
|
@property
|
|
def deviceRoot(self):
|
|
"""
|
|
The device root on the device filesystem for putting temporary
|
|
testing files.
|
|
"""
|
|
# derive deviceroot value if not set
|
|
if not self._deviceRoot or not self._isDeviceRootSetup:
|
|
self._deviceRoot = self._setupDeviceRoot(self._deviceRoot)
|
|
self._isDeviceRootSetup = True
|
|
|
|
return self._deviceRoot
|
|
|
|
@abstractmethod
|
|
def _setupDeviceRoot(self):
|
|
"""
|
|
Sets up and returns a device root location that can be written to by tests.
|
|
"""
|
|
|
|
def getDeviceRoot(self):
|
|
"""
|
|
Get the device root on the device filesystem for putting temporary
|
|
testing files.
|
|
|
|
.. deprecated:: 0.38
|
|
Use the :py:attr:`deviceRoot` property instead.
|
|
"""
|
|
return self.deviceRoot
|
|
|
|
@abstractmethod
|
|
def getTempDir(self):
|
|
"""
|
|
Returns a temporary directory we can use on this device, ensuring
|
|
also that it exists.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def shell(self, cmd, outputfile, env=None, cwd=None, timeout=None, root=False):
|
|
"""
|
|
Executes shell command on device and returns exit code.
|
|
|
|
:param cmd: Commandline list to execute
|
|
:param outputfile: File to store output
|
|
:param env: Environment to pass to exec command
|
|
:param cwd: Directory to execute command from
|
|
:param timeout: specified in seconds, defaults to 'default_timeout'
|
|
:param root: Specifies whether command requires root privileges
|
|
"""
|
|
|
|
def shellCheckOutput(self, cmd, env=None, cwd=None, timeout=None, root=False):
|
|
"""
|
|
Executes shell command on device and returns output as a string. Raises if
|
|
the return code is non-zero.
|
|
|
|
:param cmd: Commandline list to execute
|
|
:param env: Environment to pass to exec command
|
|
:param cwd: Directory to execute command from
|
|
:param timeout: specified in seconds, defaults to 'default_timeout'
|
|
:param root: Specifies whether command requires root privileges
|
|
:raises: DMError
|
|
"""
|
|
buf = StringIO.StringIO()
|
|
retval = self.shell(cmd, buf, env=env, cwd=cwd, timeout=timeout, root=root)
|
|
output = str(buf.getvalue()[0:-1]).rstrip()
|
|
buf.close()
|
|
if retval != 0:
|
|
raise DMError(
|
|
"Non-zero return code for command: %s "
|
|
"(output: '%s', retval: '%s')" % (cmd, output, retval))
|
|
return output
|
|
|
|
@abstractmethod
|
|
def getProcessList(self):
|
|
"""
|
|
Returns array of tuples representing running processes on the device.
|
|
|
|
Format of tuples is (processId, processName, userId)
|
|
"""
|
|
|
|
def processInfo(self, processName):
|
|
"""
|
|
Returns information on the process with processName.
|
|
Information on process is in tuple format: (pid, process path, user)
|
|
If a process with the specified name does not exist this function will return None.
|
|
"""
|
|
if not isinstance(processName, basestring):
|
|
raise TypeError("Process name %s is not a string" % processName)
|
|
|
|
processInfo = None
|
|
|
|
# filter out extra spaces
|
|
parts = filter(lambda x: x != '', processName.split(' '))
|
|
processName = ' '.join(parts)
|
|
|
|
# filter out the quoted env string if it exists
|
|
# ex: '"name=value;name2=value2;etc=..." process args' -> 'process args'
|
|
parts = processName.split('"')
|
|
if (len(parts) > 2):
|
|
processName = ' '.join(parts[2:]).strip()
|
|
|
|
pieces = processName.split(' ')
|
|
parts = pieces[0].split('/')
|
|
app = parts[-1]
|
|
|
|
procList = self.getProcessList()
|
|
if (procList == []):
|
|
return None
|
|
|
|
for proc in procList:
|
|
procName = proc[1].split('/')[-1]
|
|
if (procName == app):
|
|
processInfo = proc
|
|
break
|
|
return processInfo
|
|
|
|
def processExist(self, processName):
|
|
"""
|
|
Returns True if process with name processName is running on device.
|
|
"""
|
|
processInfo = self.processInfo(processName)
|
|
if processInfo:
|
|
return processInfo[0]
|
|
|
|
@abstractmethod
|
|
def killProcess(self, processName, sig=None):
|
|
"""
|
|
Kills the process named processName. If sig is not None, process is
|
|
killed with the specified signal.
|
|
|
|
:param processName: path or name of the process to kill
|
|
:param sig: signal to pass into the kill command (optional)
|
|
"""
|
|
|
|
@abstractmethod
|
|
def reboot(self, wait=False, ipAddr=None):
|
|
"""
|
|
Reboots the device.
|
|
|
|
:param wait: block on device to come back up before returning
|
|
:param ipAddr: if specified, try to make the device connect to this
|
|
specific IP address after rebooting (only works with
|
|
SUT; if None, we try to determine a reasonable address
|
|
ourselves)
|
|
"""
|
|
|
|
@abstractmethod
|
|
def installApp(self, appBundlePath, destPath=None):
|
|
"""
|
|
Installs an application onto the device.
|
|
|
|
:param appBundlePath: path to the application bundle on the device
|
|
:param destPath: destination directory of where application should be
|
|
installed to (optional)
|
|
"""
|
|
|
|
@abstractmethod
|
|
def uninstallApp(self, appName, installPath=None):
|
|
"""
|
|
Uninstalls the named application from device and DOES NOT cause a reboot.
|
|
|
|
:param appName: the name of the application (e.g org.mozilla.fennec)
|
|
:param installPath: the path to where the application was installed (optional)
|
|
"""
|
|
|
|
@abstractmethod
|
|
def uninstallAppAndReboot(self, appName, installPath=None):
|
|
"""
|
|
Uninstalls the named application from device and causes a reboot.
|
|
|
|
:param appName: the name of the application (e.g org.mozilla.fennec)
|
|
:param installPath: the path to where the application was installed (optional)
|
|
"""
|
|
|
|
@abstractmethod
|
|
def updateApp(self, appBundlePath, processName=None, destPath=None,
|
|
wait=False, ipAddr=None):
|
|
"""
|
|
Updates the application on the device and reboots.
|
|
|
|
:param appBundlePath: path to the application bundle on the device
|
|
:param processName: used to end the process if the applicaiton is
|
|
currently running (optional)
|
|
:param destPath: Destination directory to where the application should
|
|
be installed (optional)
|
|
:param wait: block on device to come back up before returning
|
|
:param ipAddr: if specified, try to make the device connect to this
|
|
specific IP address after rebooting (only works with
|
|
SUT; if None and wait is True, we try to determine a
|
|
reasonable address ourselves)
|
|
"""
|
|
|
|
@staticmethod
|
|
def _writePNG(buf, width, height):
|
|
"""
|
|
Method for writing a PNG from a buffer, used by getScreenshot on older devices,
|
|
"""
|
|
# Based on: http://code.activestate.com/recipes/577443-write-a-png-image-in-native-python/
|
|
width_byte_4 = width * 4
|
|
raw_data = b"".join(b'\x00' + buf[span:span + width_byte_4]
|
|
for span in range(0, (height - 1) * width * 4, width_byte_4))
|
|
|
|
def png_pack(png_tag, data):
|
|
chunk_head = png_tag + data
|
|
return struct.pack("!I", len(data)) \
|
|
+ chunk_head \
|
|
+ struct.pack("!I", 0xFFFFFFFF & zlib.crc32(chunk_head))
|
|
return b"".join([
|
|
b'\x89PNG\r\n\x1a\n',
|
|
png_pack(b'IHDR', struct.pack("!2I5B", width, height, 8, 6, 0, 0, 0)),
|
|
png_pack(b'IDAT', zlib.compress(raw_data, 9)),
|
|
png_pack(b'IEND', b'')])
|
|
|
|
@abstractmethod
|
|
def _getRemoteHash(self, filename):
|
|
"""
|
|
Return the md5 sum of a file on the device.
|
|
"""
|
|
|
|
@staticmethod
|
|
def _getLocalHash(filename):
|
|
"""
|
|
Return the MD5 sum of a file on the host.
|
|
"""
|
|
f = open(filename, 'rb')
|
|
if f is None:
|
|
return None
|
|
|
|
try:
|
|
mdsum = hashlib.md5()
|
|
except:
|
|
return None
|
|
|
|
while 1:
|
|
data = f.read(1024)
|
|
if not data:
|
|
break
|
|
mdsum.update(data)
|
|
|
|
f.close()
|
|
hexval = mdsum.hexdigest()
|
|
return hexval
|
|
|
|
@staticmethod
|
|
def _escapedCommandLine(cmd):
|
|
"""
|
|
Utility function to return escaped and quoted version of command line.
|
|
"""
|
|
quotedCmd = []
|
|
|
|
for arg in cmd:
|
|
arg.replace('&', '\&')
|
|
|
|
needsQuoting = False
|
|
for char in [' ', '(', ')', '"', '&']:
|
|
if arg.find(char) >= 0:
|
|
needsQuoting = True
|
|
break
|
|
if needsQuoting:
|
|
arg = '\'%s\'' % arg
|
|
|
|
quotedCmd.append(arg)
|
|
|
|
return " ".join(quotedCmd)
|
|
|
|
|
|
def _pop_last_line(file_obj):
|
|
"""
|
|
Utility function to get the last line from a file (shared between ADB and
|
|
SUT device managers). Function also removes it from the file. Intended to
|
|
strip off the return code from a shell command.
|
|
"""
|
|
bytes_from_end = 1
|
|
file_obj.seek(0, 2)
|
|
length = file_obj.tell() + 1
|
|
while bytes_from_end < length:
|
|
file_obj.seek((-1) * bytes_from_end, 2)
|
|
data = file_obj.read()
|
|
|
|
if bytes_from_end == length - 1 and len(data) == 0: # no data, return None
|
|
return None
|
|
|
|
if data[0] == '\n' or bytes_from_end == length - 1:
|
|
# found the last line, which should have the return value
|
|
if data[0] == '\n':
|
|
data = data[1:]
|
|
|
|
# truncate off the return code line
|
|
file_obj.truncate(length - bytes_from_end)
|
|
file_obj.seek(0, 2)
|
|
file_obj.write('\0')
|
|
|
|
return data
|
|
|
|
bytes_from_end += 1
|
|
|
|
return None
|
|
|
|
|
|
class ZeroconfListener(object):
|
|
|
|
def __init__(self, hwid, evt):
|
|
self.hwid = hwid
|
|
self.evt = evt
|
|
|
|
# Format is 'SUTAgent [hwid:015d2bc2825ff206] [ip:10_242_29_221]._sutagent._tcp.local.'
|
|
def addService(self, zeroconf, type, name):
|
|
# print "Found _sutagent service broadcast:", name
|
|
if not name.startswith("SUTAgent"):
|
|
return
|
|
|
|
sutname = name.split('.')[0]
|
|
m = re.search('\[hwid:([^\]]*)\]', sutname)
|
|
if m is None:
|
|
return
|
|
|
|
hwid = m.group(1)
|
|
|
|
m = re.search('\[ip:([0-9_]*)\]', sutname)
|
|
if m is None:
|
|
return
|
|
|
|
ip = m.group(1).replace("_", ".")
|
|
|
|
if self.hwid == hwid:
|
|
self.ip = ip
|
|
self.evt.set()
|
|
|
|
def removeService(self, zeroconf, type, name):
|
|
pass
|