Add streaming test suite.

 - test_userstream

diff --git a/.travis.yml b/.travis.yml
index 1f77e2ed04fe4d1b69d27980f5a8fed48ca7bcb4..6f8afb06f4d43e2cbaa4a40cbaea7f412c5d7735 100644 (file)
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,5 +1,5 @@
 ---
-script: nosetests -v tests.test_api
+script: nosetests -v tests.test_api tests.test_streaming
 language: python
 env:
   global:

diff --git a/tests/config.py b/tests/config.py
index 67ee00ab76bd1fa1f83b08e9b718222c13ed942f..ebb8e5065f3c5b1535092cbed39790c26ea5e684 100644 (file)
--- a/tests/config.py
+++ b/tests/config.py
@@ -1,8 +1,15 @@
 import os
 
+from tweepy.auth import OAuthHandler
+
 username = os.environ.get('TWITTER_USERNAME', '')
 oauth_consumer_key = os.environ.get('CONSUMER_KEY', '')
 oauth_consumer_secret = os.environ.get('CONSUMER_SECRET', '')
 oauth_token = os.environ.get('ACCESS_KEY', '')
 oauth_token_secret = os.environ.get('ACCESS_SECRET', '')
 
+def create_auth():
+    auth = OAuthHandler(oauth_consumer_key, oauth_consumer_secret)
+    auth.set_access_token(oauth_token, oauth_token_secret)
+    return auth
+

diff --git a/tests/mock.py b/tests/mock.py
new file mode 100644 (file)
index 0000000..109f9d5
--- /dev/null
+++ b/tests/mock.py
@@ -0,0 +1,8 @@
+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
new file mode 100644 (file)
index 0000000..cac0b07
--- /dev/null
+++ b/tests/test_streaming.py
@@ -0,0 +1,37 @@
+from time import sleep
+import unittest
+
+from tweepy.api import API
+from tweepy.streaming import Stream, StreamListener
+
+from config import create_auth
+from mock import mock_tweet
+
+class MockStreamListener(StreamListener):
+    def __init__(self):
+        super(MockStreamListener, self).__init__()
+        self.status_count = 0
+
+    def on_status(self, status):
+        self.status_count += 1
+        return False
+
+class TweepyStreamTests(unittest.TestCase):
+    def setUp(self):
+        self.auth = create_auth()
+        self.listener = MockStreamListener()
+        self.stream = Stream(self.auth, self.listener)
+
+    def tearDown(self):
+        self.stream.disconnect()
+
+    def test_userstream(self):
+        self.stream.userstream(async=True)
+
+        # Generate random tweet which should show up in the stream.
+        # Wait a bit of time for it to arrive before asserting.
+        API(self.auth).update_status(mock_tweet())
+        sleep(1)
+
+        self.assertEqual(self.listener.status_count, 1)
+