|
1 #!/usr/bin/env python |
|
2 |
|
3 # This Source Code Form is subject to the terms of the Mozilla Public |
|
4 # License, v. 2.0. If a copy of the MPL was not distributed with this file, |
|
5 # You can obtain one at http://mozilla.org/MPL/2.0/. |
|
6 |
|
7 import mozhttpd |
|
8 import urllib2 |
|
9 import os |
|
10 import unittest |
|
11 import re |
|
12 import json |
|
13 import tempfile |
|
14 |
|
15 here = os.path.dirname(os.path.abspath(__file__)) |
|
16 |
|
17 class ApiTest(unittest.TestCase): |
|
18 resource_get_called = 0 |
|
19 resource_post_called = 0 |
|
20 resource_del_called = 0 |
|
21 |
|
22 @mozhttpd.handlers.json_response |
|
23 def resource_get(self, request, objid): |
|
24 self.resource_get_called += 1 |
|
25 return (200, { 'called': self.resource_get_called, |
|
26 'id': objid, |
|
27 'query': request.query }) |
|
28 |
|
29 @mozhttpd.handlers.json_response |
|
30 def resource_post(self, request): |
|
31 self.resource_post_called += 1 |
|
32 return (201, { 'called': self.resource_post_called, |
|
33 'data': json.loads(request.body), |
|
34 'query': request.query }) |
|
35 |
|
36 @mozhttpd.handlers.json_response |
|
37 def resource_del(self, request, objid): |
|
38 self.resource_del_called += 1 |
|
39 return (200, { 'called': self.resource_del_called, |
|
40 'id': objid, |
|
41 'query': request.query }) |
|
42 |
|
43 def get_url(self, path, server_port, querystr): |
|
44 url = "http://127.0.0.1:%s%s" % (server_port, path) |
|
45 if querystr: |
|
46 url += "?%s" % querystr |
|
47 return url |
|
48 |
|
49 def try_get(self, server_port, querystr): |
|
50 self.resource_get_called = 0 |
|
51 |
|
52 f = urllib2.urlopen(self.get_url('/api/resource/1', server_port, querystr)) |
|
53 try: |
|
54 self.assertEqual(f.getcode(), 200) |
|
55 except AttributeError: |
|
56 pass # python 2.4 |
|
57 self.assertEqual(json.loads(f.read()), { 'called': 1, 'id': str(1), 'query': querystr }) |
|
58 self.assertEqual(self.resource_get_called, 1) |
|
59 |
|
60 def try_post(self, server_port, querystr): |
|
61 self.resource_post_called = 0 |
|
62 |
|
63 postdata = { 'hamburgers': '1234' } |
|
64 try: |
|
65 f = urllib2.urlopen(self.get_url('/api/resource/', server_port, querystr), |
|
66 data=json.dumps(postdata)) |
|
67 except urllib2.HTTPError, e: |
|
68 # python 2.4 |
|
69 self.assertEqual(e.code, 201) |
|
70 body = e.fp.read() |
|
71 else: |
|
72 self.assertEqual(f.getcode(), 201) |
|
73 body = f.read() |
|
74 self.assertEqual(json.loads(body), { 'called': 1, |
|
75 'data': postdata, |
|
76 'query': querystr }) |
|
77 self.assertEqual(self.resource_post_called, 1) |
|
78 |
|
79 def try_del(self, server_port, querystr): |
|
80 self.resource_del_called = 0 |
|
81 |
|
82 opener = urllib2.build_opener(urllib2.HTTPHandler) |
|
83 request = urllib2.Request(self.get_url('/api/resource/1', server_port, querystr)) |
|
84 request.get_method = lambda: 'DEL' |
|
85 f = opener.open(request) |
|
86 |
|
87 try: |
|
88 self.assertEqual(f.getcode(), 200) |
|
89 except AttributeError: |
|
90 pass # python 2.4 |
|
91 self.assertEqual(json.loads(f.read()), { 'called': 1, 'id': str(1), 'query': querystr }) |
|
92 self.assertEqual(self.resource_del_called, 1) |
|
93 |
|
94 def test_api(self): |
|
95 httpd = mozhttpd.MozHttpd(port=0, |
|
96 urlhandlers = [ { 'method': 'GET', |
|
97 'path': '/api/resource/([^/]+)/?', |
|
98 'function': self.resource_get }, |
|
99 { 'method': 'POST', |
|
100 'path': '/api/resource/?', |
|
101 'function': self.resource_post }, |
|
102 { 'method': 'DEL', |
|
103 'path': '/api/resource/([^/]+)/?', |
|
104 'function': self.resource_del } |
|
105 ]) |
|
106 httpd.start(block=False) |
|
107 |
|
108 server_port = httpd.httpd.server_port |
|
109 |
|
110 # GET |
|
111 self.try_get(server_port, '') |
|
112 self.try_get(server_port, '?foo=bar') |
|
113 |
|
114 # POST |
|
115 self.try_post(server_port, '') |
|
116 self.try_post(server_port, '?foo=bar') |
|
117 |
|
118 # DEL |
|
119 self.try_del(server_port, '') |
|
120 self.try_del(server_port, '?foo=bar') |
|
121 |
|
122 # GET: By default we don't serve any files if we just define an API |
|
123 f = None |
|
124 exception_thrown = False |
|
125 try: |
|
126 f = urllib2.urlopen(self.get_url('/', server_port, None)) |
|
127 except urllib2.HTTPError, e: |
|
128 self.assertEqual(e.code, 404) |
|
129 exception_thrown = True |
|
130 self.assertTrue(exception_thrown) |
|
131 |
|
132 def test_nonexistent_resources(self): |
|
133 # Create a server with a placeholder handler so we don't fall back |
|
134 # to serving local files |
|
135 httpd = mozhttpd.MozHttpd(port=0) |
|
136 httpd.start(block=False) |
|
137 server_port = httpd.httpd.server_port |
|
138 |
|
139 # GET: Return 404 for non-existent endpoint |
|
140 f = None |
|
141 exception_thrown = False |
|
142 try: |
|
143 f = urllib2.urlopen(self.get_url('/api/resource/', server_port, None)) |
|
144 except urllib2.HTTPError, e: |
|
145 self.assertEqual(e.code, 404) |
|
146 exception_thrown = True |
|
147 self.assertTrue(exception_thrown) |
|
148 |
|
149 # POST: POST should also return 404 |
|
150 f = None |
|
151 exception_thrown = False |
|
152 try: |
|
153 f = urllib2.urlopen(self.get_url('/api/resource/', server_port, None), |
|
154 data=json.dumps({})) |
|
155 except urllib2.HTTPError, e: |
|
156 self.assertEqual(e.code, 404) |
|
157 exception_thrown = True |
|
158 self.assertTrue(exception_thrown) |
|
159 |
|
160 # DEL: DEL should also return 404 |
|
161 f = None |
|
162 exception_thrown = False |
|
163 try: |
|
164 opener = urllib2.build_opener(urllib2.HTTPHandler) |
|
165 request = urllib2.Request(self.get_url('/api/resource/', server_port, |
|
166 None)) |
|
167 request.get_method = lambda: 'DEL' |
|
168 f = opener.open(request) |
|
169 except urllib2.HTTPError, e: |
|
170 self.assertEqual(e.code, 404) |
|
171 exception_thrown = True |
|
172 self.assertTrue(exception_thrown) |
|
173 |
|
174 def test_api_with_docroot(self): |
|
175 httpd = mozhttpd.MozHttpd(port=0, docroot=here, |
|
176 urlhandlers = [ { 'method': 'GET', |
|
177 'path': '/api/resource/([^/]+)/?', |
|
178 'function': self.resource_get } ]) |
|
179 httpd.start(block=False) |
|
180 server_port = httpd.httpd.server_port |
|
181 |
|
182 # We defined a docroot, so we expect a directory listing |
|
183 f = urllib2.urlopen(self.get_url('/', server_port, None)) |
|
184 try: |
|
185 self.assertEqual(f.getcode(), 200) |
|
186 except AttributeError: |
|
187 pass # python 2.4 |
|
188 self.assertTrue('Directory listing for' in f.read()) |
|
189 |
|
190 # Make sure API methods still work |
|
191 self.try_get(server_port, '') |
|
192 self.try_get(server_port, '?foo=bar') |
|
193 |
|
194 class ProxyTest(unittest.TestCase): |
|
195 |
|
196 def tearDown(self): |
|
197 # reset proxy opener in case it changed |
|
198 urllib2.install_opener(None) |
|
199 |
|
200 def test_proxy(self): |
|
201 docroot = tempfile.mkdtemp() |
|
202 hosts = ('mozilla.com', 'mozilla.org') |
|
203 unproxied_host = 'notmozilla.org' |
|
204 def url(host): return 'http://%s/' % host |
|
205 |
|
206 index_filename = 'index.html' |
|
207 def index_contents(host): return '%s index' % host |
|
208 |
|
209 index = file(os.path.join(docroot, index_filename), 'w') |
|
210 index.write(index_contents('*')) |
|
211 index.close() |
|
212 |
|
213 httpd = mozhttpd.MozHttpd(port=0, docroot=docroot) |
|
214 httpd.start(block=False) |
|
215 server_port = httpd.httpd.server_port |
|
216 |
|
217 proxy_support = urllib2.ProxyHandler({'http': 'http://127.0.0.1:%d' % |
|
218 server_port}) |
|
219 urllib2.install_opener(urllib2.build_opener(proxy_support)) |
|
220 |
|
221 for host in hosts: |
|
222 f = urllib2.urlopen(url(host)) |
|
223 try: |
|
224 self.assertEqual(f.getcode(), 200) |
|
225 except AttributeError: |
|
226 pass # python 2.4 |
|
227 self.assertEqual(f.read(), index_contents('*')) |
|
228 |
|
229 httpd.stop() |
|
230 |
|
231 # test separate directories per host |
|
232 |
|
233 httpd = mozhttpd.MozHttpd(port=0, docroot=docroot, proxy_host_dirs=True) |
|
234 httpd.start(block=False) |
|
235 server_port = httpd.httpd.server_port |
|
236 |
|
237 proxy_support = urllib2.ProxyHandler({'http': 'http://127.0.0.1:%d' % |
|
238 server_port}) |
|
239 urllib2.install_opener(urllib2.build_opener(proxy_support)) |
|
240 |
|
241 # set up dirs |
|
242 for host in hosts: |
|
243 os.mkdir(os.path.join(docroot, host)) |
|
244 file(os.path.join(docroot, host, index_filename), 'w') \ |
|
245 .write(index_contents(host)) |
|
246 |
|
247 for host in hosts: |
|
248 f = urllib2.urlopen(url(host)) |
|
249 try: |
|
250 self.assertEqual(f.getcode(), 200) |
|
251 except AttributeError: |
|
252 pass # python 2.4 |
|
253 self.assertEqual(f.read(), index_contents(host)) |
|
254 |
|
255 exc = None |
|
256 try: |
|
257 urllib2.urlopen(url(unproxied_host)) |
|
258 except urllib2.HTTPError, e: |
|
259 exc = e |
|
260 self.assertNotEqual(exc, None) |
|
261 self.assertEqual(exc.code, 404) |
|
262 |
|
263 |
|
264 if __name__ == '__main__': |
|
265 unittest.main() |