]>
Commit | Line | Data |
---|---|---|
f0603331 MV |
1 | # encoding: utf-8 |
2 | from __future__ import unicode_literals | |
e541e268 | 3 | |
9999744f | 4 | import os |
e541e268 | 5 | from random import choice |
c77b5e4b | 6 | import time |
ed23f46c MV |
7 | import pickle |
8 | import json | |
e541e268 | 9 | |
32a4811d | 10 | from twitter import Twitter, NoAuth, OAuth, read_token_file, TwitterHTTPError |
1e95ec45 | 11 | from twitter.api import TwitterDictResponse, TwitterListResponse, POST_ACTIONS, method_for_uri |
e541e268 MV |
12 | from twitter.cmdline import CONSUMER_KEY, CONSUMER_SECRET |
13 | ||
14 | noauth = NoAuth() | |
15 | oauth = OAuth(*read_token_file('tests/oauth_creds') | |
2cf5509c | 16 | + (CONSUMER_KEY, CONSUMER_SECRET)) |
17 | ||
920528cd MV |
18 | twitter11 = Twitter(domain='api.twitter.com', |
19 | auth=oauth, | |
20 | api_version='1.1') | |
e541e268 | 21 | |
759cdb65 | 22 | twitter_upl = Twitter(domain='upload.twitter.com', |
e3447bde MV |
23 | auth=oauth, |
24 | api_version='1.1') | |
759cdb65 | 25 | |
1ff50236 | 26 | twitter11_na = Twitter(domain='api.twitter.com', |
27 | auth=noauth, | |
28 | api_version='1.1') | |
e541e268 MV |
29 | |
30 | AZaz = "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ" | |
31 | ||
2b533cdc | 32 | b64_image_data = b"iVBORw0KGgoAAAANSUhEUgAAABAAAAAQCAIAAACQkWg2AAAAAXNSR0IArs4c6QAAAAlwSFlzAAALEwAACxMBAJqcGAAAAAd0SU1FB94JFhMBAJv5kaUAAAAZdEVYdENvbW1lbnQAQ3JlYXRlZCB3aXRoIEdJTVBXgQ4XAAAA4UlEQVQoz7WSIZLGIAxG6c5OFZjianBcIOfgPkju1DsEBWfAUEcNGGpY8Xe7dDoVFRvHfO8NJGRorZE39UVe1nd/WNfVObcsi3OOEAIASikAmOf5D2q/FWPUWgshKKWfiFIqhNBaxxhPjPQ05/z+Bs557xw9hBC89ymlu5BS8t6HEC5NW2sR8alRRLTWXoRSSinlSejT12M9BAAAgCeoTw9BSimlfBIu6WdYtVZEVErdaaUUItZaL/9wOsaY83YAMMb0dGtt6Jdv3/ec87ZtOWdCCGNsmibG2DiOJzP8+7b+AAOmsiPxyHWCAAAAAElFTkSuQmCC" |
2cf5509c | 33 | |
e541e268 MV |
34 | def get_random_str(): |
35 | return ''.join(choice(AZaz) for _ in range(10)) | |
36 | ||
37 | ||
c075046b R |
38 | def test_API_set_tweet(unicod=False): |
39 | random_tweet = "A random tweet %s" % \ | |
40 | ("with unicode üøπ" if unicod else "") + get_random_str() | |
2cf5509c | 41 | twitter11.statuses.update(status=random_tweet) |
f0603331 | 42 | time.sleep(5) |
64e08c8e | 43 | recent = twitter11.statuses.user_timeline() |
e541e268 | 44 | assert recent |
c77b5e4b SK |
45 | assert isinstance(recent.rate_limit_remaining, int) |
46 | assert isinstance(recent.rate_limit_reset, int) | |
f0603331 MV |
47 | texts = [tweet['text'] for tweet in recent] |
48 | assert random_tweet in texts | |
e541e268 | 49 | |
066c34e0 | 50 | def test_API_set_unicode_tweet(): |
c075046b | 51 | test_API_set_tweet(unicod=True) |
066c34e0 | 52 | |
53 | ||
9999744f | 54 | def clean_link(text): |
15ea270b | 55 | pos = text.find(" https://t.co") |
9999744f R |
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 | ||
759cdb65 MV |
63 | def _img_data(): |
64 | return open(os.path.join(__location__, "test.png"), "rb").read() | |
65 | ||
2b533cdc MV |
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} | |
9999744f R |
71 | twitter11.statuses.update_with_media(**params) |
72 | time.sleep(5) | |
64e08c8e | 73 | recent = twitter11.statuses.user_timeline() |
9999744f R |
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(): | |
2b533cdc MV |
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) | |
9999744f | 86 | |
6e12514e | 87 | def _test_upload_media(): |
759cdb65 MV |
88 | res = twitter_upl.media.upload(media=_img_data()) |
89 | assert res | |
90 | assert res["media_id"] | |
6e12514e R |
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 | |
9999744f | 105 | |
652c5402 | 106 | def test_search(): |
fcf08b18 | 107 | # In 1.1, search works on api.twitter.com not search.twitter.com |
108 | # and requires authorisation | |
ed074882 | 109 | results = twitter11.search.tweets(q='foo') |
652c5402 | 110 | assert results |
e748eed8 | 111 | |
112 | ||
113 | def test_get_trends(): | |
114 | # This is one method of inserting parameters, using named | |
115 | # underscore params. | |
c6e1e349 | 116 | world_trends = twitter11.trends.available(_woeid=1) |
e748eed8 | 117 | assert world_trends |
118 | ||
119 | ||
120 | def test_get_trends_2(): | |
121 | # This is a nicer variation of the same call as above. | |
2cf5509c | 122 | world_trends = twitter11.trends._(1) |
e748eed8 | 123 | assert world_trends |
920528cd MV |
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 | ||
ed23f46c | 130 | |
32a4811d HN |
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 | |
ed23f46c MV |
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 | |
1e95ec45 BB |
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' |