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

Better Callback.matches: give direct access to captures. Better livetest

Browse Source
This commit is contained in:
Mathieu Agopian
2013-03-26 11:04:31 +01:00
parent c4f46f3fed
commit 6a961f72ef
5 changed files with 123 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
docs/source/index.rst
View File

@@ -98,7 +98,7 @@ Providing the rules as a parameter
Here's a callback that will only be triggered if the subject matches the
pattern 'Hello ' followed by a word, anywhere in the subject (it uses
``re.search``):
``re.findall``):
.. code-block:: python
@@ -172,25 +172,29 @@ applied, and emails from any sender will potentially trigger the callback.
For each piece of data (subject, from, to, cc, body), the callback class,
once instantiated with the mail, and the ``check_rules`` method called, will
have the attribute ``self.matches[item]`` set with all the matches from the
given patterns, if any. Matches will in fact be ``re.MatchObject``.
have the attribute ``self.matches[item]`` set with all the captures from the
given patterns, if any, or the full match.
Here are example subjects for the subject rules:
[``r'^Hello (\w+), (.*)'``, ``r'[Hh]i (\w+)``]
[``r'Hello (\w+), (.*)'``, ``r'[Hh]i (\w+)``]
For each of the following examples, ``self.matches['subject']`` will be a list
of two ``re.MatchObject``, one for each regular expression.
of all the captures for all the regular expressions.
If a regular expression doesn't match, then it'll return ``None``.
If a regular expression doesn't match, then it'll return an empty list.
For each example subject, a ``re.MatchObject`` will be represented by its
matching groups:
* 'Hello Bryan, how are you?': [('Bryan', 'how are you?')]
* 'Hi Bryan, how are you?': ['Bryan']
* 'aloha, hi Bryan!': ['Bryan']
* 'aloha Bryan': rules not respected, callback not triggered, []
* 'Hello Bryan, how are you?':
[['Hello Bryan, how are you?', 'Bryan', 'how are you?'], None]
* 'Hi Bryan, how are you?': [None, ['Hi Bryan', 'Bryan']]
* 'aloha, hi Bryan!': [None, ['hi Bryan', 'Bryan']]
* 'aloha Bryan': rules not respected, callback not triggered, [None, None]
Here are example subjects for the subject rules (no captures):
[``r'Hello \w+'``, ``r'[Hh]i \w+``]
* 'Hello Bryan, how are you?': ['Hello Bryan']
* 'Hi Bryan, how are you?': ['Hi Bryan']
* 'aloha, hi Bryan!': ['hi Bryan']
* 'aloha Bryan': rules not respected, callback not triggered, []
Rules checking

8
mailbot/callback.py
View File

@@ -1,13 +1,14 @@
# -*- coding: utf-8 -*-
from re import search
from collections import defaultdict
from re import findall
class Callback(object):
"""Base class for callbacks."""
matches = {}
def __init__(self, message, rules):
self.matches = defaultdict(list)
self.message = message
self.rules = rules
@@ -48,7 +49,8 @@ class Callback(object):
# if item is not in header, then item == 'body'
value = message.get(item, self.get_email_body(message))
self.matches[item] = [search(regexp, value) for regexp in regexps]
for regexp in regexps: # store all captures for easy access
self.matches[item] += findall(regexp, value)
return any(self.matches[item])

124
mailbot/livetests/test_mail_received.py
View File

@@ -99,73 +99,85 @@ class MailReceivedTest(MailBotTestCase):
email = open(email_file, 'r').read()
self.mb.client.append(self.home_folder, email)
class MatchingCallback(Callback):
"""Callback with each rule matching the test mail.
# Callback with each rule matching the test mail
# Each rule contains a non matching regexp, which shouldn't prevent the
# callback from being triggered
matching_rules = {
'subject': [r'Task name \w+', r'Task name (\w+)', 'NOMATCH'],
'to': [r'\w+\+\w+@example.com', r'(\w+)\+(\w+)@example.com',
'NOMATCH'],
'cc': [r'\w+@example.com', r'(\w+)@example.com', 'NOMATCH'],
'from': [r'\w+\.\w+@example.com', r'(\w+)\.(\w+)@example.com',
'NOMATCH'],
'body': [r'Mail content \w+', r'Mail content (\w+)',
'NOMATCH']}
Each rule contains a non matching regexp, which shouldn't prevent
the callback from being triggered
# Callback with each rule but one matching the test mail.
# To prevent the callback from being triggered, at least one rule must
# completely fail (have 0 regexp that matches).
failing_rules = {
'subject': [r'Task name \w+', r'Task name (\w+)', 'NOMATCH'],
'to': [r'\w+\+\w+@example.com', r'(\w+)\+(\w+)@example.com',
'NOMATCH'],
'cc': [r'\w+@example.com', r'(\w+)@example.com', 'NOMATCH'],
'from': [r'\w+\.\w+@example.com', r'(\w+)\.(\w+)@example.com',
'NOMATCH'],
'body': ['NOMATCH', 'DOESNT MATCH EITHER']} # this rule fails
"""
rules = {
'subject': [r'Task name \w+', r'Task name (\w+)', 'NOMATCH'],
'to': [r'\w+\+\w+@gmail.com', r'(\w+)\+(\w+)@gmail.com',
'NOMATCH'],
'from': [r'\w+\.\w+@gmail.com', r'(\w+)\.(\w+)@gmail.com',
'NOMATCH'],
'body': [r'Mail content \w+', r'Mail content (\w+)',
'NOMATCH']}
class TestCallback(Callback):
def __init__(self, message, rules):
super(TestCallback, self).__init__(message, rules)
self.called = False
self.check_rules_result = False
self.triggered = False
def check_rules(self):
res = super(MatchingCallback, self).check_rules()
assert res, "Matching callback check_rules returned False"
res = super(TestCallback, self).check_rules()
self.called = True
self.check_rules_result = res
return res
def trigger(self):
m = self.matches['subject']
assert len(m) == 3
assert m[0].group(0) == 'Task name here'
assert m[1].group(1) == 'here'
assert m[2] is None
self.triggered = True
m = self.matches['to']
assert len(m) == 3
assert m[0].group(0) == 'testmagopian+RANDOM_KEY@gmail.com'
assert m[1].group(1) == 'testmagopian'
assert m[1].group(2) == 'RANDOM_KEY'
assert m[2] is None
matching_callback = TestCallback(message_from_string(email),
matching_rules)
m = self.matches['from']
assert len(m) == 3
assert m[0].group(0) == 'mathieu.agopian@gmail.com'
assert m[1].group(1) == 'mathieu'
assert m[1].group(2) == 'agopian'
assert m[2] is None
def make_matching_callback(email, rules):
return matching_callback
m = self.matches['body']
assert len(m) == 3
assert m[0].group(0) == 'Mail content here'
assert m[1].group(1) == 'here'
assert m[2] is None
failing_callback = TestCallback(message_from_string(email),
failing_rules)
class NonMatchingCallback(Callback):
"""Callback with each rule but one matching the test mail.
def make_failing_callback(email, rules):
return failing_callback
To prevent the callback from being triggered, at least one rule
must completely fail (have 0 regexp that matches).
"""
rules = { # only difference is that one rule doesn't match
'subject': [r'Task name \w+', r'Task name (\w+)', 'NOMATCH'],
'to': [r'\w+\+\w+@gmail.com', r'(\w+)\+(\w+)@gmail.com',
'NOMATCH'],
'from': [r'\w+\.\w+@gmail.com', r'(\w+)\.(\w+)@gmail.com',
'NOMATCH'],
'body': ['NOMATCH', 'DOESNT MATCH EITHER']}
def trigger(self):
assert False, "Non matching callback has been triggered"
register(MatchingCallback)
register(NonMatchingCallback)
register(make_matching_callback, matching_rules)
register(make_failing_callback, failing_rules)
self.mb.process_messages()
self.assertTrue(matching_callback.called)
self.assertTrue(matching_callback.check_rules_result)
self.assertTrue(matching_callback.triggered)
self.assertEqual(matching_callback.matches['subject'],
['Task name here', 'here'])
self.assertEqual(matching_callback.matches['from'],
['foo.bar@example.com', ('foo', 'bar')])
self.assertEqual(matching_callback.matches['to'],
['foo+RANDOM_KEY@example.com',
'bar+RANDOM_KEY_2@example.com',
('foo', 'RANDOM_KEY'),
('bar', 'RANDOM_KEY_2')])
self.assertEqual(matching_callback.matches['cc'],
['foo@example.com',
'bar@example.com',
'foo', 'bar'])
self.assertEqual(matching_callback.matches['body'],
['Mail content here', 'here'])
self.assertTrue(failing_callback.called)
self.assertFalse(failing_callback.check_rules_result)
self.assertFalse(failing_callback.triggered)
self.assertEqual(failing_callback.matches['body'], [])

19
mailbot/tests/mails/mail_with_attachment.txt
View File

@@ -1,16 +1,16 @@
Delivered-To: testmagopian+random_key@gmail.com
Delivered-To: foo+RANDOM_KEY@example.com
Received: by 10.194.34.7 with SMTP id v7csp101053wji;
Fri, 15 Mar 2013 02:28:52 -0700 (PDT)
Return-Path: <mathieu.agopian@gmail.com>
Received-SPF: pass (google.com: domain of mathieu.agopian@gmail.com designates 10.182.31.109 as permitted sender) client-ip=10.182.31.109
Return-Path: <foo+RANDOM_KEY@example.com>
Received-SPF: pass (example.com: domain of foo+RANDOM_KEY@example.com designates 1.2.3.4 as permitted sender) client-ip=1.2.3.4
Authentication-Results: mr.google.com;
spf=pass (google.com: domain of mathieu.agopian@gmail.com designates 10.182.31.109 as permitted sender) smtp.mail=mathieu.agopian@gmail.com;
dkim=pass header.i=@gmail.com
spf=pass (example.com: domain of foo+RANDOM_KEY@example.com designates 1.2.3.4 as permitted sender) smtp.mail=foo+RANDOM_KEY@example.com;
dkim=pass header.i=@example.com
X-Received: from mr.google.com ([10.182.31.109])
by 10.182.31.109 with SMTP id z13mr2632031obh.37.1363339731787 (num_hops = 1);
Fri, 15 Mar 2013 02:28:51 -0700 (PDT)
DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed;
d=gmail.com; s=20120113;
d=example.com; s=20120113;
h=mime-version:x-received:date:message-id:subject:from:to
:content-type;
bh=EDdIiN1bkSUqRxA5ZGCbAxWo/K7ayqdf9ZDEQqAGvDU=;
@@ -25,10 +25,11 @@ X-Received: by 10.182.31.109 with SMTP id z13mr2632031obh.37.1363339731783;
Fri, 15 Mar 2013 02:28:51 -0700 (PDT)
Received: by 10.182.98.129 with HTTP; Fri, 15 Mar 2013 02:28:51 -0700 (PDT)
Date: Fri, 15 Mar 2013 10:28:51 +0100
Message-ID: <CAB-JLVBXqYpS1GzujSAopk3cS1Xo8C8A+bQew0_jkOpAJu1pFw@mail.gmail.com>
Message-ID: <CAB-JLVBXqYpS1GzujSAopk3cS1Xo8C8A+bQew0_jkOpAJu1pFw@mail.example.com>
Subject: Task name here
From: Mathieu AGOPIAN <mathieu.agopian@gmail.com>
To: testmagopian+RANDOM_KEY@gmail.com
From: Foo Bar <foo.bar@example.com>
To: foo+RANDOM_KEY@example.com, bar+RANDOM_KEY_2@example.com
Cc: foo@example.com, bar@example.com
Content-Type: multipart/mixed; boundary=14dae93b5c806bd71504d7f3442a
--14dae93b5c806bd71504d7f3442a

37
mailbot/tests/test_callback.py
View File

@@ -2,7 +2,6 @@
from email import message_from_file, message_from_string
from os.path import dirname, join
from re import search
from mock import Mock
@@ -42,7 +41,7 @@ class CallbackTest(MailBotTestCase):
callback.rules = {'foo': True, 'bar': ['test'], 'baz': 'barf'}
self.assertEqual(callback.check_rules(), True)
def test_check_item(self):
def test_check_item_non_existent(self):
empty = message_from_string('')
callback = Callback(empty, {})
@@ -50,28 +49,38 @@ class CallbackTest(MailBotTestCase):
self.assertEqual(callback.check_item('foobar', ['.*'], empty), None)
self.assertEqual(callback.check_item('foobar', ['(.*)']), None)
# test on real mail
def test_check_item_subject(self):
email_file = join(dirname(__file__), 'mails/mail_with_attachment.txt')
email = message_from_file(open(email_file, 'r'))
callback = Callback(email, {})
# subject
self.assertFalse(callback.check_item('subject', []))
self.assertEqual(callback.matches['subject'], [])
self.assertTrue(callback.check_item('subject', ['(.*)']))
self.assertEqual(callback.matches['subject'][0].groups(),
search('(.*)', 'Task name here').groups())
self.assertFalse(callback.check_item('subject', ['foo']))
self.assertEqual(callback.matches['subject'], [])
# body
self.assertTrue(callback.check_item('subject', ['Task name (.*)']))
self.assertEqual(callback.matches['subject'], ['here'])
def test_check_item_to(self):
# "to" may be a list of several emails
email_file = join(dirname(__file__), 'mails/mail_with_attachment.txt')
email = message_from_file(open(email_file, 'r'))
callback = Callback(email, {})
self.assertTrue(callback.check_item('to', [r'\+([^@]+)@']))
self.assertEqual(callback.matches['to'],
['RANDOM_KEY', 'RANDOM_KEY_2'])
def test_check_item_body(self):
email_file = join(dirname(__file__), 'mails/mail_with_attachment.txt')
email = message_from_file(open(email_file, 'r'))
callback = Callback(email, {})
callback.get_email_body = Mock(return_value='some mail body')
self.assertFalse(callback.check_item('body', []))
self.assertEqual(callback.matches['body'], [])
callback.get_email_body.assert_called_once_with(email)
self.assertTrue(callback.check_item('body', ['(.*)']))
self.assertEqual(callback.matches['body'][0].groups(),
search('(.*)', 'some mail body').groups())
self.assertTrue(callback.check_item('body', ['.+']))
self.assertEqual(callback.matches['body'], ['some mail body'])
def test_get_email_body(self):
callback = Callback('foo', 'bar')