def tearDown(self):
self.stream.disconnect()
+ def on_connect():
+ API(self.auth).update_status(mock_tweet())
+
+
def test_userstream(self):
# Generate random tweet which should show up in the stream.
- def on_connect():
- API(self.auth).update_status(mock_tweet())
-
- self.listener.connect_cb = on_connect
+
+ self.listener.connect_cb = self.on_connect
self.listener.status_stop_count = 1
self.stream.userstream()
self.assertEqual(self.listener.status_count, 1)
+
+ def test_sitestream(self):
+ self.listener.connect_cb = self.on_connect
+ self.listener.status_stop_count = 1
+ self.sitestream(follow=[self.auth.get_username()])
+ self.assertEqual(self.listener.status_count, 1)
def test_sample(self):
self.listener.status_stop_count = 10
self.parameters['delimited'] = 'length'
self._start(async)
+ def sitestream(self, follow, stall_warnings=False, with_='user', replies=False, async=False):
+ self.parameters = {}
+ if self.running:
+ raise TweepError('Stream object already connected!')
+ self.url = '/%s/site.json' % STREAM_VERSION
+ self.parameters['follow'] = ','.join(map(str, follow))
+ self.parameters['delimited'] = 'length'
+ if stall_warnings:
+ self.parameters['stall_warnings'] = stall_warnings
+ if with_:
+ self.parameters['with'] = with_
+ if replies:
+ self.parameters['replies'] = replies
+ self.body = urlencode_noplus(self.parameters)
+ self._start(async)
+
def disconnect(self):
if self.running is False:
return