summaryrefslogtreecommitdiffstats
path: root/testing/mozbase/mozhttpd/tests/api.py
diff options
context:
space:
mode:
Diffstat (limited to 'testing/mozbase/mozhttpd/tests/api.py')
-rw-r--r--testing/mozbase/mozhttpd/tests/api.py266
1 files changed, 266 insertions, 0 deletions
diff --git a/testing/mozbase/mozhttpd/tests/api.py b/testing/mozbase/mozhttpd/tests/api.py
new file mode 100644
index 000000000..b785ac5ef
--- /dev/null
+++ b/testing/mozbase/mozhttpd/tests/api.py
@@ -0,0 +1,266 @@
+#!/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()
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-10 07:56:34 +0000