]>
Commit | Line | Data |
---|---|---|
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 | twitter11_na = Twitter(domain='api.twitter.com', | |
23 | auth=noauth, | |
24 | api_version='1.1') | |
25 | ||
26 | AZaz = "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ" | |
27 | ||
28 | ||
29 | def get_random_str(): | |
30 | return ''.join(choice(AZaz) for _ in range(10)) | |
31 | ||
32 | ||
33 | def test_API_set_tweet(unicod=False): | |
34 | random_tweet = "A random tweet %s" % \ | |
35 | ("with unicode üøπ" if unicod else "") + get_random_str() | |
36 | twitter11.statuses.update(status=random_tweet) | |
37 | time.sleep(5) | |
38 | recent = twitter11.statuses.user_timeline() | |
39 | assert recent | |
40 | assert isinstance(recent.rate_limit_remaining, int) | |
41 | assert isinstance(recent.rate_limit_reset, int) | |
42 | texts = [tweet['text'] for tweet in recent] | |
43 | assert random_tweet in texts | |
44 | ||
45 | def test_API_set_unicode_tweet(): | |
46 | test_API_set_tweet(unicod=True) | |
47 | ||
48 | ||
49 | def clean_link(text): | |
50 | pos = text.find(" http://t.co") | |
51 | if pos != -1: | |
52 | return text[:pos] | |
53 | return text | |
54 | ||
55 | __location__ = os.path.realpath( | |
56 | os.path.join(os.getcwd(), os.path.dirname(__file__))) | |
57 | ||
58 | def test_API_set_unicode_twitpic(base64=False): | |
59 | random_tweet = "A random twitpic from %s with unicode üøπ" % \ | |
60 | ("base64" if base64 else "file") + get_random_str() | |
61 | if base64: | |
62 | img = b"iVBORw0KGgoAAAANSUhEUgAAABAAAAAQCAIAAACQkWg2AAAAAXNSR0IArs4c6QAAAAlwSFlzAAALEwAACxMBAJqcGAAAAAd0SU1FB94JFhMBAJv5kaUAAAAZdEVYdENvbW1lbnQAQ3JlYXRlZCB3aXRoIEdJTVBXgQ4XAAAA4UlEQVQoz7WSIZLGIAxG6c5OFZjianBcIOfgPkju1DsEBWfAUEcNGGpY8Xe7dDoVFRvHfO8NJGRorZE39UVe1nd/WNfVObcsi3OOEAIASikAmOf5D2q/FWPUWgshKKWfiFIqhNBaxxhPjPQ05/z+Bs557xw9hBC89ymlu5BS8t6HEC5NW2sR8alRRLTWXoRSSinlSejT12M9BAAAgCeoTw9BSimlfBIu6WdYtVZEVErdaaUUItZaL/9wOsaY83YAMMb0dGtt6Jdv3/ec87ZtOWdCCGNsmibG2DiOJzP8+7b+AAOmsiPxyHWCAAAAAElFTkSuQmCC" | |
63 | else: | |
64 | with open(os.path.join(__location__, "test.png"), "rb") as f: | |
65 | img = f.read() | |
66 | params = {"status": random_tweet, "media[]": img} | |
67 | if base64: | |
68 | params["_base64"] = True | |
69 | twitter11.statuses.update_with_media(**params) | |
70 | time.sleep(5) | |
71 | recent = twitter11.statuses.user_timeline() | |
72 | assert recent | |
73 | texts = [clean_link(tweet['text']) for tweet in recent] | |
74 | assert random_tweet in texts | |
75 | ||
76 | def test_API_set_unicode_twitpic_base64(): | |
77 | test_API_set_unicode_twitpic(base64=True) | |
78 | ||
79 | ||
80 | def test_search(): | |
81 | # In 1.1, search works on api.twitter.com not search.twitter.com | |
82 | # and requires authorisation | |
83 | results = twitter11.search.tweets(q='foo') | |
84 | assert results | |
85 | ||
86 | ||
87 | def test_get_trends(): | |
88 | # This is one method of inserting parameters, using named | |
89 | # underscore params. | |
90 | world_trends = twitter11.trends.available(_woeid=1) | |
91 | assert world_trends | |
92 | ||
93 | ||
94 | def test_get_trends_2(): | |
95 | # This is a nicer variation of the same call as above. | |
96 | world_trends = twitter11.trends._(1) | |
97 | assert world_trends | |
98 | ||
99 | ||
100 | def test_get_trends_3(): | |
101 | # Of course they broke it all again in 1.1... | |
102 | assert twitter11.trends.place(_id=1) | |
103 | ||
104 | ||
105 | def test_TwitterHTTPError_raised_for_invalid_oauth(): | |
106 | test_passed = False | |
107 | try: | |
108 | twitter11_na.statuses.mentions_timeline() | |
109 | except TwitterHTTPError: | |
110 | # this is the error we are looking for :) | |
111 | test_passed = True | |
112 | assert test_passed | |
113 | ||
114 | ||
115 | def test_picklability(): | |
116 | res = TwitterDictResponse({'a': 'b'}) | |
117 | p = pickle.dumps(res) | |
118 | res2 = pickle.loads(p) | |
119 | assert res == res2 | |
120 | assert res2['a'] == 'b' | |
121 | ||
122 | res = TwitterListResponse([1, 2, 3]) | |
123 | p = pickle.dumps(res) | |
124 | res2 = pickle.loads(p) | |
125 | assert res == res2 | |
126 | assert res2[2] == 3 | |
127 | ||
128 | ||
129 | def test_jsonifability(): | |
130 | res = TwitterDictResponse({'a': 'b'}) | |
131 | p = json.dumps(res) | |
132 | res2 = json.loads(p) | |
133 | assert res == res2 | |
134 | assert res2['a'] == 'b' | |
135 | ||
136 | res = TwitterListResponse([1, 2, 3]) | |
137 | p = json.dumps(res) | |
138 | res2 = json.loads(p) | |
139 | assert res == res2 | |
140 | assert res2[2] == 3 | |
141 | ||
142 | ||
143 | def test_method_for_uri(): | |
144 | for action in POST_ACTIONS: | |
145 | assert method_for_uri(get_random_str() + '/' + action) == 'POST' | |
146 | assert method_for_uri('statuses/timeline') == 'GET' |