[catapult] - Copy Telemetry's gsutilz over to third_party.

third_party/gsutil/third_party/protorpc/protorpc/wsgi/util_test.py

Index: third_party/gsutil/third_party/protorpc/protorpc/wsgi/util_test.py
diff --git a/third_party/gsutil/third_party/protorpc/protorpc/wsgi/util_test.py b/third_party/gsutil/third_party/protorpc/protorpc/wsgi/util_test.py
new file mode 100755
index 0000000000000000000000000000000000000000..60a79afeab182472026826a965a19656ae9d5f35
--- /dev/null
+++ b/third_party/gsutil/third_party/protorpc/protorpc/wsgi/util_test.py
@@ -0,0 +1,295 @@
+#!/usr/bin/env python
+#
+# Copyright 2011 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.
+#
+
+"""WSGI utility library tests."""
+import six
+from six.moves import filter
+
+__author__ = 'rafe@google.com (Rafe Kaplan)'
+
+
+import six.moves.http_client
+import unittest
+
+from protorpc import test_util
+from protorpc import util
+from protorpc import webapp_test_util
+from protorpc.wsgi import util as wsgi_util
+
+APP1 = wsgi_util.static_page('App1')
+APP2 = wsgi_util.static_page('App2')
+NOT_FOUND = wsgi_util.error(six.moves.http_client.NOT_FOUND)
+
+
+class WsgiTestBase(webapp_test_util.WebServerTestBase):
+
+  server_thread = None
+
+  def CreateWsgiApplication(self):
+    return None
+
+  def DoHttpRequest(self,
+                    path='/',
+                    content=None,
+                    content_type='text/plain; charset=utf-8',
+                    headers=None):
+    connection = six.moves.http_client.HTTPConnection('localhost', self.port)
+    if content is None:
+      method = 'GET'
+    else:
+      method = 'POST'
+    headers = {'content=type': content_type}
+    headers.update(headers)
+    connection.request(method, path, content, headers)
+    response = connection.getresponse()
+
+    not_date_or_server = lambda header: header[0] not in ('date', 'server')
+    headers = list(filter(not_date_or_server, response.getheaders()))
+
+    return response.status, response.reason, response.read(), dict(headers)
+
+
+class StaticPageBase(WsgiTestBase):
+
+  def testDefault(self):
+    default_page = wsgi_util.static_page()
+    self.ResetServer(default_page)
+    status, reason, content, headers = self.DoHttpRequest()
+    self.assertEquals(200, status)
+    self.assertEquals('OK', reason)
+    self.assertEquals('', content)
+    self.assertEquals({'content-length': '0',
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+  def testHasContent(self):
+    default_page = wsgi_util.static_page('my content')
+    self.ResetServer(default_page)
+    status, reason, content, headers = self.DoHttpRequest()
+    self.assertEquals(200, status)
+    self.assertEquals('OK', reason)
+    self.assertEquals('my content', content)
+    self.assertEquals({'content-length': str(len('my content')),
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+  def testHasContentType(self):
+    default_page = wsgi_util.static_page(content_type='text/plain')
+    self.ResetServer(default_page)
+    status, reason, content, headers = self.DoHttpRequest()
+    self.assertEquals(200, status)
+    self.assertEquals('OK', reason)
+    self.assertEquals('', content)
+    self.assertEquals({'content-length': '0',
+                       'content-type': 'text/plain',
+                      },
+                      headers)
+
+  def testHasStatus(self):
+    default_page = wsgi_util.static_page(status='400 Not Good Request')
+    self.ResetServer(default_page)
+    status, reason, content, headers = self.DoHttpRequest()
+    self.assertEquals(400, status)
+    self.assertEquals('Not Good Request', reason)
+    self.assertEquals('', content)
+    self.assertEquals({'content-length': '0',
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+  def testHasStatusInt(self):
+    default_page = wsgi_util.static_page(status=401)
+    self.ResetServer(default_page)
+    status, reason, content, headers = self.DoHttpRequest()
+    self.assertEquals(401, status)
+    self.assertEquals('Unauthorized', reason)
+    self.assertEquals('', content)
+    self.assertEquals({'content-length': '0',
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+  def testHasStatusUnknown(self):
+    default_page = wsgi_util.static_page(status=909)
+    self.ResetServer(default_page)
+    status, reason, content, headers = self.DoHttpRequest()
+    self.assertEquals(909, status)
+    self.assertEquals('Unknown Error', reason)
+    self.assertEquals('', content)
+    self.assertEquals({'content-length': '0',
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+  def testHasStatusTuple(self):
+    default_page = wsgi_util.static_page(status=(500, 'Bad Thing'))
+    self.ResetServer(default_page)
+    status, reason, content, headers = self.DoHttpRequest()
+    self.assertEquals(500, status)
+    self.assertEquals('Bad Thing', reason)
+    self.assertEquals('', content)
+    self.assertEquals({'content-length': '0',
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+  def testHasHeaders(self):
+    default_page = wsgi_util.static_page(headers=[('x', 'foo'),
+                                                  ('a', 'bar'),
+                                                  ('z', 'bin')])
+    self.ResetServer(default_page)
+    status, reason, content, headers = self.DoHttpRequest()
+    self.assertEquals(200, status)
+    self.assertEquals('OK', reason)
+    self.assertEquals('', content)
+    self.assertEquals({'content-length': '0',
+                       'content-type': 'text/html; charset=utf-8',
+                       'x': 'foo',
+                       'a': 'bar',
+                       'z': 'bin',
+                      },
+                      headers)
+
+  def testHeadersUnicodeSafe(self):
+    default_page = wsgi_util.static_page(headers=[('x', u'foo')])
+    self.ResetServer(default_page)
+    status, reason, content, headers = self.DoHttpRequest()
+    self.assertEquals(200, status)
+    self.assertEquals('OK', reason)
+    self.assertEquals('', content)
+    self.assertEquals({'content-length': '0',
+                       'content-type': 'text/html; charset=utf-8',
+                       'x': 'foo',
+                      },
+                      headers)
+    self.assertTrue(isinstance(headers['x'], str))
+
+  def testHasHeadersDict(self):
+    default_page = wsgi_util.static_page(headers={'x': 'foo',
+                                                  'a': 'bar',
+                                                  'z': 'bin'})
+    self.ResetServer(default_page)
+    status, reason, content, headers = self.DoHttpRequest()
+    self.assertEquals(200, status)
+    self.assertEquals('OK', reason)
+    self.assertEquals('', content)
+    self.assertEquals({'content-length': '0',
+                       'content-type': 'text/html; charset=utf-8',
+                       'x': 'foo',
+                       'a': 'bar',
+                       'z': 'bin',
+                      },
+                      headers)
+
+
+class FirstFoundTest(WsgiTestBase):
+
+  def testEmptyConfiguration(self):
+    self.ResetServer(wsgi_util.first_found([]))
+    status, status_text, content, headers = self.DoHttpRequest('/')
+    self.assertEquals(six.moves.http_client.NOT_FOUND, status)
+    self.assertEquals(six.moves.http_client.responses[six.moves.http_client.NOT_FOUND], status_text)
+    self.assertEquals(util.pad_string(six.moves.http_client.responses[six.moves.http_client.NOT_FOUND]),
+                      content)
+    self.assertEquals({'content-length': '512',
+                       'content-type': 'text/plain; charset=utf-8',
+                      },
+                      headers)
+
+  def testOneApp(self):
+    self.ResetServer(wsgi_util.first_found([APP1]))
+
+    status, status_text, content, headers = self.DoHttpRequest('/')
+    self.assertEquals(six.moves.http_client.OK, status)
+    self.assertEquals(six.moves.http_client.responses[six.moves.http_client.OK], status_text)
+    self.assertEquals('App1', content)
+    self.assertEquals({'content-length': '4',
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+  def testIterator(self):
+    self.ResetServer(wsgi_util.first_found(iter([APP1])))
+
+    status, status_text, content, headers = self.DoHttpRequest('/')
+    self.assertEquals(six.moves.http_client.OK, status)
+    self.assertEquals(six.moves.http_client.responses[six.moves.http_client.OK], status_text)
+    self.assertEquals('App1', content)
+    self.assertEquals({'content-length': '4',
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+    # Do request again to make sure iterator was properly copied.
+    status, status_text, content, headers = self.DoHttpRequest('/')
+    self.assertEquals(six.moves.http_client.OK, status)
+    self.assertEquals(six.moves.http_client.responses[six.moves.http_client.OK], status_text)
+    self.assertEquals('App1', content)
+    self.assertEquals({'content-length': '4',
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+  def testTwoApps(self):
+    self.ResetServer(wsgi_util.first_found([APP1, APP2]))
+
+    status, status_text, content, headers = self.DoHttpRequest('/')
+    self.assertEquals(six.moves.http_client.OK, status)
+    self.assertEquals(six.moves.http_client.responses[six.moves.http_client.OK], status_text)
+    self.assertEquals('App1', content)
+    self.assertEquals({'content-length': '4',
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+  def testFirstNotFound(self):
+    self.ResetServer(wsgi_util.first_found([NOT_FOUND, APP2]))
+
+    status, status_text, content, headers = self.DoHttpRequest('/')
+    self.assertEquals(six.moves.http_client.OK, status)
+    self.assertEquals(six.moves.http_client.responses[six.moves.http_client.OK], status_text)
+    self.assertEquals('App2', content)
+    self.assertEquals({'content-length': '4',
+                       'content-type': 'text/html; charset=utf-8',
+                      },
+                      headers)
+
+  def testOnlyNotFound(self):
+    def current_error(environ, start_response):
+      """The variable current_status is defined in loop after ResetServer."""
+      headers = [('content-type', 'text/plain')]
+      status_line = '%03d Whatever' % current_status
+      start_response(status_line, headers)
+      return []
+
+    self.ResetServer(wsgi_util.first_found([current_error, APP2]))
+
+    statuses_to_check = sorted(httplib.responses.keys())
+    # 100, 204 and 304 have slightly different expectations, so they are left
+    # out of this test in order to keep the code simple.
+    for dont_check in (100, 200, 204, 304, 404):
+      statuses_to_check.remove(dont_check)
+    for current_status in statuses_to_check:
+      status, status_text, content, headers = self.DoHttpRequest('/')
+      self.assertEquals(current_status, status)
+      self.assertEquals('Whatever', status_text)
+
+
+if __name__ == '__main__':
+  unittest.main()