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