self.assertEquals(self.listener.status_count,
self.listener.status_stop_count)
+ def test_track_encoding(self):
+ s = Stream(None, None)
+ s._start = lambda async: None
+ s.filter(track=[u'Caf\xe9'])
+
+ # Should be UTF-8 encoded
+ self.assertEqual(u'Caf\xe9'.encode('utf8'), s.parameters['track'])
+
+ def test_follow_encoding(self):
+ s = Stream(None, None)
+ s._start = lambda async: None
+ s.filter(follow=[u'Caf\xe9'])
+
+ # Should be UTF-8 encoded
+ self.assertEqual(u'Caf\xe9'.encode('utf8'), s.parameters['follow'])
class TweepyStreamBackoffTests(unittest.TestCase):
def setUp(self):
self.url += '&count=%s' % count
self._start(async)
- def filter(self, follow=None, track=None, async=False, locations=None,
- count = None, stall_warnings=False, languages=None):
+ def filter(self, follow=None, track=None, async=False, locations=None,
+ count=None, stall_warnings=False, languages=None, encoding='utf8'):
self.parameters = {}
self.headers['Content-type'] = "application/x-www-form-urlencoded"
if self.running:
raise TweepError('Stream object already connected!')
self.url = '/%s/statuses/filter.json?delimited=length' % STREAM_VERSION
if follow:
- self.parameters['follow'] = ','.join(map(str, follow))
+ encoded_follow = [s.encode(encoding) for s in follow]
+ self.parameters['follow'] = ','.join(encoded_follow)
if track:
- self.parameters['track'] = ','.join(map(str, track))
+ encoded_track = [s.encode(encoding) for s in track]
+ self.parameters['track'] = ','.join(encoded_track)
if locations and len(locations) > 0:
assert len(locations) % 4 == 0
self.parameters['locations'] = ','.join(['%.2f' % l for l in locations])