Merge pull request #507 from judy2k/python3-tox-vcrpy-squash

Port to Python 3.3+

.gitignore

 *.pyc
 *.swp
-build/
-dist/
-tweepy.egg-info/
-.env/
+env.sh
 .coverage
-htmlcov/
+.env
+.idea
+.pdbrc
+.tox
+build
+dist
+cassettes
+htmlcov
+tweepy.egg-info

CONTRIBUTORS

@@ -31,4 +31,4 @@ Stuart Powers
 Jeff Hull (@jsh2134)
 Mike (mikeandmore)
 Kohei YOSHIDA
-
+Mark Smith (@judy2k)

MANIFEST.in

+include requirements.txt

README.md

@@ -18,8 +18,7 @@ Github and install it manually:
     cd tweepy
     python setup.py install
 
-**Note** only Python 2.6 and 2.7 are supported at
-the moment. The Python 3 family is not yet supported.
+Python 2.6 and 2.7, 3.3 & 3.4 are supported.
 
 Documentation
 -------------

requirements.txt

-requests>=2.0.0
-requests_oauthlib==0.4.0
+requests==2.3.0
+requests_oauthlib==0.4.1
+six==1.7.3

run_tests.sh

@@ -3,6 +3,5 @@
 if [[ $TRAVIS_BRANCH == "production" ]]; then
   nosetests -v --with-coverage tests.test_api tests.test_streaming tests.test_cursors tests.test_utils
 else
-  curl "https://dl.dropboxusercontent.com/u/231242/record.json" -o tests/record.json
-  USE_REPLAY=1 nosetests -v tests.test_api tests.test_utils
+  USE_REPLAY=1 nosetests -v --with-coverage tests.test_api tests.test_utils
 fi

test_requirements.txt

-httreplay==0.1.4
-coveralls==0.2
-unittest2==0.5.1
+tox>=1.7.2
+vcrpy==1.0.2
 mock==1.0.1
-boto==2.27
+unittest2   # Comment this line out if using Python 3.

tests/config.py

 import os
-import sys
-from unittest2 import TestCase
 
-from httreplay import start_replay, stop_replay
-from httreplay.utils import filter_headers_key
+import vcr
 
 from tweepy.auth import OAuthHandler
 from tweepy.api import API
 
+import six
+if six.PY3:
+    import unittest
+else:
+    import unittest2 as unittest
+
 username = os.environ.get('TWITTER_USERNAME', 'tweepytest')
 oauth_consumer_key = os.environ.get('CONSUMER_KEY', '')
 oauth_consumer_secret = os.environ.get('CONSUMER_SECRET', '')
@@ -15,23 +18,23 @@ oauth_token = os.environ.get('ACCESS_KEY', '')
 oauth_token_secret = os.environ.get('ACCESS_SECRET', '')
 use_replay = os.environ.get('USE_REPLAY', False)
 
-class TweepyTestCase(TestCase):
 
+tape = vcr.VCR(
+    cassette_library_dir='cassettes',
+    filter_headers=['Authorization'],
+    serializer='json',
+    # Either use existing cassettes, or never use recordings:
+    record_mode='new_episodes' if use_replay else 'all',
+)
+
+
+class TweepyTestCase(unittest.TestCase):
     def setUp(self):
         self.auth = create_auth()
         self.api = API(self.auth)
         self.api.retry_count = 2
         self.api.retry_delay = 5
 
-        if use_replay:
-            def filter_body(data): return ''
-            start_replay('tests/record.json',
-                         headers_key=filter_headers_key(['Authorization']),
-                         body_key=filter_body)
-
-    def tearDown(self):
-        if use_replay:
-            stop_replay()
 
 def create_auth():
     auth = OAuthHandler(oauth_consumer_key, oauth_consumer_secret)

tests/test_api.py

-import unittest2 as unittest
+import unittest
 import random
+import shutil
 from time import sleep
 import os
 
@@ -7,13 +8,14 @@ from nose import SkipTest
 
 from tweepy import Friendship, MemoryCache, FileCache, API
 from tweepy.parsers import Parser
-from config import TweepyTestCase, username, use_replay
+from .config import TweepyTestCase, username, use_replay, tape
 
 test_tweet_id = '266367358078169089'
 tweet_text = 'testing 1000'
 
 """Unit tests"""
 
+
 class TweepyErrorTests(unittest.TestCase):
 
     def testpickle(self):
@@ -27,27 +29,34 @@ class TweepyErrorTests(unittest.TestCase):
         self.assertEqual(e.reason, e2.reason)
         self.assertEqual(e.response, e2.response)
 
+
 class TweepyAPITests(TweepyTestCase):
 
     # TODO: Actually have some sort of better assertion
+    @tape.use_cassette('testgetoembed.json')
     def testgetoembed(self):
         data = self.api.get_oembed(test_tweet_id)
         self.assertEqual(data['author_name'], "Twitter")
 
+    @tape.use_cassette('testparserargumenthastobeaparserinstance.json')
     def testparserargumenthastobeaparserinstance(self):
         """ Testing the issue https://github.com/tweepy/tweepy/issues/421"""
         self.assertRaises(TypeError, API, self.auth, parser=Parser)
 
+    @tape.use_cassette('testhometimeline.json')
     def testhometimeline(self):
         self.api.home_timeline()
 
+    @tape.use_cassette('testusertimeline.json')
     def testusertimeline(self):
         self.api.user_timeline()
         self.api.user_timeline('twitter')
 
+    @tape.use_cassette('testmentionstimeline.json')
     def testmentionstimeline(self):
         self.api.mentions_timeline()
 
+    @tape.use_cassette('testretweetsofme.json')
     def testretweetsofme(self):
         self.api.retweets_of_me()
 
@@ -55,15 +64,19 @@ class TweepyAPITests(TweepyTestCase):
         # TODO(josh): Need a way to get random tweets to retweet.
         raise SkipTest()
 
+    @tape.use_cassette('testretweets.json')
     def testretweets(self):
         self.api.retweets(test_tweet_id)
 
+    @tape.use_cassette('testretweeters.json')
     def testretweeters(self):
         self.api.retweeters(test_tweet_id)
 
+    @tape.use_cassette('testgetstatus.json')
     def testgetstatus(self):
         self.api.get_status(id=test_tweet_id)
 
+    @tape.use_cassette('testupdateanddestroystatus.json')
     def testupdateanddestroystatus(self):
         # test update
         text = tweet_text if use_replay else 'testing %i' % random.randint(0, 1000)
@@ -74,10 +87,12 @@ class TweepyAPITests(TweepyTestCase):
         deleted = self.api.destroy_status(id=update.id)
         self.assertEqual(deleted.id, update.id)
 
+    @tape.use_cassette('testupdatestatuswithmedia.yaml', serializer='yaml')
     def testupdatestatuswithmedia(self):
         update = self.api.update_with_media('examples/banner.png', status=tweet_text)
         self.assertIn(tweet_text + ' http://t.co', update.text)
 
+    @tape.use_cassette('testgetuser.json')
     def testgetuser(self):
         u = self.api.get_user('twitter')
         self.assertEqual(u.screen_name, 'twitter')
@@ -85,38 +100,47 @@ class TweepyAPITests(TweepyTestCase):
         u = self.api.get_user(783214)
         self.assertEqual(u.screen_name, 'twitter')
 
+    @tape.use_cassette('testlookupusers.json')
     def testlookupusers(self):
         def check(users):
             self.assertEqual(len(users), 2)
         check(self.api.lookup_users(user_ids=[6844292, 6253282]))
         check(self.api.lookup_users(screen_names=['twitterapi', 'twitter']))
 
+    @tape.use_cassette('testsearchusers.json')
     def testsearchusers(self):
         self.api.search_users('twitter')
 
+    @tape.use_cassette('testsuggestedcategories.json')
     def testsuggestedcategories(self):
         self.api.suggested_categories()
 
+    @tape.use_cassette('testsuggestedusers.json')
     def testsuggestedusers(self):
         categories = self.api.suggested_categories()
         if len(categories) != 0:
             self.api.suggested_users(categories[0].slug)
 
+    @tape.use_cassette('testsuggesteduserstweets.json')
     def testsuggesteduserstweets(self):
         categories = self.api.suggested_categories()
         if len(categories) != 0:
             self.api.suggested_users_tweets(categories[0].slug)
 
+    @tape.use_cassette('testme.json')
     def testme(self):
         me = self.api.me()
         self.assertEqual(me.screen_name, username)
 
+    @tape.use_cassette('testdirectmessages.json')
     def testdirectmessages(self):
         self.api.direct_messages()
 
+    @tape.use_cassette('testsentdirectmessages.json')
     def testsentdirectmessages(self):
         self.api.sent_direct_messages()
 
+    @tape.use_cassette('testsendanddestroydirectmessage.json')
     def testsendanddestroydirectmessage(self):
         # send
         sent_dm = self.api.send_direct_message(username, text='test message')
@@ -131,6 +155,7 @@ class TweepyAPITests(TweepyTestCase):
         self.assertEqual(destroyed_dm.sender.screen_name, username)
         self.assertEqual(destroyed_dm.recipient.screen_name, username)
 
+    @tape.use_cassette('testcreatedestroyfriendship.json')
     def testcreatedestroyfriendship(self):
         enemy = self.api.destroy_friendship('twitter')
         self.assertEqual(enemy.screen_name, 'twitter')
@@ -142,23 +167,29 @@ class TweepyAPITests(TweepyTestCase):
         friend = self.api.create_friendship('twitter')
         self.assertEqual(friend.screen_name, 'twitter')
 
+    @tape.use_cassette('testshowfriendship.json')
     def testshowfriendship(self):
         source, target = self.api.show_friendship(target_screen_name='twitter')
         self.assert_(isinstance(source, Friendship))
         self.assert_(isinstance(target, Friendship))
 
+    @tape.use_cassette('testfriendsids.json')
     def testfriendsids(self):
         self.api.friends_ids(username)
 
+    @tape.use_cassette('testfollowersids.json')
     def testfollowersids(self):
         self.api.followers_ids(username)
 
+    @tape.use_cassette('testfriends.json')
     def testfriends(self):
         self.api.friends(username)
 
+    @tape.use_cassette('testfollowers.json')
     def testfollowers(self):
         self.api.followers(username)
 
+    @tape.use_cassette('testverifycredentials.json')
     def testverifycredentials(self):
         self.assertNotEqual(self.api.verify_credentials(), False)
 
@@ -170,6 +201,7 @@ class TweepyAPITests(TweepyTestCase):
         me = self.api.verify_credentials(skip_status=True)
         self.assertFalse(hasattr(me, 'status'))
 
+    @tape.use_cassette('testratelimitstatus.json')
     def testratelimitstatus(self):
         self.api.rate_limit_status()
 
@@ -179,6 +211,7 @@ class TweepyAPITests(TweepyTestCase):
         self.api.set_delivery_device('none')
     """
 
+    @tape.use_cassette('testupdateprofilecolors.json')
     def testupdateprofilecolors(self):
         original = self.api.me()
         updated = self.api.update_profile_colors('000', '000', '000', '000', '000')
@@ -206,9 +239,11 @@ class TweepyAPITests(TweepyTestCase):
         self.api.update_profile_background_image('examples/bg.png')
     """
 
+    @tape.use_cassette('testupdateprofilebannerimage.yaml', serializer='yaml')
     def testupdateprofilebannerimage(self):
         self.api.update_profile_banner('examples/banner.png')
 
+    @tape.use_cassette('testupdateprofile.json')
     def testupdateprofile(self):
         original = self.api.me()
         profile = {
@@ -226,21 +261,26 @@ class TweepyAPITests(TweepyTestCase):
             if k == 'email': continue
             self.assertEqual(getattr(updated, k), v)
 
+    @tape.use_cassette('testfavorites.json')
     def testfavorites(self):
         self.api.favorites()
 
+    @tape.use_cassette('testcreatedestroyfavorite.json')
     def testcreatedestroyfavorite(self):
         self.api.create_favorite(4901062372)
         self.api.destroy_favorite(4901062372)
 
+    @tape.use_cassette('testcreatedestroyblock.json')
     def testcreatedestroyblock(self):
         self.api.create_block('twitter')
         self.api.destroy_block('twitter')
         self.api.create_friendship('twitter') # restore
 
+    @tape.use_cassette('testblocks.json')
     def testblocks(self):
         self.api.blocks()
 
+    @tape.use_cassette('testblocksids.json')
     def testblocksids(self):
         self.api.blocks_ids()
 
@@ -255,21 +295,27 @@ class TweepyAPITests(TweepyTestCase):
     #     self.assertEqual(l.description, 'updated!')
     #     self.api.destroy_list(list_id=l.id)
 
+    @tape.use_cassette('testlistsall.json')
     def testlistsall(self):
         self.api.lists_all()
 
+    @tape.use_cassette('testlistsmemberships.json')
     def testlistsmemberships(self):
         self.api.lists_memberships()
 
+    @tape.use_cassette('testlistssubscriptions.json')
     def testlistssubscriptions(self):
         self.api.lists_subscriptions()
 
+    @tape.use_cassette('testlisttimeline.json')
     def testlisttimeline(self):
         self.api.list_timeline('applepie', 'stars')
 
+    @tape.use_cassette('testgetlist.json')
     def testgetlist(self):
         self.api.get_list(owner_screen_name='applepie', slug='stars')
 
+    @tape.use_cassette('testaddremovelistmember.json')
     def testaddremovelistmember(self):
         params = {
             'slug': 'test',
@@ -284,6 +330,7 @@ class TweepyAPITests(TweepyTestCase):
         sleep(3)
         assert_list(self.api.remove_list_member(**params))
 
+    @tape.use_cassette('testaddremovelistmembers.json')
     def testaddremovelistmembers(self):
         params = {
             'slug': 'test',
@@ -297,12 +344,15 @@ class TweepyAPITests(TweepyTestCase):
         assert_list(self.api.add_list_members(**params))
         assert_list(self.api.remove_list_members(**params))
 
+    @tape.use_cassette('testlistmembers.json')
     def testlistmembers(self):
         self.api.list_members('applepie', 'stars')
 
+    @tape.use_cassette('testshowlistmember.json')
     def testshowlistmember(self):
         self.assertTrue(self.api.show_list_member(owner_screen_name='applepie', slug='stars', screen_name='NathanFillion'))
 
+    @tape.use_cassette('testsubscribeunsubscribelist.json')
     def testsubscribeunsubscribelist(self):
         params = {
             'owner_screen_name': 'applepie',
@@ -311,34 +361,42 @@ class TweepyAPITests(TweepyTestCase):
         self.api.subscribe_list(**params)
         self.api.unsubscribe_list(**params)
 
+    @tape.use_cassette('testlistsubscribers.json')
     def testlistsubscribers(self):
         self.api.list_subscribers('applepie', 'stars')
 
+    @tape.use_cassette('testshowlistsubscriber.json')
     def testshowlistsubscriber(self):
         self.assertTrue(self.api.show_list_subscriber('tweepytest', 'test', 'applepie'))
 
+    @tape.use_cassette('testsavedsearches.json')
     def testsavedsearches(self):
         s = self.api.create_saved_search('test')
         self.api.saved_searches()
         self.assertEqual(self.api.get_saved_search(s.id).query, 'test')
         self.api.destroy_saved_search(s.id)
 
+    @tape.use_cassette('testsearch.json')
     def testsearch(self):
         self.api.search('tweepy')
 
+    @tape.use_cassette('testgeoapis.json')
     def testgeoapis(self):
         def place_name_in_list(place_name, place_list):
             """Return True if a given place_name is in place_list."""
             return any([x.full_name.lower() == place_name.lower() for x in place_list])
 
-        twitter_hq = self.api.geo_similar_places(lat=37, long= -122, name='Twitter HQ')
+        twitter_hq = self.api.geo_similar_places(lat=37.7821120598956,
+                                                 long=-122.400612831116,
+                                                 name='South of Market Child Care')
         # Assumes that twitter_hq is first Place returned...
-        self.assertEqual(twitter_hq[0].id, '3bdf30ed8b201f31')
+        self.assertEqual(twitter_hq[0].id, '09aff2da00000000')
         # Test various API functions using Austin, TX, USA
-        self.assertEqual(self.api.geo_id(id='c3f37afa9efcf94b').full_name, 'Austin, TX')
+        self.assertEqual(self.api.geo_id(id='1ffd3558f2e98349').full_name, 'Dogpatch, San Francisco')
         self.assertTrue(place_name_in_list('Austin, TX',
             self.api.reverse_geocode(lat=30.267370168467806, long= -97.74261474609375))) # Austin, TX, USA
 
+    @tape.use_cassette('testsupportedlanguages.json')
     def testsupportedlanguages(self):
         languages = self.api.supported_languages()
         expected_dict = {
@@ -348,6 +406,7 @@ class TweepyAPITests(TweepyTestCase):
         }
         self.assertTrue(expected_dict in languages)
 
+    @tape.use_cassette('testcachedresult.json')
     def testcachedresult(self):
         self.api.cache = MemoryCache()
         self.api.home_timeline()
@@ -355,8 +414,8 @@ class TweepyAPITests(TweepyTestCase):
         self.api.home_timeline()
         self.assertTrue(self.api.cached_result)
 
-class TweepyCacheTests(unittest.TestCase):
 
+class TweepyCacheTests(unittest.TestCase):
     timeout = 2.0
     memcache_servers = ['127.0.0.1:11211']  # must be running for test to pass
 
@@ -393,10 +452,14 @@ class TweepyCacheTests(unittest.TestCase):
 
     def testfilecache(self):
         os.mkdir('cache_test_dir')
-        self.cache = FileCache('cache_test_dir', self.timeout)
-        self._run_tests()
-        self.cache.flush()
-        os.rmdir('cache_test_dir')
+        try:
+            self.cache = FileCache('cache_test_dir', self.timeout)
+            self._run_tests()
+            self.cache.flush()
+        finally:
+            if os.path.exists('cache_test_dir'):
+                shutil.rmtree('cache_test_dir')
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_auth.py

-import unittest2 as unittest
+from __future__ import absolute_import
 
-from config import *
+from .config import *
 from tweepy import API, OAuthHandler
 
+import six
+if six.PY3:
+    import unittest
+else:
+    import unittest2 as unittest
+
 class TweepyAuthTests(unittest.TestCase):
 
     def testoauth(self):

tests/test_cursors.py

-import unittest2 as unittest
+from tweepy import Cursor
 
-from tweepy import API, Cursor
+from .config import create_auth
+from .config import TweepyTestCase, username, use_replay, tape
 
-from config import create_auth
+import six
+if six.PY3:
+    import unittest
+else:
+    import unittest2 as unittest
 
-class TweepyCursorTests(unittest.TestCase):
-
-    def setUp(self):
-        self.api = API(create_auth())
 
+class TweepyCursorTests(TweepyTestCase):
+    @tape.use_cassette('testidcursoritems.json')
     def testidcursoritems(self):
         items = list(Cursor(self.api.user_timeline).items(25))
         self.assertEqual(len(items), 25)
 
+    @tape.use_cassette('testidcursorpages.json')
     def testidcursorpages(self):
         pages = list(Cursor(self.api.user_timeline).pages(5))
         self.assertEqual(len(pages), 5)
 
+    @tape.use_cassette('testcursorcursoritems.json')
     def testcursorcursoritems(self):
         items = list(Cursor(self.api.friends_ids).items(10))
         self.assertEqual(len(items), 10)
 
-        items = list(Cursor(self.api.followers_ids, 'twitter').items(10))
+        items = list(Cursor(self.api.followers_ids, username).items(10))
         self.assertEqual(len(items), 10)
 
+    @tape.use_cassette('testcursorcursorpages.json')
     def testcursorcursorpages(self):
         pages = list(Cursor(self.api.friends_ids).pages(1))
         self.assert_(len(pages) == 1)
 
-        pages = list(Cursor(self.api.followers_ids, 'twitter').pages(1))
+        pages = list(Cursor(self.api.followers_ids, username).pages(1))
         self.assert_(len(pages) == 1)
 
+    @tape.use_cassette('testcursorsetstartcursor.json')
     def testcursorsetstartcursor(self):
         c = Cursor(self.api.friends_ids, cursor=123456)
         self.assertEqual(c.iterator.next_cursor, 123456)

tests/test_rate_limit.py

-import unittest2 as unittest
+import unittest
 import os
 
 from tweepy import API, Cursor
 from tweepy.error import TweepError
 
-from config import create_auth
+import six
+if six.PY3:
+    import unittest
+else:
+    import unittest2 as unittest
+from .config import create_auth
 
 testratelimit = 'TEST_RATE_LIMIT' in os.environ
 
@@ -24,7 +29,7 @@ class TweepyRateLimitTests(unittest.TestCase):
         for user_id in test_user_ids:
             try:
                 self.api.user_timeline(user_id=user_id, count=1, include_rts=True)
-            except TweepError, e:
+            except TweepError as e:
                 # continue if we're not autherized to access the user's timeline or she doesn't exist anymore
                 if e.response is not None and e.response.status in set([401, 404]): 
                     continue
@@ -35,4 +40,4 @@ if __name__ == '__main__':
     if testratelimit:
         unittest.TextTestRunner().run(unittest.loader.makeSuite(TweepyRateLimitTests))
     else:
-        unittest.main()
\ No newline at end of file
+        unittest.main()

tests/test_resultset.py

-import unittest2 as unittest
+from .config import TweepyTestCase
 
 from tweepy.models import ResultSet
 
@@ -10,7 +10,7 @@ class IdItem(object):
 
 ids_fixture = [1, 10, 8, 50, 2, 100, 5]
 
-class TweepyResultSetTests(unittest.TestCase):
+class TweepyResultSetTests(TweepyTestCase):
     def setUp(self):
         self.results = ResultSet()
         for i in ids_fixture:

tests/test_streaming.py

-from StringIO import StringIO
-from time import sleep
-import unittest2 as unittest
+from __future__ import absolute_import, print_function
+
+from .config import tape
+
+import six
+if six.PY3:
+    import unittest
+    from unittest.case import skip
+else:
+    import unittest2 as unittest
+    from unittest2.case import skip
 
 from tweepy.api import API
 from tweepy.auth import OAuthHandler
 from tweepy.models import Status
 from tweepy.streaming import Stream, StreamListener, ReadBuffer
 
-from config import create_auth
-from test_utils import mock_tweet
+from .config</