tests/test_sanity.py

   1 # encoding: utf-8
   2 from __future__ import unicode_literals
   3
   4 import os
   5 from random import choice
   6 import time
   7 import pickle
   8 import json
   9
  10 from twitter import Twitter, NoAuth, OAuth, read_token_file, TwitterHTTPError
  11 from twitter.api import TwitterDictResponse, TwitterListResponse, POST_ACTIONS, method_for_uri
  12 from twitter.cmdline import CONSUMER_KEY, CONSUMER_SECRET
  13
  14 noauth = NoAuth()
  15 oauth = OAuth(*read_token_file('tests/oauth_creds')
  16               + (CONSUMER_KEY, CONSUMER_SECRET))
  17
  18 twitter11 = Twitter(domain='api.twitter.com',
  19                     auth=oauth,
  20                     api_version='1.1')
  21
  22 twitter_upl = Twitter(domain='upload.twitter.com',
  23                       auth=oauth,
  24                       api_version='1.1')
  25
  26 twitter11_na = Twitter(domain='api.twitter.com',
  27                        auth=noauth,
  28                        api_version='1.1')
  29
  30 AZaz = "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ"
  31
  32 b64_image_data = b"iVBORw0KGgoAAAANSUhEUgAAABAAAAAQCAIAAACQkWg2AAAAAXNSR0IArs4c6QAAAAlwSFlzAAALEwAACxMBAJqcGAAAAAd0SU1FB94JFhMBAJv5kaUAAAAZdEVYdENvbW1lbnQAQ3JlYXRlZCB3aXRoIEdJTVBXgQ4XAAAA4UlEQVQoz7WSIZLGIAxG6c5OFZjianBcIOfgPkju1DsEBWfAUEcNGGpY8Xe7dDoVFRvHfO8NJGRorZE39UVe1nd/WNfVObcsi3OOEAIASikAmOf5D2q/FWPUWgshKKWfiFIqhNBaxxhPjPQ05/z+Bs557xw9hBC89ymlu5BS8t6HEC5NW2sR8alRRLTWXoRSSinlSejT12M9BAAAgCeoTw9BSimlfBIu6WdYtVZEVErdaaUUItZaL/9wOsaY83YAMMb0dGtt6Jdv3/ec87ZtOWdCCGNsmibG2DiOJzP8+7b+AAOmsiPxyHWCAAAAAElFTkSuQmCC"
  33
  34 def get_random_str():
  35     return ''.join(choice(AZaz) for _ in range(10))
  36
  37
  38 def test_API_set_tweet(unicod=False):
  39     random_tweet = "A random tweet %s" % \
  40         ("with unicode üøπ" if unicod else "") + get_random_str()
  41     twitter11.statuses.update(status=random_tweet)
  42     time.sleep(5)
  43     recent = twitter11.statuses.user_timeline()
  44     assert recent
  45     assert isinstance(recent.rate_limit_remaining, int)
  46     assert isinstance(recent.rate_limit_reset, int)
  47     texts = [tweet['text'] for tweet in recent]
  48     assert random_tweet in texts
  49
  50 def test_API_set_unicode_tweet():
  51     test_API_set_tweet(unicod=True)
  52
  53
  54 def clean_link(text):
  55     pos = text.find(" https://t.co")
  56     if pos != -1:
  57         return text[:pos]
  58     return text
  59
  60 __location__ = os.path.realpath(
  61     os.path.join(os.getcwd(), os.path.dirname(__file__)))
  62
  63 def _img_data():
  64     return open(os.path.join(__location__, "test.png"), "rb").read()
  65
  66 def _test_API_old_media(img, _base64):
  67     random_tweet = (
  68         "A random twitpic with unicode üøπ"
  69         + get_random_str())
  70     params = {"status": random_tweet, "media[]": img, "_base64": _base64}
  71     twitter11.statuses.update_with_media(**params)
  72     time.sleep(5)
  73     recent = twitter11.statuses.user_timeline()
  74     assert recent
  75     texts = [clean_link(tweet['text']) for tweet in recent]
  76     assert random_tweet in texts
  77
  78 def test_API_set_unicode_twitpic_base64():
  79     _test_API_old_media(b64_image_data, True)
  80
  81 def test_API_set_unicode_twitpic_base64_string():
  82     _test_API_old_media(b64_image_data.decode('utf-8'), True)
  83
  84 def test_API_set_unicode_twitpic_auto_base64_convert():
  85     _test_API_old_media(_img_data(), False)
  86
  87 def _test_upload_media():
  88     res = twitter_upl.media.upload(media=_img_data())
  89     assert res
  90     assert res["media_id"]
  91     return str(res["media_id"])
  92
  93 def test_multitwitpic():
  94     pics = [_test_upload_media(), _test_upload_media(), _test_upload_media()]
  95     random_tweet = ("I can even tweet multiple pictures at once now! ★  "
  96         + get_random_str())
  97     res = twitter11.statuses.update(status=random_tweet, media_ids=",".join(pics))
  98     assert res
  99     assert res["extended_entities"]
 100     assert len(res["extended_entities"]["media"]) == len(pics)
 101     recent = twitter11.statuses.user_timeline()
 102     assert recent
 103     texts = [clean_link(t['text']) for t in recent]
 104     assert random_tweet in texts
 105
 106 def test_search():
 107     # In 1.1, search works on api.twitter.com not search.twitter.com
 108     # and requires authorisation
 109     results = twitter11.search.tweets(q='foo')
 110     assert results
 111
 112
 113 def test_get_trends():
 114     # This is one method of inserting parameters, using named
 115     # underscore params.
 116     world_trends = twitter11.trends.available(_woeid=1)
 117     assert world_trends
 118
 119
 120 def test_get_trends_2():
 121     # This is a nicer variation of the same call as above.
 122     world_trends = twitter11.trends._(1)
 123     assert world_trends
 124
 125
 126 def test_get_trends_3():
 127     # Of course they broke it all again in 1.1...
 128     assert twitter11.trends.place(_id=1)
 129
 130
 131 def test_TwitterHTTPError_raised_for_invalid_oauth():
 132     test_passed = False
 133     try:
 134         twitter11_na.statuses.mentions_timeline()
 135     except TwitterHTTPError:
 136         # this is the error we are looking for :)
 137         test_passed = True
 138     assert test_passed
 139
 140
 141 def test_picklability():
 142     res = TwitterDictResponse({'a': 'b'})
 143     p = pickle.dumps(res)
 144     res2 = pickle.loads(p)
 145     assert res == res2
 146     assert res2['a'] == 'b'
 147
 148     res = TwitterListResponse([1, 2, 3])
 149     p = pickle.dumps(res)
 150     res2 = pickle.loads(p)
 151     assert res == res2
 152     assert res2[2] == 3
 153
 154
 155 def test_jsonifability():
 156     res = TwitterDictResponse({'a': 'b'})
 157     p = json.dumps(res)
 158     res2 = json.loads(p)
 159     assert res == res2
 160     assert res2['a'] == 'b'
 161
 162     res = TwitterListResponse([1, 2, 3])
 163     p = json.dumps(res)
 164     res2 = json.loads(p)
 165     assert res == res2
 166     assert res2[2] == 3
 167
 168
 169 def test_method_for_uri():
 170     for action in POST_ACTIONS:
 171         assert method_for_uri(get_random_str() + '/' + action) == 'POST'
 172     assert method_for_uri('statuses/timeline') == 'GET'