code/twython
1
Fork 0
Code Issues 23 Pull requests 4 Projects Releases 5 Packages Wiki Activity Actions

Merge remote-tracking branch 'upstream/master' into cursor-pages

Browse source
This commit is contained in:
Ian 2014-02-21 10:24:44 +11:00
commit 3c278efd4e
14 changed files with 845 additions and 505 deletions
Download patch file Download diff file Expand all files Collapse all files

4
.travis.yml
View file

@ -17,7 +17,9 @@ env:
- TEST_LIST_SLUG=team
- TEST_LIST_OWNER_SCREEN_NAME=twitterapi
- ACCESS_TOKEN_B64=U2FsdGVkX18QdBhvMNshM4PGy04tU3HLwKP+nNSoNZHKsvGLjELcWEXN2LIu/T+yngX1vGONf9lo14ElnfS4k7sfhiru8phR4+rZuBVP3bDvC2A6fXJuhuLqNhBrWqg32WQewvxLWDWBoKmnvRHg5b74GHh+IN/12tU0cBF2HK8=
install: pip install -r requirements.txt
install:
- pip install -r requirements.txt
- if [[ $TRAVIS_PYTHON_VERSION == '2.6' ]]; then pip install unittest2; fi
script: nosetests -v -w tests/ --logging-filter="twython" --with-cov --cov twython --cov-config .coveragerc --cov-report term-missing
notifications:
email: false

13
HISTORY.rst
View file

@ -3,6 +3,19 @@
History
-------
3.1.2 (2013-12-05)
++++++++++++++++++
- Fixed Changelog (HISTORY.rst)
3.1.1 (2013-12-05)
++++++++++++++++++
- Update `requests` version to 2.1.0.
- Fixed: Streaming issue where `Exceptions` in handlers or `on_success` which subclass `ValueError` would previously be caught and reported as a JSON decoding problem, and `on_error()` would be called (with status_code=200)
- Fixed issue where XML was returned when bad tokens were passed to `get_authorized_tokens`
- Fixed import for `setup` causing installation to fail on some devices (eg. Nokia N9/MeeGo)
3.1.0 (2013-09-25)
++++++++++++++++++

4
docs/conf.py
View file

@ -50,9 +50,9 @@ copyright = u'2013, Ryan McGrath'
# built documents.
#
# The short X.Y version.
version = '3.1.0'
version = '3.1.2'
# The full version, including alpha/beta/rc tags.
release = '3.1.0'
release = '3.1.2'
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.

2
docs/usage/starting_out.rst
View file

@ -98,7 +98,7 @@ Once you have the final user tokens, store them in a database for later use!
.. code-block:: python
OAUTH_TOKEN = final_step['oauth_token']
OAUTH_TOKEN_SECERT = final_step['oauth_token_secret']
OAUTH_TOKEN_SECRET = final_step['oauth_token_secret']
.. _oauth2:

3
requirements.txt
View file

@ -1,5 +1,6 @@
coverage==3.6.0
requests==2.0.1
requests==2.1.0
requests_oauthlib==0.4.0
python-coveralls==2.1.0
nose-cov==1.6
responses==0.2.0

4
setup.py
View file

@ -9,7 +9,7 @@ except ImportError:
from distutils.core import setup
__author__ = 'Ryan McGrath <ryan@venodesigns.net>'
__version__ = '3.1.0'
__version__ = '3.1.2'
packages = [
'twython',
@ -23,7 +23,7 @@ if sys.argv[-1] == 'publish':
setup(
name='twython',
version=__version__,
install_requires=['requests==2.0.1', 'requests_oauthlib==0.4.0'],
install_requires=['requests==2.1.0', 'requests_oauthlib==0.4.0'],
author='Ryan McGrath',
author_email='ryan@venodesigns.net',
license=open('LICENSE').read(),

6
tests/config.py
View file

@ -1,5 +1,11 @@
import os
import sys
if sys.version_info[0] == 2 and sys.version_info[1] == 6:
import unittest2 as unittest
else:
import unittest
app_key = os.environ.get('APP_KEY')
app_secret = os.environ.get('APP_SECRET')
oauth_token = os.environ.get('OAUTH_TOKEN')

13
tests/test_auth.py
View file

@ -1,8 +1,6 @@
from twython import Twython, TwythonError, TwythonAuthError
from .config import app_key, app_secret, screen_name
import unittest
from .config import app_key, app_secret, screen_name, unittest
class TwythonAuthTestCase(unittest.TestCase):
@ -16,48 +14,57 @@ class TwythonAuthTestCase(unittest.TestCase):
self.oauth2_bad_api = Twython('BAD_APP_KEY', 'BAD_APP_SECRET',
oauth_version=2)
@unittest.skip('skipping non-updated test')
def test_get_authentication_tokens(self):
"""Test getting authentication tokens works"""
self.api.get_authentication_tokens(callback_url='http://google.com/',
force_login=True,
screen_name=screen_name)
@unittest.skip('skipping non-updated test')
def test_get_authentication_tokens_bad_tokens(self):
"""Test getting authentication tokens with bad tokens
raises TwythonAuthError"""
self.assertRaises(TwythonAuthError, self.bad_api.get_authentication_tokens,
callback_url='http://google.com/')
@unittest.skip('skipping non-updated test')
def test_get_authorized_tokens_bad_tokens(self):
"""Test getting final tokens fails with wrong tokens"""
self.assertRaises(TwythonError, self.bad_api.get_authorized_tokens,
'BAD_OAUTH_VERIFIER')
@unittest.skip('skipping non-updated test')
def test_get_authorized_tokens_invalid_or_expired_tokens(self):
"""Test getting final token fails when invalid or expired tokens have been passed"""
self.assertRaises(TwythonError, self.bad_api_invalid_tokens.get_authorized_tokens,
'BAD_OAUTH_VERIFIER')
@unittest.skip('skipping non-updated test')
def test_get_authentication_tokens_raises_error_when_oauth2(self):
"""Test when API is set for OAuth 2, get_authentication_tokens raises
a TwythonError"""
self.assertRaises(TwythonError, self.oauth2_api.get_authentication_tokens)
@unittest.skip('skipping non-updated test')
def test_get_authorization_tokens_raises_error_when_oauth2(self):
"""Test when API is set for OAuth 2, get_authorized_tokens raises
a TwythonError"""
self.assertRaises(TwythonError, self.oauth2_api.get_authorized_tokens,
'BAD_OAUTH_VERIFIER')
@unittest.skip('skipping non-updated test')
def test_obtain_access_token(self):
"""Test obtaining an Application Only OAuth 2 access token succeeds"""
self.oauth2_api.obtain_access_token()
@unittest.skip('skipping non-updated test')
def test_obtain_access_token_bad_tokens(self):
"""Test obtaining an Application Only OAuth 2 access token using bad app tokens fails"""
self.assertRaises(TwythonAuthError,
self.oauth2_bad_api.obtain_access_token)
@unittest.skip('skipping non-updated test')
def test_obtain_access_token_raises_error_when_oauth1(self):
"""Test when API is set for OAuth 1, obtain_access_token raises a
TwythonError"""

731
tests/test_core.py
View file

@ -1,84 +1,289 @@
from twython import Twython, TwythonError, TwythonAuthError
from twython import Twython, TwythonError, TwythonAuthError, TwythonRateLimitError
from .config import (
app_key, app_secret, oauth_token, oauth_token_secret,
protected_twitter_1, protected_twitter_2, screen_name,
test_tweet_id, test_list_slug, test_list_owner_screen_name,
access_token, test_tweet_object, test_tweet_html
test_tweet_object, test_tweet_html, unittest
)
import time
import unittest
import responses
import requests
from twython.compat import is_py2
if is_py2:
from StringIO import StringIO
else:
from io import StringIO
try:
import unittest.mock as mock
except ImportError:
import mock
class TwythonAPITestCase(unittest.TestCase):
def setUp(self):
self.api = Twython('', '', '', '')
client_args = {
'headers': {
'User-Agent': '__twython__ Test'
},
'allow_redirects': False
}
def get_url(self, endpoint):
"""Convenience function for mapping from endpoint to URL"""
return '%s/%s.json' % (self.api.api_url % self.api.api_version, endpoint)
oauth2_client_args = {
'headers': {} # This is so we can hit coverage that Twython sets User-Agent for us if none is supplied
}
def register_response(self, method, url, body='{}', match_querystring=False,
status=200, adding_headers=None, stream=False,
content_type='application/json; charset=utf-8'):
"""Wrapper function for responses for simpler unit tests"""
self.api = Twython(app_key, app_secret,
oauth_token, oauth_token_secret,
client_args=client_args)
# responses uses BytesIO to hold the body so it needs to be in bytes
if not is_py2:
body = bytes(body, 'UTF-8')
self.oauth2_api = Twython(app_key, access_token=access_token,
client_args=oauth2_client_args)
responses.add(method, url, body, match_querystring,
status, adding_headers, stream, content_type)
@responses.activate
def test_request_should_handle_full_endpoint(self):
"""Test that request() accepts a full URL for the endpoint argument"""
url = 'https://api.twitter.com/1.1/search/tweets.json'
self.register_response(responses.GET, url)
self.api.request(url)
self.assertEqual(1, len(responses.calls))
self.assertEqual(url, responses.calls[0].request.url)
@responses.activate
def test_request_should_handle_relative_endpoint(self):
"""Test that request() accepts a twitter endpoint name for the endpoint argument"""
url = 'https://api.twitter.com/1.1/search/tweets.json'
self.register_response(responses.GET, url)
self.api.request('search/tweets', version='1.1')
self.assertEqual(1, len(responses.calls))
self.assertEqual(url, responses.calls[0].request.url)
@responses.activate
def test_request_should_post_request_regardless_of_case(self):
"""Test that request() accepts the HTTP method name regardless of case"""
url = 'https://api.twitter.com/1.1/statuses/update.json'
self.register_response(responses.POST, url)
self.api.request(url, method='POST')
self.api.request(url, method='post')
self.assertEqual(2, len(responses.calls))
self.assertEqual('POST', responses.calls[0].request.method)
self.assertEqual('POST', responses.calls[1].request.method)
@responses.activate
def test_request_should_throw_exception_with_invalid_http_method(self):
"""Test that request() throws an exception when an invalid HTTP method is passed"""
#TODO(cash): should Twython catch the AttributeError and throw a TwythonError
self.assertRaises(AttributeError, self.api.request, endpoint='search/tweets', method='INVALID')
@responses.activate
def test_request_should_encode_boolean_as_lowercase_string(self):
"""Test that request() encodes a boolean parameter as a lowercase string"""
endpoint = 'search/tweets'
url = self.get_url(endpoint)
self.register_response(responses.GET, url)
self.api.request(endpoint, params={'include_entities': True})
self.api.request(endpoint, params={'include_entities': False})
self.assertEqual(url + '?include_entities=true', responses.calls[0].request.url)
self.assertEqual(url + '?include_entities=false', responses.calls[1].request.url)
@responses.activate
def test_request_should_handle_string_or_number_parameter(self):
"""Test that request() encodes a numeric or string parameter correctly"""
endpoint = 'search/tweets'
url = self.get_url(endpoint)
self.register_response(responses.GET, url)
self.api.request(endpoint, params={'lang': 'es'})
self.api.request(endpoint, params={'count': 50})
self.assertEqual(url + '?lang=es', responses.calls[0].request.url)
self.assertEqual(url + '?count=50', responses.calls[1].request.url)
@responses.activate
def test_request_should_encode_list_of_strings_as_string(self):
"""Test that request() encodes a list of strings as a comma-separated string"""
endpoint = 'search/tweets'
url = self.get_url(endpoint)
location = ['37.781157', '-122.39872', '1mi']
self.register_response(responses.GET, url)
self.api.request(endpoint, params={'geocode': location})
# requests url encodes the parameters so , is %2C
self.assertEqual(url + '?geocode=37.781157%2C-122.39872%2C1mi', responses.calls[0].request.url)
@responses.activate
def test_request_should_encode_numeric_list_as_string(self):
"""Test that request() encodes a list of numbers as a comma-separated string"""
endpoint = 'search/tweets'
url = self.get_url(endpoint)
location = [37.781157, -122.39872, '1mi']
self.register_response(responses.GET, url)
self.api.request(endpoint, params={'geocode': location})
self.assertEqual(url + '?geocode=37.781157%2C-122.39872%2C1mi', responses.calls[0].request.url)
@responses.activate
def test_request_should_ignore_bad_parameter(self):
"""Test that request() ignores unexpected parameter types"""
endpoint = 'search/tweets'
url = self.get_url(endpoint)
self.register_response(responses.GET, url)
self.api.request(endpoint, params={'geocode': self})
self.assertEqual(url, responses.calls[0].request.url)
@responses.activate
def test_request_should_handle_file_as_parameter(self):
"""Test that request() pulls a file out of params for requests lib"""
endpoint = 'account/update_profile_image'
url = self.get_url(endpoint)
self.register_response(responses.POST, url)
mock_file = StringIO("Twython test image")
self.api.request(endpoint, method='POST', params={'image': mock_file})
self.assertIn(b'filename="image"', responses.calls[0].request.body)
self.assertIn(b"Twython test image", responses.calls[0].request.body)
@responses.activate
def test_request_should_put_params_in_body_when_post(self):
"""Test that request() passes params as data when the request is a POST"""
endpoint = 'statuses/update'
url = self.get_url(endpoint)
self.register_response(responses.POST, url)
self.api.request(endpoint, method='POST', params={'status': 'this is a test'})
self.assertIn(b'status=this+is+a+test', responses.calls[0].request.body)
self.assertNotIn('status=this+is+a+test', responses.calls[0].request.url)
@responses.activate
def test_get_uses_get_method(self):
"""Test Twython generic GET request works"""
endpoint = 'account/verify_credentials'
url = self.get_url(endpoint)
self.register_response(responses.GET, url)
self.api.get(endpoint)
self.assertEqual(1, len(responses.calls))
self.assertEqual(url, responses.calls[0].request.url)
@responses.activate
def test_post_uses_post_method(self):
"""Test Twython generic POST request works"""
endpoint = 'statuses/update'
url = self.get_url(endpoint)
self.register_response(responses.POST, url)
self.api.post(endpoint, params={'status': 'I love Twython!'})
self.assertEqual(1, len(responses.calls))
self.assertEqual(url, responses.calls[0].request.url)
def test_raise_twython_error_on_request_exception(self):
"""Test if TwythonError is raised by a RequestException"""
with mock.patch.object(requests.Session, 'get') as get_mock:
# mocking an ssl cert error
get_mock.side_effect = requests.RequestException("hostname 'example.com' doesn't match ...")
self.assertRaises(TwythonError, self.api.get, 'https://example.com')
@responses.activate
def test_request_should_get_convert_json_to_data(self):
"""Test that Twython converts JSON data to a Python object"""
endpoint = 'statuses/show'
url = self.get_url(endpoint)
self.register_response(responses.GET, url, body='{"id": 210462857140252672}')
data = self.api.request(endpoint, params={'id': 210462857140252672})
self.assertEqual({'id': 210462857140252672}, data)
@responses.activate
def test_request_should_raise_exception_with_invalid_json(self):
"""Test that Twython handles invalid JSON (though Twitter should not return it)"""
endpoint = 'statuses/show'
url = self.get_url(endpoint)
self.register_response(responses.GET, url, body='{"id: 210462857140252672}')
self.assertRaises(TwythonError, self.api.request, endpoint, params={'id': 210462857140252672})
@responses.activate
def test_request_should_handle_401(self):
"""Test that Twython raises an auth error on 401 error"""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url, body='{"errors":[{"message":"Error"}]}', status=401)
self.assertRaises(TwythonAuthError, self.api.request, endpoint)
@responses.activate
def test_request_should_handle_400_for_missing_auth_data(self):
"""Test that Twython raises an auth error on 400 error when no oauth data sent"""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url,
body='{"errors":[{"message":"Bad Authentication data"}]}', status=400)
self.assertRaises(TwythonAuthError, self.api.request, endpoint)
@responses.activate
def test_request_should_handle_400_that_is_not_auth_related(self):
"""Test that Twython raises a normal error on 400 error when unrelated to authorization"""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url,
body='{"errors":[{"message":"Bad request"}]}', status=400)
self.assertRaises(TwythonError, self.api.request, endpoint)
@responses.activate
def test_request_should_handle_rate_limit(self):
"""Test that Twython raises an rate limit error on 429"""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url,
body='{"errors":[{"message":"Rate Limit"}]}', status=429)
self.assertRaises(TwythonRateLimitError, self.api.request, endpoint)
@responses.activate
def test_get_lastfunction_header_should_return_header(self):
"""Test getting last specific header of the last API call works"""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url, adding_headers={'x-rate-limit-remaining': 37})
self.api.get(endpoint)
value = self.api.get_lastfunction_header('x-rate-limit-remaining')
self.assertEqual(37, value)
value2 = self.api.get_lastfunction_header('does-not-exist')
self.assertIsNone(value2)
value3 = self.api.get_lastfunction_header('not-there-either', 96)
self.assertEqual(96, value3)
def test_get_lastfunction_header_should_raise_error_when_no_previous_call(self):
"""Test attempting to get a header when no API call was made raises a TwythonError"""
self.assertRaises(TwythonError, self.api.get_lastfunction_header, 'no-api-call-was-made')
# Static methods
def test_construct_api_url(self):
"""Test constructing a Twitter API url works as we expect"""
url = 'https://api.twitter.com/1.1/search/tweets.json'
constructed_url = self.api.construct_api_url(url, q='#twitter')
self.assertEqual(constructed_url, 'https://api.twitter.com/1.1/search/tweets.json?q=%23twitter')
def test_get(self):
"""Test Twython generic GET request works"""
self.api.get('account/verify_credentials')
def test_post(self):
"""Test Twython generic POST request works, with a full url and
with just an endpoint"""
update_url = 'https://api.twitter.com/1.1/statuses/update.json'
status = self.api.post(update_url, params={'status': 'I love Twython! %s' % int(time.time())})
self.api.post('statuses/destroy/%s' % status['id_str'])
def test_get_lastfunction_header(self):
"""Test getting last specific header of the last API call works"""
self.api.get('statuses/home_timeline')
self.api.get_lastfunction_header('x-rate-limit-remaining')
def test_get_lastfunction_header_not_present(self):
"""Test getting specific header that does not exist from the last call returns None"""
self.api.get('statuses/home_timeline')
header = self.api.get_lastfunction_header('does-not-exist')
self.assertEqual(header, None)
def test_get_lastfunction_header_no_last_api_call(self):
"""Test attempting to get a header when no API call was made raises a TwythonError"""
self.assertRaises(TwythonError, self.api.get_lastfunction_header,
'no-api-call-was-made')
def test_cursor(self):
"""Test looping through the generator results works, at least once that is"""
search = self.api.cursor(self.api.search, q='twitter', count=1)
counter = 0
while counter < 2:
counter += 1
result = next(search)
new_id_str = int(result['id_str'])
if counter == 1:
prev_id_str = new_id_str
time.sleep(1) # Give time for another tweet to come into search
if counter == 2:
self.assertTrue(new_id_str > prev_id_str)
def test_encode(self):
"""Test encoding UTF-8 works"""
self.api.encode('Twython is awesome!')
@ -98,406 +303,6 @@ class TwythonAPITestCase(unittest.TestCase):
def test_html_for_tweet_short_url(self):
"""Test using expanded url in HTML for Tweet displays full urls"""
tweet_text = self.api.html_for_tweet(test_tweet_object, False)
# Make sure HTML doesn't contain the display OR exapanded url
# Make sure HTML doesn't contain the display OR expanded url
self.assertTrue(not 'http://google.com' in tweet_text)
self.assertTrue(not 'google.com' in tweet_text)
def test_raise_error_on_bad_ssl_cert(self):
"""Test TwythonError is raised by a RequestException when an actual HTTP happens"""
self.assertRaises(TwythonError, self.api.get, 'https://example.com')
# Timelines
def test_get_mentions_timeline(self):
"""Test returning mentions timeline for authenticated user succeeds"""
self.api.get_mentions_timeline()
def test_get_user_timeline(self):
"""Test returning timeline for authenticated user and random user
succeeds"""
self.api.get_user_timeline() # Authenticated User Timeline
self.api.get_user_timeline(screen_name='twitter') # Random User Timeline
def test_get_protected_user_timeline_following(self):
"""Test returning a protected user timeline who you are following
succeeds"""
self.api.get_user_timeline(screen_name=protected_twitter_1)
def test_get_protected_user_timeline_not_following(self):
"""Test returning a protected user timeline who you are not following
fails and raise a TwythonAuthError"""
self.assertRaises(TwythonAuthError, self.api.get_user_timeline,
screen_name=protected_twitter_2)
def test_retweeted_of_me(self):
"""Test that getting recent tweets by authenticated user that have
been retweeted by others succeeds"""
self.api.retweeted_of_me()
def test_get_home_timeline(self):
"""Test returning home timeline for authenticated user succeeds"""
self.api.get_home_timeline()
# Tweets
def test_get_retweets(self):
"""Test getting retweets of a specific tweet succeeds"""
self.api.get_retweets(id=test_tweet_id)
def test_show_status(self):
"""Test returning a single status details succeeds"""
self.api.show_status(id=test_tweet_id)
def test_update_and_destroy_status(self):
"""Test updating and deleting a status succeeds"""
status = self.api.update_status(status='Test post just to get deleted :( %s' % int(time.time()))
self.api.destroy_status(id=status['id_str'])
def test_get_oembed_tweet(self):
"""Test getting info to embed tweet on Third Party site succeeds"""
self.api.get_oembed_tweet(id='99530515043983360')
def test_get_retweeters_ids(self):
"""Test getting ids for people who retweeted a tweet succeeds"""
self.api.get_retweeters_ids(id='99530515043983360')
# Search
def test_search(self):
"""Test searching tweets succeeds"""
self.api.search(q='twitter')
# Direct Messages
def test_get_direct_messages(self):
"""Test getting the authenticated users direct messages succeeds"""
self.api.get_direct_messages()
def test_get_sent_messages(self):
"""Test getting the authenticated users direct messages they've
sent succeeds"""
self.api.get_sent_messages()
def test_send_get_and_destroy_direct_message(self):
"""Test sending, getting, then destory a direct message succeeds"""
message = self.api.send_direct_message(screen_name=protected_twitter_1,
text='Hey d00d! %s' % int(time.time()))
self.api.get_direct_message(id=message['id_str'])
self.api.destroy_direct_message(id=message['id_str'])
def test_send_direct_message_to_non_follower(self):
"""Test sending a direct message to someone who doesn't follow you
fails"""
self.assertRaises(TwythonError, self.api.send_direct_message,
screen_name=protected_twitter_2, text='Yo, man! %s' % int(time.time()))
# Friends & Followers
def test_get_user_ids_of_blocked_retweets(self):
"""Test that collection of user_ids that the authenticated user does
not want to receive retweets from succeeds"""
self.api.get_user_ids_of_blocked_retweets(stringify_ids=True)
def test_get_friends_ids(self):
"""Test returning ids of users the authenticated user and then a random
user is following succeeds"""
self.api.get_friends_ids()
self.api.get_friends_ids(screen_name='twitter')
def test_get_followers_ids(self):
"""Test returning ids of users the authenticated user and then a random
user are followed by succeeds"""
self.api.get_followers_ids()
self.api.get_followers_ids(screen_name='twitter')
def test_lookup_friendships(self):
"""Test returning relationships of the authenticating user to the
comma-separated list of up to 100 screen_names or user_ids provided
succeeds"""
self.api.lookup_friendships(screen_name='twitter,ryanmcgrath')
def test_get_incoming_friendship_ids(self):
"""Test returning incoming friendship ids succeeds"""
self.api.get_incoming_friendship_ids()
def test_get_outgoing_friendship_ids(self):
"""Test returning outgoing friendship ids succeeds"""
self.api.get_outgoing_friendship_ids()
def test_create_friendship(self):
"""Test creating a friendship succeeds"""
self.api.create_friendship(screen_name='justinbieber')
def test_destroy_friendship(self):
"""Test destroying a friendship succeeds"""
self.api.destroy_friendship(screen_name='justinbieber')
def test_update_friendship(self):
"""Test updating friendships succeeds"""
self.api.update_friendship(screen_name=protected_twitter_1,
retweets='true')
self.api.update_friendship(screen_name=protected_twitter_1,
retweets=False)
def test_show_friendships(self):
"""Test showing specific friendship succeeds"""
self.api.show_friendship(target_screen_name=protected_twitter_1)
def test_get_friends_list(self):
"""Test getting list of users authenticated user then random user is
following succeeds"""
self.api.get_friends_list()
self.api.get_friends_list(screen_name='twitter')
def test_get_followers_list(self):
"""Test getting list of users authenticated user then random user are
followed by succeeds"""
self.api.get_followers_list()
self.api.get_followers_list(screen_name='twitter')
# Users
def test_get_account_settings(self):
"""Test getting the authenticated user account settings succeeds"""
self.api.get_account_settings()
def test_verify_credentials(self):
"""Test representation of the authenticated user call succeeds"""
self.api.verify_credentials()
def test_update_account_settings(self):
"""Test updating a user account settings succeeds"""
self.api.update_account_settings(lang='en')
def test_update_delivery_service(self):
"""Test updating delivery settings fails because we don't have
a mobile number on the account"""
self.assertRaises(TwythonError, self.api.update_delivery_service,
device='none')
def test_update_profile(self):
"""Test updating profile succeeds"""
self.api.update_profile(include_entities='true')
def test_update_profile_colors(self):
"""Test updating profile colors succeeds"""
self.api.update_profile_colors(profile_background_color='3D3D3D')
def test_list_blocks(self):
"""Test listing users who are blocked by the authenticated user
succeeds"""
self.api.list_blocks()
def test_list_block_ids(self):
"""Test listing user ids who are blocked by the authenticated user
succeeds"""
self.api.list_block_ids()
def test_create_block(self):
"""Test blocking a user succeeds"""
self.api.create_block(screen_name='justinbieber')
def test_destroy_block(self):
"""Test unblocking a user succeeds"""
self.api.destroy_block(screen_name='justinbieber')
def test_lookup_user(self):
"""Test listing a number of user objects succeeds"""
self.api.lookup_user(screen_name='twitter,justinbieber')
def test_show_user(self):
"""Test showing one user works"""
self.api.show_user(screen_name='twitter')
def test_search_users(self):
"""Test that searching for users succeeds"""
self.api.search_users(q='Twitter API')
def test_get_contributees(self):
"""Test returning list of accounts the specified user can
contribute to succeeds"""
self.api.get_contributees(screen_name='TechCrunch')
def test_get_contributors(self):
"""Test returning list of accounts that contribute to the
authenticated user fails because we are not a Contributor account"""
self.assertRaises(TwythonError, self.api.get_contributors,
screen_name=screen_name)
def test_remove_profile_banner(self):
"""Test removing profile banner succeeds"""
self.api.remove_profile_banner()
def test_get_profile_banner_sizes(self):
"""Test getting list of profile banner sizes fails because
we have not uploaded a profile banner"""
self.assertRaises(TwythonError, self.api.get_profile_banner_sizes)
# Suggested Users
def test_get_user_suggestions_by_slug(self):
"""Test getting user suggestions by slug succeeds"""
self.api.get_user_suggestions_by_slug(slug='twitter')
def test_get_user_suggestions(self):
"""Test getting user suggestions succeeds"""
self.api.get_user_suggestions()
def test_get_user_suggestions_statuses_by_slug(self):
"""Test getting status of suggested users succeeds"""
self.api.get_user_suggestions_statuses_by_slug(slug='funny')
# Favorites
def test_get_favorites(self):
"""Test getting list of favorites for the authenticated
user succeeds"""
self.api.get_favorites()
def test_create_and_destroy_favorite(self):
"""Test creating and destroying a favorite on a tweet succeeds"""
self.api.create_favorite(id=test_tweet_id)
self.api.destroy_favorite(id=test_tweet_id)
# Lists
def test_show_lists(self):
"""Test show lists for specified user"""
self.api.show_lists(screen_name='twitter')
def test_get_list_statuses(self):
"""Test timeline of tweets authored by members of the
specified list succeeds"""
self.api.get_list_statuses(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
def test_create_update_destroy_list_add_remove_list_members(self):
"""Test create a list, adding and removing members then
deleting the list succeeds"""
the_list = self.api.create_list(name='Stuff %s' % int(time.time()))
list_id = the_list['id_str']
self.api.update_list(list_id=list_id, name='Stuff Renamed %s' % int(time.time()))
screen_names = ['johncena', 'xbox']
# Multi add/delete members
self.api.create_list_members(list_id=list_id,
screen_name=screen_names)
self.api.delete_list_members(list_id=list_id,
screen_name=screen_names)
# Single add/delete member
self.api.add_list_member(list_id=list_id, screen_name='justinbieber')
self.api.delete_list_member(list_id=list_id, screen_name='justinbieber')
self.api.delete_list(list_id=list_id)
def test_get_list_memberships(self):
"""Test list of memberhips the authenticated user succeeds"""
self.api.get_list_memberships()
def test_get_list_subscribers(self):
"""Test list of subscribers of a specific list succeeds"""
self.api.get_list_subscribers(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
def test_subscribe_is_subbed_and_unsubscribe_to_list(self):
"""Test subscribing, is a list sub and unsubbing to list succeeds"""
self.api.subscribe_to_list(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
# Returns 404 if user is not a subscriber
self.api.is_list_subscriber(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name,
screen_name=screen_name)
self.api.unsubscribe_from_list(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
def test_is_list_member(self):
"""Test returning if specified user is member of a list succeeds"""
# Returns 404 if not list member
self.api.is_list_member(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name,
screen_name='themattharris')
def test_get_list_members(self):
"""Test listing members of the specified list succeeds"""
self.api.get_list_members(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
def test_get_specific_list(self):
"""Test getting specific list succeeds"""
self.api.get_specific_list(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
def test_get_list_subscriptions(self):
"""Test collection of the lists the specified user is
subscribed to succeeds"""
self.api.get_list_subscriptions(screen_name='twitter')
def test_show_owned_lists(self):
"""Test collection of lists the specified user owns succeeds"""
self.api.show_owned_lists(screen_name='twitter')
# Saved Searches
def test_get_saved_searches(self):
"""Test getting list of saved searches for authenticated
user succeeds"""
self.api.get_saved_searches()
def test_create_get_destroy_saved_search(self):
"""Test getting list of saved searches for authenticated
user succeeds"""
saved_search = self.api.create_saved_search(query='#Twitter')
saved_search_id = saved_search['id_str']
self.api.show_saved_search(id=saved_search_id)
self.api.destroy_saved_search(id=saved_search_id)
# Places & Geo
def test_get_geo_info(self):
"""Test getting info about a geo location succeeds"""
self.api.get_geo_info(place_id='df51dec6f4ee2b2c')
def test_reverse_geo_code(self):
"""Test reversing geocode succeeds"""
self.api.reverse_geocode(lat='37.76893497', long='-122.42284884')
def test_search_geo(self):
"""Test search for places that can be attached
to a statuses/update succeeds"""
self.api.search_geo(query='Toronto')
def test_get_similar_places(self):
"""Test locates places near the given coordinates which
are similar in name succeeds"""
self.api.get_similar_places(lat='37', long='-122', name='Twitter HQ')
# Trends
def test_get_place_trends(self):
"""Test getting the top 10 trending topics for a specific
WOEID succeeds"""
self.api.get_place_trends(id=1)
def test_get_available_trends(self):
"""Test returning locations that Twitter has trending
topic information for succeeds"""
self.api.get_available_trends()
def test_get_closest_trends(self):
"""Test getting the locations that Twitter has trending topic
information for, closest to a specified location succeeds"""
self.api.get_closest_trends(lat='37', long='-122')
# Help
def test_get_twitter_configuration(self):
"""Test getting Twitter's configuration succeeds"""
self.api.get_twitter_configuration()
def test_get_supported_languages(self):
"""Test getting languages supported by Twitter succeeds"""
self.api.get_supported_languages()
def test_privacy_policy(self):
"""Test getting Twitter's Privacy Policy succeeds"""
self.api.get_privacy_policy()
def test_get_tos(self):
"""Test getting the Twitter Terms of Service succeeds"""
self.api.get_tos()
def test_get_application_rate_limit_status(self):
"""Test getting application rate limit status succeeds"""
self.oauth2_api.get_application_rate_limit_status()

503
tests/test_endpoints.py Normal file
View file

@ -0,0 +1,503 @@
from twython import Twython, TwythonError, TwythonAuthError
from .config import (
app_key, app_secret, oauth_token, oauth_token_secret,
protected_twitter_1, protected_twitter_2, screen_name,
test_tweet_id, test_list_slug, test_list_owner_screen_name,
access_token, test_tweet_object, test_tweet_html, unittest
)
import time
class TwythonEndpointsTestCase(unittest.TestCase):
def setUp(self):
client_args = {
'headers': {
'User-Agent': '__twython__ Test'
},
'allow_redirects': False
}
oauth2_client_args = {
'headers': {} # This is so we can hit coverage that Twython sets User-Agent for us if none is supplied
}
self.api = Twython(app_key, app_secret,
oauth_token, oauth_token_secret,
client_args=client_args)
self.oauth2_api = Twython(app_key, access_token=access_token,
client_args=oauth2_client_args)
# Timelines
@unittest.skip('skipping non-updated test')
def test_get_mentions_timeline(self):
"""Test returning mentions timeline for authenticated user succeeds"""
self.api.get_mentions_timeline()
@unittest.skip('skipping non-updated test')
def test_get_user_timeline(self):
"""Test returning timeline for authenticated user and random user
succeeds"""
self.api.get_user_timeline() # Authenticated User Timeline
self.api.get_user_timeline(screen_name='twitter') # Random User Timeline
@unittest.skip('skipping non-updated test')
def test_get_protected_user_timeline_following(self):
"""Test returning a protected user timeline who you are following
succeeds"""
self.api.get_user_timeline(screen_name=protected_twitter_1)
@unittest.skip('skipping non-updated test')
def test_get_protected_user_timeline_not_following(self):
"""Test returning a protected user timeline who you are not following
fails and raise a TwythonAuthError"""
self.assertRaises(TwythonAuthError, self.api.get_user_timeline,
screen_name=protected_twitter_2)
@unittest.skip('skipping non-updated test')
def test_retweeted_of_me(self):
"""Test that getting recent tweets by authenticated user that have
been retweeted by others succeeds"""
self.api.retweeted_of_me()
@unittest.skip('skipping non-updated test')
def test_get_home_timeline(self):
"""Test returning home timeline for authenticated user succeeds"""
self.api.get_home_timeline()
# Tweets
@unittest.skip('skipping non-updated test')
def test_get_retweets(self):
"""Test getting retweets of a specific tweet succeeds"""
self.api.get_retweets(id=test_tweet_id)
@unittest.skip('skipping non-updated test')
def test_show_status(self):
"""Test returning a single status details succeeds"""
self.api.show_status(id=test_tweet_id)
@unittest.skip('skipping non-updated test')
def test_update_and_destroy_status(self):
"""Test updating and deleting a status succeeds"""
status = self.api.update_status(status='Test post just to get deleted :( %s' % int(time.time()))
self.api.destroy_status(id=status['id_str'])
@unittest.skip('skipping non-updated test')
def test_get_oembed_tweet(self):
"""Test getting info to embed tweet on Third Party site succeeds"""
self.api.get_oembed_tweet(id='99530515043983360')
@unittest.skip('skipping non-updated test')
def test_get_retweeters_ids(self):
"""Test getting ids for people who retweeted a tweet succeeds"""
self.api.get_retweeters_ids(id='99530515043983360')
# Search
@unittest.skip('skipping non-updated test')
def test_search(self):
"""Test searching tweets succeeds"""
self.api.search(q='twitter')
# Direct Messages
@unittest.skip('skipping non-updated test')
def test_get_direct_messages(self):
"""Test getting the authenticated users direct messages succeeds"""
self.api.get_direct_messages()
@unittest.skip('skipping non-updated test')
def test_get_sent_messages(self):
"""Test getting the authenticated users direct messages they've
sent succeeds"""
self.api.get_sent_messages()
@unittest.skip('skipping non-updated test')
def test_send_get_and_destroy_direct_message(self):
"""Test sending, getting, then destory a direct message succeeds"""
message = self.api.send_direct_message(screen_name=protected_twitter_1,
text='Hey d00d! %s' % int(time.time()))
self.api.get_direct_message(id=message['id_str'])
self.api.destroy_direct_message(id=message['id_str'])
@unittest.skip('skipping non-updated test')
def test_send_direct_message_to_non_follower(self):
"""Test sending a direct message to someone who doesn't follow you
fails"""
self.assertRaises(TwythonError, self.api.send_direct_message,
screen_name=protected_twitter_2, text='Yo, man! %s' % int(time.time()))
# Friends & Followers
@unittest.skip('skipping non-updated test')
def test_get_user_ids_of_blocked_retweets(self):
"""Test that collection of user_ids that the authenticated user does
not want to receive retweets from succeeds"""
self.api.get_user_ids_of_blocked_retweets(stringify_ids=True)
@unittest.skip('skipping non-updated test')
def test_get_friends_ids(self):
"""Test returning ids of users the authenticated user and then a random
user is following succeeds"""
self.api.get_friends_ids()
self.api.get_friends_ids(screen_name='twitter')
@unittest.skip('skipping non-updated test')
def test_get_followers_ids(self):
"""Test returning ids of users the authenticated user and then a random
user are followed by succeeds"""
self.api.get_followers_ids()
self.api.get_followers_ids(screen_name='twitter')
@unittest.skip('skipping non-updated test')
def test_lookup_friendships(self):
"""Test returning relationships of the authenticating user to the
comma-separated list of up to 100 screen_names or user_ids provided
succeeds"""
self.api.lookup_friendships(screen_name='twitter,ryanmcgrath')
@unittest.skip('skipping non-updated test')
def test_get_incoming_friendship_ids(self):
"""Test returning incoming friendship ids succeeds"""
self.api.get_incoming_friendship_ids()
@unittest.skip('skipping non-updated test')
def test_get_outgoing_friendship_ids(self):
"""Test returning outgoing friendship ids succeeds"""
self.api.get_outgoing_friendship_ids()
@unittest.skip('skipping non-updated test')
def test_create_friendship(self):
"""Test creating a friendship succeeds"""
self.api.create_friendship(screen_name='justinbieber')
@unittest.skip('skipping non-updated test')
def test_destroy_friendship(self):
"""Test destroying a friendship succeeds"""
self.api.destroy_friendship(screen_name='justinbieber')
@unittest.skip('skipping non-updated test')
def test_update_friendship(self):
"""Test updating friendships succeeds"""
self.api.update_friendship(screen_name=protected_twitter_1,
retweets='true')
self.api.update_friendship(screen_name=protected_twitter_1,
retweets=False)
@unittest.skip('skipping non-updated test')
def test_show_friendships(self):
"""Test showing specific friendship succeeds"""
self.api.show_friendship(target_screen_name=protected_twitter_1)
@unittest.skip('skipping non-updated test')
def test_get_friends_list(self):
"""Test getting list of users authenticated user then random user is
following succeeds"""
self.api.get_friends_list()
self.api.get_friends_list(screen_name='twitter')
@unittest.skip('skipping non-updated test')
def test_get_followers_list(self):
"""Test getting list of users authenticated user then random user are
followed by succeeds"""
self.api.get_followers_list()
self.api.get_followers_list(screen_name='twitter')
# Users
@unittest.skip('skipping non-updated test')
def test_get_account_settings(self):
"""Test getting the authenticated user account settings succeeds"""
self.api.get_account_settings()
@unittest.skip('skipping non-updated test')
def test_verify_credentials(self):
"""Test representation of the authenticated user call succeeds"""
self.api.verify_credentials()
@unittest.skip('skipping non-updated test')
def test_update_account_settings(self):
"""Test updating a user account settings succeeds"""
self.api.update_account_settings(lang='en')
@unittest.skip('skipping non-updated test')
def test_update_delivery_service(self):
"""Test updating delivery settings fails because we don't have
a mobile number on the account"""
self.assertRaises(TwythonError, self.api.update_delivery_service,
device='none')
@unittest.skip('skipping non-updated test')
def test_update_profile(self):
"""Test updating profile succeeds"""
self.api.update_profile(include_entities='true')
@unittest.skip('skipping non-updated test')
def test_update_profile_colors(self):
"""Test updating profile colors succeeds"""
self.api.update_profile_colors(profile_background_color='3D3D3D')
@unittest.skip('skipping non-updated test')
def test_list_blocks(self):
"""Test listing users who are blocked by the authenticated user
succeeds"""
self.api.list_blocks()
@unittest.skip('skipping non-updated test')
def test_list_block_ids(self):
"""Test listing user ids who are blocked by the authenticated user
succeeds"""
self.api.list_block_ids()
@unittest.skip('skipping non-updated test')
def test_create_block(self):
"""Test blocking a user succeeds"""
self.api.create_block(screen_name='justinbieber')
@unittest.skip('skipping non-updated test')
def test_destroy_block(self):
"""Test unblocking a user succeeds"""
self.api.destroy_block(screen_name='justinbieber')
@unittest.skip('skipping non-updated test')
def test_lookup_user(self):
"""Test listing a number of user objects succeeds"""
self.api.lookup_user(screen_name='twitter,justinbieber')
@unittest.skip('skipping non-updated test')
def test_show_user(self):
"""Test showing one user works"""
self.api.show_user(screen_name='twitter')
@unittest.skip('skipping non-updated test')
def test_search_users(self):
"""Test that searching for users succeeds"""
self.api.search_users(q='Twitter API')
@unittest.skip('skipping non-updated test')
def test_get_contributees(self):
"""Test returning list of accounts the specified user can
contribute to succeeds"""
self.api.get_contributees(screen_name='TechCrunch')
@unittest.skip('skipping non-updated test')
def test_get_contributors(self):
"""Test returning list of accounts that contribute to the
authenticated user fails because we are not a Contributor account"""
self.assertRaises(TwythonError, self.api.get_contributors,
screen_name=screen_name)
@unittest.skip('skipping non-updated test')
def test_remove_profile_banner(self):
"""Test removing profile banner succeeds"""
self.api.remove_profile_banner()
@unittest.skip('skipping non-updated test')
def test_get_profile_banner_sizes(self):
"""Test getting list of profile banner sizes fails because
we have not uploaded a profile banner"""
self.assertRaises(TwythonError, self.api.get_profile_banner_sizes)
# Suggested Users
@unittest.skip('skipping non-updated test')
def test_get_user_suggestions_by_slug(self):
"""Test getting user suggestions by slug succeeds"""
self.api.get_user_suggestions_by_slug(slug='twitter')
@unittest.skip('skipping non-updated test')
def test_get_user_suggestions(self):
"""Test getting user suggestions succeeds"""
self.api.get_user_suggestions()
@unittest.skip('skipping non-updated test')
def test_get_user_suggestions_statuses_by_slug(self):
"""Test getting status of suggested users succeeds"""
self.api.get_user_suggestions_statuses_by_slug(slug='funny')
# Favorites
@unittest.skip('skipping non-updated test')
def test_get_favorites(self):
"""Test getting list of favorites for the authenticated
user succeeds"""
self.api.get_favorites()
@unittest.skip('skipping non-updated test')
def test_create_and_destroy_favorite(self):
"""Test creating and destroying a favorite on a tweet succeeds"""
self.api.create_favorite(id=test_tweet_id)
self.api.destroy_favorite(id=test_tweet_id)
# Lists
@unittest.skip('skipping non-updated test')
def test_show_lists(self):
"""Test show lists for specified user"""
self.api.show_lists(screen_name='twitter')
@unittest.skip('skipping non-updated test')
def test_get_list_statuses(self):
"""Test timeline of tweets authored by members of the
specified list succeeds"""
self.api.get_list_statuses(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
@unittest.skip('skipping non-updated test')
def test_create_update_destroy_list_add_remove_list_members(self):
"""Test create a list, adding and removing members then
deleting the list succeeds"""
the_list = self.api.create_list(name='Stuff %s' % int(time.time()))
list_id = the_list['id_str']
self.api.update_list(list_id=list_id, name='Stuff Renamed %s' % int(time.time()))
screen_names = ['johncena', 'xbox']
# Multi add/delete members
self.api.create_list_members(list_id=list_id,
screen_name=screen_names)
self.api.delete_list_members(list_id=list_id,
screen_name=screen_names)
# Single add/delete member
self.api.add_list_member(list_id=list_id, screen_name='justinbieber')
self.api.delete_list_member(list_id=list_id, screen_name='justinbieber')
self.api.delete_list(list_id=list_id)
@unittest.skip('skipping non-updated test')
def test_get_list_memberships(self):
"""Test list of memberhips the authenticated user succeeds"""
self.api.get_list_memberships()
@unittest.skip('skipping non-updated test')
def test_get_list_subscribers(self):
"""Test list of subscribers of a specific list succeeds"""
self.api.get_list_subscribers(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
@unittest.skip('skipping non-updated test')
def test_subscribe_is_subbed_and_unsubscribe_to_list(self):
"""Test subscribing, is a list sub and unsubbing to list succeeds"""
self.api.subscribe_to_list(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
# Returns 404 if user is not a subscriber
self.api.is_list_subscriber(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name,
screen_name=screen_name)
self.api.unsubscribe_from_list(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
@unittest.skip('skipping non-updated test')
def test_is_list_member(self):
"""Test returning if specified user is member of a list succeeds"""
# Returns 404 if not list member
self.api.is_list_member(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name,
screen_name='themattharris')
@unittest.skip('skipping non-updated test')
def test_get_list_members(self):
"""Test listing members of the specified list succeeds"""
self.api.get_list_members(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
@unittest.skip('skipping non-updated test')
def test_get_specific_list(self):
"""Test getting specific list succeeds"""
self.api.get_specific_list(slug=test_list_slug,
owner_screen_name=test_list_owner_screen_name)
@unittest.skip('skipping non-updated test')
def test_get_list_subscriptions(self):
"""Test collection of the lists the specified user is
subscribed to succeeds"""
self.api.get_list_subscriptions(screen_name='twitter')
@unittest.skip('skipping non-updated test')
def test_show_owned_lists(self):
"""Test collection of lists the specified user owns succeeds"""
self.api.show_owned_lists(screen_name='twitter')
# Saved Searches
@unittest.skip('skipping non-updated test')
def test_get_saved_searches(self):
"""Test getting list of saved searches for authenticated
user succeeds"""
self.api.get_saved_searches()
@unittest.skip('skipping non-updated test')
def test_create_get_destroy_saved_search(self):
"""Test getting list of saved searches for authenticated
user succeeds"""
saved_search = self.api.create_saved_search(query='#Twitter')
saved_search_id = saved_search['id_str']
self.api.show_saved_search(id=saved_search_id)
self.api.destroy_saved_search(id=saved_search_id)
# Places & Geo
@unittest.skip('skipping non-updated test')
def test_get_geo_info(self):
"""Test getting info about a geo location succeeds"""
self.api.get_geo_info(place_id='df51dec6f4ee2b2c')
@unittest.skip('skipping non-updated test')
def test_reverse_geo_code(self):
"""Test reversing geocode succeeds"""
self.api.reverse_geocode(lat='37.76893497', long='-122.42284884')
@unittest.skip('skipping non-updated test')
def test_search_geo(self):
"""Test search for places that can be attached
to a statuses/update succeeds"""
self.api.search_geo(query='Toronto')
@unittest.skip('skipping non-updated test')
def test_get_similar_places(self):
"""Test locates places near the given coordinates which
are similar in name succeeds"""
self.api.get_similar_places(lat='37', long='-122', name='Twitter HQ')
# Trends
@unittest.skip('skipping non-updated test')
def test_get_place_trends(self):
"""Test getting the top 10 trending topics for a specific
WOEID succeeds"""
self.api.get_place_trends(id=1)
@unittest.skip('skipping non-updated test')
def test_get_available_trends(self):
"""Test returning locations that Twitter has trending
topic information for succeeds"""
self.api.get_available_trends()
@unittest.skip('skipping non-updated test')
def test_get_closest_trends(self):
"""Test getting the locations that Twitter has trending topic
information for, closest to a specified location succeeds"""
self.api.get_closest_trends(lat='37', long='-122')
# Help
@unittest.skip('skipping non-updated test')
def test_get_twitter_configuration(self):
"""Test getting Twitter's configuration succeeds"""
self.api.get_twitter_configuration()
@unittest.skip('skipping non-updated test')
def test_get_supported_languages(self):
"""Test getting languages supported by Twitter succeeds"""
self.api.get_supported_languages()
@unittest.skip('skipping non-updated test')
def test_privacy_policy(self):
"""Test getting Twitter's Privacy Policy succeeds"""
self.api.get_privacy_policy()
@unittest.skip('skipping non-updated test')
def test_get_tos(self):
"""Test getting the Twitter Terms of Service succeeds"""
self.api.get_tos()
@unittest.skip('skipping non-updated test')
def test_get_application_rate_limit_status(self):
"""Test getting application rate limit status succeeds"""
self.oauth2_api.get_application_rate_limit_status()

9
tests/test_streaming.py
View file

@ -1,11 +1,9 @@
from twython import TwythonStreamer, TwythonStreamError
from .config import (
app_key, app_secret, oauth_token, oauth_token_secret
app_key, app_secret, oauth_token, oauth_token_secret, unittest
)
import unittest
class TwythonStreamTestCase(unittest.TestCase):
def setUp(self):
@ -29,19 +27,24 @@ class TwythonStreamTestCase(unittest.TestCase):
oauth_token, oauth_token_secret,
client_args=client_args)
@unittest.skip('skipping non-updated test')
def test_stream_status_filter(self):
self.api.statuses.filter(track='twitter')
@unittest.skip('skipping non-updated test')
def test_stream_status_sample(self):
self.api.statuses.sample()
@unittest.skip('skipping non-updated test')
def test_stream_status_firehose(self):
self.assertRaises(TwythonStreamError, self.api.statuses.firehose,
track='twitter')
@unittest.skip('skipping non-updated test')
def test_stream_site(self):
self.assertRaises(TwythonStreamError, self.api.site,
follow='twitter')
@unittest.skip('skipping non-updated test')
def test_stream_user(self):
self.api.user(track='twitter')

2
twython/__init__.py
View file

@ -18,7 +18,7 @@ Questions, comments? ryan@venodesigns.net
"""
__author__ = 'Ryan McGrath <ryan@venodesigns.net>'
__version__ = '3.1.0'
__version__ = '3.1.2'
from .api import Twython
from .streaming import TwythonStreamer

54
twython/api.py
View file

@ -143,7 +143,6 @@ class Twython(EndpointsMixin, object):
response = func(url, **requests_args)
except requests.RequestException as e:
raise TwythonError(str(e))
content = response.content.decode('utf-8')
# create stash for last function intel
self._last_call = {
@ -153,37 +152,18 @@ class Twython(EndpointsMixin, object):
'headers': response.headers,
'status_code': response.status_code,
'url': response.url,
'content': content,
'content': response.text,
}
# Wrap the json loads in a try, and defer an error
# Twitter will return invalid json with an error code in the headers
json_error = False
try:
try:
# try to get json
content = content.json()
except AttributeError:
# if unicode detected
content = json.loads(content)
except ValueError:
json_error = True
content = {}
# greater than 304 (not modified) is an error
if response.status_code > 304:
# If there is no error message, use a default.
errors = content.get('errors',
[{'message': 'An error occurred processing your request.'}])
if errors and isinstance(errors, list):
error_message = errors[0]['message']
else:
error_message = errors # pragma: no cover
error_message = self._get_error_message(response)
self._last_call['api_error'] = error_message
ExceptionType = TwythonError
if response.status_code == 429:
# Twitter API 1.1, always return 429 when rate limit is exceeded
ExceptionType = TwythonRateLimitError # pragma: no cover
ExceptionType = TwythonRateLimitError
elif response.status_code == 401 or 'Bad Authentication data' in error_message:
# Twitter API 1.1, returns a 401 Unauthorized or
# a 400 "Bad Authentication data" for invalid/expired app keys/user tokens
@ -193,12 +173,30 @@ class Twython(EndpointsMixin, object):
error_code=response.status_code,
retry_after=response.headers.get('retry-after'))
# if we have a json error here, then it's not an official Twitter API error
if json_error and not response.status_code in (200, 201, 202): # pragma: no cover
raise TwythonError('Response was not valid JSON, unable to decode.')
try:
content = response.json()
except ValueError:
raise TwythonError('Response was not valid JSON. Unable to decode.')
return content
def _get_error_message(self, response):
"""Parse and return the first error message"""
error_message = 'An error occurred processing your request.'
try:
content = response.json()
# {"errors":[{"code":34,"message":"Sorry, that page does not exist"}]}
error_message = content['errors'][0]['message']
except ValueError:
# bad json data from Twitter for an error
pass
except (KeyError, IndexError):
# missing data so fallback to default message
pass
return error_message
def request(self, endpoint, method='GET', params=None, version='1.1'):
"""Return dict of response received from Twitter's API
@ -392,7 +390,7 @@ class Twython(EndpointsMixin, object):
"""Returns a generator for results that match a specified query.
:param function: Instance of a Twython function (Twython.get_home_timeline, Twython.search)
:param \*\*params: Extra parameters to send with your request (usually parameters excepted by the Twitter API endpoint)
:param \*\*params: Extra parameters to send with your request (usually parameters accepted by the Twitter API endpoint)
:rtype: generator
Usage::

2
twython/exceptions.py
View file

@ -53,6 +53,8 @@ class TwythonRateLimitError(TwythonError): # pragma: no cover
msg = '%s (Retry after %d seconds)' % (msg, retry_after)
TwythonError.__init__(self, msg, error_code=error_code)
self.retry_after = retry_after
class TwythonStreamError(TwythonError):
"""Raised when an invalid response from the Stream API is received"""