.. _fbf.plugs.core.test:
test
~~~~
.. automodule:: fbf.plugs.core.test
:show-inheritance:
:members:
:undoc-members:
CODE
----
::
# fbf/plugs/core/test.py
# encoding: utf-8
#
#
""" test plugin. """
import fbf
from fbf.utils.exception import exceptionmsg, handle_exception, exceptionevents, exceptionlist
from fbf.lib.commands import cmnds
from fbf.lib.examples import examples
from fbf.lib.eventbase import EventBase
from fbf.lib.users import users
from fbf.lib.threads import start_new_thread
from fbf.utils.generic import waitforqueue, waitevents
from fbf.lib.runner import cmndrunner, defaultrunner, threadrunner
from fbf.lib.errors import NoSuchCommand
from fbf import setglobal, getglobal
.. _fbf.plugs.core.test_basic_imports:
basic imports
----------------
::
import time
import random
import copy
import logging
import queue
.. _fbf.plugs.core.test_defines_:
defines
----------
::
cpy = copy.deepcopy
"""
donot = ['relay', 'silent', 'spider', 'chan-token', 'bug', 'test-deadline', 'test-plugs', 'stats', 'geo', 'whatis', 'urlinfo', 'privmsg', 'notice', 'disable', 'deadline', 'twitter', 'stop', 'admin', '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', 'ln', 'markov-learn', 'pit', 'bugtracker', 'tu', 'banner', 'cloud', 'dispatch', 'lns', 'loglevel', \
'test-plugs', 'cloneurl', 'clone', 'hb', 'rss-all', 'rss-get', 'rss-sync', 'rss-add', 'rss-register', 'rss-cloneurl', 'rss-scan']
"""
donot = ['tinyurl', 'deadline', 'imdb', 'wiki', 'translate', 'rss', 'docmnd', 'exception', 'forward', 'admin', 'urlinfo', 'spider', 'fleet', 'validate', 'silent', 'test-plugs', 'markov-learn', 'loglevel', 'twitter', 'snarf', 'relay', 'reboot', 'quit']
disabled = ['tinyurl', 'imdb', 'wikipedia', 'translate', 'rss', 'forward', 'admin', 'urlinfo', 'spider', 'fleet', 'twitter', 'snarf', 'relay']
errors = {}
teller = 0
def dummy(a, b=None):
return ""
.. _fbf.plugs.core.test_dotest_function:
dotest function
------------------
::
def dotest(bot, event, filter="", direct=False):
""" do 1 test loop on the bot, e.g. execute all the examples found. """
global teller
global errors
match = filter
waiting = []
examplez = examples.getexamples(disabled)
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
e = bot.make_event(event.auth, event.channel, ";" + example, event, cbtype=event.cbtype)
e.speed = teller % 5
e.nodispatch = False
e.iscommand = True
e.globalerrors = True
try:
if direct: e.execute()
else:
bot.put(e)
waiting.append(e)
except NoSuchCommand as ex: logging.error("failed to find %s command" % str(ex)) ; continue
except Exception as ex: altar.testerrors[example] = str(ex)
teller += 1
time.sleep(0.01)
for e in waiting: e.join()
event.reply("%s commands executed" % teller)
.. _fbf.plugs.core.test_test-plugs_command:
test-plugs command
---------------------
::
def handle_testplugs(bot, event):
""" no arguments - test the plugins by executing all the available examples. """
#bot.plugs.loadall(force=True)
global teller
try: threaded = event.args[0]
except (ValueError, IndexError): threaded = 1
try: loop = int(event.args[1])
except (ValueError, IndexError): loop = 1
try: filter = event.args[2]
except (ValueError, IndexError): filter = ""
threads = []
teller = 0
jobs = []
bot.plugs.disable(disabled)
#event.dontclose = True
for i in range(loop):
if threaded:
job = threadrunner.put(5, "dotest", dotest, bot, event, filter)
jobs.append(job)
else: dotest(bot, event)
if jobs:
logging.error("%s jobs launched" % len(jobs))
for job in jobs:
if job: logging.info("trying to join %s" % str(job)) ; job.join()
testerrors = getglobal("testerrors")
if testerrors:
event.reply("there are %s test errors .. " % len(testerrors))
for cmnd, error in testerrors.items(): event.reply("%s - %s" % (cmnd, error))
else: event.reply("no errors")
errors = getglobal("errors")
if errors:
event.reply("there are %s errors .. " % len(errors))
for cmnd, error in errors.items(): event.reply("%s - %s" % (cmnd, error))
else: event.reply("no errors")
for (event, msg) in exceptionevents: event.reply("EXCEPTION: %s - %s" % (event.txt,msg))
for msg in exceptionlist: event.reply("EXCEPTION: %s" % msg)
#event.outqueue.append(None)
cmnds.add('test-plugs', handle_testplugs, ['TEST', 'USER'], speed=3, threaded=True)
examples.add('test-plugs', 'test all plugins by running there examples', 'test-plugs')
.. _fbf.plugs.core.test_test-forcedconnection_command:
test-forcedconnection command
--------------------------------
::
def handle_forcedreconnect(bot, ievent):
""" no arguments - do a forced reconnect. """
if not bot.cfg.ssl: bot.sock.shutdown(2)
else: bot.sock.shutdown()
cmnds.add('test-forcedreconnect', handle_forcedreconnect, 'TEST')
.. _fbf.plugs.core.test_test-forcedexception_command:
test-forcedexception command
-------------------------------
::
def handle_forcedexception(bot, ievent):
""" no arguments - raise a exception. """
ievent.reply("raising exception.")
raise Exception('test exception')
cmnds.add('test-forcedexception', handle_forcedexception, 'TEST')
examples.add('test-forcedexception', 'throw an exception as test', 'test-forcedexception')
.. _fbf.plugs.core.test_test-forcedexceptionthreaded_command:
test-forcedexceptionthreaded command
---------------------------------------
::
def handle_forcedexceptionthreaded(bot, ievent):
""" no arguments - raise a exception. """
ievent.reply("raising exception")
raise Exception('test exception')
cmnds.add('test-forcedexceptionthreaded', handle_forcedexceptionthreaded, 'TEST', threaded=True)
examples.add('test-forcedexceptionthreaded', 'throw an exception as test in a thread', 'test-forcedexceptionthreaded')
.. _fbf.plugs.core.test_test-wrongxml_command:
test-wrongxml command
------------------------
::
def handle_testwrongxml(bot, ievent):
""" no arguments - try sending borked xml. """
if not bot.type == "sxmpp":
ievent.reply('only sxmpp')
return
ievent.reply('sending bork xml')
bot._raw('')
cmnds.add('test-wrongxml', handle_testwrongxml, 'TEST')
.. _fbf.plugs.core.test_test-unicode_command:
test-unicode command
-----------------------
::
def handle_testunicode(bot, ievent):
""" no arguments - send unicode test down the output paths. """
outtxt = "Đíť ìš éèñ ëņċøďıńğŧęŝţ· .. にほんごがはなせません .. ₀0⁰₁1¹₂2²₃3³₄4⁴₅5⁵₆6⁶₇7⁷₈8⁸₉9⁹ .. ▁▂▃▄▅▆▇▉▇▆▅▄▃▂▁ .. .. uǝʌoqǝʇsɹǝpuo pɐdı ǝɾ ʇpnoɥ ǝɾ"
ievent.reply(outtxt)
bot.say(ievent.channel, outtxt, event=ievent)
cmnds.add('test-unicode', handle_testunicode, 'TEST')
examples.add('test-unicode', 'test if unicode output path is clear', 'test-unicode')
.. _fbf.plugs.core.test_test-docmnd_command:
test-docmnd command
----------------------
::
def handle_testdocmnd(bot, ievent):
""" no arguments - call bot.docmnd(). """
if ievent.rest: bot.docmnd(ievent.origin or ievent.userhost, ievent.channel, ievent.rest, event=ievent)
else: ievent.missing("") ; return
ievent.done()
cmnds.add('test-docmnd', handle_testdocmnd, 'TEST')
examples.add('test-docmnd', 'test the bot.docmnd() method', 'test-docmnd version')
.. _fbf.plugs.core.test_test-say_command:
test-say command
-------------------
::
def handle_testsay(bot, ievent):
""" arguments: - call the say command on the current bot. """
if not ievent.rest:
ievent.missing("")
return
bot.say(ievent.channel, ievent.rest)
cmnds.add('test-say', handle_testsay, 'TEST')
examples.add('test-say', 'use bot.say()', 'test-say')
.. _fbf.plugs.core.test_test-options_command:
test-options command
-----------------------
::
def handle_testoptions(bot, ievent):
""" no arguments - show options in current event. """
ievent.reply('"%s" - %s' % (ievent.txt, str(ievent.options)))
cmnds.add('test-options', handle_testoptions, 'TEST')
examples.add('test-options', "test event options", "test-options")
.. _fbf.plugs.core.test_test-deadline_command:
test-deadline command
------------------------
::
def handle_testdeadline(bot, ievent):
""" no arguments - slee for 40 seconds in the mainloop. """
ievent.reply('starting 40 sec sleep')
time.sleep(40)
cmnds.add('test-deadline', handle_testdeadline, 'TEST')
examples.add('test-deadline', "sleep 40 sec to trigger deadlineexceeded exception (GAE)", "test-deadline")
.. _fbf.plugs.core.test_test-html_command:
test-html command
--------------------
::
def handle_testhtml(bot, ievent):
""" arguments: [] - test the html=True option to event.reply(). """
if not ievent.rest: data = 'YOOOO BROEDERS'
else: data = ievent.rest
ievent.reply(data, html=True)
cmnds.add('test-html', handle_testhtml, 'TEST')
examples.add('test-html', 'test html output', '1) test-html 2) test-html ')
.. _fbf.plugs.core.test_test-uuid_command:
test-uuid command
--------------------
::
def handle_testuuid(bot, ievent):
""" no arguments - show a uuid4. """
import uuid
ievent.reply(str(uuid.uuid4()))
cmnds.add('test-uuid', handle_testuuid, 'TEST')
examples.add("test-uuid", "show a uuid4.", "test-uuid")
.. _fbf.plugs.core.test_test-threaded_command:
test-threaded command
------------------------
::
def handle_testthreaded(bot, ievent):
""" no arguments - run a threaded command. """
ievent.reply("yoooo!")
cmnds.add("test-threaded", handle_testthreaded, "TEST", threaded=True)
examples.add("test-threaded", "run a threaded command.", "test-threaded")
.. _fbf.plugs.core.test_test-backend_command:
test-backend command
-----------------------
::
def handle_testbackend(bot, ievent):
""" no arguments - run a threaded command. """
ievent.reply("backend yoooo!")
cmnds.add("test-backend", handle_testthreaded, "TEST", threaded="backend")
examples.add("test-backend", "run a test command on the backend", "test-backend")
.. _fbf.plugs.core.test_test-re_command:
test-re command
------------------
::
def handle_testre(bot, event):
event.reply(str(event.groups))
cmnds.add("test-re (.*)$", handle_testre, "TEST", regex=True)
examples.add("test-re", "regular expression as command test", "test-re mekker")
.. _fbf.plugs.core.test_test-multi_command:
test-multi command
---------------------
::
def handle_testmulti(bot, event):
try: nr = int(event.rest)
except: event.missing("") ; return
if nr > 5: nr = 5
for i in range(nr): event.reply("bla %s" % i)
cmnds.add("test-multi", handle_testmulti, ["OPER",])