Revision f7eb83d9dca3ba0ee93e629ba5322732f99a3a30 authored by Jeremy Low on 30 September 2018, 13:23:40 UTC, committed by Jeremy Low on 30 September 2018, 13:23:40 UTC
1 parent 8b65eed
test_api_30.py
# encoding: utf-8
from __future__ import unicode_literals, print_function
import json
import os
import re
import sys
from tempfile import NamedTemporaryFile
import unittest
try:
from unittest.mock import patch
except ImportError:
from mock import patch
import warnings
import twitter
import responses
from responses import GET, POST
warnings.filterwarnings('ignore', category=DeprecationWarning)
DEFAULT_URL = re.compile(r'https?://.*\.twitter.com/1\.1/.*')
class ErrNull(object):
""" Suppress output of tests while writing to stdout or stderr. This just
takes in data and does nothing with it.
"""
def write(self, data):
pass
class ApiTest(unittest.TestCase):
def setUp(self):
self.api = twitter.Api(
consumer_key='test',
consumer_secret='test',
access_token_key='test',
access_token_secret='test',
sleep_on_rate_limit=False,
chunk_size=500 * 1024)
self.base_url = 'https://api.twitter.com/1.1'
self._stderr = sys.stderr
sys.stderr = ErrNull()
def tearDown(self):
sys.stderr = self._stderr
pass
def testApiSetUp(self):
self.assertRaises(
twitter.TwitterError,
lambda: twitter.Api(consumer_key='test'))
def testSetAndClearCredentials(self):
api = twitter.Api()
api.SetCredentials(consumer_key='test',
consumer_secret='test',
access_token_key='test',
access_token_secret='test')
self.assertEqual(api._consumer_key, 'test')
self.assertEqual(api._consumer_secret, 'test')
self.assertEqual(api._access_token_key, 'test')
self.assertEqual(api._access_token_secret, 'test')
api.ClearCredentials()
self.assertFalse(all([
api._consumer_key,
api._consumer_secret,
api._access_token_key,
api._access_token_secret
]))
@responses.activate
def testApiRaisesAuthErrors(self):
responses.add(GET, DEFAULT_URL, body='')
api = twitter.Api()
api.SetCredentials(consumer_key='test',
consumer_secret='test',
access_token_key='test',
access_token_secret='test')
api._Api__auth = None
self.assertRaises(twitter.TwitterError, lambda: api.GetFollowers())
@responses.activate
def testAppOnlyAuth(self):
responses.add(method=POST,
url='https://api.twitter.com/oauth2/token',
body='{"token_type":"bearer","access_token":"testing"}')
api = twitter.Api(
consumer_key='test',
consumer_secret='test',
application_only_auth=True)
self.assertEqual(api._bearer_token['access_token'], "testing")
@responses.activate
def testGetHelpConfiguration(self):
with open('testdata/get_help_configuration.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetHelpConfiguration()
self.assertEqual(resp.get('short_url_length_https'), 23)
@responses.activate
def testGetShortUrlLength(self):
with open('testdata/get_help_configuration.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetShortUrlLength()
self.assertEqual(resp, 23)
resp = self.api.GetShortUrlLength(https=True)
self.assertEqual(resp, 23)
@responses.activate
def testGetTrendsCurrent(self):
with open('testdata/get_trends_current.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetTrendsCurrent()
self.assertTrue(type(resp[0]) is twitter.Trend)
@responses.activate
def testGetHomeTimeline(self):
with open('testdata/get_home_timeline.json') as f:
resp_data = f.read()
responses.add(
GET, 'https://api.twitter.com/1.1/statuses/home_timeline.json?tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetHomeTimeline()
status = resp[0]
self.assertEqual(type(status), twitter.Status)
self.assertEqual(status.id, 674674925823787008)
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetHomeTimeline(count='literally infinity'))
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetHomeTimeline(count=4000))
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetHomeTimeline(max_id='also infinity'))
self.assertRaises(twitter.TwitterError,
lambda: self.api.GetHomeTimeline(
since_id='still infinity'))
@responses.activate
def testGetHomeTimelineWithExclusions(self):
with open('testdata/get_home_timeline.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
self.assertTrue(self.api.GetHomeTimeline(count=100,
trim_user=True,
max_id=674674925823787008))
@responses.activate
def testGetUserTimelineByUserID(self):
with open('testdata/get_user_timeline.json') as f:
resp_data = f.read()
responses.add(responses.GET, DEFAULT_URL, body=resp_data, status=200)
resp = self.api.GetUserTimeline(user_id=673483)
self.assertTrue(type(resp[0]) is twitter.Status)
self.assertTrue(type(resp[0].user) is twitter.User)
self.assertEqual(resp[0].user.id, 673483)
@responses.activate
def testGetUserTimelineByScreenName(self):
with open('testdata/get_user_timeline.json') as f:
resp_data = f.read()
responses.add(
GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetUserTimeline(screen_name='dewitt')
self.assertEqual(resp[0].id, 675055636267298821)
self.assertTrue(resp)
@responses.activate
def testGetRetweets(self):
with open('testdata/get_retweets.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetRetweets(statusid=397)
self.assertTrue(type(resp[0]) is twitter.Status)
self.assertTrue(type(resp[0].user) is twitter.User)
@responses.activate
def testGetRetweetsCount(self):
with open('testdata/get_retweets_count.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetRetweets(statusid=312, count=63)
self.assertTrue(len(resp), 63)
@responses.activate
def testGetRetweeters(self):
with open('testdata/get_retweeters.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetRetweeters(status_id=397)
self.assertTrue(type(resp) is list)
self.assertTrue(type(resp[0]) is int)
@responses.activate
def testGetBlocks(self):
with open('testdata/get_blocks_0.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/blocks/list.json?cursor=-1&stringify_ids=False&include_entities=False&skip_status=False&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
with open('testdata/get_blocks_1.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/blocks/list.json?cursor=1524574483549312671&stringify_ids=False&include_entities=False&skip_status=False&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetBlocks()
self.assertTrue(
isinstance(resp, list),
"Expected resp type to be list, got {0}".format(type(resp)))
self.assertTrue(
isinstance(resp[0], twitter.User),
"Expected type of first obj in resp to be twitter.User, got {0}".format(
type(resp[0])))
self.assertEqual(
len(resp), 2,
"Expected len of resp to be 2, got {0}".format(len(resp)))
self.assertEqual(
resp[0].screen_name, 'RedScareBot',
"Expected screen_name of 1st blocked user to be RedScareBot, was {0}".format(
resp[0].screen_name))
self.assertEqual(
resp[0].screen_name, 'RedScareBot',
"Expected screen_name of 2nd blocked user to be RedScareBot, was {0}".format(
resp[0].screen_name))
@responses.activate
def testGetBlocksPaged(self):
with open('testdata/get_blocks_1.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
ncur, pcur, resp = self.api.GetBlocksPaged(cursor=1524574483549312671)
self.assertTrue(
isinstance(resp, list),
"Expected list, got {0}".format(type(resp)))
self.assertTrue(
isinstance(resp[0], twitter.User),
"Expected twitter.User, got {0}".format(type(resp[0])))
self.assertEqual(
len(resp), 1,
"Expected len of resp to be 1, got {0}".format(len(resp)))
self.assertEqual(
resp[0].screen_name, 'RedScareBot',
"Expected username of blocked user to be RedScareBot, got {0}".format(
resp[0].screen_name))
@responses.activate
def testGetBlocksIDs(self):
with open('testdata/get_blocks_ids_0.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/blocks/ids.json?cursor=-1&stringify_ids=False&include_entities=True&skip_status=False&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
with open('testdata/get_blocks_ids_1.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/blocks/ids.json?cursor=1524566179872860311&stringify_ids=False&include_entities=True&skip_status=False&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetBlocksIDs()
self.assertTrue(
isinstance(resp, list),
"Expected list, got {0}".format(type(resp)))
self.assertTrue(
isinstance(resp[0], int),
"Expected list, got {0}".format(type(resp)))
self.assertEqual(
len(resp), 2,
"Expected len of resp to be 2, got {0}".format(len(resp)))
@responses.activate
def testGetBlocksIDsPaged(self):
with open('testdata/get_blocks_ids_1.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
_, _, resp = self.api.GetBlocksIDsPaged(cursor=1524566179872860311)
self.assertTrue(
isinstance(resp, list),
"Expected list, got {0}".format(type(resp)))
self.assertTrue(
isinstance(resp[0], int),
"Expected list, got {0}".format(type(resp)))
self.assertEqual(
len(resp), 1,
"Expected len of resp to be 1, got {0}".format(len(resp)))
@responses.activate
def testGetFriendIDs(self):
# First request for first 5000 friends
with open('testdata/get_friend_ids_0.json') as f:
resp_data = f.read()
responses.add(
GET,
'https://api.twitter.com/1.1/friends/ids.json?count=5000&cursor=-1&stringify_ids=False&screen_name=EricHolthaus&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
# Second (last) request for remaining friends
with open('testdata/get_friend_ids_1.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/friends/ids.json?stringify_ids=False&count=5000&cursor=1417903878302254556&screen_name=EricHolthaus&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetFriendIDs(screen_name='EricHolthaus')
self.assertTrue(type(resp) is list)
self.assertEqual(len(resp), 6452)
self.assertTrue(type(resp[0]) is int)
# Error checking
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetFriendIDs(total_count='infinity'))
@responses.activate
def testGetFriendIDsPaged(self):
with open('testdata/get_friend_ids_0.json') as f:
resp_data = f.read()
responses.add(responses.GET, DEFAULT_URL, body=resp_data, status=200)
ncursor, pcursor, resp = self.api.GetFriendIDsPaged(screen_name='EricHolthaus')
self.assertLessEqual(len(resp), 5000)
self.assertTrue(ncursor)
self.assertFalse(pcursor)
@responses.activate
def testGetFriendsPaged(self):
with open('testdata/get_friends_paged.json') as f:
resp_data = f.read()
responses.add(responses.GET, DEFAULT_URL, body=resp_data, status=200)
ncursor, pcursor, resp = self.api.GetFriendsPaged(screen_name='codebear', count=200)
self.assertEqual(ncursor, 1494734862149901956)
self.assertEqual(pcursor, 0)
self.assertEqual(len(resp), 200)
self.assertTrue(type(resp[0]) is twitter.User)
@responses.activate
def testGetFriendsPagedUID(self):
with open('testdata/get_friends_paged_uid.json') as f:
resp_data = f.read()
responses.add(responses.GET, DEFAULT_URL, body=resp_data, status=200)
ncursor, pcursor, resp = self.api.GetFriendsPaged(user_id=12, count=200)
self.assertEqual(ncursor, 1510410423140902959)
self.assertEqual(pcursor, 0)
self.assertEqual(len(resp), 200)
self.assertTrue(type(resp[0]) is twitter.User)
@responses.activate
def testGetFriendsAdditionalParams(self):
with open('testdata/get_friends_paged_additional_params.json') as f:
resp_data = f.read()
responses.add(responses.GET, DEFAULT_URL, body=resp_data, status=200)
ncursor, pcursor, resp = self.api.GetFriendsPaged(user_id=12,
count=200,
skip_status=True,
include_user_entities=True)
self.assertEqual(ncursor, 1510492845088954664)
self.assertEqual(pcursor, 0)
self.assertEqual(len(resp), 200)
self.assertTrue(type(resp[0]) is twitter.User)
@responses.activate
def testGetFriends(self):
"""
This is tedious, but the point is to add a responses endpoint for
each call that GetFriends() is going to make against the API and
have it return the appropriate json data.
"""
cursor = -1
for i in range(0, 5):
with open('testdata/get_friends_{0}.json'.format(i)) as f:
resp_data = f.read()
endpoint = 'https://api.twitter.com/1.1/friends/list.json?count=200&tweet_mode=compat&include_user_entities=True&screen_name=codebear&skip_status=False&cursor={0}'.format(cursor)
responses.add(GET, endpoint, body=resp_data, match_querystring=True)
cursor = json.loads(resp_data)['next_cursor']
resp = self.api.GetFriends(screen_name='codebear')
self.assertEqual(len(resp), 819)
@responses.activate
def testGetFriendsWithLimit(self):
with open('testdata/get_friends_0.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetFriends(screen_name='codebear', total_count=200)
self.assertEqual(len(resp), 200)
def testFriendsErrorChecking(self):
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetFriends(screen_name='jack',
total_count='infinity'))
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetFriendsPaged(screen_name='jack',
count='infinity'))
@responses.activate
def testGetFollowersIDs(self):
# First request for first 5000 followers
with open('testdata/get_follower_ids_0.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/followers/ids.json?tweet_mode=compat&cursor=-1&stringify_ids=False&count=5000&screen_name=GirlsMakeGames',
body=resp_data,
match_querystring=True,
status=200)
# Second (last) request for remaining followers
with open('testdata/get_follower_ids_1.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/followers/ids.json?tweet_mode=compat&count=5000&screen_name=GirlsMakeGames&cursor=1482201362283529597&stringify_ids=False',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetFollowerIDs(screen_name='GirlsMakeGames')
self.assertTrue(type(resp) is list)
self.assertEqual(len(resp), 7885)
self.assertTrue(type(resp[0]) is int)
# Error checking
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetFollowerIDs(total_count='infinity'))
@responses.activate
def testGetFollowers(self):
# First request for first 200 followers
with open('testdata/get_followers_0.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'{base_url}/followers/list.json?tweet_mode=compat&include_user_entities=True&count=200&screen_name=himawari8bot&skip_status=False&cursor=-1'.format(
base_url=self.api.base_url),
body=resp_data,
match_querystring=True,
status=200)
# Second (last) request for remaining followers
with open('testdata/get_followers_1.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'{base_url}/followers/list.json?tweet_mode=compat&include_user_entities=True&skip_status=False&count=200&screen_name=himawari8bot&cursor=1516850034842747602'.format(
base_url=self.api.base_url),
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetFollowers(screen_name='himawari8bot')
self.assertTrue(type(resp) is list)
self.assertTrue(type(resp[0]) is twitter.User)
self.assertEqual(len(resp), 335)
@responses.activate
def testGetFollowersPaged(self):
with open('testdata/get_followers_0.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
ncursor, pcursor, resp = self.api.GetFollowersPaged(screen_name='himawari8bot')
self.assertTrue(type(resp) is list)
self.assertTrue(type(resp[0]) is twitter.User)
self.assertEqual(len(resp), 200)
@responses.activate
def testGetFollowerIDsPaged(self):
with open('testdata/get_follower_ids_0.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/followers/ids.json?tweet_mode=compat&count=5000&stringify_ids=False&cursor=-1&screen_name=himawari8bot',
body=resp_data,
match_querystring=True,
status=200)
ncursor, pcursor, resp = self.api.GetFollowerIDsPaged(
screen_name='himawari8bot')
self.assertTrue(type(resp) is list)
self.assertTrue(type(resp[0]) is int)
self.assertEqual(len(resp), 5000)
with open('testdata/get_follower_ids_stringify.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/followers/ids.json?tweet_mode=compat&count=5000&stringify_ids=True&user_id=12&cursor=-1',
body=resp_data,
match_querystring=True,
status=200)
ncursor, pcursor, resp = self.api.GetFollowerIDsPaged(
user_id=12,
stringify_ids=True)
self.assertTrue(type(resp) is list)
if sys.version_info.major >= 3:
self.assertTrue(type(resp[0]) is str)
else:
self.assertTrue(type(resp[0]) is unicode)
self.assertEqual(len(resp), 5000)
@responses.activate
def testUsersLookup(self):
with open('testdata/users_lookup.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.UsersLookup(user_id=[718443])
self.assertTrue(type(resp) is list)
self.assertEqual(len(resp), 1)
user = resp[0]
self.assertTrue(type(user) is twitter.User)
self.assertEqual(user.screen_name, 'kesuke')
self.assertEqual(user.id, 718443)
@responses.activate
def testGetUser(self):
with open('testdata/get_user.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetUser(user_id=718443)
self.assertTrue(type(resp) is twitter.User)
self.assertEqual(resp.screen_name, 'kesuke')
self.assertEqual(resp.id, 718443)
@responses.activate
def testGetFavorites(self):
with open('testdata/get_favorites.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetFavorites(user_id=12, count=1, since_id=10, max_id=200)
self.assertTrue(type(resp) is list)
fav = resp[0]
self.assertEqual(fav.id, 677180133447372800)
self.assertIn("Extremely", fav.text)
@responses.activate
def testGetMentions(self):
with open('testdata/get_mentions.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetMentions(count=1)
self.assertTrue(type(resp) is list)
self.assertTrue([type(mention) is twitter.Status for mention in resp])
self.assertEqual(resp[0].id, 676148312349609985)
@responses.activate
def testGetListTimeline(self):
with open('testdata/get_list_timeline.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetListTimeline(list_id=None,
slug='space-bots',
owner_screen_name='inky')
self.assertTrue(type(resp) is list)
self.assertTrue([type(status) is twitter.Status for status in resp])
self.assertEqual(resp[0].id, 693191602957852676)
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetListTimeline(
list_id=None,
slug=None,
owner_id=None))
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetListTimeline(
list_id=None,
slug=None,
owner_screen_name=None))
@responses.activate
def testPostUpdate(self):
with open('testdata/post_update.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/statuses/update.json',
body=resp_data,
status=200)
post = self.api.PostUpdate(
status="blah Longitude coordinate of the tweet in degrees.")
self.assertTrue(type(post) is twitter.Status)
self.assertEqual(
post.text, "blah Longitude coordinate of the tweet in degrees.")
self.assertTrue(post.geo is None)
self.assertEqual(post.user.screen_name, 'notinourselves')
@responses.activate
def testPostUpdateExtraParams(self):
with open('testdata/post_update_extra_params.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/statuses/update.json',
body=resp_data,
status=200)
post = self.api.PostUpdate(
status="Not a dupe. Longitude coordinate of the tweet in degrees.",
in_reply_to_status_id=681496308251754496,
latitude=37.781157,
longitude=-122.398720,
place_id="1",
display_coordinates=True,
trim_user=True)
self.assertEqual(post.in_reply_to_status_id, 681496308251754496)
self.assertIsNotNone(post.coordinates)
@responses.activate
def testVerifyCredentials(self):
with open('testdata/verify_credentials.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.VerifyCredentials()
self.assertEqual(type(resp), twitter.User)
self.assertEqual(resp.name, 'notinourselves')
@responses.activate
def testVerifyCredentialsIncludeEmail(self):
with open('testdata/get_verify_credentials_include_email.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.VerifyCredentials(skip_status=True, include_email=True)
self.assertTrue(isinstance(resp, twitter.User))
self.assertEqual(resp.email, 'test@example.com')
@responses.activate
def testUpdateBanner(self):
responses.add(
POST,
'{0}/account/update_profile_banner.json'.format(self.api.base_url),
body=b'',
status=201
)
resp = self.api.UpdateBanner(image='testdata/168NQ.jpg')
self.assertTrue(resp)
@responses.activate
def testUpdateBanner422Error(self):
responses.add(
POST,
'{0}/account/update_profile_banner.json'.format(self.api.base_url),
body=b'',
status=422
)
self.assertRaises(
twitter.TwitterError,
lambda: self.api.UpdateBanner(image='testdata/168NQ.jpg')
)
try:
self.api.UpdateBanner(image='testdata/168NQ.jpg')
except twitter.TwitterError as e:
self.assertTrue("The image could not be resized or is too large." in str(e))
@responses.activate
def testUpdateBanner400Error(self):
responses.add(
POST,
'{0}/account/update_profile_banner.json'.format(self.api.base_url),
body=b'',
status=400
)
try:
self.api.UpdateBanner(image='testdata/168NQ.jpg')
except twitter.TwitterError as e:
self.assertTrue("Image data could not be processed" in str(e))
@responses.activate
def testGetMemberships(self):
with open('testdata/get_memberships.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetMemberships()
self.assertTrue(type(resp) is list)
self.assertTrue([type(lst) is twitter.List for lst in resp])
self.assertEqual(resp[0].id, 210635540)
@responses.activate
def testGetListsList(self):
with open('testdata/get_lists_list.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/list.json?tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetListsList()
self.assertTrue(type(resp) is list)
self.assertTrue([type(lst) is twitter.List for lst in resp])
self.assertEqual(resp[0].id, 189643778)
with open('testdata/get_lists_list_screen_name.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/list.json?tweet_mode=compat&screen_name=inky',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetListsList(screen_name='inky')
self.assertTrue(type(resp) is list)
self.assertTrue([type(lst) is twitter.List for lst in resp])
self.assertEqual(resp[0].id, 224581495)
with open('testdata/get_lists_list_user_id.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/list.json?tweet_mode=compat&user_id=13148',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetListsList(user_id=13148)
self.assertTrue(type(resp) is list)
self.assertTrue([type(lst) is twitter.List for lst in resp])
self.assertEqual(resp[0].id, 224581495)
@responses.activate
def testGetLists(self):
with open('testdata/get_lists.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetLists()
self.assertTrue(resp)
lst = resp[0]
self.assertEqual(lst.id, 229581524)
self.assertTrue(type(lst), twitter.List)
self.assertEqual(lst.full_name, "@notinourselves/test")
self.assertEqual(lst.slug, "test")
@responses.activate
def testGetListMembers(self):
with open('testdata/get_list_members_0.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/members.json?count=100&include_entities=False&skip_status=False&list_id=93527328&cursor=-1&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
with open('testdata/get_list_members_1.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/members.json?list_id=93527328&skip_status=False&include_entities=False&count=100&tweet_mode=compat&cursor=4611686020936348428',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetListMembers(list_id=93527328)
self.assertTrue(type(resp[0]) is twitter.User)
self.assertEqual(resp[0].id, 4048395140)
@responses.activate
def testGetListMembersPaged(self):
with open('testdata/get_list_members_0.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/members.json?count=100&include_entities=True&cursor=4611686020936348428&list_id=93527328&skip_status=False&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetListMembersPaged(list_id=93527328, cursor=4611686020936348428)
self.assertTrue([isinstance(u, twitter.User) for u in resp])
with open('testdata/get_list_members_extra_params.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/members.json?count=100&tweet_mode=compat&cursor=4611686020936348428&list_id=93527328&skip_status=True&include_entities=False',
body=resp_data,
match_querystring=True,
status=200)
_, _, resp = self.api.GetListMembersPaged(list_id=93527328,
cursor=4611686020936348428,
skip_status=True,
include_entities=False,
count=100)
self.assertFalse(resp[0].status)
@responses.activate
def testGetListTimeline(self):
with open('testdata/get_list_timeline.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/statuses.json?&list_id=229581524&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetListTimeline(list_id=229581524)
self.assertTrue(type(resp[0]) is twitter.Status)
with open('testdata/get_list_timeline_max_since.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/statuses.json?owner_screen_name=notinourselves&slug=test&max_id=692980243339071488&tweet_mode=compat&since_id=692829211019575296',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetListTimeline(slug='test',
owner_screen_name='notinourselves',
max_id=692980243339071488,
since_id=692829211019575296)
self.assertTrue([isinstance(s, twitter.Status) for s in resp])
self.assertEqual(len(resp), 7)
self.assertTrue([s.id >= 692829211019575296 for s in resp])
self.assertTrue([s.id <= 692980243339071488 for s in resp])
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetListTimeline(slug='test'))
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetListTimeline())
# 4012966701
with open('testdata/get_list_timeline_count_rts_ent.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/statuses.json?include_rts=False&count=13&tweet_mode=compat&include_entities=False&slug=test&owner_id=4012966701',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetListTimeline(slug='test',
owner_id=4012966701,
count=13,
include_entities=False,
include_rts=False)
self.assertEqual(len(resp), 13)
@responses.activate
def testCreateList(self):
with open('testdata/post_create_list.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/lists/create.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.CreateList(
name='test2',
mode='private',
description='test for python-twitter')
self.assertEqual(resp.id, 233452137)
self.assertEqual(resp.description, 'test for python-twitter')
self.assertEqual(resp.mode, 'private')
@responses.activate
def testDestroyList(self):
with open('testdata/post_destroy_list.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/lists/destroy.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.DestroyList(list_id=233452137)
self.assertEqual(resp.id, 233452137)
self.assertEqual(resp.member_count, 0)
@responses.activate
def testCreateSubscription(self):
with open('testdata/post_create_subscription.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/lists/subscribers/create.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.CreateSubscription(list_id=225486809)
self.assertEqual(resp.id, 225486809)
self.assertEqual(resp.name, 'my-bots')
@responses.activate
def testDestroySubscription(self):
with open('testdata/post_destroy_subscription.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/lists/subscribers/destroy.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.DestroySubscription(list_id=225486809)
self.assertEqual(resp.id, 225486809)
self.assertEqual(resp.name, 'my-bots')
@responses.activate
def testShowSubscription(self):
# User not a subscriber to the list.
with open('testdata/get_show_subscription_not_subscriber.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/subscribers/show.json?tweet_mode=compat&user_id=4040207472&list_id=189643778',
body=resp_data,
match_querystring=True,
status=200)
try:
self.api.ShowSubscription(list_id=189643778, user_id=4040207472)
except twitter.TwitterError as e:
self.assertIn(
"The specified user is not a subscriber of this list.",
str(e.message))
# User is a subscriber to list
with open('testdata/get_show_subscription.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/subscribers/show.json?list_id=189643778&tweet_mode=compat&screen_name=__jcbl__',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.ShowSubscription(list_id=189643778,
screen_name='__jcbl__')
self.assertEqual(resp.id, 372018022)
self.assertEqual(resp.screen_name, '__jcbl__')
self.assertTrue(resp.status)
# User is subscriber, using extra params
with open('testdata/get_show_subscription_extra_params.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/subscribers/show.json?include_entities=True&tweet_mode=compat&list_id=18964377&skip_status=True&screen_name=__jcbl__',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.ShowSubscription(list_id=18964377,
screen_name='__jcbl__',
include_entities=True,
skip_status=True)
self.assertFalse(resp.status)
@responses.activate
def testGetSubscriptions(self):
with open('testdata/get_get_subscriptions.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetSubscriptions()
self.assertEqual(len(resp), 1)
self.assertEqual(resp[0].name, 'space bots')
@responses.activate
def testGetSubscriptionsSN(self):
with open('testdata/get_get_subscriptions_uid.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetSubscriptions(screen_name='inky')
self.assertEqual(len(resp), 20)
self.assertTrue([isinstance(l, twitter.List) for l in resp])
@responses.activate
def testGetMemberships(self):
with open('testdata/get_get_memberships.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/memberships.json?count=20&cursor=-1&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetMemberships()
self.assertEqual(len(resp), 1)
self.assertEqual(resp[0].name, 'my-bots')
with open('testdata/get_get_memberships_himawari8bot.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/lists/memberships.json?count=20&cursor=-1&screen_name=himawari8bot&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetMemberships(screen_name='himawari8bot')
self.assertEqual(len(resp), 20)
self.assertTrue([isinstance(lst, twitter.List) for lst in resp])
@responses.activate
def testCreateListsMember(self):
with open('testdata/post_create_lists_member.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/lists/members/create.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.CreateListsMember(list_id=229581524, user_id=372018022)
self.assertTrue(isinstance(resp, twitter.List))
self.assertEqual(resp.name, 'test')
self.assertEqual(resp.member_count, 2)
@responses.activate
def testCreateListsMemberMultiple(self):
with open('testdata/post_create_lists_member_multiple.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/lists/members/create_all.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.CreateListsMember(list_id=229581524,
user_id=[372018022, 4040207472])
self.assertTrue(isinstance(resp, twitter.List))
self.assertEqual(resp.name, 'test')
self.assertEqual(resp.member_count, 3)
@responses.activate
def testDestroyListsMember(self):
with open('testdata/post_destroy_lists_member.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/lists/members/destroy.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.DestroyListsMember(list_id=229581524, user_id=372018022)
self.assertTrue(isinstance(resp, twitter.List))
self.assertEqual(resp.name, 'test')
self.assertEqual(resp.member_count, 1)
@responses.activate
def testDestroyListsMemberMultiple(self):
with open('testdata/post_destroy_lists_member_multiple.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/lists/members/destroy_all.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.DestroyListsMember(list_id=229581524,
user_id=[372018022, 4040207472])
self.assertEqual(resp.member_count, 0)
self.assertEqual(resp.name, 'test')
self.assertTrue(isinstance(resp, twitter.List))
@responses.activate
def testPostUpdateWithMedia(self):
# API will first make a POST request to upload the file.
with open('testdata/post_upload_media_simple.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://upload.twitter.com/1.1/media/upload.json',
body=resp_data,
match_querystring=True,
status=200)
# Then the POST request to post a status with the media id attached.
with open('testdata/post_update_media_id.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/statuses/update.json?media_ids=697007311538229248',
body=resp_data,
match_querystring=True,
status=200)
# Local file
resp = self.api.PostUpdate(media='testdata/168NQ.jpg', status='test')
self.assertEqual(697007311538229248, resp.AsDict()['media'][0]['id'])
self.assertEqual(resp.text, "hi this is a test for media uploads with statuses https://t.co/FHgqb6iLOX")
# File object
with open('testdata/168NQ.jpg', 'rb') as f:
resp = self.api.PostUpdate(media=[f], status='test')
self.assertEqual(697007311538229248, resp.AsDict()['media'][0]['id'])
self.assertEqual(resp.text, "hi this is a test for media uploads with statuses https://t.co/FHgqb6iLOX")
# Media ID as int
resp = self.api.PostUpdate(media=697007311538229248, status='test')
# Media ID as list of ints
resp = self.api.PostUpdate(media=[697007311538229248], status='test')
responses.add(
POST,
"https://api.twitter.com/1.1/statuses/update.json?media_ids=697007311538229248,697007311538229249",
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.PostUpdate(
media=[697007311538229248, 697007311538229249], status='test')
@responses.activate
def testLookupFriendship(self):
with open('testdata/get_friendships_lookup_none.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/friendships/lookup.json?user_id=12&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
responses.add(
responses.GET,
'https://api.twitter.com/1.1/friendships/lookup.json?user_id=12,6385432&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
responses.add(
responses.GET,
'https://api.twitter.com/1.1/friendships/lookup.json?screen_name=jack&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
responses.add(
responses.GET,
'https://api.twitter.com/1.1/friendships/lookup.json?screen_name=jack,dickc&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.LookupFriendship(user_id=12)
self.assertTrue(isinstance(resp, list))
self.assertTrue(isinstance(resp[0], twitter.UserStatus))
self.assertEqual(resp[0].following, False)
self.assertEqual(resp[0].followed_by, False)
# If any of the following produce an unexpected result, the test will
# fail on a request to a URL that hasn't been set by responses:
test_user = twitter.User(id=12, screen_name='jack')
test_user2 = twitter.User(id=6385432, screen_name='dickc')
resp = self.api.LookupFriendship(screen_name='jack')
resp = self.api.LookupFriendship(screen_name=['jack'])
resp = self.api.LookupFriendship(screen_name=test_user)
resp = self.api.LookupFriendship(screen_name=[test_user, test_user2])
resp = self.api.LookupFriendship(user_id=12)
resp = self.api.LookupFriendship(user_id=[12])
resp = self.api.LookupFriendship(user_id=test_user)
resp = self.api.LookupFriendship(user_id=[test_user, test_user2])
self.assertRaises(
twitter.TwitterError,
lambda: self.api.LookupFriendship())
@responses.activate
def testLookupFriendshipMute(self):
with open('testdata/get_friendships_lookup_muting.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.LookupFriendship(screen_name='dickc')
self.assertEqual(resp[0].blocking, False)
self.assertEqual(resp[0].muting, True)
@responses.activate
def testLookupFriendshipBlockMute(self):
with open('testdata/get_friendships_lookup_muting_blocking.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.LookupFriendship(screen_name='dickc')
self.assertEqual(resp[0].muting, True)
self.assertEqual(resp[0].blocking, True)
@responses.activate
def testPostMediaMetadata(self):
responses.add(
POST,
'https://upload.twitter.com/1.1/media/metadata/create.json',
body=b'',
status=200)
resp = self.api.PostMediaMetadata(media_id=718561981427396608, alt_text='test')
# At the moment, all we can do is test if the method call works. The response
# body should be blank, with a 200 status on success.
self.assertTrue(resp)
@responses.activate
def testGetStatusWithExtAltText(self):
with open('testdata/get_status_ext_alt.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetStatus(status_id=724441953534877696)
self.assertEqual(resp.media[0].ext_alt_text, "\u201cJon Snow is dead.\u2026\u201d from \u201cGAME OF THRONES SEASON 6 EPISODES\u201d by HBO PR.")
@responses.activate
def testGetStatus(self):
with open('testdata/get_status.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetStatus(status_id=397)
self.assertTrue(type(resp) is twitter.Status)
self.assertEqual(resp.id, 397)
self.assertEqual(resp.user.id, 12)
self.assertFalse(resp != resp)
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetStatus(status_id='test'))
@responses.activate
def testGetStatusExtraParams(self):
with open('testdata/get_status_extra_params.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetStatus(status_id=397, trim_user=True, include_entities=False)
self.assertFalse(resp.user.screen_name)
@responses.activate
def testGetStatuses(self):
with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
with open('testdata/get_statuses.1.json') as f:
resp_data = f.read()
rsps.add(GET, DEFAULT_URL, body=resp_data)
with open('testdata/get_statuses.2.json') as f:
resp_data = f.read()
rsps.add(GET, DEFAULT_URL, body=resp_data)
with open('testdata/get_statuses.ids.txt') as f:
status_ids = [int(l) for l in f]
resp = self.api.GetStatuses(status_ids)
self.assertTrue(type(resp) is list)
self.assertEqual(set(respitem.id for respitem in resp), set(status_ids))
self.assertFalse(resp != resp)
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetStatuses(['test']))
@responses.activate
def testGetStatusesMap(self):
with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
with open('testdata/get_statuses.map.1.json') as f:
resp_data = f.read()
rsps.add(GET, DEFAULT_URL, body=resp_data)
with open('testdata/get_statuses.map.2.json') as f:
resp_data = f.read()
rsps.add(GET, DEFAULT_URL, body=resp_data)
with open('testdata/get_statuses.ids.txt') as f:
status_ids = [int(l) for l in f]
resp = self.api.GetStatuses(status_ids, map=True)
self.assertTrue(type(resp) is dict)
self.assertTrue(all([resp.get(status_id) for status_id in status_ids]))
self.assertFalse(resp != resp)
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetStatuses(['test'], map=True))
@responses.activate
def testGetStatusOembed(self):
with open('testdata/get_status_oembed.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/statuses/oembed.json?tweet_mode=compat&id=397',
body=resp_data,
match_querystring=True,
status=200)
responses.add(
responses.GET,
'https://api.twitter.com/1.1/statuses/oembed.json?tweet_mode=compat&url=https://twitter.com/jack/statuses/397',
body=resp_data,
match_querystring=True,
status=200)
resp_id = self.api.GetStatusOembed(status_id=397)
self.assertEqual(resp_id['url'], 'https://twitter.com/jack/statuses/397')
self.assertEqual(resp_id['provider_url'], 'https://twitter.com')
self.assertEqual(resp_id['provider_name'], 'Twitter')
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetStatusOembed(status_id='test'))
resp_url = self.api.GetStatusOembed(url="https://twitter.com/jack/statuses/397")
self.assertEqual(resp_id, resp_url)
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetStatusOembed(status_id=None, url=None))
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetStatusOembed(status_id=397, align='test'))
@responses.activate
def testGetMutes(self):
# First iteration of the loop to get all the user's mutes
with open('testdata/get_mutes_users_list_loop_0.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/mutes/users/list.json?cursor=-1&stringify_ids=False&include_entities=True&skip_status=False&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
# Last interation of that loop.
with open('testdata/get_mutes_users_list_loop_1.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/mutes/users/list.json?cursor=1535206520056388207&stringify_ids=False&include_entities=True&skip_status=False&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetMutes(include_entities=True)
self.assertEqual(len(resp), 82)
self.assertTrue(isinstance(resp[0], twitter.User))
@responses.activate
def testGetMutesIDs(self):
# First iteration of the loop to get all the user's mutes
with open('testdata/get_mutes_users_ids_loop_0.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/mutes/users/ids.json?cursor=-1&stringify_ids=False&include_entities=True&skip_status=False&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
# Last interation of that loop.
with open('testdata/get_mutes_users_ids_loop_1.json') as f:
resp_data = f.read()
responses.add(
responses.GET,
'https://api.twitter.com/1.1/mutes/users/ids.json?cursor=1535206520056565155&stringify_ids=False&include_entities=True&skip_status=False&tweet_mode=compat',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.GetMutesIDs()
self.assertEqual(len(resp), 82)
self.assertTrue(isinstance(resp[0], int))
@responses.activate
def testCreateBlock(self):
with open('testdata/post_blocks_create.json') as f:
resp_data = f.read()
responses.add(
POST,
DEFAULT_URL,
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.CreateBlock(screen_name='jack')
self.assertTrue(isinstance(resp, twitter.User))
self.assertEqual(resp.screen_name, 'jack')
resp = self.api.CreateBlock(user_id=12)
self.assertTrue(isinstance(resp, twitter.User))
self.assertEqual(resp.id, 12)
@responses.activate
def testDestroyBlock(self):
with open('testdata/post_blocks_destroy.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/blocks/destroy.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.DestroyBlock(screen_name='jack')
self.assertTrue(isinstance(resp, twitter.User))
self.assertEqual(resp.screen_name, 'jack')
resp = self.api.DestroyBlock(user_id=12)
self.assertTrue(isinstance(resp, twitter.User))
self.assertEqual(resp.id, 12)
@responses.activate
def testCreateMute(self):
with open('testdata/post_mutes_users_create.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/mutes/users/create.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.CreateMute(screen_name='jack')
self.assertTrue(isinstance(resp, twitter.User))
self.assertEqual(resp.screen_name, 'jack')
resp = self.api.CreateMute(user_id=12)
self.assertTrue(isinstance(resp, twitter.User))
self.assertEqual(resp.id, 12)
@responses.activate
def testDestroyMute(self):
with open('testdata/post_mutes_users_destroy.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/mutes/users/destroy.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.DestroyMute(screen_name='jack')
self.assertTrue(isinstance(resp, twitter.User))
self.assertEqual(resp.screen_name, 'jack')
resp = self.api.DestroyMute(user_id=12)
self.assertTrue(isinstance(resp, twitter.User))
self.assertEqual(resp.id, 12)
@responses.activate
def testMuteBlockParamsAndErrors(self):
# Basic type/error checking
self.assertRaises(
twitter.TwitterError,
lambda: self.api.CreateMute(user_id='test'))
self.assertRaises(
twitter.TwitterError,
lambda: self.api.CreateMute())
with open('testdata/post_mutes_users_create_skip_status.json') as f:
resp_data = f.read()
responses.add(
POST,
'https://api.twitter.com/1.1/mutes/users/create.json',
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.CreateMute(screen_name='jack', skip_status=True)
self.assertTrue(isinstance(resp, twitter.User))
self.assertFalse(resp.status)
@responses.activate
def testPostUploadMediaChunkedInit(self):
with open('testdata/post_upload_chunked_INIT.json') as f:
resp_data = f.read()
responses.add(POST, DEFAULT_URL, body=resp_data, status=200)
with open('testdata/corgi.gif', 'rb') as fp:
resp = self.api._UploadMediaChunkedInit(fp)
self.assertEqual(len(responses.calls), 1)
self.assertEqual(resp[0], 737956420046356480)
@responses.activate
def testPostUploadMediaChunkedAppend(self):
media_fp, filename, _, _ = twitter.twitter_utils.parse_media_file(
'testdata/corgi.gif')
responses.add(POST, DEFAULT_URL, body='', status=200)
resp = self.api._UploadMediaChunkedAppend(media_id=737956420046356480,
media_fp=media_fp,
filename=filename)
self.assertEqual(len(responses.calls), 7)
self.assertTrue(resp)
@responses.activate
def testPostUploadMediaChunkedAppendNonASCIIFilename(self):
media_fp, filename, _, _ = twitter.twitter_utils.parse_media_file(
'testdata/corgi.gif')
filename = "عَرَبِيّ"
responses.add(responses.POST, DEFAULT_URL, body='', status=200)
resp = self.api._UploadMediaChunkedAppend(media_id=737956420046356480,
media_fp=media_fp,
filename=filename)
self.assertEqual(len(responses.calls), 7)
@responses.activate
def testPostUploadMediaChunkedFinalize(self):
with open('testdata/post_upload_chunked_FINAL.json') as f:
resp_data = f.read()
responses.add(POST, DEFAULT_URL, body=resp_data, status=200)
resp = self.api._UploadMediaChunkedFinalize(media_id=737956420046356480)
self.assertEqual(len(responses.calls), 1)
self.assertTrue(resp)
@responses.activate
def testGetUserSuggestionCategories(self):
with open('testdata/get_user_suggestion_categories.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetUserSuggestionCategories()
self.assertTrue(type(resp[0]) is twitter.Category)
@responses.activate
def testGetUserSuggestion(self):
with open('testdata/get_user_suggestion.json') as f:
resp_data = f.read()
responses.add(responses.GET, DEFAULT_URL, body=resp_data, status=200)
category = twitter.Category(name='Funny', slug='funny', size=20)
resp = self.api.GetUserSuggestion(category=category)
self.assertTrue(type(resp[0]) is twitter.User)
@responses.activate
def testGetUserTimeSinceMax(self):
with open('testdata/get_user_timeline_sincemax.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetUserTimeline(user_id=12, since_id=757782013914951680, max_id=758097930670645248)
self.assertEqual(len(resp), 6)
@responses.activate
def testGetUserTimelineCount(self):
with open('testdata/get_user_timeline_count.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetUserTimeline(user_id=12, count=63)
self.assertEqual(len(resp), 63)
@responses.activate
def testDestroyStatus(self):
with open('testdata/post_destroy_status.json') as f:
resp_data = f.read()
responses.add(
POST,
DEFAULT_URL,
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.DestroyStatus(status_id=746507834003578880)
self.assertTrue(isinstance(resp, twitter.models.Status))
self.assertEqual(resp.id, 746507834003578880)
@responses.activate
def testCreateFavorite(self):
with open('testdata/post_create_favorite.json') as f:
resp_data = f.read()
responses.add(POST, DEFAULT_URL, body=resp_data, status=200)
resp = self.api.CreateFavorite(status_id=757283981683412992)
self.assertEqual(resp.id, 757283981683412992)
status = twitter.models.Status(id=757283981683412992)
resp = self.api.CreateFavorite(status)
self.assertEqual(resp.id, 757283981683412992)
self.assertRaises(
twitter.TwitterError,
lambda: self.api.CreateFavorite(status=None, status_id=None))
@responses.activate
def testDestroyFavorite(self):
with open('testdata/post_destroy_favorite.json') as f:
resp_data = f.read()
responses.add(POST, DEFAULT_URL, body=resp_data, status=200)
resp = self.api.DestroyFavorite(status_id=757283981683412992)
self.assertEqual(resp.id, 757283981683412992)
status = twitter.models.Status(id=757283981683412992)
resp = self.api.DestroyFavorite(status)
self.assertEqual(resp.id, 757283981683412992)
self.assertRaises(
twitter.TwitterError,
lambda: self.api.DestroyFavorite(status=None, status_id=None))
@responses.activate
def testPostDirectMessage(self):
with open('testdata/post_post_direct_message.json') as f:
resp_data = f.read()
responses.add(
POST,
DEFAULT_URL,
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.PostDirectMessage(text="test message", user_id=372018022)
self.assertEqual(resp.text, "test message")
self.assertEqual(resp.sender_id, "4012966701")
self.assertEqual(resp.recipient_id, "372018022")
self.assertTrue(resp._json)
@responses.activate
def testDestroyDirectMessage(self):
with open('testdata/post_destroy_direct_message.json') as f:
resp_data = f.read()
responses.add(
POST,
DEFAULT_URL,
body=resp_data,
match_querystring=True,
status=200)
resp = self.api.DestroyDirectMessage(message_id=761517675243679747)
@responses.activate
def testShowFriendship(self):
with open('testdata/get_show_friendship.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.ShowFriendship(source_user_id=4012966701, target_user_id=372018022)
self.assertTrue(resp['relationship']['target'].get('following', None))
resp = self.api.ShowFriendship(source_screen_name='notinourselves', target_screen_name='__jcbl__')
self.assertRaises(
twitter.TwitterError,
lambda: self.api.ShowFriendship(source_screen_name='notinourselves')
)
self.assertRaises(
twitter.TwitterError,
lambda: self.api.ShowFriendship(target_screen_name='__jcbl__')
)
@responses.activate
@patch('twitter.api.Api.UploadMediaChunked')
def test_UploadSmallVideoUsesChunkedData(self, mocker):
responses.add(POST, DEFAULT_URL, body='{}')
video = NamedTemporaryFile(suffix='.mp4')
video.write(b'10' * 1024)
video.seek(0, 0)
resp = self.api.PostUpdate('test', media=video)
assert os.path.getsize(video.name) <= 1024 * 1024
assert isinstance(resp, twitter.Status)
assert twitter.api.Api.UploadMediaChunked.called
@responses.activate
def test_post_retweet(self):
with open('testdata/post_retweet.json') as f:
resp_data = f.read()
responses.add(POST, DEFAULT_URL, body=resp_data)
resp = self.api.PostRetweet(status_id=967413349473574913, trim_user=True)
assert resp
assert resp.id == 967465567773839360
self.assertRaises(
twitter.TwitterError,
lambda: self.api.PostRetweet(status_id=0))
self.assertRaises(
twitter.TwitterError,
lambda: self.api.PostRetweet(status_id='asdf'))
@responses.activate
def test_get_retweets_of_me(self):
with open('testdata/get_retweets_of_me.json') as f:
resp_data = f.read()
responses.add(GET, DEFAULT_URL, body=resp_data)
resp = self.api.GetRetweetsOfMe(
count=1,
since_id=0,
max_id=100,
trim_user=True,
include_entities=True,
include_user_entities=True)
assert resp
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetRetweetsOfMe(count=200))
self.assertRaises(
twitter.TwitterError,
lambda: self.api.GetRetweetsOfMe(count='asdf'))
@responses.activate
def test_incoming_friendships(self):
with open('testdata/get_incoming_friendships.json') as f:
responses.add(GET, DEFAULT_URL, f.read())
resp = self.api.IncomingFriendship(cursor=1, stringify_ids=True)
assert resp
assert isinstance(resp, list)
assert resp[0] == 12
@responses.activate
def test_outgoing_friendships(self):
with open('testdata/get_outgoing_friendships.json') as f:
responses.add(GET, DEFAULT_URL, f.read())
resp = self.api.OutgoingFriendship(cursor=1, stringify_ids=True)
assert resp
assert isinstance(resp, list)
assert resp[0] == 12
@responses.activate
def test_update_profile(self):
with open('testdata/update_profile.json') as f:
responses.add(POST, DEFAULT_URL, f.read())
resp = self.api.UpdateProfile(
name='jeremy',
location='philly',
profileURL='example.com',
description='test',
profile_link_color='#e35',
include_entities=True,
skip_status=True)
assert resp
assert isinstance(resp, twitter.User)
Computing file changes ...
