Commit e44bf2a3 authored by Timo Ewalds's avatar Timo Ewalds
Browse files

Add tests for the streaming connection backoffs.

Move mock.py contents into test_utils.py to fix conflict between local mock.py and mock library.
parent 447f69cd
import random
import string
def mock_tweet():
"""Generate some random tweet text."""
count = random.randint(70, 140)
return ''.join([random.choice(string.letters) for i in xrange(count)])
......@@ -2,11 +2,13 @@ from time import sleep
import unittest
from tweepy.api import API
from tweepy.auth import OAuthHandler
from tweepy.models import Status
from tweepy.streaming import Stream, StreamListener
from config import create_auth
from mock import mock_tweet
from test_utils import mock_tweet
from mock import MagicMock, patch
class MockStreamListener(StreamListener):
def __init__(self, test_case):
......@@ -24,6 +26,10 @@ class MockStreamListener(StreamListener):
self.test_case.fail('timeout')
return False
def on_error(self, code):
print "response: %s" % code
return True
def on_status(self, status):
self.status_count += 1
self.test_case.assertIsInstance(status, Status)
......@@ -62,3 +68,38 @@ class TweepyStreamTests(unittest.TestCase):
self.assertEquals(self.listener.status_count,
self.listener.status_stop_count)
class TweepyStreamBackoffTests(unittest.TestCase):
def setUp(self):
#bad auth causes twitter to return 401 errors
self.auth = OAuthHandler("bad-key", "bad-secret")
self.auth.set_access_token("bad-token", "bad-token-secret")
self.listener = MockStreamListener(self)
self.stream = Stream(self.auth, self.listener)
def tearDown(self):
self.stream.disconnect()
def test_exp_backoff(self):
self.stream = Stream(self.auth, self.listener, timeout=3.0,
retry_count=1, retry_time=1.0, retry_time_cap=100.0)
self.stream.sample()
# 1 retry, should be 4x the retry_time
self.assertEqual(self.stream.retry_time, 4.0)
def test_exp_backoff_cap(self):
self.stream = Stream(self.auth, self.listener, timeout=3.0,
retry_count=1, retry_time=1.0, retry_time_cap=3.0)
self.stream.sample()
# 1 retry, but 4x the retry_time exceeds the cap, so should be capped
self.assertEqual(self.stream.retry_time, 3.0)
mock_resp = MagicMock()
mock_resp.return_value.status = 420
@patch('httplib.HTTPConnection.getresponse', mock_resp)
def test_420(self):
self.stream = Stream(self.auth, self.listener, timeout=3.0, retry_count=0,
retry_time=1.0, retry_420=1.5, retry_time_cap=20.0)
self.stream.sample()
# no retries, but error 420, should be double the retry_420, not double the retry_time
self.assertEqual(self.stream.retry_time, 3.0)
......@@ -2,6 +2,15 @@ from unittest import TestCase
from tweepy.utils import *
import random
import string
def mock_tweet():
"""Generate some random tweet text."""
count = random.randint(70, 140)
return ''.join([random.choice(string.letters) for i in xrange(count)])
class TweepyUtilsTests(TestCase):
def testparse_datetime(self):
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment