From 3a061c4b0053b4e145fe35c1b30a5451a27c0963 Mon Sep 17 00:00:00 2001 From: Will Thompson Date: Tue, 6 Oct 2015 23:14:55 +0100 Subject: [PATCH] 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. --- tests/test_streaming.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 8b7abf8..008dbb3 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -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): -- 2.25.1