Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Refactoring in progress: move http stuff out of handle_mailin.py and …

…into utils.py. Haven't refactored twitter yet
  • Loading branch information...
commit bd6b0a2ae17e0fed751634c8fba132c28525d707 1 parent 262f54d
Paul Winkler authored
131 fixcity/bmabr/management/commands/handle_mailin.py
View
@@ -18,26 +18,24 @@
from datetime import datetime
from optparse import make_option
-from poster.encode import multipart_encode
from stat import S_IRWXU, S_IRWXG, S_IRWXO
+
import email.Header
-import httplib2
import mimetypes
import os
import re
-import socket
import sys
import tempfile
import time
import traceback
import unicodedata
-import urlparse
from django.core.files.uploadedfile import SimpleUploadedFile
from django.core.mail import send_mail
from django.core.management.base import BaseCommand
-from django.utils import simplejson as json
+from utils import FixcityHttp
+
from django.conf import settings
logger = settings.LOGGER
@@ -384,10 +382,13 @@ def get_photos(self, message_parts):
class RackMaker(object):
- def __init__(self, notifier, options):
+ # XXX Maybe this class doesn't need to exist anymore?
+ # FixcityHttp might be directly usable for email, twitter, etc?
+
+ def __init__(self, notifier, options, error_adapter):
self.notifier = notifier
self.options = options
- self.error_adapter = ErrorAdapter()
+ self.error_adapter = error_adapter
def submit(self, data):
"""
@@ -397,100 +398,13 @@ def submit(self, data):
logger.debug("would save rack here")
return
- photos = data.pop('photos', {})
- url = settings.RACK_POSTING_URL
- result = self.do_post_json(url, data)
- if not result:
- return
+ return FixcityHttp(self.notifier, self.error_adapter).submit(data)
- # Lots of rack-specific stuff below
- parsed_url = urlparse.urlparse(url)
- base_url = parsed_url[0] + '://' + parsed_url[1]
- photo_url = base_url + result['photo_post_url']
- rack_url = base_url + result['rack_url']
- rack_user = result.get('user')
-
- if photos.has_key('photo'):
- datagen, headers = multipart_encode({'photo': photos['photo']})
- # httplib2 doesn't like poster's integer headers.
- headers['Content-Length'] = str(headers['Content-Length'])
- body = ''.join([s for s in datagen])
- result = self.do_post(photo_url, body, headers=headers)
- logger.debug("result from photo upload:")
- logger.debug(result)
- # XXX need to add links per
- # https://projects.openplans.org/fixcity/wiki/EmailText
- # ... will need an HTML version.
- reply = "Thanks for your rack suggestion!\n\n"
- reply += "You must verify that your spot meets DOT requirements\n"
- reply += "before we can submit it.\n"
- reply += "To verify, go to: %(rack_url)sedit/\n\n"
- if not rack_user:
- # XXX Create an inactive account and add a confirmation link.
- reply += "To create an account, go to %(base_url)s/accounts/register/ .\n\n"
- reply += "Thanks!\n\n"
- reply += "- The Open Planning Project & Livable Streets Initiative\n"
- reply = reply % locals()
- self.notifier.reply("FixCity Rack Confirmation", reply)
-
-
- def do_post(self, url, body, headers={}):
- """POST the body to the URL. Returns response body
- on success, or None on failure.
- """
- error_subject = "Unsuccessful Rack Request"
- http = httplib2.Http()
- try:
- response, content = http.request(url, 'POST',
- headers=headers,
- body=body)
- except (socket.error, AttributeError):
- # it's absurd that we have to catch AttributeError here,
- # but apparently that's what httplib2 0.5.0 raises because
- # the socket ends up being None. Lame!
- # Known issue: http://code.google.com/p/httplib2/issues/detail?id=62&can=1&q=AttributeError
- logger.debug('Server down? %r' % url)
- self.notifier.bounce(
- error_subject,
- self.error_adapter.server_error_retry,
- notify_admin='Server down??'
- )
- return None
-
- logger.debug("server %r responded with:\n%s" % (url, content))
-
- if response.status >= 500:
- err_msg = self.error_adapter.server_error_permanent
- self.notifier.bounce(
- error_subject, err_msg, notify_admin='500 Server error',
- notify_admin_body=content)
- return None
- return content
-
- def do_post_json(self, url, data, headers={}):
- """Post some data as json to the given URL.
- Expect the response to be JSON data, and return it decoded.
-
- If the server detects validation errors, it should include an
- 'errors' key in the response data. The value for 'errors'
- should be a mapping of field name to a list of error strings
- for that field. (Not coincidentally, django forms yield
- validation errors in that format.)
- """
- body = json.dumps(data)
- error_subject = "Unsuccessful Rack Request"
- headers.setdefault('Content-type', 'application/json')
- response_body = self.do_post(url, body, headers)
- if response_body:
- result = json.loads(response_body)
- else:
- result = {}
- if result.has_key('errors'):
- err_msg = self.error_adapter.validation_errors(result['errors'])
- self.notifier.bounce(error_subject, err_msg)
- result = None
- return result
+ def do_post(self, *args, **kw):
+ return FixcityHttp(self.notifier, self.error_adapter).do_post(*args, **kw)
+ def do_post_json(self, *args, **kw):
+ return FixcityHttp(self.notifier, self.error_adapter).do_post_json(*args, **kw)
class Notifier(object):
@@ -543,6 +457,25 @@ def notify_admin(self, subject, body):
from_addr = 'racks@fixcity.org'
send_mail(subject, body, from_addr, [admin_email], fail_silently=False)
+ def on_submit_success(self, vars):
+ """
+ Callback that the submitter will use to notify the user of success.
+ """
+ # XXX need to add links per
+ # https://projects.openplans.org/fixcity/wiki/EmailText
+ # ... will need an HTML version.
+ reply = "Thanks for your rack suggestion!\n\n"
+ reply += "You must verify that your spot meets DOT requirements\n"
+ reply += "before we can submit it.\n"
+ reply += "To verify, go to: %(rack_url)sedit/\n\n"
+ if not vars['rack_user']:
+ # XXX Create an inactive account and add a confirmation link.
+ reply += "To create an account, go to %(base_url)s/accounts/register/ .\n\n"
+ reply += "Thanks!\n\n"
+ reply += "- OpenPlans & Livable Streets Initiative\n"
+ reply = reply % vars
+ self.reply("FixCity Rack Confirmation", reply)
+
class ErrorAdapter(object):
72 fixcity/bmabr/management/commands/tweeter.py
View
@@ -21,9 +21,10 @@
class TwitterFetcher(object):
- def __init__(self, twitter_api, username):
+ def __init__(self, twitter_api, username, notifier):
self.twitter_api = twitter_api
self.username = username
+ self.notifier = notifier
def parse(self, tweet):
msg = tweet.text.replace('@' + self.username, '')
@@ -73,12 +74,13 @@ class RackMaker(object):
"format e.g. http://bit.ly/76pXSi and try again "
"or @ us w/questions.")
- def __init__(self, config, api):
+ def __init__(self, config, api, notifier):
self.url = config.RACK_POSTING_URL
self.username = config.TWITTER_USER
self.password = config.TWITTER_PASSWORD
self.twitter_api = api
self.status_file_path = config.TWITTER_STATUS_PATH
+ self.notifier = notifier
def load_last_status(self, recent_only):
last_processed_id = None
@@ -111,7 +113,7 @@ def main(self, recent_only=True):
# Twitter is feeling sad again.
# Let's bail out and hope they're back soon.
return
- twit = TwitterFetcher(self.twitter_api, self.username)
+ twit = TwitterFetcher(self.twitter_api, self.username, self.notifier)
all_tweets = twit.get_tweets(last_processed_id)
for tweet in reversed(all_tweets):
@@ -119,9 +121,9 @@ def main(self, recent_only=True):
user = tweet.user.screen_name
if parsed:
- self.new_rack(**parsed)
+ self.submit(**parsed)
else:
- self.bounce(user, self.general_error_message)
+ self.notifier.bounce(user, self.general_error_message)
# XXX We should lock the status file in case this script
# ever takes so long that it overlaps with the next
# run. Or something.
@@ -150,7 +152,7 @@ def main(self, recent_only=True):
# and notify user if they eventually succeed?
- def new_rack(self, title, address, user, date, tweetid):
+ def submit(self, title, address, user, date, tweetid):
url = self.url
# XXX UGH, copy-pasted from handle_mailin.py. Refactoring time!
description = ''
@@ -173,8 +175,9 @@ def new_rack(self, title, address, user, date, tweetid):
headers=headers,
body=jsondata)
except socket.error:
- _notify_admin('Server down??',
- 'Could not post some tweets, fixcity.org dead?')
+ self.notifier.notify_admin(
+ 'Server down??',
+ 'Could not post some tweets, fixcity.org dead?')
# Important to re-raise here, to prevent storing this tweet's
# ID as the last successfully processed one.
raise
@@ -183,7 +186,7 @@ def new_rack(self, title, address, user, date, tweetid):
# XXX give a URL to a help page w/ more info?
# Maybe even a private URL to a page w/ this user's exact errors?
err_msg = self.general_error_message
- self.bounce(
+ self.notifier.bounce(
user, err_msg,
notify_admin='fixcity: twitter: 500 Server error',
notify_admin_body=content)
@@ -194,18 +197,32 @@ def new_rack(self, title, address, user, date, tweetid):
## errors = adapt_errors(result['errors'])
## for k, v in sorted(errors.items()):
## err_msg += "%s: %s\n" % (k, '; '.join(v))
-
- self.bounce(user, err_msg)
+
+ self.notifier.bounce(user, err_msg)
return
else:
# XXX handle errors from bitly.
shortened_url = shorten_url('%s%s/' % (self.url, result['rack']))
- self.bounce(user,
- "Thank you! Here's the rack request %s; now you can "
- "register to verify your request "
- "http://bit.ly/84Myis" % shortened_url)
+ self.notifier.bounce(
+ user,
+ "Thank you! Here's the rack request %s; now you can register "
+ "to verify your request "
+ % shortened_url)
+
+
+class Notifier(object):
+
+ def __init__(self, twitter_api):
+ self.twitter_api = twitter_api
def bounce(self, user, message, notify_admin='', notify_admin_body=''):
+ """Bounce a message, with additional info for explanation.
+
+ If the notify_admin string is non-empty, the site admin will
+ be notified, with that string appended to the subject.
+ If notify_admin_body is non-empty, it will be added to the body
+ sent to the admin.
+ """
message = '@%s %s' % (user, message)
message = message[:140]
try:
@@ -222,20 +239,22 @@ def bounce(self, user, message, notify_admin='', notify_admin_body=''):
if notify_admin_body:
admin_body += 'Additional info:\n'
admin_body += notify_admin_body
- _notify_admin(admin_subject, admin_body)
+ self.notify_admin(admin_subject, admin_body)
-def _notify_admin(subject, body):
- admin_email = settings.SERVICE_FAILURE_EMAIL
- from_addr = 'racks@fixcity.org'
- send_mail(subject, body, from_addr, [admin_email], fail_silently=False)
- # person receiving cron messages will get stdout
- logger.info(body)
+ @staticmethod
+ def notify_admin(subject, body):
+ admin_email = settings.SERVICE_FAILURE_EMAIL
+ from_addr = 'racks@fixcity.org'
+ send_mail(subject, body, from_addr, [admin_email], fail_silently=False)
+ # person receiving cron messages will get stdout
+ logger.info(body)
-def adapt_errors(adict):
- # XXX TODO
- return adict
+class ErrorAdapter(object):
+ def adapt_errors(self, adict):
+ # XXX TODO
+ return adict
def api_factory(settings):
@@ -249,5 +268,6 @@ class Command(BaseCommand):
def handle(self, *args, **options):
api = api_factory(settings)
- builder = RackMaker(settings, api)
+ notifier = Notifier(api)
+ builder = RackMaker(settings, api, notifier)
builder.main(recent_only=True)
35 fixcity/bmabr/tests/__init__.py
View
@@ -1,7 +1,28 @@
-from test_models import *
-from test_templatetags import *
-from test_tweeter import *
-from test_utils import *
-from test_bulkorder import *
-from test_views import *
-from test_handle_mailin import *
+"""
+Work around Django only supporting a single tests.py file by default.
+
+This will load test suites from all files named test*py in the tests package.
+
+Could have been done by creating a custom test runner as per
+http://docs.djangoproject.com/en/dev/topics/testing/#using-different-testing-frameworks
+but this is easier.
+
+"""
+
+import glob
+import os
+import unittest
+
+def suite():
+ # for some reason django does not expect this to be a TestSuite instance;
+ # rather it must be a zero-arg callable that returns a TestSuite.
+ here = os.path.abspath(os.path.dirname(__file__))
+ testfiles = glob.glob(here + '/test*py')
+ testmodules = [os.path.splitext(os.path.basename(name))[0]
+ for name in testfiles]
+ testmodules = [__name__ + '.' + name for name in testmodules]
+
+ #_suite = unittest.TestSuite()
+ return unittest.defaultTestLoader.loadTestsFromNames(testmodules)
+ #return _suite
+
50 fixcity/bmabr/tests/test_handle_mailin.py
View
@@ -70,7 +70,7 @@ def test_parse__apple_weirdness(self, mock_notifier, mock_debug):
# What else to test??
-class TestRackMaker(TestCase):
+class TestMailinRackMaker(TestCase):
@mock.patch('httplib2.Response')
@mock.patch('httplib2.Http.request')
@@ -82,7 +82,7 @@ def test_do_post__success(self, mock_notifier, mock_debug, mock_request,
notifier = mock_notifier()
response.status = 200
mock_request.return_value = (response, 'hello POST world')
- maker = handle_mailin.RackMaker(mock_notifier(), {})
+ maker = handle_mailin.RackMaker(mock_notifier(), {}, handle_mailin.ErrorAdapter())
content = maker.do_post('http://example.com', 'test body')
self.assertEqual(content, 'hello POST world')
@@ -99,7 +99,7 @@ def test_do_post__500_error(self, mock_notifier, mock_debug, mock_request,
notifier = mock_notifier()
response.status = 500
mock_request.return_value = (response, 'hello POST world')
- maker = handle_mailin.RackMaker(notifier, {})
+ maker = handle_mailin.RackMaker(notifier, {}, handle_mailin.ErrorAdapter())
content = maker.do_post('http://example.com', 'test body')
self.assertEqual(content, None)
@@ -118,7 +118,7 @@ def test_do_post__socket_error(self, mock_notifier, mock_debug,
import socket
notifier = mock_notifier()
mock_request.side_effect = socket.error("kaboom")
- maker = handle_mailin.RackMaker(notifier, {})
+ maker = handle_mailin.RackMaker(notifier, {}, handle_mailin.ErrorAdapter())
content = maker.do_post('http://example.com', 'test body')
self.assertEqual(content, None)
@@ -137,7 +137,7 @@ def test_do_post_json(self, mock_notifier, mock_debug, mock_request,
notifier = mock_notifier()
response.status = 200
mock_request.return_value = (response, '{"foo": "bar"}')
- maker = handle_mailin.RackMaker(mock_notifier(), {})
+ maker = handle_mailin.RackMaker(mock_notifier(), {}, handle_mailin.ErrorAdapter())
content = maker.do_post_json('http://example.com',
"{'some key': 'some value'}")
@@ -158,7 +158,7 @@ def test_do_post_json__validation_errors(self, mock_notifier, mock_debug,
error_body = json.dumps(
{'errors': {'title': ['This field is required.']}})
mock_request.return_value = (response, error_body)
- maker = handle_mailin.RackMaker(mock_notifier(), {})
+ maker = handle_mailin.RackMaker(mock_notifier(), {}, handle_mailin.ErrorAdapter())
content = maker.do_post_json('http://example.com',
"{'some key': 'some value'}")
@@ -168,7 +168,7 @@ def test_do_post_json__validation_errors(self, mock_notifier, mock_debug,
@mock.patch('logging.Logger.debug')
def test_submit__dry_run(self, mock_debug):
- maker = handle_mailin.RackMaker(None, {'dry-run': True})
+ maker = handle_mailin.RackMaker(None, {'dry-run': True}, None)
self.assertEqual(maker.submit({}), None)
@mock.patch('httplib2.Response')
@@ -181,36 +181,50 @@ def test_submit__no_result(self, mock_notifier, mock_debug,
notifier = mock_notifier()
response.status = 200
mock_request.return_value = (response, '')
- maker = handle_mailin.RackMaker(notifier, {})
+ maker = handle_mailin.RackMaker(notifier, {}, handle_mailin.ErrorAdapter())
self.assertEqual(maker.submit({}), None)
@mock.patch('httplib2.Response')
- @mock.patch('fixcity.bmabr.management.commands.handle_mailin.RackMaker.do_post')
+ @mock.patch('fixcity.bmabr.management.commands.utils.FixcityHttp.do_post')
@mock.patch('logging.Logger.debug')
@mock.patch('fixcity.bmabr.management.commands.handle_mailin.Notifier')
def test_submit__with_photos_and_user(self, mock_notifier, mock_debug,
- mock_do_post, mock_response):
- mock_do_post.return_value = '''{
+ mock_do_post,
+ mock_response):
+ # XXX Most of this functionality is now in utils. MOve test
+ # somewhere more appropriate.
+
+ # Mock typically uses side_effect() to specify multiple return
+ # value; clunky API but works fine.
+ do_post_return_values = [
+ '''{
"user": "bob",
- "photo_post_url": "http://example.com/photos/",
- "rack_url": "http://example.com/racks/1"
- }'''
+ "photo_post_url": "/photos/",
+ "rack_url": "/racks/1"
+ }''',
+ 'OK']
+ def side_effect(*args, **kw):
+ return do_post_return_values.pop(0)
+ mock_do_post.side_effect = side_effect
notifier = mock_notifier()
- maker = handle_mailin.RackMaker(notifier, {})
-
+ maker = handle_mailin.RackMaker(notifier, {}, handle_mailin.ErrorAdapter())
# Mock photo needs to be just file-like enough.
mock_photo_file = mock.Mock()
mock_photo_file.name = 'foo.jpg'
mock_photo_file.fileno.side_effect = AttributeError()
mock_photo_file.tell.return_value = 12345
mock_photo_file.read.return_value = ''
+
self.assertEqual(maker.submit({'photos': {'photo': mock_photo_file}}),
None)
- self.assertEqual(notifier.reply.call_args[0][0], "FixCity Rack Confirmation")
+ self.assertEqual(notifier.on_submit_success.call_count, 1)
+ vars = notifier.on_submit_success.call_args[0][0]
+ self.assert_(vars.has_key('rack_url'))
+ self.assert_(vars.has_key('rack_user'))
-class TestNotifier(TestCase):
+class TestMailinNotifier(TestCase):
@staticmethod
def _make_one():
158 fixcity/bmabr/tests/test_tweeter.py
View
@@ -32,7 +32,7 @@ class user:
screen_name = 'bob'
def test_parse(self):
- fetcher = tweeter.TwitterFetcher(None, self.username)
+ fetcher = tweeter.TwitterFetcher(None, self.username, None)
self.assertEqual(fetcher.parse(self.StubTweet),
{'date': '1970-01-01 00:00:00',
'address': 'an address',
@@ -42,7 +42,7 @@ def test_parse(self):
@mock.patch('logging.Logger.warn')
def test_parse_invalid(self, mock_logger_warn):
- fetcher = tweeter.TwitterFetcher(None, self.username)
+ fetcher = tweeter.TwitterFetcher(None, self.username, None)
self.StubTweet.text = 'invalid format'
self.assertEqual(fetcher.parse(self.StubTweet), None)
@@ -78,7 +78,7 @@ class MockRequest:
def test_get_tweets__server_error(self, mock_mentions):
import tweepy
mock_mentions.side_effect = tweepy.error.TweepError('500 or something')
- fetcher = tweeter.TwitterFetcher(tweepy.API(), self.username)
+ fetcher = tweeter.TwitterFetcher(tweepy.API(), self.username, None)
self.assertEqual(fetcher.get_tweets(), [])
@@ -86,7 +86,7 @@ def test_get_tweets__server_error(self, mock_mentions):
def test_get_tweets__empty(self, mock_mentions):
import tweepy
mock_mentions.return_value = []
- fetcher = tweeter.TwitterFetcher(tweepy.API(), self.username)
+ fetcher = tweeter.TwitterFetcher(tweepy.API(), self.username, None)
self.assertEqual(fetcher.get_tweets(), [])
@mock.patch('tweepy.API.mentions')
@@ -110,7 +110,7 @@ def get_mock_tweet_results(count=1, page=1, *args, **kw):
return results
mock_mentions.side_effect = get_mock_tweet_results
- fetcher = tweeter.TwitterFetcher(tweepy.API(), self.username)
+ fetcher = tweeter.TwitterFetcher(tweepy.API(), self.username, None)
results = fetcher.get_tweets()
self.assertEqual(len(results), 207)
self.assertEqual(mock_mentions.call_count, 2)
@@ -121,7 +121,7 @@ def test_load_last_status(self, mock_open, MockTweepyAPI):
import StringIO, pickle
mock_open.return_value = StringIO.StringIO(
pickle.dumps({'last_processed_id': 99}))
- builder = tweeter.RackMaker(settings, MockTweepyAPI())
+ builder = tweeter.RackMaker(settings, MockTweepyAPI(), None)
self.assertEqual(builder.load_last_status(True), 99)
self.assertEqual(builder.load_last_status(False), None)
@@ -134,7 +134,7 @@ def test_main__twitter_down(self, MockTweepyAPI):
tweepy_mock = MockTweepyAPI()
tweepy_mock.rate_limit_status.side_effect = tweepy.error.TweepError(
"server down?")
- builder = tweeter.RackMaker(settings, tweepy_mock)
+ builder = tweeter.RackMaker(settings, tweepy_mock, None)
builder.main()
self.assertEqual(tweepy_mock.get_tweets.call_count, 0)
@@ -143,15 +143,15 @@ def test_main__over_limit(self, MockTweepyAPI):
tweepy_mock = MockTweepyAPI()
tweepy_mock.rate_limit_status.return_value = {
'remaining_hits': 0, 'reset_time': 'tomorrow'}
- builder = tweeter.RackMaker(settings, tweepy_mock)
+ builder = tweeter.RackMaker(settings, tweepy_mock, None)
self.assertRaises(Exception, builder.main)
- @mock.patch('fixcity.bmabr.management.commands.tweeter.RackMaker.new_rack')
+ @mock.patch('fixcity.bmabr.management.commands.tweeter.RackMaker.submit')
@mock.patch('tweepy.API')
- def test_main(self, MockTweepyAPI, mock_new_rack):
+ def test_main(self, MockTweepyAPI, mock_submit):
tweepy_mock = MockTweepyAPI()
user = settings.TWITTER_USER
- builder = tweeter.RackMaker(settings, tweepy_mock)
+ builder = tweeter.RackMaker(settings, tweepy_mock, None)
# The Mock API works OK but setting attrs is a bit tedious...
# i wish you could pass a dict as the spec argument.
status = mock.Mock(['id', 'text', 'user', 'created_at'])
@@ -165,8 +165,8 @@ def test_main(self, MockTweepyAPI, mock_new_rack):
tweepy_mock.rate_limit_status.return_value = {'remaining_hits': 999}
builder.main(False)
- self.assertEqual(mock_new_rack.call_count, 1)
- self.assertEqual(mock_new_rack.call_args,
+ self.assertEqual(mock_submit.call_count, 1)
+ self.assertEqual(mock_submit.call_args,
((),
{'address': '13 thames st, brooklyn, ny',
'date': '1970-01-01 00:00:00',
@@ -175,58 +175,23 @@ def test_main(self, MockTweepyAPI, mock_new_rack):
'user': 'some twitter user',
}))
- @mock.patch('tweepy.API')
- def test_bounce(self, MockTweepyAPI):
- tweepy_mock = MockTweepyAPI()
- builder = tweeter.RackMaker(settings, tweepy_mock)
- builder.bounce('somebody', 'an interesting message')
- self.assertEqual(tweepy_mock.update_status.call_args,
- (('@somebody an interesting message',), {}))
-
- @mock.patch('tweepy.API')
- def test_bounce__twitter_down(self, MockTweepyAPI):
- tweepy_mock = MockTweepyAPI()
- builder = tweeter.RackMaker(settings, tweepy_mock)
- import tweepy
- tweepy_mock.update_status.side_effect = tweepy.error.TweepError(
- "server down?")
- builder.bounce('somebody else', 'twitter down?')
- # ... umm... nothing interesting to test here?
-
- @mock.patch('logging.Logger.info')
- @mock.patch('fixcity.bmabr.management.commands.tweeter.send_mail')
- @mock.patch('tweepy.API')
- def test_bounce__notify_admin(self, MockTweepyAPI, mock_send_mail,
- mock_info):
- tweepy_mock = MockTweepyAPI()
- builder = tweeter.RackMaker(settings, tweepy_mock)
- message = 'a message!'
- subject = 'this is not my day.'
- builder.bounce('somebody', message, notify_admin=subject)
- args = mock_send_mail.call_args
- self.assertEqual(args[0][0], 'FixCity tweeter bounce! %s' % subject)
- self.failUnless(args[0][1].count('Bouncing to: somebody'))
-
- builder.bounce('somebody', message, notify_admin=subject,
- notify_admin_body='more body')
- args = mock_send_mail.call_args
- self.failUnless(args[0][1].count('more body'))
-
-
+ @mock.patch('fixcity.bmabr.management.commands.tweeter.Notifier')
@mock.patch('fixcity.bmabr.management.commands.tweeter.shorten_url')
@mock.patch('logging.Logger.info')
@mock.patch('fixcity.bmabr.management.commands.tweeter.http')
@mock.patch('tweepy.API')
- def test_new_rack(self, MockTweepyAPI, mock_http, mock_info, mock_shorten):
+ def test_submit(self, MockTweepyAPI, mock_http, mock_info, mock_shorten,
+ mock_notifier):
tweepy_mock = MockTweepyAPI()
- builder = tweeter.RackMaker(settings, tweepy_mock)
+ mock_notifier.twitter_api = tweepy_mock
+ builder = tweeter.RackMaker(settings, tweepy_mock, mock_notifier)
class StubResponse:
status = 200
mock_http.request.return_value = (StubResponse(), '{"rack": 99}')
mock_shorten.return_value = 'http://short_url/'
- builder.new_rack('TITLE', 'ADDRESS', 'USER', 'DATE', 123)
+ builder.submit('TITLE', 'ADDRESS', 'USER', 'DATE', 123)
self.assertEqual(mock_http.request.call_count, 1)
args = mock_http.request.call_args
self.assert_(args[0][0].startswith('http'))
@@ -245,60 +210,101 @@ class StubResponse:
self.assertEqual(decoded['twitter_user'], 'USER')
# We notified the user too.
- self.assertEqual(tweepy_mock.update_status.call_count, 1)
+ self.assertEqual(mock_notifier.bounce.call_count, 1)
+ @mock.patch('fixcity.bmabr.management.commands.tweeter.Notifier')
@mock.patch('fixcity.bmabr.management.commands.tweeter.shorten_url')
@mock.patch('logging.Logger.info')
@mock.patch('fixcity.bmabr.management.commands.tweeter.http')
@mock.patch('tweepy.API')
- def test_new_rack__errors(self, MockTweepyAPI, mock_http, mock_info,
- mock_shorten):
+ def test_submit__errors(self, MockTweepyAPI, mock_http, mock_info,
+ mock_shorten, mock_notifier):
tweepy_mock = MockTweepyAPI()
- builder = tweeter.RackMaker(settings, tweepy_mock)
+ builder = tweeter.RackMaker(settings, tweepy_mock, mock_notifier)
class StubResponse:
status = 200
mock_http.request.return_value = (StubResponse(), '{"errors": "any"}')
mock_shorten.return_value = 'http://short_url/'
- builder.new_rack('TITLE', 'ADDRESS', 'USER', 'DATE', 123)
+ builder.submit('TITLE', 'ADDRESS', 'USER', 'DATE', 123)
self.assertEqual(mock_http.request.call_count, 1)
-
# We notified the user too.
- self.assertEqual(tweepy_mock.update_status.call_count, 1)
- notify_args = tweepy_mock.update_status.call_args
- self.assert_(notify_args[0][0].count('something went wrong'))
+ self.assertEqual(mock_notifier.bounce.call_count, 1)
+ notify_args, notify_kwargs = mock_notifier.bounce.call_args
+ self.assert_(notify_args[1].count('something went wrong'))
+ @mock.patch('fixcity.bmabr.management.commands.tweeter.Notifier')
@mock.patch('logging.Logger.info')
- @mock.patch('fixcity.bmabr.management.commands.tweeter.RackMaker.bounce')
@mock.patch('fixcity.bmabr.management.commands.tweeter.http')
@mock.patch('tweepy.API')
- def test_new_rack__server_error(self, MockTweepyAPI, mock_http,
- mock_bounce, mock_info):
+ def test_submit__server_error(self, MockTweepyAPI, mock_http, mock_info,
+ mock_notifier):
tweepy_mock = MockTweepyAPI()
- builder = tweeter.RackMaker(settings, tweepy_mock)
+ builder = tweeter.RackMaker(settings, tweepy_mock, mock_notifier)
class StubResponse:
status = 500
mock_http.request.return_value = (StubResponse(), 'content')
- builder.new_rack('TITLE', 'ADDRESS', 'USER', 'DATE', 123)
- self.assertEqual(mock_bounce.call_count, 1)
+ builder.submit('TITLE', 'ADDRESS', 'USER', 'DATE', 123)
+ self.assertEqual(mock_notifier.bounce.call_count, 1)
+ @mock.patch('fixcity.bmabr.management.commands.tweeter.Notifier')
@mock.patch('logging.Logger.info')
- @mock.patch('fixcity.bmabr.management.commands.tweeter._notify_admin')
@mock.patch('fixcity.bmabr.management.commands.tweeter.http')
@mock.patch('tweepy.API')
- def test_new_rack__network_error(self, MockTweepyAPI, mock_http,
- mock_notify_admin, mock_info):
+ def test_submit__network_error(self, MockTweepyAPI, mock_http, mock_info,
+ mock_notifier):
tweepy_mock = MockTweepyAPI()
- builder = tweeter.RackMaker(settings, tweepy_mock)
+ builder = tweeter.RackMaker(settings, tweepy_mock, mock_notifier)
import socket
mock_http.request.side_effect = socket.error('oops')
- self.assertRaises(socket.error, builder.new_rack,
+ self.assertRaises(socket.error, builder.submit,
'TITLE', 'ADDRESS', 'USER', 'DATE', 123)
- self.assertEqual(mock_notify_admin.call_count, 1)
+ self.assertEqual(mock_notifier.notify_admin.call_count, 1)
+
+
+class TestTweeterNotifier(TestCase):
+
+ @mock.patch('tweepy.API')
+ def test_bounce(self, MockTweepyAPI):
+ tweepy_mock = MockTweepyAPI()
+ notifier = tweeter.Notifier(tweepy_mock)
+ notifier.bounce('somebody', 'an interesting message')
+ self.assertEqual(tweepy_mock.update_status.call_args,
+ (('@somebody an interesting message',), {}))
+
+ @mock.patch('tweepy.API')
+ def test_bounce__twitter_down(self, MockTweepyAPI):
+ tweepy_mock = MockTweepyAPI()
+ notifier = tweeter.Notifier(tweepy_mock)
+ import tweepy
+ tweepy_mock.update_status.side_effect = tweepy.error.TweepError(
+ "server down?")
+ notifier.bounce('somebody else', 'twitter down?')
+ # ... umm... nothing interesting to test here?
+
+ @mock.patch('logging.Logger.info')
+ @mock.patch('fixcity.bmabr.management.commands.tweeter.send_mail')
+ @mock.patch('tweepy.API')
+ def test_bounce__notify_admin(self, MockTweepyAPI, mock_send_mail,
+ mock_info):
+ tweepy_mock = MockTweepyAPI()
+ notifier = tweeter.Notifier(tweepy_mock)
+ message = 'a message!'
+ subject = 'this is not my day.'
+ notifier.bounce('somebody', message, notify_admin=subject)
+ args = mock_send_mail.call_args
+ self.assertEqual(args[0][0], 'FixCity tweeter bounce! %s' % subject)
+ self.failUnless(args[0][1].count('Bouncing to: somebody'))
+
+ notifier.bounce('somebody', message, notify_admin=subject,
+ notify_admin_body='more body')
+ args = mock_send_mail.call_args
+ self.failUnless(args[0][1].count('more body'))
+
-class TestCommand(TestCase):
+class TestTweeterCommand(TestCase):
@mock.patch('fixcity.bmabr.management.commands.tweeter.api_factory')
@mock.patch('fixcity.bmabr.management.commands.tweeter.RackMaker.main')
1  fixcity/bmabr/tests/test_utils.py
View
@@ -35,3 +35,4 @@ def test_rotate_image_by_exif(self):
# Unfortunately PIL doesn't save exif info on new images...
self.assertEqual(exif_utils.get_exif_info(rotated), {})
+
Please sign in to comment.
Something went wrong with that request. Please try again.