Package ZenEvents :: Module xtest
[hide private]
[frames] | no frames]

Source Code for Module ZenEvents.xtest

  1  ########################################################################### 
  2  # 
  3  # This program is part of Zenoss Core, an open source monitoring platform. 
  4  # Copyright (C) 2007, Zenoss Inc. 
  5  # 
  6  # This program is free software; you can redistribute it and/or modify it 
  7  # under the terms of the GNU General Public License version 2 as published by 
  8  # the Free Software Foundation. 
  9  # 
 10  # For complete information please visit: http://www.zenoss.com/oss/ 
 11  # 
 12  ########################################################################### 
 13  #! /usr/bin/env python  
 14   
 15  __doc__='''xtest.py 
 16   
 17  Sends test events to zenoss via xml-rpc. 
 18  Events can be specified on the command line or read from a file. 
 19  ''' 
 20   
 21  import Globals 
 22  from Products.ZenUtils.CmdBase import CmdBase 
 23  import xmlrpclib 
 24  import time 
 25  import sys 
 26  from Products.ZenEvents.ZenEventClasses import Status_Perf 
 27   
 28  # Input files must be in python format and contain a list named 'events' 
 29  # that contains one dictionary per test event. 
 30  # Each of these dictionaries should have values for device, summary, component 
 31  # and severity. 
 32  #Example: 
 33  # 
 34  #events = [ 
 35  #    { 
 36  #        'device': 'Device1a', 
 37  #        'summary': 'This is the summary.', 
 38  #        'component': 'Some component', 
 39  #        'severity': 4, 
 40  #    }, 
 41  #    { 
 42  #        'device': 'Device2a', 
 43  #        'summary': 'This is the summary.', 
 44  #        'component': 'Some component', 
 45  #        'severity': 4, 
 46  #    }, 
 47  #] 
 48   
 49   
50 -class XTest(CmdBase):
51 52 # Sample event and corresponding clear event used by several methods. 53 sampleEvent = dict(device='Sample device', 54 summary='Test event at %s' % time.time(), 55 eventClass=Status_Perf, 56 severity=4, 57 component='Sample component') 58 sampleClear = sampleEvent.copy() 59 sampleClear.update(dict( 60 severity=0, 61 summary='Clear event')) 62 63
64 - def __init__(self, noopts=0):
65 CmdBase.__init__(self, noopts) 66 self.proxy = None
67 68
69 - def buildOptions(self):
70 """basic options setup sub classes can add more options here""" 71 CmdBase.buildOptions(self) 72 self.parser.add_option('--file', 73 dest="filepath",default=None, 74 help="file containing event details") 75 self.parser.add_option('--sample', 76 dest='dosample', default=False, 77 action='store_true', 78 help='Send sample event and clear event') 79 self.parser.add_option('-d', '--device', 80 dest="device",default='', 81 help="device to use for event") 82 self.parser.add_option('-s', '--summary', 83 dest="summary",default='', 84 help="summary to use for event") 85 self.parser.add_option('-c', '--component', 86 dest="component",default='', 87 help="component to use for event") 88 self.parser.add_option('-y', '--severity', 89 dest="severity",default=4, 90 type='int', 91 help="severity to use for event") 92 self.parser.add_option('--rpchost', 93 dest="rpchost",default='localhost', 94 help="host for xml-rpc request") 95 self.parser.add_option('--rpcport', 96 dest="rpcport",default='8081', 97 help="port for xml-rpc request")
98 99
100 - def parseEventsFile(self, filepath=None):
101 ''' Not much actual parsing going on here, just importing 102 the given file. 103 ''' 104 if not filepath: 105 filepath = self.options.filepath 106 args = {} 107 execfile(filepath, {}, args) 108 if args.has_key('events'): 109 events = args['events'] 110 else: 111 events = [] 112 sys.stderr.write('%s has no value for events\n' % filepath) 113 return events
114 115
116 - def getXmlRpcProxy(self):
117 ''' Returns xmlrpc proxy, creating on if the instance doesn't 118 already have one. 119 ''' 120 if not self.proxy: 121 self.proxy = xmlrpclib.ServerProxy( 122 'http://%s:%s/' % (self.options.rpchost, self.options.rpcport), 123 #verbose=1, 124 encoding='iso-8859-1') 125 return self.proxy
126 127
128 - def sendEvents(self, events):
129 ''' events is a list of dictionaries with details of events to send. 130 This sends those events via the xmlrpc proxy. 131 ''' 132 proxy = self.getXmlRpcProxy() 133 proxy.sendEvents(events)
134 135
136 - def sendSampleEvents(self, repeat=1):
137 ''' Sends the sample event and corresponding clear event. 138 Repeats this as many times as specified by repeat. 139 ''' 140 for i in range(repeat): 141 self.sendEvents([self.sampleEvent, self.sampleClear])
142 143
144 - def sendSampleWithDelayedClear(self, repeat=1, delay=30):
145 ''' Sends sample event then after a delay sends the clear event, 146 repeating as specified. 147 ''' 148 self.sendEvents([self.sampleEvent]) 149 time.sleep(30) 150 self.sendEvents([self.sampleClear])
151 152
153 - def sendEventsFromFile(self, filepath=None):
154 ''' Parse the given file (or the file specified at init) and 155 send the files. 156 ''' 157 events = self.parseEventsFile(filepath=filepath) 158 self.sendEvents(events)
159 160
161 -def main():
162 "performance test" 163 xt = XTest() 164 xt.sendSampleEvents(repeat=100)
165
166 -def coverage():
167 xt = XTest() 168 xt.sendSampleEvents() 169 xt.sendEvents([xt.sampleEvent]) 170 proxy = xt.getXmlRpcProxy() 171 issues = proxy.getDevicePingIssues() 172 for i in issues: 173 print i
174
175 -def simple():
176 xt = XTest() 177 xt.sendSmapleWithDelayedClear()
178 179 if __name__ == '__main__': 180 xt = XTest() 181 if xt.options.dosample: 182 xt.sendSampleEvents() 183 elif xt.options.filepath: 184 xt.sendEventsFromFile() 185 elif xt.options.device: 186 event = dict(device=xt.options.device, 187 summary=xt.options.summary, 188 component=xt.options.component, 189 severity=xt.options.severity) 190 xt.sendEvents([event]) 191 else: 192 xt.parser.print_help() 193