Commit 0661dbf5 authored by Mark Smith's avatar Mark Smith
Browse files

Port to Python 3.3+

parent 5b76b365
*.pyc
*.swp
build/
dist/
tweepy.egg-info/
.env/
env.sh
.coverage
htmlcov/
.env
.idea
.pdbrc
.tox
build
dist
cassettes
htmlcov
tweepy.egg-info
......@@ -31,4 +31,4 @@ Stuart Powers
Jeff Hull (@jsh2134)
Mike (mikeandmore)
Kohei YOSHIDA
Mark Smith (@judy2k)
include requirements.txt
......@@ -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
-------------
......
......@@ -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
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.
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)
......
This diff is collapsed.
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()
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):
......
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)
......
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()
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:
......
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