Allura/allura/lib/spam/akismetservice.py to Allura/allura/lib/spam/akismetfilter.py
--- a/Allura/allura/lib/spam/akismetservice.py +++ b/Allura/allura/lib/spam/akismetfilter.py @@ -4,12 +4,32 @@ from pylons import tmpl_context as c from allura.lib import helpers as h +from allura.lib.spam import SpamFilter import akismet + log = logging.getLogger(__name__) -class Akismet(akismet.Akismet): + +class AkismetSpamFilter(SpamFilter): + """Spam checking implementation via Akismet service. + + To enable Akismet spam filtering in your Allura instance, first + enable the entry point in setup.py:: + + [allura.spam] + akismet = allura.lib.spam.akismetfilter:AkismetSpamFilter + + Then include the following parameters in your .ini file:: + + spam.method = akismet + spam.key = <your Akismet key here> + """ + def __init__(self, config): + self.service = akismet.Akismet(config.get('spam.key'), config.get('base_url')) + self.service.verify_key() + def check(self, text, artifact=None, user=None, content_type='comment', **kw): log_msg = text kw['comment_content'] = text @@ -28,6 +48,6 @@ # kw will be urlencoded, need to utf8-encode for k, v in kw.items(): kw[k] = h.really_unicode(v).encode('utf8') - res = self.comment_check(text, data=kw, build_data=False) + res = self.service.comment_check(text, data=kw, build_data=False) log.info("spam=%s (akismet): %s" % (str(res), log_msg)) return res
Allura/allura/tests/unit/test_spam.py to Allura/allura/tests/unit/spam/test_akismet.py
--- a/Allura/allura/tests/unit/test_spam.py +++ b/Allura/allura/tests/unit/spam/test_akismet.py @@ -5,20 +5,21 @@ import urllib try: - from allura.lib.spam.akismetservice import Akismet + from allura.lib.spam.akismetfilter import AkismetSpamFilter except ImportError: - Akismet = None + AkismetSpamFilter = None -@unittest.skipIf(Akismet is None, "Can't import Akismet") +@unittest.skipIf(AkismetSpamFilter is None, "Can't import AkismetSpamFilter") class TestAkismet(unittest.TestCase): - def setUp(self): - self.akismet = Akismet() + @mock.patch('allura.lib.spam.akismetfilter.akismet') + def setUp(self, akismet_lib): + self.akismet = AkismetSpamFilter({}) def side_effect(*args, **kw): # side effect to test that data being sent to # akismet can be successfully urlencoded urllib.urlencode(kw.get('data', {})) - self.akismet.comment_check = mock.Mock(side_effect=side_effect) + self.akismet.service.comment_check = mock.Mock(side_effect=side_effect) self.fake_artifact = mock.Mock(**{'url.return_value': 'artifact url'}) self.fake_user = mock.Mock(display_name=u'S��me User', email_addresses=['user@domain']) @@ -35,38 +36,38 @@ user_agent='some browser', referrer='some url') - @mock.patch('allura.lib.spam.akismetservice.c') - @mock.patch('allura.lib.spam.akismetservice.request') + @mock.patch('allura.lib.spam.akismetfilter.c') + @mock.patch('allura.lib.spam.akismetfilter.request') def test_check(self, request, c): request.headers = self.fake_headers c.user = None self.akismet.check(self.content) - self.akismet.comment_check.assert_called_once_with(self.content, + self.akismet.service.comment_check.assert_called_once_with(self.content, data=self.expected_data, build_data=False) - @mock.patch('allura.lib.spam.akismetservice.c') - @mock.patch('allura.lib.spam.akismetservice.request') + @mock.patch('allura.lib.spam.akismetfilter.c') + @mock.patch('allura.lib.spam.akismetfilter.request') def test_check_with_explicit_content_type(self, request, c): request.headers = self.fake_headers c.user = None self.akismet.check(self.content, content_type='some content type') self.expected_data['comment_type'] = 'some content type' - self.akismet.comment_check.assert_called_once_with(self.content, + self.akismet.service.comment_check.assert_called_once_with(self.content, data=self.expected_data, build_data=False) - @mock.patch('allura.lib.spam.akismetservice.c') - @mock.patch('allura.lib.spam.akismetservice.request') + @mock.patch('allura.lib.spam.akismetfilter.c') + @mock.patch('allura.lib.spam.akismetfilter.request') def test_check_with_artifact(self, request, c): request.headers = self.fake_headers c.user = None self.akismet.check(self.content, artifact=self.fake_artifact) expected_data = self.expected_data expected_data['permalink'] = 'artifact url' - self.akismet.comment_check.assert_called_once_with(self.content, + self.akismet.service.comment_check.assert_called_once_with(self.content, data=expected_data, build_data=False) - @mock.patch('allura.lib.spam.akismetservice.c') - @mock.patch('allura.lib.spam.akismetservice.request') + @mock.patch('allura.lib.spam.akismetfilter.c') + @mock.patch('allura.lib.spam.akismetfilter.request') def test_check_with_user(self, request, c): request.headers = self.fake_headers c.user = None @@ -74,11 +75,11 @@ expected_data = self.expected_data expected_data.update(comment_author=u'S��me User'.encode('utf8'), comment_author_email='user@domain') - self.akismet.comment_check.assert_called_once_with(self.content, + self.akismet.service.comment_check.assert_called_once_with(self.content, data=expected_data, build_data=False) - @mock.patch('allura.lib.spam.akismetservice.c') - @mock.patch('allura.lib.spam.akismetservice.request') + @mock.patch('allura.lib.spam.akismetfilter.c') + @mock.patch('allura.lib.spam.akismetfilter.request') def test_check_with_implicit_user(self, request, c): request.headers = self.fake_headers c.user = self.fake_user @@ -86,11 +87,11 @@ expected_data = self.expected_data expected_data.update(comment_author=u'S��me User'.encode('utf8'), comment_author_email='user@domain') - self.akismet.comment_check.assert_called_once_with(self.content, + self.akismet.service.comment_check.assert_called_once_with(self.content, data=expected_data, build_data=False) - @mock.patch('allura.lib.spam.akismetservice.c') - @mock.patch('allura.lib.spam.akismetservice.request') + @mock.patch('allura.lib.spam.akismetfilter.c') + @mock.patch('allura.lib.spam.akismetfilter.request') def test_check_with_fallback_ip(self, request, c): self.expected_data['user_ip'] = 'fallback ip' self.fake_headers.pop('X_FORWARDED_FOR') @@ -98,5 +99,5 @@ request.remote_addr = self.fake_headers['REMOTE_ADDR'] c.user = None self.akismet.check(self.content) - self.akismet.comment_check.assert_called_once_with(self.content, + self.akismet.service.comment_check.assert_called_once_with(self.content, data=self.expected_data, build_data=False)