| #!/usr/bin/env python |
| # |
| # Copyright 2007 Google Inc. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # |
| """Tests for google.apphosting.tools.devappserver2.wsgi_request_info.""" |
| |
| |
| |
| import re |
| import unittest |
| import wsgiref.util |
| |
| from google.appengine.tools.devappserver2 import wsgi_request_info |
| |
| |
| class TestWSGIRequestInfo(unittest.TestCase): |
| """Tests for WSGIRequestInfo.""" |
| |
| def setUp(self): |
| self.dispatcher = object() |
| self.request_info = wsgi_request_info.WSGIRequestInfo(self.dispatcher) |
| |
| def _assert_request_id(self, request_id): |
| self.assertTrue(re.match('[a-zA-Z]{10}$', request_id), |
| 'invalid request id: %r' % request_id) |
| |
| def _create_environ(self, scheme, host, path='', query=''): |
| environ = {'wsgi.url_scheme': scheme, |
| 'HTTP_HOST': host, |
| 'PATH_INFO': path, |
| 'QUERY_STRING': query} |
| wsgiref.util.setup_testing_defaults(environ) |
| return environ |
| |
| def _create_module_configuration(self, module_name, version_id): |
| class ModuleConfiguration(object): |
| pass |
| |
| config = ModuleConfiguration() |
| config.major_version = version_id |
| config.module_name = module_name |
| return config |
| |
| def test_get_request_url(self): |
| with self.request_info.request( |
| self._create_environ('https', 'machine:8080', |
| '/foo', 'bar=baz'), |
| self._create_module_configuration('default', '1')) as request_id: |
| self._assert_request_id(request_id) |
| self.assertEqual('https://machine:8080/foo?bar=baz', |
| self.request_info.get_request_url(request_id)) |
| |
| def test_get_request_environ(self): |
| environ = object() |
| with self.request_info.request( |
| environ, |
| self._create_module_configuration('default', '1')) as request_id: |
| self._assert_request_id(request_id) |
| self.assertIs(environ, self.request_info.get_request_environ(request_id)) |
| |
| def test_get_dispatcher(self): |
| with self.request_info.request( |
| self._create_environ('https', 'machine:8080', |
| '/foo', 'bar=baz'), |
| self._create_module_configuration('default', '1')) as request_id: |
| self._assert_request_id(request_id) |
| self.assertEqual(self.dispatcher, |
| self.request_info.get_dispatcher()) |
| |
| def test_get_module(self): |
| with self.request_info.request( |
| self._create_environ('https', 'machine:8080', |
| '/foo', 'bar=baz'), |
| self._create_module_configuration('default', '1')) as request_id: |
| self._assert_request_id(request_id) |
| self.assertEqual('default', self.request_info.get_module(request_id)) |
| |
| def test_get_version(self): |
| with self.request_info.request( |
| self._create_environ('https', 'machine:8080', |
| '/foo', 'bar=baz'), |
| self._create_module_configuration('default', '1')) as request_id: |
| self._assert_request_id(request_id) |
| self.assertEqual('1', self.request_info.get_version(request_id)) |
| |
| def test_get_instance_unset(self): |
| with self.request_info.request( |
| self._create_environ('https', 'machine:8080', |
| '/foo', 'bar=baz'), |
| self._create_module_configuration('default', '1')) as request_id: |
| self._assert_request_id(request_id) |
| self.assertIsNone(self.request_info.get_instance(request_id)) |
| |
| def test_get_instance(self): |
| with self.request_info.request( |
| self._create_environ('https', 'machine:8080', |
| '/foo', 'bar=baz'), |
| self._create_module_configuration('default', '1')) as request_id: |
| instance = object() |
| self.request_info.set_request_instance(request_id, instance) |
| self._assert_request_id(request_id) |
| self.assertEqual(instance, self.request_info.get_instance(request_id)) |
| |
| def test_concurrent_requests(self): |
| request_id1 = self.request_info.start_request( |
| self._create_environ('http', 'machine:8081'), |
| self._create_module_configuration('default', '1')) |
| request_id2 = self.request_info.start_request( |
| self._create_environ('http', 'machine:8082'), |
| self._create_module_configuration('default', '2')) |
| request_id3 = self.request_info.start_request( |
| self._create_environ('http', 'machine:8083'), |
| self._create_module_configuration('other', '1')) |
| |
| self._assert_request_id(request_id1) |
| self._assert_request_id(request_id2) |
| self._assert_request_id(request_id3) |
| self.assertTrue(request_id1 != request_id2 != request_id3) |
| |
| self.assertEqual('http://machine:8081/', |
| self.request_info.get_request_url(request_id1)) |
| self.assertEqual(self.dispatcher, |
| self.request_info.get_dispatcher()) |
| self.assertEqual('default', self.request_info.get_module(request_id1)) |
| self.assertEqual('1', self.request_info.get_version(request_id1)) |
| self.assertEqual('http://machine:8082/', |
| self.request_info.get_request_url(request_id2)) |
| self.assertEqual(self.dispatcher, |
| self.request_info.get_dispatcher()) |
| self.assertEqual('default', self.request_info.get_module(request_id2)) |
| self.assertEqual('2', self.request_info.get_version(request_id2)) |
| self.request_info.end_request(request_id1) |
| self.request_info.end_request(request_id2) |
| self.assertEqual('http://machine:8083/', |
| self.request_info.get_request_url(request_id3)) |
| self.assertEqual(self.dispatcher, |
| self.request_info.get_dispatcher()) |
| self.assertEqual('other', self.request_info.get_module(request_id3)) |
| self.assertEqual('1', self.request_info.get_version(request_id3)) |
| |
| if __name__ == '__main__': |
| unittest.main() |
