michael@0: #!/usr/bin/env python michael@0: michael@0: # This Source Code Form is subject to the terms of the Mozilla Public michael@0: # License, v. 2.0. If a copy of the MPL was not distributed with this file, michael@0: # You can obtain one at http://mozilla.org/MPL/2.0/. michael@0: michael@0: import mozhttpd michael@0: import urllib2 michael@0: import os michael@0: import unittest michael@0: import re michael@0: import json michael@0: import tempfile michael@0: michael@0: here = os.path.dirname(os.path.abspath(__file__)) michael@0: michael@0: class ApiTest(unittest.TestCase): michael@0: resource_get_called = 0 michael@0: resource_post_called = 0 michael@0: resource_del_called = 0 michael@0: michael@0: @mozhttpd.handlers.json_response michael@0: def resource_get(self, request, objid): michael@0: self.resource_get_called += 1 michael@0: return (200, { 'called': self.resource_get_called, michael@0: 'id': objid, michael@0: 'query': request.query }) michael@0: michael@0: @mozhttpd.handlers.json_response michael@0: def resource_post(self, request): michael@0: self.resource_post_called += 1 michael@0: return (201, { 'called': self.resource_post_called, michael@0: 'data': json.loads(request.body), michael@0: 'query': request.query }) michael@0: michael@0: @mozhttpd.handlers.json_response michael@0: def resource_del(self, request, objid): michael@0: self.resource_del_called += 1 michael@0: return (200, { 'called': self.resource_del_called, michael@0: 'id': objid, michael@0: 'query': request.query }) michael@0: michael@0: def get_url(self, path, server_port, querystr): michael@0: url = "http://127.0.0.1:%s%s" % (server_port, path) michael@0: if querystr: michael@0: url += "?%s" % querystr michael@0: return url michael@0: michael@0: def try_get(self, server_port, querystr): michael@0: self.resource_get_called = 0 michael@0: michael@0: f = urllib2.urlopen(self.get_url('/api/resource/1', server_port, querystr)) michael@0: try: michael@0: self.assertEqual(f.getcode(), 200) michael@0: except AttributeError: michael@0: pass # python 2.4 michael@0: self.assertEqual(json.loads(f.read()), { 'called': 1, 'id': str(1), 'query': querystr }) michael@0: self.assertEqual(self.resource_get_called, 1) michael@0: michael@0: def try_post(self, server_port, querystr): michael@0: self.resource_post_called = 0 michael@0: michael@0: postdata = { 'hamburgers': '1234' } michael@0: try: michael@0: f = urllib2.urlopen(self.get_url('/api/resource/', server_port, querystr), michael@0: data=json.dumps(postdata)) michael@0: except urllib2.HTTPError, e: michael@0: # python 2.4 michael@0: self.assertEqual(e.code, 201) michael@0: body = e.fp.read() michael@0: else: michael@0: self.assertEqual(f.getcode(), 201) michael@0: body = f.read() michael@0: self.assertEqual(json.loads(body), { 'called': 1, michael@0: 'data': postdata, michael@0: 'query': querystr }) michael@0: self.assertEqual(self.resource_post_called, 1) michael@0: michael@0: def try_del(self, server_port, querystr): michael@0: self.resource_del_called = 0 michael@0: michael@0: opener = urllib2.build_opener(urllib2.HTTPHandler) michael@0: request = urllib2.Request(self.get_url('/api/resource/1', server_port, querystr)) michael@0: request.get_method = lambda: 'DEL' michael@0: f = opener.open(request) michael@0: michael@0: try: michael@0: self.assertEqual(f.getcode(), 200) michael@0: except AttributeError: michael@0: pass # python 2.4 michael@0: self.assertEqual(json.loads(f.read()), { 'called': 1, 'id': str(1), 'query': querystr }) michael@0: self.assertEqual(self.resource_del_called, 1) michael@0: michael@0: def test_api(self): michael@0: httpd = mozhttpd.MozHttpd(port=0, michael@0: urlhandlers = [ { 'method': 'GET', michael@0: 'path': '/api/resource/([^/]+)/?', michael@0: 'function': self.resource_get }, michael@0: { 'method': 'POST', michael@0: 'path': '/api/resource/?', michael@0: 'function': self.resource_post }, michael@0: { 'method': 'DEL', michael@0: 'path': '/api/resource/([^/]+)/?', michael@0: 'function': self.resource_del } michael@0: ]) michael@0: httpd.start(block=False) michael@0: michael@0: server_port = httpd.httpd.server_port michael@0: michael@0: # GET michael@0: self.try_get(server_port, '') michael@0: self.try_get(server_port, '?foo=bar') michael@0: michael@0: # POST michael@0: self.try_post(server_port, '') michael@0: self.try_post(server_port, '?foo=bar') michael@0: michael@0: # DEL michael@0: self.try_del(server_port, '') michael@0: self.try_del(server_port, '?foo=bar') michael@0: michael@0: # GET: By default we don't serve any files if we just define an API michael@0: f = None michael@0: exception_thrown = False michael@0: try: michael@0: f = urllib2.urlopen(self.get_url('/', server_port, None)) michael@0: except urllib2.HTTPError, e: michael@0: self.assertEqual(e.code, 404) michael@0: exception_thrown = True michael@0: self.assertTrue(exception_thrown) michael@0: michael@0: def test_nonexistent_resources(self): michael@0: # Create a server with a placeholder handler so we don't fall back michael@0: # to serving local files michael@0: httpd = mozhttpd.MozHttpd(port=0) michael@0: httpd.start(block=False) michael@0: server_port = httpd.httpd.server_port michael@0: michael@0: # GET: Return 404 for non-existent endpoint michael@0: f = None michael@0: exception_thrown = False michael@0: try: michael@0: f = urllib2.urlopen(self.get_url('/api/resource/', server_port, None)) michael@0: except urllib2.HTTPError, e: michael@0: self.assertEqual(e.code, 404) michael@0: exception_thrown = True michael@0: self.assertTrue(exception_thrown) michael@0: michael@0: # POST: POST should also return 404 michael@0: f = None michael@0: exception_thrown = False michael@0: try: michael@0: f = urllib2.urlopen(self.get_url('/api/resource/', server_port, None), michael@0: data=json.dumps({})) michael@0: except urllib2.HTTPError, e: michael@0: self.assertEqual(e.code, 404) michael@0: exception_thrown = True michael@0: self.assertTrue(exception_thrown) michael@0: michael@0: # DEL: DEL should also return 404 michael@0: f = None michael@0: exception_thrown = False michael@0: try: michael@0: opener = urllib2.build_opener(urllib2.HTTPHandler) michael@0: request = urllib2.Request(self.get_url('/api/resource/', server_port, michael@0: None)) michael@0: request.get_method = lambda: 'DEL' michael@0: f = opener.open(request) michael@0: except urllib2.HTTPError, e: michael@0: self.assertEqual(e.code, 404) michael@0: exception_thrown = True michael@0: self.assertTrue(exception_thrown) michael@0: michael@0: def test_api_with_docroot(self): michael@0: httpd = mozhttpd.MozHttpd(port=0, docroot=here, michael@0: urlhandlers = [ { 'method': 'GET', michael@0: 'path': '/api/resource/([^/]+)/?', michael@0: 'function': self.resource_get } ]) michael@0: httpd.start(block=False) michael@0: server_port = httpd.httpd.server_port michael@0: michael@0: # We defined a docroot, so we expect a directory listing michael@0: f = urllib2.urlopen(self.get_url('/', server_port, None)) michael@0: try: michael@0: self.assertEqual(f.getcode(), 200) michael@0: except AttributeError: michael@0: pass # python 2.4 michael@0: self.assertTrue('Directory listing for' in f.read()) michael@0: michael@0: # Make sure API methods still work michael@0: self.try_get(server_port, '') michael@0: self.try_get(server_port, '?foo=bar') michael@0: michael@0: class ProxyTest(unittest.TestCase): michael@0: michael@0: def tearDown(self): michael@0: # reset proxy opener in case it changed michael@0: urllib2.install_opener(None) michael@0: michael@0: def test_proxy(self): michael@0: docroot = tempfile.mkdtemp() michael@0: hosts = ('mozilla.com', 'mozilla.org') michael@0: unproxied_host = 'notmozilla.org' michael@0: def url(host): return 'http://%s/' % host michael@0: michael@0: index_filename = 'index.html' michael@0: def index_contents(host): return '%s index' % host michael@0: michael@0: index = file(os.path.join(docroot, index_filename), 'w') michael@0: index.write(index_contents('*')) michael@0: index.close() michael@0: michael@0: httpd = mozhttpd.MozHttpd(port=0, docroot=docroot) michael@0: httpd.start(block=False) michael@0: server_port = httpd.httpd.server_port michael@0: michael@0: proxy_support = urllib2.ProxyHandler({'http': 'http://127.0.0.1:%d' % michael@0: server_port}) michael@0: urllib2.install_opener(urllib2.build_opener(proxy_support)) michael@0: michael@0: for host in hosts: michael@0: f = urllib2.urlopen(url(host)) michael@0: try: michael@0: self.assertEqual(f.getcode(), 200) michael@0: except AttributeError: michael@0: pass # python 2.4 michael@0: self.assertEqual(f.read(), index_contents('*')) michael@0: michael@0: httpd.stop() michael@0: michael@0: # test separate directories per host michael@0: michael@0: httpd = mozhttpd.MozHttpd(port=0, docroot=docroot, proxy_host_dirs=True) michael@0: httpd.start(block=False) michael@0: server_port = httpd.httpd.server_port michael@0: michael@0: proxy_support = urllib2.ProxyHandler({'http': 'http://127.0.0.1:%d' % michael@0: server_port}) michael@0: urllib2.install_opener(urllib2.build_opener(proxy_support)) michael@0: michael@0: # set up dirs michael@0: for host in hosts: michael@0: os.mkdir(os.path.join(docroot, host)) michael@0: file(os.path.join(docroot, host, index_filename), 'w') \ michael@0: .write(index_contents(host)) michael@0: michael@0: for host in hosts: michael@0: f = urllib2.urlopen(url(host)) michael@0: try: michael@0: self.assertEqual(f.getcode(), 200) michael@0: except AttributeError: michael@0: pass # python 2.4 michael@0: self.assertEqual(f.read(), index_contents(host)) michael@0: michael@0: exc = None michael@0: try: michael@0: urllib2.urlopen(url(unproxied_host)) michael@0: except urllib2.HTTPError, e: michael@0: exc = e michael@0: self.assertNotEqual(exc, None) michael@0: self.assertEqual(exc.code, 404) michael@0: michael@0: michael@0: if __name__ == '__main__': michael@0: unittest.main()