testing/mozbase/mozhttpd/tests/api.py@b8a032363ba2
testing/mozbase/mozhttpd/tests/api.py
Thu, 22 Jan 2015 13:21:57 +0100
- author
- Michael Schloh von Bennewitz <michael@schloh.com>
- date
- Thu, 22 Jan 2015 13:21:57 +0100
- branch
- TOR_BUG_9701
- changeset 15
- b8a032363ba2
- permissions
- -rw-r--r--
Incorporate requested changes from Mozilla in review:
https://bugzilla.mozilla.org/show_bug.cgi?id=1123480#c6
1 #!/usr/bin/env python
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/.
7 import mozhttpd
8 import urllib2
9 import os
10 import unittest
11 import re
12 import json
13 import tempfile
15 here = os.path.dirname(os.path.abspath(__file__))
17 class ApiTest(unittest.TestCase):
18 resource_get_called = 0
19 resource_post_called = 0
20 resource_del_called = 0
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 })
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 })
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 })
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
49 def try_get(self, server_port, querystr):
50 self.resource_get_called = 0
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)
60 def try_post(self, server_port, querystr):
61 self.resource_post_called = 0
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)
79 def try_del(self, server_port, querystr):
80 self.resource_del_called = 0
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)
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)
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)
108 server_port = httpd.httpd.server_port
110 # GET
111 self.try_get(server_port, '')
112 self.try_get(server_port, '?foo=bar')
114 # POST
115 self.try_post(server_port, '')
116 self.try_post(server_port, '?foo=bar')
118 # DEL
119 self.try_del(server_port, '')
120 self.try_del(server_port, '?foo=bar')
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)
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
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)
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)
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)
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
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())
190 # Make sure API methods still work
191 self.try_get(server_port, '')
192 self.try_get(server_port, '?foo=bar')
194 class ProxyTest(unittest.TestCase):
196 def tearDown(self):
197 # reset proxy opener in case it changed
198 urllib2.install_opener(None)
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
206 index_filename = 'index.html'
207 def index_contents(host): return '%s index' % host
209 index = file(os.path.join(docroot, index_filename), 'w')
210 index.write(index_contents('*'))
211 index.close()
213 httpd = mozhttpd.MozHttpd(port=0, docroot=docroot)
214 httpd.start(block=False)
215 server_port = httpd.httpd.server_port
217 proxy_support = urllib2.ProxyHandler({'http': 'http://127.0.0.1:%d' %
218 server_port})
219 urllib2.install_opener(urllib2.build_opener(proxy_support))
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('*'))
229 httpd.stop()
231 # test separate directories per host
233 httpd = mozhttpd.MozHttpd(port=0, docroot=docroot, proxy_host_dirs=True)
234 httpd.start(block=False)
235 server_port = httpd.httpd.server_port
237 proxy_support = urllib2.ProxyHandler({'http': 'http://127.0.0.1:%d' %
238 server_port})
239 urllib2.install_opener(urllib2.build_opener(proxy_support))
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))
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))
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)
264 if __name__ == '__main__':
265 unittest.main()
