gozerbot test plugin.
do a forced reconnect on the bot.
run tests repetatively.
test loaded plugins by running all examples through the bot.
do the test run in a thread.
do a run on the tests registered in the plugins.
# gozerplugs/test.py # # """ gozerbot test plugin. """
from gozerbot.utils.exception import exceptionmsg, handle_exception from gozerbot.tests import tests from gozerbot.commands import cmnds from gozerbot.examples import examples from gozerbot.irc.ircevent import Ircevent from gozerbot.users import users from gozerbot.plughelp import plughelp from gozerbot.partyline import partyline from gozerbot.threads.thr import start_new_thread
import time import random
plughelp.add('test', 'plugin to do tests')
donot = ['bash', 'quit', 'reboot', 'shutdown', 'exit', 'delete', 'halt', 'upgrade', 'install', 'reconnect', 'wiki', 'weather', 'sc', 'jump', 'disable', 'dict', \ 'snarf', 'validate', 'popcon', 'twitter', 'tinyurl', 'whois', 'rblcheck', 'wowwiki', 'wikipedia', 'tr', 'translate', 'serie', 'sc', 'shoutcast', 'mash', \ 'gcalc', 'identi', 'mail', 'part', 'cycle', 'exception', 'fleet', 'rss', 'ln', 'markov-learn', 'pit', 'bugtracker', 'tu', 'banner', 'test', 'cloud', 'dispatch', 'lns'] def dummy(a, b=None): return ""
import gozerbot.utils.url import gozerbot.generic oldgeturl = gozerbot.utils.url.geturl oldgeturl2 = gozerbot.utils.url.geturl2
def handle_testplugs(bot, msg): """ test loaded plugins by running all examples through the bot. """ if bot.cfg['type'] == 'irc' and not msg.isdcc: msg.reply('use this command in a /dcc chat with the bot') ; return gozerbot.utils.url.geturl = dummy gozerbot.utils.url.geturl2 = dummy gozerbot.generic.geturl = dummy gozerbot.generic.geturl2 = dummy if msg.rest: match = msg.rest else: match = "" try: users.add('test', ['test@test',], ['USER', 'OPER']) except Exception, ex: pass bot.channels.setdefault('test', {}) try: loop = int(msg.options['--loop']) except (KeyError, ValueError): loop = 1 teller = 0 for i in range(loop): msg.reply('starting loop %s' % str(i)) examplez = examples.getexamples() random.shuffle(examplez) for example in examplez: if match and match not in example: continue skip = False for dont in donot: if dont in example: skip = True if skip: continue teller += 1 if bot.jabber: from gozerbot.xmpp.message import Message newmessage = Message(msg) newmessage.txt = example newmessage.onlyqueues = False msg.reply('command: ' + example) try: bot.domsg(newmessage) except Exception, ex: handle_exception() else: newmessage = Ircevent(msg) newmessage.txt = '!' + example newmessage.onlyqueues = False msg.reply('command: ' + example) try: bot.domsg(newmessage) except Exception, ex: handle_exception() try: time.sleep(int(msg.options['--sleep'])) except (KeyError, ValueError): pass msg.reply('%s tests run' % teller) gozerbot.utils.url.geturl = oldgeturl gozerbot.utils.url.geturl2 = oldgeturl2 gozerbot.generic.geturl = oldgeturl gozerbot.generic.geturl2 = oldgeturl2 cmnds.add('test-plugs', handle_testplugs, ['OPER', ], options={'--sleep': '0', '--loop': '1'}, threaded=True) tests.add('test-plugs')
def handle_testsrun(bot, ievent): """ do a run on the tests registered in the plugins. """ if bot.cfg['type'] == 'irc' and not ievent.isdcc: ievent.reply('use this command in a /dcc chat with the bot') ; return toolate = [] err = {} try: loop = ievent.options['--loop'] loop = int(loop) except (KeyError, ValueError): loop = 1 tests.dotests(bot, ievent) ievent.reply('%s tests run .. %s errors' % (tests.teller, len(tests.err))) if tests.err: teller = 0 for test in tests.err: ievent.reply('error #%s %s' % (teller, str(test))) ; teller += 1 if tests.toolate: ievent.reply('toolate: ', tests.toolate, dot=True) cmnds.add('test-run', handle_testsrun, 'OPER', threaded=True, options={'--loop': '1', '--threaded': 'False'}) examples.add('test-run', 'run the tests', 'test-run --sleep 0')
def handle_testrunthread(bot, ievent): """ do the test run in a thread. """ if bot.cfg['type'] == 'irc' and not ievent.isdcc: ievent.reply('use this command in a /dcc chat with the bot') ; return toolate = [] err = {} try: plug = ievent.options['--plug'] except (KeyError, ValueError): plug = None tests.dotests(bot, ievent, threaded=True, plug=plug) ievent.reply('%s tests run .. %s errors' % (tests.teller, len(tests.err))) if tests.err: teller = 0 for execstring, test in tests.err.iteritems(): ievent.reply('error #%s .. %s' % (teller, str(test))) ; teller += 1 if tests.toolate: ievent.reply('toolate: ', tests.toolate, dot=True) cmnds.add('test-runthread', handle_testrunthread, 'OPER', threaded=True, options={'--plug': None})
def handle_forcedreconnect(bot, ievent): """ do a forced reconnect on the bot. """ if bot.jabber: bot.disconnectHandler(Exception('test exception for reconnect')) else: bot.sock.close() cmnds.add('test-forcedreconnect', handle_forcedreconnect, 'OPER')
def handle_forcedexception(bot, ievent): raise Exception('test exception') cmnds.add('test-forcedexception', handle_forcedexception, 'OPER') tests.add('test-forcedexception')
def handle_testwrongxml(bot, ievent): if not bot.jabber: ievent.reply('only xmpp') ; return ievent.reply('sending bork xml') bot._raw('<message asdfadf/>') cmnds.add('test-wrongxml', handle_testwrongxml, 'OPER')
def handle_testhammer(bot, msg): """ run tests repetatively. """ if bot.cfg['type'] == 'irc' and not msg.isdcc: msg.reply('use this command in a /dcc chat with the bot') ; return gozerbot.utils.url.geturl = dummy gozerbot.utils.url.geturl2 = dummy gozerbot.generic.geturl = dummy gozerbot.generic.geturl2 = dummy if msg.rest: match = msg.rest else: match = "" try: users.add('test', ['test@test',], ['USER', 'OPER']) except Exception, ex: pass bot.channels.setdefault('test', {}) try: loop = int(msg.options['--loop']) except (KeyError, ValueError): loop = 1 try: hammer = int(msg.options['--hammer']) except (KeyError, ValueError): hammer = 10 teller = 0 for i in range(loop): msg.reply('starting loop %s' % str(i)) examplez = examples.getexamples() random.shuffle(examplez) for example in examplez: if match and match not in example: continue skip = False for dont in donot: if dont in example: skip = True if skip: continue teller += 1 if bot.jabber: from gozerbot.xmpp.message import Message newmessage = Message(msg) newmessage.txt = example newmessage.onlyqueues = False msg.reply('command: ' + example) for hammernr in range(hammer): try: bot.domsg(newmessage) except Exception, ex: handle_exception() else: newmessage = Ircevent(msg) newmessage.txt = '!' + example newmessage.onlyqueues = False msg.reply('command: ' + example) for hammernr in range(hammer): try: bot.domsg(newmessage) except Exception, ex: handle_exception() try: time.sleep(int(msg.options['--sleep'])) except (KeyError, ValueError): pass msg.reply('%s tests run' % teller) gozerbot.utils.url.geturl = oldgeturl gozerbot.utils.url.geturl2 = oldgeturl2 gozerbot.generic.geturl = oldgeturl gozerbot.generic.geturl2 = oldgeturl2 cmnds.add('test-hammer', handle_testhammer, 'OPER') cmnds.add('test-hammer', handle_testhammer, 'OPER', threaded=True) examples.add('test-hammer', 'run the hammer tests', 'test-hammer') tests.add('test-hammer')
def handle_tojson(bot, ievent): ievent.reply(ievent.tojson()) cmnds.add('test-json', handle_tojson, 'OPER')