Commit ab1e69f5 authored by drevicko's avatar drevicko
Browse files

Merge rebased branch 'master' into fix-missing-params-metadata

parents e0fefa20 638c1ccd
source = tweepy
......@@ -3,3 +3,6 @@
script: nosetests -v tests.test_api tests.test_streaming tests.test_cursors
language: python
- "2.6"
- "2.7"
- pip install -r test_requirements.txt
script: ./
- python: "2.6"
- TWITTER_USERNAME="tweepytest"
......@@ -20,8 +28,5 @@ env:
- "2.7"
- master
after_success: 'if [[ "$TRAVIS_PULL_REQUEST" == "false" ]]; then coveralls; fi'
Version 2.2
- Added update_profile_banner endpoint.
- Don't treat HTTP status codes in 200 range as errors.
- Tests no longer packaged into egg releases.
- Improve test stability and enable CI testing on pull requests.
- Removed Basic Auth.
- Use built-in timeout feature of httplib to fix appengine.
- Added retweeters() endpoint.
- Removed deprecated retweeted_by and retweeted_by_ids.
- Improved datetime parsing. Should be more thread safe.
- Enable coverage reporting. Upload reports to Coveralls.
- Removed deprecated test() endpoint.
- New stream listeners callback on_disconnect(). Called whenever
"disconnect" messages arrive from Twitter before connection is killed.
- [Compare View](
Version 2.1
- Added get_oembed().
......@@ -10,8 +29,7 @@ Version 2.1
- Added on_connect() callback to StreamListener.
- Switched API search() to v1.1 endpoint. Some breaking changes.
- Drop "page" based cursors and use "ID" based ones instead.
- [Compare
- [Compare View](
Version 2.0
Thank you to all who have contributed to this project!
If you contributed and not listed below please let me know.
Aaron Hill
Aaron Swartz
Adam Miskiewicz
MIT License
Copyright (c) 2009-2010 Joshua Roesslein
Copyright (c) 2013-2014 Joshua Roesslein
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
Tweepy [![Build Status](](
Tweepy: Twitter for Python!
*Twitter for Python!*
[![Build Status](](
[![Downloads](]( [![Downloads](](
[![Coverage Status](](
import tweepy
# === Basic Authentication ===
# *Note: Basic Authentication is deprecated and no longer supported on Twitter.
# It is still provided for use in services like which still suppports it.*
# This mode of authentication requires the user provide their username and plain text password.
# These credentials will then be provided for each request to the API for authentication.
# You would normally fetch this in your application
# by asking the user or loading from some dark place.
username = ""
password = ""
# Create an authentication handler passing it the username and password.
# We will use this object later on when creating our API object.
auth = tweepy.auth.BasicAuthHandler(username, password)
# Create the API object providing it the authentication handler to use.
# Each request will then be authenticated using this handler.
api = tweepy.API(auth)
api.update_status('Updating using basic authentication via Tweepy!')
#! /usr/bin/env bash
if [[ $TRAVIS_SECURE_ENV_VARS == "false" ]]; then
curl "" -o tests/record.json
USE_REPLAY=1 nosetests -v tests.test_api tests.test_utils
nosetests -v --with-coverage tests.test_api tests.test_streaming tests.test_cursors tests.test_utils
......@@ -10,6 +10,6 @@ setup(name="tweepy",
author="Joshua Roesslein",
packages = find_packages(),
keywords= "twitter library",
zip_safe = True)
keywords="twitter library",
import os
import sys
from unittest2 import TestCase
from httreplay import start_replay, stop_replay
from httreplay.utils import filter_headers_key
from tweepy.auth import OAuthHandler
from tweepy.api import API
username = os.environ.get('TWITTER_USERNAME', '')
username = os.environ.get('TWITTER_USERNAME', 'tweepytest')
oauth_consumer_key = os.environ.get('CONSUMER_KEY', '')
oauth_consumer_secret = os.environ.get('CONSUMER_SECRET', '')
oauth_token = os.environ.get('ACCESS_KEY', '')
oauth_token_secret = os.environ.get('ACCESS_SECRET', '')
use_replay = os.environ.get('USE_REPLAY', False)
class TweepyTestCase(TestCase):
def setUp(self):
self.auth = create_auth()
self.api = API(self.auth)
self.api.retry_count = 2
self.api.retry_delay = 5
if use_replay:
def filter_body(data): return ''
def tearDown(self):
if use_replay:
def create_auth():
auth = OAuthHandler(oauth_consumer_key, oauth_consumer_secret)
import random
import string
def mock_tweet():
"""Generate some random tweet text."""
count = random.randint(70, 140)
return ''.join([random.choice(string.letters) for i in xrange(count)])
import unittest
import unittest2 as unittest
import random
from time import sleep
import os
from nose import SkipTest
from tweepy import (API, OAuthHandler, Friendship, Cursor,
MemoryCache, FileCache)
from config import *
from tweepy import Friendship, MemoryCache, FileCache
from config import TweepyTestCase, username, use_replay
test_tweet_id = '266367358078169089'
tweet_text = 'testing 1000'
"""Unit tests"""
......@@ -27,14 +26,7 @@ class TweepyErrorTests(unittest.TestCase):
self.assertEqual(e.reason, e2.reason)
self.assertEqual(e.response, e2.response)
class TweepyAPITests(unittest.TestCase):
def setUp(self):
auth = OAuthHandler(oauth_consumer_key, oauth_consumer_secret)
auth.set_access_token(oauth_token, oauth_token_secret)
self.api = API(auth)
self.api.retry_count = 2
self.api.retry_delay = 5
class TweepyAPITests(TweepyTestCase):
# TODO: Actually have some sort of better assertion
def testgetoembed(self):
......@@ -62,12 +54,15 @@ class TweepyAPITests(unittest.TestCase):
def testretweets(self):
def testretweeters(self):
def testgetstatus(self):
def testupdateanddestroystatus(self):
# test update
text = 'testing %i' % random.randint(0, 1000)
text = tweet_text if use_replay else 'testing %i' % random.randint(0, 1000)
update = self.api.update_status(status=text)
self.assertEqual(update.text, text)
......@@ -75,6 +70,10 @@ class TweepyAPITests(unittest.TestCase):
deleted = self.api.destroy_status(
def testupdatestatuswithmedia(self):
update = self.api.update_with_media('examples/banner.png', status=tweet_text)
self.assertIn(tweet_text + '', update.text)
def testgetuser(self):
u = self.api.get_user('twitter')
self.assertEqual(u.screen_name, 'twitter')
......@@ -82,6 +81,12 @@ class TweepyAPITests(unittest.TestCase):
u = self.api.get_user(783214)
self.assertEqual(u.screen_name, 'twitter')
def testlookupusers(self):
def check(users):
self.assertEqual(len(users), 2)
check(self.api.lookup_users(user_ids=[6844292, 6253282]))
check(self.api.lookup_users(screen_names=['twitterapi', 'twitter']))
def testsearchusers(self):
......@@ -134,7 +139,7 @@ class TweepyAPITests(unittest.TestCase):
self.assertEqual(friend.screen_name, 'twitter')
def testshowfriendship(self):
source, target = self.api.show_friendship(target_screen_name='twtiter')
source, target = self.api.show_friendship(target_screen_name='twitter')
self.assert_(isinstance(source, Friendship))
self.assert_(isinstance(target, Friendship))
......@@ -197,6 +202,9 @@ class TweepyAPITests(unittest.TestCase):
def testupdateprofilebannerimage(self):
def testupdateprofile(self):
original =
profile = {
......@@ -268,6 +276,7 @@ class TweepyAPITests(unittest.TestCase):
self.assertEqual(, params['slug'])
def testlistmembers(self):
......@@ -288,7 +297,7 @@ class TweepyAPITests(unittest.TestCase):
self.api.list_subscribers('applepie', 'stars')
def testshowlistsubscriber(self):
self.assertTrue(self.api.show_list_subscriber('twitter', 'team', username))
self.assertTrue(self.api.show_list_subscriber('tweepytest', 'test', 'applepie'))
def testsavedsearches(self):
s = self.api.create_saved_search('test')
......@@ -312,6 +321,22 @@ class TweepyAPITests(unittest.TestCase):
self.assertTrue(place_name_in_list('Austin, TX',
self.api.reverse_geocode(lat=30.267370168467806, long= -97.74261474609375))) # Austin, TX, USA
def testsupportedlanguages(self):
languages = self.api.supported_languages()
expected_dict = {
"name": "English",
"code": "en",
"status": "production"
self.assertTrue(expected_dict in languages)
def testcachedresult(self):
self.api.cache = MemoryCache()
class TweepyCacheTests(unittest.TestCase):
timeout = 2.0
import unittest
import unittest2 as unittest
from config import *
from tweepy import API, OAuthHandler
import unittest
import unittest2 as unittest
from tweepy import API, Cursor
import unittest
import unittest2 as unittest
from tweepy.models import ResultSet
from time import sleep
import unittest
import unittest2 as unittest
from tweepy.api import API
from tweepy.auth import OAuthHandler
from tweepy.models import Status
from tweepy.streaming import Stream, StreamListener
from config import create_auth
from mock import mock_tweet
from test_utils import mock_tweet
from mock import MagicMock, patch
class MockStreamListener(StreamListener):
def __init__(self, test_case):
......@@ -24,6 +26,10 @@ class MockStreamListener(StreamListener):'timeout')
return False
def on_error(self, code):
print "response: %s" % code
return True
def on_status(self, status):
self.status_count += 1
self.test_case.assertIsInstance(status, Status)
......@@ -62,3 +68,53 @@ class TweepyStreamTests(unittest.TestCase):
def test_track_encoding(self):
s = Stream(None, None)
s._start = lambda async: None
# Should be UTF-8 encoded
self.assertEqual(u'Caf\xe9'.encode('utf8'), s.parameters['track'])
def test_follow_encoding(self):
s = Stream(None, None)
s._start = lambda async: None
# Should be UTF-8 encoded
self.assertEqual(u'Caf\xe9'.encode('utf8'), s.parameters['follow'])
class TweepyStreamBackoffTests(unittest.TestCase):
def setUp(self):
#bad auth causes twitter to return 401 errors
self.auth = OAuthHandler("bad-key", "bad-secret")
self.auth.set_access_token("bad-token", "bad-token-secret")
self.listener = MockStreamListener(self) = Stream(self.auth, self.listener)
def tearDown(self):
def test_exp_backoff(self): = Stream(self.auth, self.listener, timeout=3.0,
retry_count=1, retry_time=1.0, retry_time_cap=100.0)
# 1 retry, should be 4x the retry_time
self.assertEqual(, 4.0)
def test_exp_backoff_cap(self): = Stream(self.auth, self.listener, timeout=3.0,
retry_count=1, retry_time=1.0, retry_time_cap=3.0)
# 1 retry, but 4x the retry_time exceeds the cap, so should be capped
self.assertEqual(, 3.0)
mock_resp = MagicMock()
mock_resp.return_value.status = 420
@patch('httplib.HTTPConnection.getresponse', mock_resp)
def test_420(self): = Stream(self.auth, self.listener, timeout=3.0, retry_count=0,
retry_time=1.0, retry_420=1.5, retry_time_cap=20.0)
# no retries, but error 420, should be double the retry_420, not double the retry_time
self.assertEqual(, 3.0)
from unittest2 import TestCase
from tweepy.utils import *
import random
import string
def mock_tweet():
"""Generate some random tweet text."""
count = random.randint(70, 140)
return ''.join([random.choice(string.letters) for i in xrange(count)])
class TweepyUtilsTests(TestCase):
def testparse_datetime(self):
result = parse_datetime("Wed Aug 27 13:08:45 +0000 2008")
self.assertEqual(datetime(2008, 8, 27, 13, 8, 45), result)
def testlist_to_csv(self):
self.assertEqual("1,2,3", list_to_csv([1,2,3]))
list_to_csv(["bird", "tweet", "nest", "egg"]))
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment