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

# Source Code for Module Products.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 or (at your
      8  # option) any later version as published by 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 'events' in args:
    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