mozilla
/
pjs
зеркало из https://github.com/mozilla/pjs.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность
pjs/build/mobile/devicemanager.py

539 строки
15 KiB
Python
Executable File
Исходник Ответственный История

# ***** BEGIN LICENSE BLOCK *****
# Version: MPL 1.1/GPL 2.0/LGPL 2.1
#
# The contents of this file are subject to the Mozilla Public License Version
# 1.1 (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
# http://www.mozilla.org/MPL/
#
# Software distributed under the License is distributed on an "AS IS" basis,
# WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
# for the specific language governing rights and limitations under the
# License.
#
# The Original Code is Test Automation Framework.
#
# The Initial Developer of the Original Code is Joel Maher.
#
# Portions created by the Initial Developer are Copyright (C) 2009
# the Initial Developer. All Rights Reserved.
#
# Contributor(s):
# Joel Maher <joel.maher@gmail.com> (Original Developer)
# Clint Talbert <cmtalbert@gmail.com>
# Mark Cote <mcote@mozilla.com>
#
# Alternatively, the contents of this file may be used under the terms of
# either the GNU General Public License Version 2 or later (the "GPL"), or
# the GNU Lesser General Public License Version 2.1 or later (the "LGPL"),
# in which case the provisions of the GPL or the LGPL are applicable instead
# of those above. If you wish to allow use of your version of this file only
# under the terms of either the GPL or the LGPL, and not to allow others to
# use your version of this file under the terms of the MPL, indicate your
# decision by deleting the provisions above and replace them with the notice
# and other provisions required by the GPL or the LGPL. If you do not delete
# the provisions above, a recipient may use your version of this file under
# the terms of any one of the MPL, the GPL or the LGPL.
#
# ***** END LICENSE BLOCK *****
import time
import hashlib
import socket
import os
import re
class FileError(Exception):
" Signifies an error which occurs while doing a file operation."
def __init__(self, msg = ''):
self.msg = msg
def __str__(self):
return self.msg
class DMError(Exception):
"generic devicemanager exception."
def __init__(self, msg= ''):
self.msg = msg
def __str__(self):
return self.msg
class DeviceManager:
# external function
# returns:
# success: True
# failure: False
def pushFile(self, localname, destname):
assert 0 == 1
return False
# external function
# returns:
# success: directory name
# failure: None
def mkDir(self, name):
assert 0 == 1
return None
# make directory structure on the device
# external function
# returns:
# success: directory structure that we created
# failure: None
def mkDirs(self, filename):
assert 0 == 1
return None
# push localDir from host to remoteDir on the device
# external function
# returns:
# success: remoteDir
# failure: None
def pushDir(self, localDir, remoteDir):
assert 0 == 1
return None
# external function
# returns:
# success: True
# failure: False
def dirExists(self, dirname):
assert 0 == 1
return False
# Because we always have / style paths we make this a lot easier with some
# assumptions
# external function
# returns:
# success: True
# failure: False
def fileExists(self, filepath):
assert 0 == 1
return False
# list files on the device, requires cd to directory first
# external function
# returns:
# success: array of filenames, ['file1', 'file2', ...]
# failure: []
def listFiles(self, rootdir):
assert 0 == 1
return []
# external function
# returns:
# success: output of telnet, i.e. "removing file: /mnt/sdcard/tests/test.txt"
# failure: None
def removeFile(self, filename):
assert 0 == 1
return False
# does a recursive delete of directory on the device: rm -Rf remoteDir
# external function
# returns:
# success: output of telnet, i.e. "removing file: /mnt/sdcard/tests/test.txt"
# failure: None
def removeDir(self, remoteDir):
assert 0 == 1
return None
# external function
# returns:
# success: array of process tuples
# failure: []
def getProcessList(self):
assert 0 == 1
return []
# external function
# returns:
# success: pid
# failure: None
def fireProcess(self, appname, failIfRunning=False):
assert 0 == 1
return None
# external function
# returns:
# success: output filename
# failure: None
def launchProcess(self, cmd, outputFile = "process.txt", cwd = '', env = '', failIfRunning=False):
assert 0 == 1
return None
# loops until 'process' has exited or 'timeout' seconds is reached
# loop sleeps for 'interval' seconds between iterations
# external function
# returns:
# success: [file contents, None]
# failure: [None, None]
def communicate(self, process, timeout = 600, interval = 5):
timed_out = True
if (timeout > 0):
total_time = 0
while total_time < timeout:
time.sleep(interval)
if self.processExist(process) == None:
timed_out = False
break
total_time += interval
if (timed_out == True):
return [None, None]
return [self.getFile(process, "temp.txt"), None]
# iterates process list and returns pid if exists, otherwise None
# external function
# returns:
# success: pid
# failure: None
def processExist(self, appname):
pid = None
#filter out extra spaces
parts = filter(lambda x: x != '', appname.split(' '))
appname = ' '.join(parts)
#filter out the quoted env string if it exists
#ex: '"name=value;name2=value2;etc=..." process args' -> 'process args'
parts = appname.split('"')
if (len(parts) > 2):
appname = ' '.join(parts[2:]).strip()
pieces = appname.split(' ')
parts = pieces[0].split('/')
app = parts[-1]
procre = re.compile('.*' + app + '.*')
procList = self.getProcessList()
if (procList == []):
return None
for proc in procList:
if (procre.match(proc[1])):
pid = proc[0]
break
return pid
# external function
# returns:
# success: output from testagent
# failure: None
def killProcess(self, appname):
assert 0 == 1
return None
# external function
# returns:
# success: filecontents
# failure: None
def catFile(self, remoteFile):
assert 0 == 1
return None
# external function
# returns:
# success: output of pullfile, string
# failure: None
def pullFile(self, remoteFile):
assert 0 == 1
return None
# copy file from device (remoteFile) to host (localFile)
# external function
# returns:
# success: output of pullfile, string
# failure: None
def getFile(self, remoteFile, localFile = ''):
assert 0 == 1
return None
# copy directory structure from device (remoteDir) to host (localDir)
# external function
# checkDir exists so that we don't create local directories if the
# remote directory doesn't exist but also so that we don't call isDir
# twice when recursing.
# returns:
# success: list of files, string
# failure: None
def getDirectory(self, remoteDir, localDir, checkDir=True):
assert 0 == 1
return None
# external function
# returns:
# success: True
# failure: False
# Throws a FileError exception when null (invalid dir/filename)
def isDir(self, remotePath):
assert 0 == 1
return False
# true/false check if the two files have the same md5 sum
# external function
# returns:
# success: True
# failure: False
def validateFile(self, remoteFile, localFile):
assert 0 == 1
return False
# return the md5 sum of a remote file
# internal function
# returns:
# success: MD5 hash for given filename
# failure: None
def getRemoteHash(self, filename):
assert 0 == 1
return None
# return the md5 sum of a file on the host
# internal function
# returns:
# success: MD5 hash for given filename
# failure: None
def getLocalHash(self, filename):
file = open(filename, 'rb')
if (file == None):
return None
try:
mdsum = hashlib.md5()
except:
return None
while 1:
data = file.read(1024)
if not data:
break
mdsum.update(data)
file.close()
hexval = mdsum.hexdigest()
if (self.debug >= 3): print "local hash returned: '" + hexval + "'"
return hexval
# Gets the device root for the testing area on the device
# For all devices we will use / type slashes and depend on the device-agent
# to sort those out. The agent will return us the device location where we
# should store things, we will then create our /tests structure relative to
# that returned path.
# Structure on the device is as follows:
# /tests
# /<fennec>|<firefox> --> approot
# /profile
# /xpcshell
# /reftest
# /mochitest
#
# external function
# returns:
# success: path for device root
# failure: None
def getDeviceRoot(self):
assert 0 == 1
return None
# Either we will have /tests/fennec or /tests/firefox but we will never have
# both. Return the one that exists
# TODO: ensure we can support org.mozilla.firefox
# external function
# returns:
# success: path for app root
# failure: None
def getAppRoot(self):
devroot = self.getDeviceRoot()
if (devroot == None):
return None
if (self.dirExists(devroot + '/fennec')):
return devroot + '/fennec'
elif (self.dirExists(devroot + '/firefox')):
return devroot + '/firefox'
elif (self.dirExsts('/data/data/org.mozilla.fennec')):
return 'org.mozilla.fennec'
elif (self.dirExists('/data/data/org.mozilla.firefox')):
return 'org.mozilla.firefox'
elif (self.dirExists('/data/data/org.mozilla.fennec_aurora')):
return 'org.mozilla.fennec_aurora'
elif (self.dirExists('/data/data/org.mozilla.firefox_beta')):
return 'org.mozilla.firefox_beta'
# Failure (either not installed or not a recognized platform)
return None
# Gets the directory location on the device for a specific test type
# Type is one of: xpcshell|reftest|mochitest
# external function
# returns:
# success: path for test root
# failure: None
def getTestRoot(self, type):
devroot = self.getDeviceRoot()
if (devroot == None):
return None
if (re.search('xpcshell', type, re.I)):
self.testRoot = devroot + '/xpcshell'
elif (re.search('?(i)reftest', type)):
self.testRoot = devroot + '/reftest'
elif (re.search('?(i)mochitest', type)):
self.testRoot = devroot + '/mochitest'
return self.testRoot
# Sends a specific process ID a signal code and action.
# For Example: SIGINT and SIGDFL to process x
def signal(self, processID, signalType, signalAction):
# currently not implemented in device agent - todo
pass
# Get a return code from process ending -- needs support on device-agent
def getReturnCode(self, processID):
# TODO: make this real
return 0
# external function
# returns:
# success: output of unzip command
# failure: None
def unpackFile(self, filename):
return None
# external function
# returns:
# success: status from test agent
# failure: None
def reboot(self, ipAddr=None, port=30000):
assert 0 == 1
return None
# validate localDir from host to remoteDir on the device
# external function
# returns:
# success: True
# failure: False
def validateDir(self, localDir, remoteDir):
if (self.debug >= 2): print "validating directory: " + localDir + " to " + remoteDir
for root, dirs, files in os.walk(localDir):
parts = root.split(localDir)
for file in files:
remoteRoot = remoteDir + '/' + parts[1]
remoteRoot = remoteRoot.replace('/', '/')
if (parts[1] == ""): remoteRoot = remoteDir
remoteName = remoteRoot + '/' + file
if (self.validateFile(remoteName, os.path.join(root, file)) <> True):
return False
return True
# Returns information about the device:
# Directive indicates the information you want to get, your choices are:
# os - name of the os
# id - unique id of the device
# uptime - uptime of the device
# systime - system time of the device
# screen - screen resolution
# memory - memory stats
# process - list of running processes (same as ps)
# disk - total, free, available bytes on disk
# power - power status (charge, battery temp)
# all - all of them - or call it with no parameters to get all the information
# returns:
# success: dict of info strings by directive name
# failure: {}
def getInfo(self, directive=None):
assert 0 == 1
return {}
# external function
# returns:
# success: output from agent for inst command
# failure: None
def installApp(self, appBundlePath, destPath=None):
assert 0 == 1
return None
# external function
# returns:
# success: True
# failure: None
def uninstallAppAndReboot(self, appName, installPath=None):
assert 0 == 1
return None
# external function
# returns:
# success: text status from command or callback server
# failure: None
def updateApp(self, appBundlePath, processName=None, destPath=None, ipAddr=None, port=30000):
assert 0 == 1
return None
# external function
# returns:
# success: time in ms
# failure: None
def getCurrentTime(self):
assert 0 == 1
return None
class NetworkTools:
def __init__(self):
pass
# Utilities to get the local ip address
def getInterfaceIp(self, ifname):
if os.name != "nt":
import fcntl
import struct
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
return socket.inet_ntoa(fcntl.ioctl(
s.fileno(),
0x8915, # SIOCGIFADDR
struct.pack('256s', ifname[:15])
)[20:24])
else:
return None
def getLanIp(self):
ip = socket.gethostbyname(socket.gethostname())
if ip.startswith("127.") and os.name != "nt":
interfaces = ["eth0","eth1","eth2","wlan0","wlan1","wifi0","ath0","ath1","ppp0"]
for ifname in interfaces:
try:
ip = self.getInterfaceIp(ifname)
break;
except IOError:
pass
return ip
# Gets an open port starting with the seed by incrementing by 1 each time
def findOpenPort(self, ip, seed):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
connected = False
if isinstance(seed, basestring):
seed = int(seed)
maxportnum = seed + 5000 # We will try at most 5000 ports to find an open one
while not connected:
try:
s.bind((ip, seed))
connected = True
s.close()
break
except:
if seed > maxportnum:
print "Could not find open port after checking 5000 ports"
raise
seed += 1
except:
print "Socket error trying to find open port"
return seed
Ссылка в новой задаче Показать git blame Копировать постоянную ссылку