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