zev/mailbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wip: register, mailbot

Browse Source
This commit is contained in:
Mathieu Agopian
2013-03-15 17:19:49 +01:00
parent 473c2f81a8
commit 56e0511490
6 changed files with 209 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -13,3 +13,4 @@ docs/build/
.coverage
.tox/
build/
share/

8
mailbot/callback.py
View File

@@ -3,3 +3,11 @@
class Callback(object):
"""Base class for callbacks."""
def __init__(self, message, rules):
self.message = message
self.rules = rules
def check_rules(self):
"""Does this message conform to the rules provided?"""
raise NotImplementedError

50
mailbot/mailbot.py
View File

@@ -1,5 +1,53 @@
# -*- coding: utf-8 -*-
from email import message_from_string
from imapclient import IMAPClient
class MailBot(object):
"""MailBot mail class, where the magic is happening."""
"""MailBot mail class, where the magic is happening.
Connect to the SMTP server using the IMAP protocol, for each unflagged
message check which callbacks should be triggered, if any, but testing
against the registered rules for each of them.
"""
imapclient = IMAPClient
message_constructor = message_from_string # easier for testing
def __init__(self, host, username, password, port=None, use_uid=True,
ssl=False, stream=False):
self.client = self.imapclient(host, port=port, use_uid=use_uid,
ssl=ssl, stream=stream)
self.client.login(username, password)
def get_message_ids(self):
"""Return the list of IDs of messages to process."""
return self.client.search(['UNFLAGGED'])
def get_messages(self):
"""Return the list of messages to process."""
ids = self.get_message_ids()
return self.client.fetch(ids, ['RFC822'])
def process_message(self, message, callback_class, rules):
"""Check if callback matches rules, and if so, trigger."""
callback = callback_class(message, rules)
if callback.check_rules():
return callback.callback()
def process_messages(self):
"""Process messages: check which callbacks should be triggered."""
from . import CALLBACKS_MAP
messages = self.get_messages()
for uid, msg in messages.items():
message = self.message_constructor(msg['RFC822'])
for callback_class, rules in CALLBACKS_MAP.items():
self.process_message(message, callback_class, rules)
self.mark_processed(uid)
def mark_processed(self, uid):
"""Mark the message corresponding to uid as processed."""
self.client.set_flags([uid], ['FLAGGED'])

16
mailbot/tests/test_callback.py Normal file
View File

@@ -0,0 +1,16 @@
# -*- coding: utf-8 -*-
from . import MailBotTestCase
from .. import Callback
class CallbackTest(MailBotTestCase):
def test_init(self):
callback = Callback('foo', 'bar')
self.assertEqual(callback.message, 'foo')
self.assertEqual(callback.rules, 'bar')
def test_check_rules(self):
callback = Callback('foo', 'bar')
self.assertEqual(callback.check_rules(), False)

43
mailbot/tests/test_init.py Normal file
View File

@@ -0,0 +1,43 @@
# -*- coding: utf-8 -*-
from . import MailBotTestCase
from .. import register, CALLBACKS_MAP, RegisterException
class RegisterTest(MailBotTestCase):
def setUp(self):
super(RegisterTest, self).setUp()
class EmptyCallback(object):
pass
class WithRulesCallback(object):
rules = {'foo': 'bar', 'baz': 'bat'}
self.empty_callback = EmptyCallback
self.with_rules_callback = WithRulesCallback
def test_register(self):
before = len(CALLBACKS_MAP)
register(self.empty_callback)
self.assertEqual(len(CALLBACKS_MAP), before + 1)
def test_register_existing(self):
register(self.empty_callback)
self.assertRaises(RegisterException, register, self.empty_callback)
self.assertTrue(register(self.with_rules_callback))
def test_register_without_rules_callback_with_rules(self):
register(self.with_rules_callback)
self.assertEqual(CALLBACKS_MAP[self.with_rules_callback],
self.with_rules_callback.rules)
def test_register_with_rules_callback_without_rules(self):
register(self.empty_callback, {'one': 'two'})
self.assertEqual(CALLBACKS_MAP[self.empty_callback], {'one': 'two'})
def test_register_with_rules_callback_with_rules(self):
register(self.with_rules_callback, {'baz': 'wow'})
self.assertEqual(CALLBACKS_MAP[self.with_rules_callback],
{'foo': 'bar', 'baz': 'wow'})

120
mailbot/tests/test_mailbot.py
View File

@@ -1,43 +1,107 @@
# -*- coding: utf-8 -*-
from mock import patch, sentinel, Mock, DEFAULT, call
from . import MailBotTestCase
from .. import register, CALLBACKS_MAP, RegisterException
from .. import CALLBACKS_MAP, MailBot
class RegisterTest(MailBotTestCase):
class TestableMailBot(MailBot):
def __init__(self, *args, **kwargs):
self.client = Mock()
class MailBotClientTest(MailBotTestCase):
def setUp(self):
super(RegisterTest, self).setUp()
super(MailBotClientTest, self).setUp()
self.bot = TestableMailBot('somehost', 'john', 'doe')
class EmptyCallback(object):
pass
class WithRulesCallback(object):
rules = {'foo': 'bar', 'baz': 'bat'}
class MailBotTest(MailBotClientTest):
self.empty_callback = EmptyCallback
self.with_rules_callback = WithRulesCallback
@patch.multiple('imapclient.imapclient.IMAPClient',
login=DEFAULT, __init__=DEFAULT)
def test_init(self, login, __init__):
__init__.return_value = None
def test_register(self):
before = len(CALLBACKS_MAP)
register(self.empty_callback)
self.assertEqual(len(CALLBACKS_MAP), before + 1)
kwargs = {'port': sentinel.port,
'ssl': sentinel.ssl,
'use_uid': sentinel.use_uid,
'stream': sentinel.use_stream}
MailBot('somehost', 'john', 'doe', **kwargs)
def test_register_existing(self):
register(self.empty_callback)
self.assertRaises(RegisterException, register, self.empty_callback)
self.assertTrue(register(self.with_rules_callback))
__init__.assert_called_once_with('somehost', **kwargs)
login.assert_called_once_with('john', 'doe')
def test_register_without_rules_callback_with_rules(self):
register(self.with_rules_callback)
self.assertEqual(CALLBACKS_MAP[self.with_rules_callback],
self.with_rules_callback.rules)
def test_get_message_ids(self):
self.bot.client.search.return_value = sentinel.id_list
def test_register_with_rules_callback_without_rules(self):
register(self.empty_callback, {'one': 'two'})
self.assertEqual(CALLBACKS_MAP[self.empty_callback], {'one': 'two'})
res = self.bot.get_message_ids()
def test_register_with_rules_callback_with_rules(self):
register(self.with_rules_callback, {'baz': 'wow'})
self.assertEqual(CALLBACKS_MAP[self.with_rules_callback],
{'foo': 'bar', 'baz': 'wow'})
self.bot.client.search.assert_called_once_with(['UNFLAGGED'])
self.assertEqual(res, sentinel.id_list)
def test_get_messages(self):
self.bot.get_message_ids = Mock(return_value=sentinel.ids)
self.bot.client.fetch.return_value = sentinel.message_list
messages = self.bot.get_messages()
self.bot.get_message_ids.assert_called_once_with()
self.bot.client.fetch.assert_called_once_with(sentinel.ids, ['RFC822'])
self.assertEqual(messages, sentinel.message_list)
def test_process_message_trigger(self):
callback = Mock()
callback.check_rules.return_value = True
callback.callback.return_value = sentinel.callback_result
callback_class = Mock(return_value=callback)
res = self.bot.process_message(sentinel.message, callback_class,
sentinel.rules)
callback_class.assert_called_once_with(sentinel.message,
sentinel.rules)
callback.check_rules.assert_called_once_with()
callback.callback.assert_called_once_with()
self.assertEqual(res, sentinel.callback_result)
def test_process_message_no_trigger(self):
callback = Mock()
callback.check_rules.return_value = False
callback_class = Mock(return_value=callback)
res = self.bot.process_message(sentinel.message, callback_class,
sentinel.rules)
callback.check_rules.assert_called_once_with()
self.assertEqual(res, None)
def test_process_messages(self):
messages = {1: {'RFC822': sentinel.mail1},
2: {'RFC822': sentinel.mail2}}
self.bot.get_messages = Mock(return_value=messages)
# message constructor will return exactly what it's given
# to be used in the "self.bot.process_message.assert_has_calls" below
self.bot.message_constructor = Mock(side_effect=lambda m: m)
self.bot.process_message = Mock()
self.bot.mark_processed = Mock()
CALLBACKS_MAP.update({sentinel.callback1: sentinel.rules1,
sentinel.callback2: sentinel.rules2})
self.bot.process_messages()
self.bot.get_messages.assert_called_once_with()
self.bot.process_message.assert_has_calls(
[call(sentinel.mail1, sentinel.callback1, sentinel.rules1),
call(sentinel.mail2, sentinel.callback1, sentinel.rules1),
call(sentinel.mail1, sentinel.callback2, sentinel.rules2),
call(sentinel.mail2, sentinel.callback2, sentinel.rules2)],
any_order=True)
def test_mark_processed(self):
self.bot.mark_processed(sentinel.id)
self.bot.client.set_flags.assert_called_once_with([sentinel.id],
['FLAGGED'])