Commit a592b1b3 authored by Joshua Roesslein's avatar Joshua Roesslein
Browse files

Add streaming test_sample test.

parent 5e068137
......@@ -2,36 +2,52 @@ from time import sleep
import unittest
from tweepy.api import API
from tweepy.models import Status
from tweepy.streaming import Stream, StreamListener
from config import create_auth
from mock import mock_tweet
class MockStreamListener(StreamListener):
def __init__(self):
def __init__(self, test_case):
super(MockStreamListener, self).__init__()
self.test_case = test_case
self.status_count = 0
self.status_stop_count = 0
self.connect_cb = None
def on_connect(self):
if self.connect_cb:
def on_status(self, status):
self.status_count += 1
return False
self.test_case.assertIsInstance(status, Status)
if self.status_stop_count == self.status_count:
return False
class TweepyStreamTests(unittest.TestCase):
def setUp(self):
self.auth = create_auth()
self.listener = MockStreamListener()
self.listener = MockStreamListener(self) = Stream(self.auth, self.listener)
def tearDown(self):
def test_userstream(self):
# Generate random tweet which should show up in the stream.
# Wait a bit of time for it to arrive before asserting.
def on_connect():
self.listener.connect_cb = on_connect
self.listener.status_stop_count = 1
self.assertEqual(self.listener.status_count, 1)
def test_sample(self):
self.listener.status_stop_count = 10
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