2.10.1
- More test coverage! - Fix ``search_gen`` - Fixed ``get_lastfunction_header`` to actually do what its docstring says, returns ``None`` if header is not found - Updated some internal API code, ``__init__`` didn't need to have ``self.auth`` and ``self.headers`` because they were never used anywhere else but the ``__init__``
This commit is contained in:
parent
05ca86805f
commit
894e94a4cd
5 changed files with 102 additions and 24 deletions
|
|
@ -3,6 +3,13 @@
|
||||||
History
|
History
|
||||||
-------
|
-------
|
||||||
|
|
||||||
|
2.10.1 (2013-05-xx)
|
||||||
|
++++++++++++++++++
|
||||||
|
- More test coverage!
|
||||||
|
- Fix ``search_gen``
|
||||||
|
- Fixed ``get_lastfunction_header`` to actually do what its docstring says, returns ``None`` if header is not found
|
||||||
|
- Updated some internal API code, ``__init__`` didn't need to have ``self.auth`` and ``self.headers`` because they were never used anywhere else but the ``__init__``
|
||||||
|
|
||||||
2.10.0 (2013-05-21)
|
2.10.0 (2013-05-21)
|
||||||
++++++++++++++++++
|
++++++++++++++++++
|
||||||
- Added ``get_retweeters_ids`` method
|
- Added ``get_retweeters_ids`` method
|
||||||
|
|
|
||||||
4
setup.py
4
setup.py
|
|
@ -1,10 +1,12 @@
|
||||||
|
#!/usr/bin/env python
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
from setuptools import setup
|
from setuptools import setup
|
||||||
|
|
||||||
__author__ = 'Ryan McGrath <ryan@venodesigns.net>'
|
__author__ = 'Ryan McGrath <ryan@venodesigns.net>'
|
||||||
__version__ = '2.10.0'
|
__version__ = '2.10.1'
|
||||||
|
|
||||||
packages = [
|
packages = [
|
||||||
'twython',
|
'twython',
|
||||||
|
|
|
||||||
|
|
@ -24,6 +24,7 @@ test_list_id = os.environ.get('TEST_LIST_ID', '574') # 574 is @twitter/team
|
||||||
class TwythonAuthTestCase(unittest.TestCase):
|
class TwythonAuthTestCase(unittest.TestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.api = Twython(app_key, app_secret)
|
self.api = Twython(app_key, app_secret)
|
||||||
|
self.bad_api = Twython('BAD_APP_KEY', 'BAD_APP_SECRET')
|
||||||
|
|
||||||
def test_get_authentication_tokens(self):
|
def test_get_authentication_tokens(self):
|
||||||
'''Test getting authentication tokens works'''
|
'''Test getting authentication tokens works'''
|
||||||
|
|
@ -31,11 +32,76 @@ class TwythonAuthTestCase(unittest.TestCase):
|
||||||
force_login=True,
|
force_login=True,
|
||||||
screen_name=screen_name)
|
screen_name=screen_name)
|
||||||
|
|
||||||
|
def test_get_authentication_tokens_bad_tokens(self):
|
||||||
|
'''Test getting authentication tokens with bad tokens
|
||||||
|
raises TwythonAuthError'''
|
||||||
|
self.assertRaises(TwythonAuthError, self.api.get_authentication_tokens,
|
||||||
|
callback_url='http://google.com/')
|
||||||
|
|
||||||
|
def test_get_authorized_tokens_bad_tokens(self):
|
||||||
|
'''Test getting final tokens fails with wrong tokens'''
|
||||||
|
self.assertRaises(TwythonError, self.api.get_authorized_tokens,
|
||||||
|
'BAD_OAUTH_VERIFIER')
|
||||||
|
|
||||||
|
|
||||||
class TwythonAPITestCase(unittest.TestCase):
|
class TwythonAPITestCase(unittest.TestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.api = Twython(app_key, app_secret,
|
self.api = Twython(app_key, app_secret,
|
||||||
oauth_token, oauth_token_secret)
|
oauth_token, oauth_token_secret,
|
||||||
|
headers={'User-Agent': '__twython__ Test'})
|
||||||
|
|
||||||
|
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_shorten_url(self):
|
||||||
|
'''Test shortening a url works'''
|
||||||
|
self.api.shorten_url('http://google.com')
|
||||||
|
|
||||||
|
def test_shorten_url_no_shortner(self):
|
||||||
|
'''Test shortening a url with no shortener provided raises TwythonError'''
|
||||||
|
self.assertRaises(TwythonError, self.api.shorten_url,
|
||||||
|
'http://google.com', '')
|
||||||
|
|
||||||
|
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!'})
|
||||||
|
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_search_gen(self):
|
||||||
|
'''Test looping through the generator results works, at least once that is'''
|
||||||
|
search = self.api.search_gen('python')
|
||||||
|
for result in search:
|
||||||
|
if result:
|
||||||
|
break
|
||||||
|
|
||||||
|
def test_encode(self):
|
||||||
|
'''Test encoding UTF-8 works'''
|
||||||
|
self.api.encode('Twython is awesome!')
|
||||||
|
|
||||||
# Timelines
|
# Timelines
|
||||||
def test_get_mentions_timeline(self):
|
def test_get_mentions_timeline(self):
|
||||||
|
|
@ -84,9 +150,11 @@ class TwythonAPITestCase(unittest.TestCase):
|
||||||
|
|
||||||
def test_retweet_twice(self):
|
def test_retweet_twice(self):
|
||||||
'''Test that trying to retweet a tweet twice raises a TwythonError'''
|
'''Test that trying to retweet a tweet twice raises a TwythonError'''
|
||||||
retweet = self.api.retweet(id='99530515043983360')
|
tweets = self.api.search(q='twitter').get('statuses')
|
||||||
self.assertRaises(TwythonError, self.api.retweet,
|
if tweets:
|
||||||
id='99530515043983360')
|
retweet = self.api.retweet(id=tweets[0]['id_str'])
|
||||||
|
self.assertRaises(TwythonError, self.api.retweet,
|
||||||
|
id=tweets[0]['id_str'])
|
||||||
|
|
||||||
# Then clean up
|
# Then clean up
|
||||||
self.api.destroy_status(id=retweet['id_str'])
|
self.api.destroy_status(id=retweet['id_str'])
|
||||||
|
|
@ -132,7 +200,7 @@ class TwythonAPITestCase(unittest.TestCase):
|
||||||
def test_get_user_ids_of_blocked_retweets(self):
|
def test_get_user_ids_of_blocked_retweets(self):
|
||||||
'''Test that collection of user_ids that the authenticated user does
|
'''Test that collection of user_ids that the authenticated user does
|
||||||
not want to receive retweets from succeeds'''
|
not want to receive retweets from succeeds'''
|
||||||
self.api.get_user_ids_of_blocked_retweets(stringify_ids='true')
|
self.api.get_user_ids_of_blocked_retweets(stringify_ids=True)
|
||||||
|
|
||||||
def test_get_friends_ids(self):
|
def test_get_friends_ids(self):
|
||||||
'''Test returning ids of users the authenticated user and then a random
|
'''Test returning ids of users the authenticated user and then a random
|
||||||
|
|
|
||||||
|
|
@ -18,7 +18,7 @@ Questions, comments? ryan@venodesigns.net
|
||||||
"""
|
"""
|
||||||
|
|
||||||
__author__ = 'Ryan McGrath <ryan@venodesigns.net>'
|
__author__ = 'Ryan McGrath <ryan@venodesigns.net>'
|
||||||
__version__ = '2.10.0'
|
__version__ = '2.10.1'
|
||||||
|
|
||||||
from .twython import Twython
|
from .twython import Twython
|
||||||
from .streaming import TwythonStreamer
|
from .streaming import TwythonStreamer
|
||||||
|
|
|
||||||
|
|
@ -59,27 +59,27 @@ class Twython(object):
|
||||||
stacklevel=2
|
stacklevel=2
|
||||||
)
|
)
|
||||||
|
|
||||||
self.headers = {'User-Agent': 'Twython v' + __version__}
|
req_headers = {'User-Agent': 'Twython v' + __version__}
|
||||||
if headers:
|
if headers:
|
||||||
self.headers.update(headers)
|
req_headers.update(headers)
|
||||||
|
|
||||||
# Generate OAuth authentication object for the request
|
# Generate OAuth authentication object for the request
|
||||||
# If no keys/tokens are passed to __init__, self.auth=None allows for
|
# If no keys/tokens are passed to __init__, auth=None allows for
|
||||||
# unauthenticated requests, although I think all v1.1 requests need auth
|
# unauthenticated requests, although I think all v1.1 requests need auth
|
||||||
self.auth = None
|
auth = None
|
||||||
if self.app_key is not None and self.app_secret is not None and \
|
if self.app_key is not None and self.app_secret is not None and \
|
||||||
self.oauth_token is None and self.oauth_token_secret is None:
|
self.oauth_token is None and self.oauth_token_secret is None:
|
||||||
self.auth = OAuth1(self.app_key, self.app_secret)
|
auth = OAuth1(self.app_key, self.app_secret)
|
||||||
|
|
||||||
if self.app_key is not None and self.app_secret is not None and \
|
if self.app_key is not None and self.app_secret is not None and \
|
||||||
self.oauth_token is not None and self.oauth_token_secret is not None:
|
self.oauth_token is not None and self.oauth_token_secret is not None:
|
||||||
self.auth = OAuth1(self.app_key, self.app_secret,
|
auth = OAuth1(self.app_key, self.app_secret,
|
||||||
self.oauth_token, self.oauth_token_secret)
|
self.oauth_token, self.oauth_token_secret)
|
||||||
|
|
||||||
self.client = requests.Session()
|
self.client = requests.Session()
|
||||||
self.client.headers = self.headers
|
self.client.headers = req_headers
|
||||||
self.client.proxies = proxies
|
self.client.proxies = proxies
|
||||||
self.client.auth = self.auth
|
self.client.auth = auth
|
||||||
self.client.verify = ssl_verify
|
self.client.verify = ssl_verify
|
||||||
|
|
||||||
# register available funcs to allow listing name when debugging.
|
# register available funcs to allow listing name when debugging.
|
||||||
|
|
@ -208,7 +208,7 @@ class Twython(object):
|
||||||
|
|
||||||
def request(self, endpoint, method='GET', params=None, version='1.1'):
|
def request(self, endpoint, method='GET', params=None, version='1.1'):
|
||||||
# In case they want to pass a full Twitter URL
|
# In case they want to pass a full Twitter URL
|
||||||
# i.e. https://search.twitter.com/
|
# i.e. https://api.twitter.com/1.1/search/tweets.json
|
||||||
if endpoint.startswith('http://') or endpoint.startswith('https://'):
|
if endpoint.startswith('http://') or endpoint.startswith('https://'):
|
||||||
url = endpoint
|
url = endpoint
|
||||||
else:
|
else:
|
||||||
|
|
@ -241,9 +241,11 @@ class Twython(object):
|
||||||
"""
|
"""
|
||||||
if self._last_call is None:
|
if self._last_call is None:
|
||||||
raise TwythonError('This function must be called after an API call. It delivers header information.')
|
raise TwythonError('This function must be called after an API call. It delivers header information.')
|
||||||
|
|
||||||
if header in self._last_call['headers']:
|
if header in self._last_call['headers']:
|
||||||
return self._last_call['headers'][header]
|
return self._last_call['headers'][header]
|
||||||
return self._last_call
|
else:
|
||||||
|
return None
|
||||||
|
|
||||||
def get_authentication_tokens(self, callback_url=None, force_login=False, screen_name=''):
|
def get_authentication_tokens(self, callback_url=None, force_login=False, screen_name=''):
|
||||||
"""Returns a dict including an authorization URL (auth_url) to direct a user to
|
"""Returns a dict including an authorization URL (auth_url) to direct a user to
|
||||||
|
|
@ -325,7 +327,7 @@ class Twython(object):
|
||||||
stacklevel=2
|
stacklevel=2
|
||||||
)
|
)
|
||||||
|
|
||||||
if shortener == '':
|
if not shortener:
|
||||||
raise TwythonError('Please provide a URL shortening service.')
|
raise TwythonError('Please provide a URL shortening service.')
|
||||||
|
|
||||||
request = requests.get(shortener, params={
|
request = requests.get(shortener, params={
|
||||||
|
|
@ -336,7 +338,7 @@ class Twython(object):
|
||||||
if request.status_code in [301, 201, 200]:
|
if request.status_code in [301, 201, 200]:
|
||||||
return request.text
|
return request.text
|
||||||
else:
|
else:
|
||||||
raise TwythonError('shortenURL() failed with a %s error code.' % request.status_code)
|
raise TwythonError('shorten_url failed with a %s error code.' % request.status_code)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def constructApiURL(base_url, params):
|
def constructApiURL(base_url, params):
|
||||||
|
|
@ -373,17 +375,16 @@ class Twython(object):
|
||||||
|
|
||||||
See Twython.search() for acceptable parameters
|
See Twython.search() for acceptable parameters
|
||||||
|
|
||||||
e.g search = x.searchGen('python')
|
e.g search = x.search_gen('python')
|
||||||
for result in search:
|
for result in search:
|
||||||
print result
|
print result
|
||||||
"""
|
"""
|
||||||
kwargs['q'] = search_query
|
|
||||||
content = self.search(q=search_query, **kwargs)
|
content = self.search(q=search_query, **kwargs)
|
||||||
|
|
||||||
if not content['results']:
|
if not content.get('statuses'):
|
||||||
raise StopIteration
|
raise StopIteration
|
||||||
|
|
||||||
for tweet in content['results']:
|
for tweet in content['statuses']:
|
||||||
yield tweet
|
yield tweet
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue