1 from collections
import namedtuple
6 from twitter
.util
import find_links
, follow_redirects
, expand_line
, parse_host_list
9 import http
.server
as BaseHTTPServer
10 import socketserver
as SocketServer
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"])
27 Response
= namedtuple('Response', 'path code headers')
29 @contextlib.contextmanager
30 def start_server(*resp
):
31 """HTTP server replying with the given responses to the expected
34 return 'http://%s:%s%s' % (socket
.gethostname(), port
, path
)
36 responses
= list(reversed(resp
))
38 class MyHandler(BaseHTTPServer
.BaseHTTPRequestHandler
):
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
)
47 httpd
= SocketServer
.TCPServer(("", 0), MyHandler
)
48 t
= threading
.Thread(target
=httpd
.serve_forever
)
51 port
= httpd
.server_address
[1]
52 yield functools
.partial(url
, port
)
55 def test_follow_redirects_direct_link():
57 with start_server(Response(link
, 200, {})) as url
:
58 assert url(link
) == follow_redirects(url(link
))
60 def test_follow_redirects_redirected_link():
61 redirected
= "/redirected"
64 Response(link
, 301, {"Location": redirected}
),
65 Response(redirected
, 200, {})) as url
:
66 assert url(redirected
) == follow_redirects(url(link
))
68 def test_follow_redirects_unavailable():
70 with start_server(Response(link
, 404, {})) as url
:
71 assert url(link
) == follow_redirects(url(link
))
73 def test_follow_redirects_link_to_last_available():
74 unavailable
= "/unavailable"
77 Response(link
, 301, {"Location": unavailable}
),
78 Response(unavailable
, 404, {})) as url
:
79 assert url(unavailable
) == follow_redirects(url(link
))
82 def test_follow_redirects_no_where():
83 link
= "http://links.nowhere/"
84 assert link
== follow_redirects(link
)
86 def test_follow_redirects_link_to_nowhere():
87 unavailable
= "http://links.nowhere/"
90 Response(link
, 301, {"Location": unavailable}
)) as url
:
91 assert unavailable
== follow_redirects(url(link
))
93 def test_follow_redirects_filtered_by_site():
95 with start_server() as url
:
96 assert url(link
) == follow_redirects(url(link
), ["other_host"])
99 def test_follow_redirects_filtered_by_site_after_redirect():
101 redirected
= "/redirected"
102 filtered
= "http://dont-follow/"
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
)
109 def test_follow_redirects_filtered_by_site_allowed():
110 redirected
= "/redirected"
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
)
118 def test_expand_line():
119 redirected
= "/redirected"
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)
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 ")