axmol/tools/jenkins-scripts/emptytest.py

#!/usr/bin/python
# check crash of cppemptytest
# install/open/uninstall apk
# perpose: for emptytest.
# now support: android
# will add: autotest-win7

import os
import sys
import json
import time
import socket
import platform
import threading
import codecs
import re
import paramiko

payload = {}
#get payload from os env
if os.environ.has_key('payload'):
	payload_str = os.environ['payload']
	#parse to json obj
	payload = json.loads(payload_str)
print 'payload:',payload
pr_num = 6326
#get pull number
if payload.has_key('number'):
	pr_num = payload['number']
print 'pr_num:' + str(pr_num)
run_app_time = 5
if os.environ.has_key('RUN_APP_TIME'):
	run_app_time = os.environ['RUN_APP_TIME']
print 'run_app_time:', run_app_time

test_name = ['cpp_empty_test']
if os.environ.has_key('TESTS_NAME'):
	temp_var = os.environ['TESTS_NAME']
	test_name = temp_var.split(', ')
package_name = ['org.cocos2dx.cpp_empty_test']
if os.environ.has_key('PACKAGE_NAME'):
	temp_var = os.environ['PACKAGE_NAME']
	package_name = temp_var.split(', ')
activity_name = ['org.cocos2dx.cpp_empty_test.AppActivity']
if os.environ.has_key('ACTIVITY_NAME'):
	temp_var = os.environ['ACTIVITY_NAME']
	activity_name = temp_var.split(', ')
gIdx = 0
if os.environ.has_key('TEST_INDEX'):
	gIdx = os.environ('TEST_INDEX')

current_platform = platform.system()
print 'current platform is:', current_platform

arrDevices = []
def getDevices():
	cmd = 'adb devices'
	info_devices = os.popen(cmd).read()
	arr_info = info_devices.split('\n')
	del arr_info[0]
	count = 0
	for device in arr_info:
		if len(device) > 0:
			count += 1
			print 'device ', count,device
			deviceInfo = device.split('	')
			global arrDevices
			obj = {}
			obj['name']  = deviceInfo[0]
			arrDevices.append(obj)
	return count

def getADBDeviceIP(device_name):
	output = os.popen("adb -s "+device_name+" shell netcfg")
	configs = output.read().split('\r\n')
	output.close()
	for l in configs:
		items = l.split()
		if len(items)>1 and items[1] == 'UP':
			if items[2].find('127.0.0.1') < 0 and items[2].find('0.0.0.0') < 0:
				return items[2]
	return False
def mapIP():
	for device in arrDevices:
		ip_d = getADBDeviceIP(device['name'])
		device['ip'] = ip_d

allThreadIsRunning = {}
def setThreadStatus():
	for device in arrDevices:
		allThreadIsRunning[device['name']] = 1

devices_info = {}
info_list = '{"product":["model","brand","name","cpu.abi","cpu.abi2","manufacturer","locale.language","locale.region"],"build":["id","version.sdk","version.release"]}'
if os.environ.has_key('DEVICE_INFO_LIST'):
	info_list = os.environ['DEVICE_INFO_LIST']
info_list = eval(info_list)
def getDeviceInfoByName(name):
	cmd = ''
	dev_name = name
	if len(name) > 0:
		cmd = 'adb -s '+name+' shell cat /system/build.prop'
	else:
		cmd = 'adb shell cat /system/build.prop'
		dev_name = 'device one'
	pip_cat = os.popen(cmd)
	read_info = pip_cat.read()
	read_info_list = read_info.split('\r\n')
	device_info_one = {}
	def checkProperty(item_str, device_name):
		for argv in info_list:
			for item in info_list[argv]:
				prop = argv+'.'+item
				if item_str.find(prop) > -1:
					arr_item = item_str.split('=')
					device_info_one[prop] = arr_item[1]
					break
	for item in read_info_list:
		checkProperty(item, dev_name)
	devices_info[dev_name] = device_info_one

#getDeviceInfoByName('')
#print 'device_info:',device_info
def getDeviceInfomation():
	for device in arrDevices:
		getDeviceInfoByName(device['name'])

info_empty_test = {}
info_empty_test_pro = ['install','open','socket','uninstall']
def init_info_empty_test():
	for item in info_empty_test_pro:
		info_empty_test[item] = {}
init_info_empty_test()

apk_name = 'apks/'+test_name[gIdx]+'/'+test_name[gIdx]+'_'+str(pr_num)+'.apk'
def install_apk_on_device(device):
	name = device['name']
	cmd = 'adb -s '+name+' install '+apk_name
	print 'install on '+name
	info_install = os.popen(cmd).read()
	print 'infomation of install apk:', info_install
	info_install_arr = info_install.split('\r\n')
	info_install_result = False
	for item in info_install_arr:
		if item.find('Success') > -1:
			info_install_result = True
	info_empty_test['install'][name] = info_install_result
	return True

def open_apk_on_device(device):
	print 'will open activity:'
	name = device['name']
	cmd = 'adb -s '+name+' shell am start -n '+package_name[gIdx]+'/'+activity_name[gIdx]
	# print 'start activity:', cmd
	info_start = os.popen(cmd).read()
	info_start = info_start.split('\r\n')
	# print 'info_start:', info_start
	info_start_result = True
	for info in info_start:
		if info.find('Error:') > -1:
			print 'infomation of open activity:',info
			info_start_result = False
	info_empty_test['open'][name] = info_start_result
	return True

PORT = 5678
def socket_status_on_device(device):
	name = device['name']
	ip = device['ip']
	soc = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
	status_socket = False
	info_of_socket_result = ''
	try:
		print 'telnet ', ip, PORT
		soc.connect((ip, PORT))
		cmd = 'resolution\r\n'
		print 'connected successfully.'
		print 'send console command: resolution'
		soc.send(cmd)
		while True:
			data = soc.recv(1024)
			if len(data):
				print data
				if data.find('size:') > -1:
					info_of_socket_result = 'OK'
					print 'close', test_name[gIdx]
					soc.send('director end')
					status_socket = True
					break
			if not data:
				info_of_socket_result = test_name[gIdx]+' is crashed!'
				break
	except Exception, e:
		info_of_socket_result = test_name[gIdx]+' is crashed!'
	time.sleep(2)
	soc.close()
	time.sleep(2)
	info_empty_test['socket'][name] = info_of_socket_result
	return status_socket

def uninstall_apk_on_device(device):
	# adb shell pm uninstall -n org.cocos2dx.hellolua
	print 'uninstall ', test_name[gIdx]
	name = device['name']
	cmd = 'adb -s '+name+' shell pm uninstall -n '+package_name[gIdx]
	info_uninstall = os.popen(cmd).read()
	info_uninstall_result = ''
	if info_uninstall.find('Success') > -1:
		info_uninstall_result = 'OK'
	else:
		info_uninstall_result = 'uninstall Failed!'
	info_empty_test['uninstall'][name] = info_uninstall_result
	return True

def excute_test_on_device(device):
	uninstall_apk_on_device(device)
	print device
	info_install = install_apk_on_device(device)
	print 'install:', info_install
	info_open = open_apk_on_device(device)
	print 'open:', info_open
	time.sleep(3)
	info_socket = socket_status_on_device(device)
	print 'socket:', info_socket
	info_uninstall = uninstall_apk_on_device(device)
	print 'uninstall:', info_uninstall
	allThreadIsRunning[device['name']] = 0

def send_result_to_master():
	if not os.environ.has_key('REMOTE_IP'):
		return false
	remote_ip = os.environ['REMOTE_IP']
	remote_port = os.environ['REMOTE_PORT']
	remote_user = os.environ['REMOTE_USER']
	remote_pwd = os.environ['REMOTE_PWD']
	remote_dir = os.environ['REMOTE_DIR']
	print remote_dir
	ssh = paramiko.SSHClient()
	print 'ssh:',ssh
	ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
	ssh.connect(remote_ip, int(remote_port), remote_user, remote_pwd)
	# create dir
	stdin, stdout, stderr = ssh.exec_command("mkdir "+remote_dir)
	ssh.close()
	trans = paramiko.Transport((remote_ip,int(remote_port)))
	trans.connect(username = remote_user, password = remote_pwd)
	sftp = paramiko.SFTPClient.from_transport(trans)
	remotepath = remote_dir+test_name[gIdx]+'_'+str(pr_num)+'.html'
	localpath = 'html/cpp_empty_test/cpp_empty_test_'+str(pr_num)+'.html'
	sftp.put(localpath, remotepath)
	sftp.close()

def check_thread_is_running():
	while 1:
		in_running = 0
		for key in allThreadIsRunning:
			in_running = in_running + allThreadIsRunning[key]
		# print 'in_running:', in_running
		time.sleep(3)
		if not in_running:
			break
	print 'all cpp emptytest is finished.'
	log_emptytest_result()
	print 'will send result:'
	send_result_to_master()
	print 'end of check thread is running.'

def generate_html_with_result(result):
	target_path = 'html/cpp_empty_test/cpp_empty_test_'+str(pr_num)+'.html'
	data = codecs.open('hello.html', encoding='UTF-8').read()
	data = re.sub("content", result, data)
	codecs.open(target_path, 'wb', encoding='UTF-8').write(data)

empty_test_result = True
str_result = ''
def log_emptytest_result():
	def appendToResult(one_str):
		global str_result
		str_result = str_result + one_str + '\r\n'
	appendToResult('<pre>')
	appendToResult('empty test start:')
	for device in arrDevices:
		name = device['name']
		appendToResult('device infomation of : '+name)
		for key in devices_info[name]:
			appendToResult('\t'+key+':'+devices_info[name][key])
		appendToResult('install: ' + str(info_empty_test['install'][name]))
		appendToResult('open: ' + str(info_empty_test['open'][name]))
		appendToResult('telnet ' + str(device['ip'])+' : ' + str(info_empty_test['socket'][name]))
		appendToResult('close: ' + str(info_empty_test['socket'][name]))
		appendToResult('uninstall: ' + str(info_empty_test['uninstall'][name]))
		if not info_empty_test['install'][name] or not info_empty_test['open'][name] or not info_empty_test['socket'][name] or not info_empty_test['uninstall'][name]:
			empty_test_result = False
			appendToResult('empty test is failed!')
		else: appendToResult('empty test is successfully!')
		appendToResult('')
	appendToResult('empty test end.</pre>')
	generate_html_with_result(str_result)
	print 'log empty test end'

class myThread(threading.Thread):
    def __init__(self,threadname):
        threading.Thread.__init__(self,name=threadname)
    def run(self):
        time.sleep(2)
        device = self.getName()
        print type(device)
        device = eval(device)
        print type(device), device
        excute_test_on_device(device)

def run_emptytest():
	for device in arrDevices:
		th = myThread(device)
		th.start()

def main():
	print 'in main:'
	getDevices()
	if len(arrDevices):
		mapIP()
		setThreadStatus()
		print 'arrDevices:',arrDevices
		time.sleep(1)
	else:
		print 'there is no device for emptytest, please check devices!'
		return 1
	if len(arrDevices):
		getDeviceInfomation()
	run_emptytest()
	check_thread_is_running()
	print 'info_empty_test:', info_empty_test
	print 'empty test end', empty_test_result
	if empty_test_result:
		return 0
	else:
		return 1

# -------------- main --------------
if __name__ == '__main__':
    sys_ret = 0
    try:
    	sys_ret = main()
    except:
		traceback.print_exc()
		sys_ret = 1
    finally:
		sys.exit(sys_ret)