2 from __future__
import unicode_literals
4 from collections
import namedtuple
9 from twitter
.util
import find_links
, follow_redirects
, expand_line
, parse_host_list
12 import http
.server
as BaseHTTPServer
13 import socketserver
as SocketServer
19 def test_find_links():
20 assert find_links("nix") == ("nix", [])
21 assert find_links("http://abc") == ("%s", ["http://abc"])
22 assert find_links("t http://abc") == ("t %s", ["http://abc"])
23 assert find_links("http://abc t") == ("%s t", ["http://abc"])
24 assert find_links("1 http://a 2 http://b 3") == ("1 %s 2 %s 3",
25 ["http://a", "http://b"])
26 assert find_links("%") == ("%%", [])
27 assert find_links("(http://abc)") == ("(%s)", ["http://abc"])
30 Response
= namedtuple('Response', 'path code headers')
32 @contextlib.contextmanager
33 def start_server(*resp
):
34 """HTTP server replying with the given responses to the expected
37 return 'http://%s:%s%s' % (socket
.gethostname(), port
, path
)
39 responses
= list(reversed(resp
))
41 class MyHandler(BaseHTTPServer
.BaseHTTPRequestHandler
):
43 response
= responses
.pop()
44 assert response
.path
== self
.path
45 self
.send_response(response
.code
)
46 for header
, value
in list(response
.headers
.items()):
47 self
.send_header(header
, value
)
50 httpd
= SocketServer
.TCPServer(("", 0), MyHandler
)
51 t
= threading
.Thread(target
=httpd
.serve_forever
)
54 port
= httpd
.server_address
[1]
55 yield functools
.partial(url
, port
)
58 def test_follow_redirects_direct_link():
60 with start_server(Response(link
, 200, {})) as url
:
61 assert url(link
) == follow_redirects(url(link
))
63 def test_follow_redirects_redirected_link():
64 redirected
= "/redirected"
67 Response(link
, 301, {"Location": redirected}
),
68 Response(redirected
, 200, {})) as url
:
69 assert url(redirected
) == follow_redirects(url(link
))
71 def test_follow_redirects_unavailable():
73 with start_server(Response(link
, 404, {})) as url
:
74 assert url(link
) == follow_redirects(url(link
))
76 def test_follow_redirects_link_to_last_available():
77 unavailable
= "/unavailable"
80 Response(link
, 301, {"Location": unavailable}
),
81 Response(unavailable
, 404, {})) as url
:
82 assert url(unavailable
) == follow_redirects(url(link
))
85 def test_follow_redirects_no_where():
86 link
= "http://links.nowhere/"
87 assert link
== follow_redirects(link
)
89 def test_follow_redirects_link_to_nowhere():
90 unavailable
= "http://links.nowhere/"
93 Response(link
, 301, {"Location": unavailable}
)) as url
:
94 assert unavailable
== follow_redirects(url(link
))
96 def test_follow_redirects_filtered_by_site():
98 with start_server() as url
:
99 assert url(link
) == follow_redirects(url(link
), ["other_host"])
102 def test_follow_redirects_filtered_by_site_after_redirect():
104 redirected
= "/redirected"
105 filtered
= "http://dont-follow/"
107 Response(link
, 301, {"Location": redirected}
),
108 Response(redirected
, 301, {"Location": filtered}
)) as url
:
109 hosts
= [socket
.gethostname()]
110 assert filtered
== follow_redirects(url(link
), hosts
)
112 def test_follow_redirects_filtered_by_site_allowed():
113 redirected
= "/redirected"
116 Response(link
, 301, {"Location": redirected}
),
117 Response(redirected
, 200, {})) as url
:
118 hosts
= [socket
.gethostname()]
119 assert url(redirected
) == follow_redirects(url(link
), hosts
)
121 def test_expand_line():
122 redirected
= "/redirected"
125 Response(link
, 301, {"Location": redirected}
),
126 Response(redirected
, 200, {})) as url
:
127 fmt
= "before %s after"
128 line
= fmt
% url(link
)
129 expected
= fmt
% url(redirected
)
130 assert expected
== expand_line(line
, None)
132 def test_parse_host_config():
133 assert set() == parse_host_list("")
134 assert set("h") == parse_host_list("h")
135 assert set(["1", "2"]) == parse_host_list("1,2")
136 assert set(["1", "2"]) == parse_host_list(" 1 , 2 ")