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