303 lines
10 KiB
Python
303 lines
10 KiB
Python
"""
|
|
To be executed with nose
|
|
|
|
TODO: test all plugins, maybe declare their sample input somewhere near the code
|
|
"""
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from collections import namedtuple
|
|
|
|
from common import buckets, rate_limit, RATE_GLOBAL
|
|
|
|
|
|
class TestEventlooper(unittest.TestCase):
|
|
def test_broken_url(self):
|
|
"""
|
|
Test that broken socket calls are not breaking
|
|
"""
|
|
from common import fetch_page
|
|
broken_url = 'http://foo'
|
|
result = fetch_page(url=broken_url)
|
|
self.assertEqual(result, (None, None))
|
|
|
|
|
|
Bucket = namedtuple("BucketConfig", ["history", "period", "max_hist_len"])
|
|
|
|
|
|
class TestRateLimiting(unittest.TestCase):
|
|
def setUp(self):
|
|
# just for assertions
|
|
self.called = {
|
|
RATE_GLOBAL: [],
|
|
}
|
|
|
|
def say(self, msg, rate_class=RATE_GLOBAL):
|
|
if rate_limit(rate_class):
|
|
self.called[rate_class].append(msg)
|
|
# print(msg)
|
|
time.sleep(0.1)
|
|
|
|
def test_simple_burst(self):
|
|
messages = ["x_%d" % i for i in range(1, 9)]
|
|
for m in messages:
|
|
self.say(msg=m)
|
|
self.assertEqual(messages[:buckets[RATE_GLOBAL].max_hist_len], self.called[RATE_GLOBAL])
|
|
|
|
def test_msg_two_bursts(self):
|
|
# custom bucket, just for testing
|
|
buckets[0x42] = Bucket(history=[], period=1, max_hist_len=5)
|
|
self.called[0x42] = []
|
|
|
|
bucket = buckets[0x42]
|
|
messages = ["x_%d" % i for i in range(0, 15)]
|
|
for i, m in enumerate(messages):
|
|
if i % bucket.max_hist_len == 0:
|
|
time.sleep(bucket.period)
|
|
self.say(msg=m, rate_class=0x42)
|
|
self.assertEqual(messages, self.called[0x42])
|
|
|
|
|
|
class TestPlugins(unittest.TestCase):
|
|
def setUp(self):
|
|
import config
|
|
self.bot_nickname = config.conf_get("bot_nickname")
|
|
pass
|
|
|
|
def shortDescription(self):
|
|
return None
|
|
|
|
def test_all_commands_no_message(self):
|
|
"""
|
|
By specification, currently not possible! should be revisited after settings words[2:].
|
|
Using '' as empty first word as workaround
|
|
"""
|
|
import plugins
|
|
plugin_functions = filter(
|
|
lambda x: x[0].startswith("command") and x[1].plugin_type == 'command',
|
|
plugins.__dict__.items()
|
|
)
|
|
|
|
msg_obj = {
|
|
'body': '',
|
|
'mucnick': 'hans'
|
|
}
|
|
|
|
plugin_arguments = {
|
|
'msg_obj': msg_obj,
|
|
'reply_user': msg_obj['mucnick'],
|
|
'data': msg_obj['body'],
|
|
'cmd_list': [],
|
|
'parser_list': [],
|
|
}
|
|
|
|
for p in plugin_functions:
|
|
func = p[1]
|
|
result = func([''], **plugin_arguments)
|
|
self.assertEqual(result, None)
|
|
|
|
def test_all_commands_garbage(self):
|
|
"""
|
|
message is rubbish
|
|
:return:
|
|
"""
|
|
import plugins
|
|
plugin_functions = filter(
|
|
lambda x: x[0].startswith("command") and x[1].plugin_type == 'command',
|
|
plugins.__dict__.items()
|
|
)
|
|
|
|
msg_obj = {
|
|
'body': 'urlbot: dewzd hweufweufgw ufzgwufgw ezfgweuf guwegf',
|
|
'mucnick': 'hans'
|
|
}
|
|
|
|
plugin_arguments = {
|
|
'msg_obj': msg_obj,
|
|
'reply_user': msg_obj['mucnick'],
|
|
'data': msg_obj['body'],
|
|
'cmd_list': [],
|
|
'parser_list': [],
|
|
}
|
|
words = msg_obj['body'].split()[1:]
|
|
|
|
for p in plugin_functions:
|
|
func = p[1]
|
|
result = func(words, **plugin_arguments)
|
|
self.assertEqual(result, None)
|
|
|
|
def test_all_commands_with_command(self):
|
|
"""
|
|
Call plugins with their name, expect None or some action dict
|
|
"""
|
|
import plugins
|
|
plugin_functions = filter(
|
|
lambda x: x[0].startswith("command") and x[1].plugin_type == 'command',
|
|
plugins.__dict__.items()
|
|
)
|
|
|
|
for p in plugin_functions:
|
|
func = p[1]
|
|
msg_obj = {
|
|
'body': '{}: {}'.format(self.bot_nickname, func.plugin_name),
|
|
'mucnick': 'hans'
|
|
}
|
|
|
|
plugin_arguments = {
|
|
'msg_obj': msg_obj,
|
|
'reply_user': msg_obj['mucnick'],
|
|
'data': msg_obj['body'],
|
|
'cmd_list': [],
|
|
'parser_list': [],
|
|
}
|
|
words = msg_obj['body'].split()[1:]
|
|
|
|
result = func(words, **plugin_arguments)
|
|
|
|
import inspect
|
|
source = inspect.getsourcelines(func)[0][2:10]
|
|
|
|
# assert that check on the right name as long as we're doing that.
|
|
self.assertTrue(func.plugin_name in ''.join(source), '{} not in {}'.format(func.plugin_name, source))
|
|
|
|
self.assertTrue(result is None or isinstance(result, dict))
|
|
if result:
|
|
self.assertTrue(any(['msg' in result, 'event' in result, 'presence' in result]))
|
|
if 'event' in result:
|
|
self.assertTrue(any(['msg' in result['event'], 'command' in result['event']]))
|
|
if 'presence' in result:
|
|
self.assertTrue(all(['msg' in result['presence'], 'status' in result['presence']]))
|
|
|
|
def test_commands_with_params(self):
|
|
"""
|
|
Test known commands with params so that
|
|
they don't return immidiately
|
|
"""
|
|
|
|
import plugins, config
|
|
plugin_functions = filter(
|
|
lambda x: hasattr(x[1], 'plugin_type') and x[1].plugin_type == 'command',
|
|
plugins.__dict__.items()
|
|
)
|
|
|
|
# fixture
|
|
config.runtime_config_store['other_bots'].append('pork')
|
|
|
|
for p in plugin_functions:
|
|
func = p[1]
|
|
msg_obj = {
|
|
'body': '{}: {}'.format(self.bot_nickname, func.plugin_name),
|
|
'mucnick': config.conf_get('bot_owner')
|
|
}
|
|
|
|
plugin_arguments = {
|
|
'msg_obj': msg_obj,
|
|
'reply_user': msg_obj['mucnick'],
|
|
'data': msg_obj['body'],
|
|
'cmd_list': [],
|
|
'parser_list': [],
|
|
}
|
|
words = msg_obj['body'].split()[1:]
|
|
param_mapping = {
|
|
'plugin': ['disable', 'cake'],
|
|
'wp': ['Wikipedia'],
|
|
'wp-en': ['Wikipedia'],
|
|
'choose': ['A', 'B'],
|
|
'decode': ['@'],
|
|
'record': ['urlbug', 'this', 'is', 'sparta'],
|
|
'flausch': ['hans'],
|
|
'set-status': ['unmute'],
|
|
'remove-from-botlist': ['pork'],
|
|
|
|
}
|
|
|
|
if func.plugin_name in param_mapping:
|
|
words.extend(param_mapping[func.plugin_name])
|
|
|
|
result = func(words, **plugin_arguments)
|
|
|
|
import inspect
|
|
source = inspect.getsourcelines(func)[0][2:10]
|
|
|
|
# assert that check on the right name as long as we're doing that.
|
|
self.assertTrue(func.plugin_name in ''.join(source), '{} not in {}'.format(func.plugin_name, source))
|
|
|
|
print(func.plugin_name, result)
|
|
self.assertIsInstance(result, dict, msg=func.plugin_name)
|
|
if result:
|
|
self.assertTrue(any(['msg' in result, 'event' in result, 'presence' in result]))
|
|
if 'event' in result:
|
|
self.assertTrue(any(['msg' in result['event'], 'command' in result['event']]))
|
|
if 'presence' in result:
|
|
self.assertTrue(all(['msg' in result['presence'], 'status' in result['presence']]))
|
|
|
|
def test_teatimer(self):
|
|
from plugins.commands import command_teatimer
|
|
result = command_teatimer(['teatimer'], reply_user='hans')
|
|
self.assertIn('event', result)
|
|
self.assertIn('time', result['event'])
|
|
self.assertIn('msg', result['event'])
|
|
self.assertIn('msg', result)
|
|
|
|
def test_botlist(self):
|
|
import config
|
|
|
|
def test_in_actual_file(filename):
|
|
with open(filename) as f:
|
|
filecontent = str(f.read())
|
|
self.assertIn('DERPDERP', filecontent)
|
|
|
|
def test_in_file():
|
|
with tempfile.NamedTemporaryFile() as f:
|
|
filename = config.runtime_config_store.filename
|
|
config.runtime_config_store.write(outfile=f)
|
|
config.runtime_config_store.filename = filename
|
|
f.seek(0)
|
|
filecontent = str(f.read())
|
|
self.assertIn('DERPDERP', filecontent)
|
|
|
|
def test_in_memory():
|
|
self.assertEqual(config.runtime_config_store['other_bots'], ['DERPDERP'])
|
|
|
|
def test_in_write():
|
|
filename = config.runtime_config_store.filename
|
|
filecontent = config.runtime_config_store.write()
|
|
config.runtime_config_store.filename = filename
|
|
self.assertIn('DERPDERP', '\n'.join(filecontent))
|
|
|
|
orig_filename = config.runtime_config_store.filename
|
|
config.runtime_config_store.filename = None
|
|
|
|
# empty the botlist
|
|
config.runtime_config_store['other_bots'] = []
|
|
with open(orig_filename, 'wb') as f:
|
|
f.write(b'')
|
|
|
|
# only append to the dict, asserting it's in memory
|
|
self.assertEqual(config.runtime_config_store['other_bots'], [])
|
|
config.runtime_config_store['other_bots'].append('DERPDERP')
|
|
test_in_memory()
|
|
test_in_write()
|
|
|
|
# using the setter (with write usage), assert it's in the file
|
|
# config.runtimeconf_set('other_bots', ['DERPDERP'])
|
|
test_in_memory()
|
|
test_in_write()
|
|
test_in_file()
|
|
|
|
# reloading the config, assuming in memory AND file
|
|
test_in_memory()
|
|
test_in_write()
|
|
test_in_file()
|
|
|
|
config.runtime_config_store.filename = orig_filename
|
|
config.runtime_config_store.write()
|
|
|
|
test_in_actual_file(orig_filename)
|
|
|
|
def tearDown(self):
|
|
import config
|
|
if 'DERPDERP' in config.runtime_config_store['other_bots']:
|
|
config.runtime_config_store['other_bots'].remove('DERPDERP')
|
|
config.runtime_config_store.write()
|