]>
Commit | Line | Data |
---|---|---|
1 | from collections import namedtuple | |
2 | import contextlib | |
3 | import functools | |
4 | import socket | |
5 | import threading | |
6 | from twitter.util import find_links, follow_redirects, expand_line, parse_host_list | |
7 | ||
8 | try: | |
9 | import http.server as BaseHTTPServer | |
10 | import socketserver as SocketServer | |
11 | except ImportError: | |
12 | import BaseHTTPServer | |
13 | import SocketServer | |
14 | ||
15 | ||
16 | def test_find_links(): | |
17 | assert find_links("nix") == ("nix", []) | |
18 | assert find_links("http://abc") == ("%s", ["http://abc"]) | |
19 | assert find_links("t http://abc") == ("t %s", ["http://abc"]) | |
20 | assert find_links("http://abc t") == ("%s t", ["http://abc"]) | |
21 | assert find_links("1 http://a 2 http://b 3") == ("1 %s 2 %s 3", | |
22 | ["http://a", "http://b"]) | |
23 | assert find_links("%") == ("%%", []) | |
24 | assert find_links("(http://abc)") == ("(%s)", ["http://abc"]) | |
25 | ||
26 | ||
27 | Response = namedtuple('Response', 'path code headers') | |
28 | ||
29 | @contextlib.contextmanager | |
30 | def start_server(*resp): | |
31 | """HTTP server replying with the given responses to the expected | |
32 | requests.""" | |
33 | def url(port, path): | |
34 | return 'http://%s:%s%s' % (socket.gethostname(), port, path) | |
35 | ||
36 | responses = list(reversed(resp)) | |
37 | ||
38 | class MyHandler(BaseHTTPServer.BaseHTTPRequestHandler): | |
39 | def do_HEAD(self): | |
40 | response = responses.pop() | |
41 | assert response.path == self.path | |
42 | self.send_response(response.code) | |
43 | for header, value in list(response.headers.items()): | |
44 | self.send_header(header, value) | |
45 | self.end_headers() | |
46 | ||
47 | httpd = SocketServer.TCPServer(("", 0), MyHandler) | |
48 | t = threading.Thread(target=httpd.serve_forever) | |
49 | t.setDaemon(True) | |
50 | t.start() | |
51 | port = httpd.server_address[1] | |
52 | yield functools.partial(url, port) | |
53 | httpd.shutdown() | |
54 | ||
55 | def test_follow_redirects_direct_link(): | |
56 | link = "/resource" | |
57 | with start_server(Response(link, 200, {})) as url: | |
58 | assert url(link) == follow_redirects(url(link)) | |
59 | ||
60 | def test_follow_redirects_redirected_link(): | |
61 | redirected = "/redirected" | |
62 | link = "/resource" | |
63 | with start_server( | |
64 | Response(link, 301, {"Location": redirected}), | |
65 | Response(redirected, 200, {})) as url: | |
66 | assert url(redirected) == follow_redirects(url(link)) | |
67 | ||
68 | def test_follow_redirects_unavailable(): | |
69 | link = "/resource" | |
70 | with start_server(Response(link, 404, {})) as url: | |
71 | assert url(link) == follow_redirects(url(link)) | |
72 | ||
73 | def test_follow_redirects_link_to_last_available(): | |
74 | unavailable = "/unavailable" | |
75 | link = "/resource" | |
76 | with start_server( | |
77 | Response(link, 301, {"Location": unavailable}), | |
78 | Response(unavailable, 404, {})) as url: | |
79 | assert url(unavailable) == follow_redirects(url(link)) | |
80 | ||
81 | ||
82 | def test_follow_redirects_no_where(): | |
83 | link = "http://links.nowhere/" | |
84 | assert link == follow_redirects(link) | |
85 | ||
86 | def test_follow_redirects_link_to_nowhere(): | |
87 | unavailable = "http://links.nowhere/" | |
88 | link = "/resource" | |
89 | with start_server( | |
90 | Response(link, 301, {"Location": unavailable})) as url: | |
91 | assert unavailable == follow_redirects(url(link)) | |
92 | ||
93 | def test_follow_redirects_filtered_by_site(): | |
94 | link = "/resource" | |
95 | with start_server() as url: | |
96 | assert url(link) == follow_redirects(url(link), ["other_host"]) | |
97 | ||
98 | ||
99 | def test_follow_redirects_filtered_by_site_after_redirect(): | |
100 | link = "/resource" | |
101 | redirected = "/redirected" | |
102 | filtered = "http://dont-follow/" | |
103 | with start_server( | |
104 | Response(link, 301, {"Location": redirected}), | |
105 | Response(redirected, 301, {"Location": filtered})) as url: | |
106 | hosts = [socket.gethostname()] | |
107 | assert filtered == follow_redirects(url(link), hosts) | |
108 | ||
109 | def test_follow_redirects_filtered_by_site_allowed(): | |
110 | redirected = "/redirected" | |
111 | link = "/resource" | |
112 | with start_server( | |
113 | Response(link, 301, {"Location": redirected}), | |
114 | Response(redirected, 200, {})) as url: | |
115 | hosts = [socket.gethostname()] | |
116 | assert url(redirected) == follow_redirects(url(link), hosts) | |
117 | ||
118 | def test_expand_line(): | |
119 | redirected = "/redirected" | |
120 | link = "/resource" | |
121 | with start_server( | |
122 | Response(link, 301, {"Location": redirected}), | |
123 | Response(redirected, 200, {})) as url: | |
124 | fmt = "before %s after" | |
125 | line = fmt % url(link) | |
126 | expected = fmt % url(redirected) | |
127 | assert expected == expand_line(line, None) | |
128 | ||
129 | def test_parse_host_config(): | |
130 | assert set() == parse_host_list("") | |
131 | assert set("h") == parse_host_list("h") | |
132 | assert set(["1", "2"]) == parse_host_list("1,2") | |
133 | assert set(["1", "2"]) == parse_host_list(" 1 , 2 ") | |
134 |