Page MenuHomeClusterLabs Projects
Files F7631758

OCFIPraTest.py
No OneTemporary
Actions

OCFIPraTest.py
View Options

#!/usr/bin/python
'''OCF IPaddr/IPaddr2 Resource Agent Test'''
__copyright__='''
Author: Huang Zhen <zhenhltc@cn.ibm.com>
Copyright (C) 2004 International Business Machines
Licensed under the GNU GPL.
'''
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
import string,sys,struct,os,random,time,syslog
from cts.CTSvars import *
def usage():
print "usage: " + sys.argv[0] \
+ " [-2]"\
+ " [--ipbase|-i first-test-ip]"\
+ " [--ipnum|-n test-ip-num]"\
+ " [--help|-h]"\
+ " [--perform|-p op]"\
+ " [number-of-iterations]"
sys.exit(1)
def perform_op(ra, ip, op):
os.environ["OCF_RA_VERSION_MAJOR"] = "1"
os.environ["OCF_RA_VERSION_MINOR"] = "0"
os.environ["OCF_ROOT"] = CTSvars.OCF_ROOT_DIR
os.environ["OCF_RESOURCE_INSTANCE"] = ip
os.environ["OCF_RESOURCE_TYPE"] = ra
os.environ["OCF_RESKEY_ip"] = ip
os.environ["HA_LOGFILE"] = "/dev/null"
os.environ["HA_LOGFACILITY"] = "local7"
path = CTSvars.OCF_ROOT_DIR +"/resource.d/heartbeat/" + ra
return os.spawnvpe(os.P_WAIT, path, [ra, op], os.environ)
def audit(ra, iplist, ipstatus, summary):
passed = 1
for ip in iplist:
ret = perform_op(ra, ip, "monitor")
if ret != ipstatus[ip]:
passed = 0
log("audit: status of %s should be %d but it is %d\t [failure]"%
(ip,ipstatus[ip],ret))
ipstatus[ip] = ret
summary["audit"]["called"] += 1;
if passed :
summary["audit"]["success"] += 1
else :
summary["audit"]["failure"] += 1
def log(towrite):
t = time.strftime("%Y/%m/%d_%H:%M:%S\t", time.localtime(time.time()))
logstr = t + " "+str(towrite)
syslog.syslog(logstr)
print logstr
if __name__ == '__main__':
ra = "IPaddr"
ipbase = "127.0.0.10"
ipnum = 1
itnum = 50
perform = None
summary = {
"start":{"called":0,"success":0,"failure":0},
"stop" :{"called":0,"success":0,"failure":0},
"audit":{"called":0,"success":0,"failure":0}
}
syslog.openlog(sys.argv[0], 0, syslog.LOG_LOCAL7)
# Process arguments...
skipthis = None
args = sys.argv[1:]
for i in range(0, len(args)) :
if skipthis :
skipthis = None
continue
elif args[i] == "-2" :
ra = "IPaddr2"
elif args[i] == "--ip" or args[i] == "-i" :
skipthis = 1
ipbase = args[i+1]
elif args[i] == "--ipnum" or args[i] == "-n" :
skipthis = 1
ipnum = int(args[i+1])
elif args[i] == "--perform" or args[i] == "-p" :
skipthis = 1
perform = args[i+1]
elif args[i] == "--help" or args[i] == "-h" :
usage()
else:
itnum = int(args[i])
log("Begin OCF IPaddr/IPaddr2 Test")
# Generate the test ips
iplist = []
ipstatus = {}
fields = string.split(ipbase, '.')
for i in range(0, ipnum) :
ip = string.join(fields, '.')
iplist.append(ip)
ipstatus[ip]=perform_op(ra,ip,"monitor")
fields[3] = str(int(fields[3])+1)
log("Test ip:" + str(iplist))
# If use ask perform an operation
if perform != None:
log("Perform opeartion %s"%perform)
for ip in iplist:
perform_op(ra, ip, perform)
log("Done")
sys.exit()
log("RA Type:" + ra)
log("Test Count:" + str(itnum))
# Prepare Random
f = open("/dev/urandom", "r")
seed=struct.unpack("BBB", f.read(3))
f.close()
#seed=(123,321,231)
rand = random.Random()
rand.seed(seed[0])
log("Test Random Seed:" + str(seed))
#
# Begin Tests
log(">>>>>>>>>>>>>>>>>>>>>>>>")
for i in range(0, itnum):
ip = rand.choice(iplist)
if ipstatus[ip] == 0:
op = "stop"
elif ipstatus[ip] == 7:
op = "start"
else :
op = rand.choice(["start","stop"])
ret = perform_op(ra, ip, op)
# update status
if op == "start" and ret == 0:
ipstatus[ip] = 0
elif op == "stop" and ret == 0:
ipstatus[ip] = 7
else :
ipstatus[ip] = 1
result = ""
if ret == 0:
result = "success"
else :
result = "failure"
summary[op]["called"] += 1
summary[op][result] += 1
log( "%d:%s %s \t[%s]"%(i, op, ip, result))
audit(ra, iplist, ipstatus, summary)
log("<<<<<<<<<<<<<<<<<<<<<<<<")
log("start:\t" + str(summary["start"]))
log("stop: \t" + str(summary["stop"]))
log("audit:\t" + str(summary["audit"]))

File Metadata

Mime Type
text/x-script.python
Expires
Thu, Oct 16, 3:07 PM (21 h, 12 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
2506629
Default Alt Text
OCFIPraTest.py (5 KB)

Event Timeline