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.

diff --git a/test_requirements.txt b/test_requirements.txt
index 626ca45e3e6059b56c3870ba1fbd1b1db918f0da..bad5cc606b1bbdcadb27a962b235cd0e626c6728 100644 (file)
--- a/test_requirements.txt
+++ b/test_requirements.txt
@@ -1,2 +1,3 @@
 httreplay==0.1.4
 coveralls==0.2
+mock==1.0.1

diff --git a/tests/mock.py b/tests/mock.py
deleted file mode 100644 (file)
index 109f9d5..0000000
--- a/tests/mock.py
+++ /dev/null
@@ -1,8 +0,0 @@
-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)])
-

diff --git a/tests/test_streaming.py b/tests/test_streaming.py
index d3fdefc6c961dbaaf4d84dbcb55e277669fd748c..07f02af89bcfbbf22b3650a4fb013d2f13f369b4 100644 (file)
--- a/tests/test_streaming.py
+++ b/tests/test_streaming.py
@@ -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)

diff --git a/tests/test_utils.py b/tests/test_utils.py
index 09208397dc5daa1ab2f36e17aa304a1f5b7a1989..57968d073497dd51ee9447a6401065e0d7d0e254 100644 (file)
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -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):