Commit 3a061c4b authored by Will Thompson's avatar Will Thompson
Browse files

Fix ReadBuffer tests to run and test the right things

Without 'Test' in the class name, the test did not run at all. It would
have caught #615.

StringIO.closed is not assignable under Python 3, but we can just
close() the fake stream to make it True.

Reading from response.raw gives back bytes, not unicode, so the fake
stream in the test case should do the same.

And finally, ReadBuffer itself yields unicode, not UTF-8-encoded bytes.

See https://github.com/tweepy/tweepy/pull/635 for the actual fix that
this tests.
parent 22c0bc02
......@@ -120,13 +120,13 @@ class TweepyStreamTests(unittest.TestCase):
self.assertEqual(u'Caf\xe9'.encode('utf8'), s.session.params['follow'])
class TweepyStreamReadBuffer(unittest.TestCase):
class TweepyStreamReadBufferTests(unittest.TestCase):
stream = """11\n{id:12345}\n\n24\n{id:23456, test:"blah"}\n"""
stream = six.b("""11\n{id:12345}\n\n24\n{id:23456, test:"blah"}\n""")
def test_read_tweet(self):
for length in [1, 2, 5, 10, 20, 50]:
buf = ReadBuffer(six.StringIO(self.stream), length)
buf = ReadBuffer(six.BytesIO(self.stream), length)
self.assertEqual('11\n', buf.read_line())
self.assertEqual('{id:12345}\n', buf.read_len(11))
self.assertEqual('\n', buf.read_line())
......@@ -157,13 +157,14 @@ class TweepyStreamReadBuffer(unittest.TestCase):
return ""
# Create a fake stream
stream = six.StringIO('')
stream = six.BytesIO(six.b(''))
# Mock it's read function so it can't be called too many times
mock_read = MagicMock(side_effect=on_read)
try:
with patch.multiple(stream, create=True, read=mock_read, closed=True):
stream.close()
with patch.multiple(stream, create=True, read=mock_read):
# Now the stream can't call 'read' more than call_limit times
# and it looks like a requests stream that is closed
buf = ReadBuffer(stream, 50)
......@@ -175,14 +176,14 @@ class TweepyStreamReadBuffer(unittest.TestCase):
self.assertEqual(mock_read.call_count, 0)
def test_read_unicode_tweet(self):
stream = '11\n{id:12345}\n\n23\n{id:23456, test:"\xe3\x81\x93"}\n\n'
stream = six.b('11\n{id:12345}\n\n23\n{id:23456, test:"\xe3\x81\x93"}\n\n')
for length in [1, 2, 5, 10, 20, 50]:
buf = ReadBuffer(six.StringIO(stream), length)
buf = ReadBuffer(six.BytesIO(stream), length)
self.assertEqual('11\n', buf.read_line())
self.assertEqual('{id:12345}\n', buf.read_len(11))
self.assertEqual('\n', buf.read_line())
self.assertEqual('23\n', buf.read_line())
self.assertEqual('{id:23456, test:"\xe3\x81\x93"}\n', buf.read_len(23))
self.assertEqual(u'{id:23456, test:"\u3053"}\n', buf.read_len(23))
class TweepyStreamBackoffTests(unittest.TestCase):
......
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