Add google-endpoints to third_party/.

third_party/google-endpoints/apitools/base/protorpclite/test_util.py

+#!/usr/bin/env python
+#
+# Copyright 2010 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.
+#
+
+"""Test utilities for message testing.
+
+Includes module interface test to ensure that public parts of module are
+correctly declared in __all__.
+
+Includes message types that correspond to those defined in
+services_test.proto.
+
+Includes additional test utilities to make sure encoding/decoding libraries
+conform.
+"""
+import cgi
+import datetime
+import inspect
+import os
+import re
+import socket
+import types
+
+import six
+from six.moves import range  # pylint: disable=redefined-builtin
+import unittest2 as unittest
+
+from apitools.base.protorpclite import message_types
+from apitools.base.protorpclite import messages
+from apitools.base.protorpclite import util
+
+# Unicode of the word "Russian" in cyrillic.
+RUSSIAN = u'\u0440\u0443\u0441\u0441\u043a\u0438\u0439'
+
+# All characters binary value interspersed with nulls.
+BINARY = b''.join(six.int2byte(value) + b'\0' for value in range(256))
+
+
+class TestCase(unittest.TestCase):
+
+    def assertRaisesWithRegexpMatch(self,
+                                    exception,
+                                    regexp,
+                                    function,
+                                    *params,
+                                    **kwargs):
+        """Check that exception is raised and text matches regular expression.
+
+        Args:
+          exception: Exception type that is expected.
+          regexp: String regular expression that is expected in error message.
+          function: Callable to test.
+          params: Parameters to forward to function.
+          kwargs: Keyword arguments to forward to function.
+        """
+        try:
+            function(*params, **kwargs)
+            self.fail('Expected exception %s was not raised' %
+                      exception.__name__)
+        except exception as err:
+            match = bool(re.match(regexp, str(err)))
+            self.assertTrue(match, 'Expected match "%s", found "%s"' % (regexp,
+                                                                        err))
+
+    def assertHeaderSame(self, header1, header2):
+        """Check that two HTTP headers are the same.
+
+        Args:
+          header1: Header value string 1.
+          header2: header value string 2.
+        """
+        value1, params1 = cgi.parse_header(header1)
+        value2, params2 = cgi.parse_header(header2)
+        self.assertEqual(value1, value2)
+        self.assertEqual(params1, params2)
+
+    def assertIterEqual(self, iter1, iter2):
+        """Check that two iterators or iterables are equal independent of order.
+
+        Similar to Python 2.7 assertItemsEqual.  Named differently in order to
+        avoid potential conflict.
+
+        Args:
+          iter1: An iterator or iterable.
+          iter2: An iterator or iterable.
+        """
+        list1 = list(iter1)
+        list2 = list(iter2)
+
+        unmatched1 = list()
+
+        while list1:
+            item1 = list1[0]
+            del list1[0]
+            for index in range(len(list2)):
+                if item1 == list2[index]:
+                    del list2[index]
+                    break
+            else:
+                unmatched1.append(item1)
+
+        error_message = []
+        for item in unmatched1:
+            error_message.append(
+                '  Item from iter1 not found in iter2: %r' % item)
+        for item in list2:
+            error_message.append(
+                '  Item from iter2 not found in iter1: %r' % item)
+        if error_message:
+            self.fail('Collections not equivalent:\n' +
+                      '\n'.join(error_message))
+
+
+class ModuleInterfaceTest(object):
+    """Test to ensure module interface is carefully constructed.
+
+    A module interface is the set of public objects listed in the
+    module __all__ attribute. Modules that that are considered public
+    should have this interface carefully declared. At all times, the
+    __all__ attribute should have objects intended to be publically
+    used and all other objects in the module should be considered
+    unused.
+
+    Protected attributes (those beginning with '_') and other imported
+    modules should not be part of this set of variables. An exception
+    is for variables that begin and end with '__' which are implicitly
+    part of the interface (eg. __name__, __file__, __all__ itself,
+    etc.).
+
+    Modules that are imported in to the tested modules are an
+    exception and may be left out of the __all__ definition. The test
+    is done by checking the value of what would otherwise be a public
+    name and not allowing it to be exported if it is an instance of a
+    module. Modules that are explicitly exported are for the time
+    being not permitted.
+
+    To use this test class a module should define a new class that
+    inherits first from ModuleInterfaceTest and then from
+    test_util.TestCase. No other tests should be added to this test
+    case, making the order of inheritance less important, but if setUp
+    for some reason is overidden, it is important that
+    ModuleInterfaceTest is first in the list so that its setUp method
+    is invoked.
+
+    Multiple inheritance is required so that ModuleInterfaceTest is
+    not itself a test, and is not itself executed as one.
+
+    The test class is expected to have the following class attributes
+    defined:
+
+      MODULE: A reference to the module that is being validated for interface
+        correctness.
+
+    Example:
+      Module definition (hello.py):
+
+        import sys
+
+        __all__ = ['hello']
+
+        def _get_outputter():
+          return sys.stdout
+
+        def hello():
+          _get_outputter().write('Hello\n')
+
+      Test definition:
+
+        import unittest
+        from protorpc import test_util
+
+        import hello
+
+        class ModuleInterfaceTest(test_util.ModuleInterfaceTest,
+                                  test_util.TestCase):
+
+          MODULE = hello
+
+
+        class HelloTest(test_util.TestCase):
+          ... Test 'hello' module ...
+
+
+        if __name__ == '__main__':
+          unittest.main()
+
+    """
+
+    def setUp(self):
+        """Set up makes sure that MODULE and IMPORTED_MODULES is defined.
+
+        This is a basic configuration test for the test itself so does not
+        get it's own test case.
+        """
+        if not hasattr(self, 'MODULE'):
+            self.fail(
+                "You must define 'MODULE' on ModuleInterfaceTest sub-class "
+                "%s." % type(self).__name__)
+
+    def testAllExist(self):
+        """Test that all attributes defined in __all__ exist."""
+        missing_attributes = []
+        for attribute in self.MODULE.__all__:
+            if not hasattr(self.MODULE, attribute):
+                missing_attributes.append(attribute)
+        if missing_attributes:
+            self.fail('%s of __all__ are not defined in module.' %
+                      missing_attributes)
+
+    def testAllExported(self):
+        """Test that all public attributes not imported are in __all__."""
+        missing_attributes = []
+        for attribute in dir(self.MODULE):
+            if not attribute.startswith('_'):
+                if (attribute not in self.MODULE.__all__ and
+                        not isinstance(getattr(self.MODULE, attribute),
+                                       types.ModuleType) and
+                        attribute != 'with_statement'):
+                    missing_attributes.append(attribute)
+        if missing_attributes:
+            self.fail('%s are not modules and not defined in __all__.' %
+                      missing_attributes)
+
+    def testNoExportedProtectedVariables(self):
+        """Test that there are no protected variables listed in __all__."""
+        protected_variables = []
+        for attribute in self.MODULE.__all__:
+            if attribute.startswith('_'):
+                protected_variables.append(attribute)
+        if protected_variables:
+            self.fail('%s are protected variables and may not be exported.' %
+                      protected_variables)
+
+    def testNoExportedModules(self):
+        """Test that no modules exist in __all__."""
+        exported_modules = []
+        for attribute in self.MODULE.__all__:
+            try:
+                value = getattr(self.MODULE, attribute)
+            except AttributeError:
+                # This is a different error case tested for in testAllExist.
+                pass
+            else:
+                if isinstance(value, types.ModuleType):
+                    exported_modules.append(attribute)
+        if exported_modules:
+            self.fail('%s are modules and may not be exported.' %
+                      exported_modules)
+
+
+class NestedMessage(messages.Message):
+    """Simple message that gets nested in another message."""
+
+    a_value = messages.StringField(1, required=True)
+
+
+class HasNestedMessage(messages.Message):
+    """Message that has another message nested in it."""
+
+    nested = messages.MessageField(NestedMessage, 1)
+    repeated_nested = messages.MessageField(NestedMessage, 2, repeated=True)
+
+
+class HasDefault(messages.Message):
+    """Has a default value."""
+
+    a_value = messages.StringField(1, default=u'a default')
+
+
+class OptionalMessage(messages.Message):
+    """Contains all message types."""
+
+    class SimpleEnum(messages.Enum):
+        """Simple enumeration type."""
+        VAL1 = 1
+        VAL2 = 2
+
+    double_value = messages.FloatField(1, variant=messages.Variant.DOUBLE)
+    float_value = messages.FloatField(2, variant=messages.Variant.FLOAT)
+    int64_value = messages.IntegerField(3, variant=messages.Variant.INT64)
+    uint64_value = messages.IntegerField(4, variant=messages.Variant.UINT64)
+    int32_value = messages.IntegerField(5, variant=messages.Variant.INT32)
+    bool_value = messages.BooleanField(6, variant=messages.Variant.BOOL)
+    string_value = messages.StringField(7, variant=messages.Variant.STRING)
+    bytes_value = messages.BytesField(8, variant=messages.Variant.BYTES)
+    enum_value = messages.EnumField(SimpleEnum, 10)
+
+
+class RepeatedMessage(messages.Message):
+    """Contains all message types as repeated fields."""
+
+    class SimpleEnum(messages.Enum):
+        """Simple enumeration type."""
+        VAL1 = 1
+        VAL2 = 2
+
+    double_value = messages.FloatField(1,
+                                       variant=messages.Variant.DOUBLE,
+                                       repeated=True)
+    float_value = messages.FloatField(2,
+                                      variant=messages.Variant.FLOAT,
+                                      repeated=True)
+    int64_value = messages.IntegerField(3,
+                                        variant=messages.Variant.INT64,
+                                        repeated=True)
+    uint64_value = messages.IntegerField(4,
+                                         variant=messages.Variant.UINT64,
+                                         repeated=True)
+    int32_value = messages.IntegerField(5,
+                                        variant=messages.Variant.INT32,
+                                        repeated=True)
+    bool_value = messages.BooleanField(6,
+                                       variant=messages.Variant.BOOL,
+                                       repeated=True)
+    string_value = messages.StringField(7,
+                                        variant=messages.Variant.STRING,
+                                        repeated=True)
+    bytes_value = messages.BytesField(8,
+                                      variant=messages.Variant.BYTES,
+                                      repeated=True)
+    enum_value = messages.EnumField(SimpleEnum,
+                                    10,
+                                    repeated=True)
+
+
+class HasOptionalNestedMessage(messages.Message):
+
+    nested = messages.MessageField(OptionalMessage, 1)
+    repeated_nested = messages.MessageField(OptionalMessage, 2, repeated=True)
+
+
+# pylint:disable=anomalous-unicode-escape-in-string
+class ProtoConformanceTestBase(object):
+    """Protocol conformance test base class.
+
+    Each supported protocol should implement two methods that support encoding
+    and decoding of Message objects in that format:
+
+      encode_message(message) - Serialize to encoding.
+      encode_message(message, encoded_message) - Deserialize from encoding.
+
+    Tests for the modules where these functions are implemented should extend
+    this class in order to support basic behavioral expectations.  This ensures
+    that protocols correctly encode and decode message transparently to the
+    caller.
+
+    In order to support these test, the base class should also extend
+    the TestCase class and implement the following class attributes
+    which define the encoded version of certain protocol buffers:
+
+      encoded_partial:
+        <OptionalMessage
+          double_value: 1.23
+          int64_value: -100000000000
+          string_value: u"a string"
+          enum_value: OptionalMessage.SimpleEnum.VAL2
+          >
+
+      encoded_full:
+        <OptionalMessage
+          double_value: 1.23
+          float_value: -2.5
+          int64_value: -100000000000
+          uint64_value: 102020202020
+          int32_value: 1020
+          bool_value: true
+          string_value: u"a string\u044f"
+          bytes_value: b"a bytes\xff\xfe"
+          enum_value: OptionalMessage.SimpleEnum.VAL2
+          >
+
+      encoded_repeated:
+        <RepeatedMessage
+          double_value: [1.23, 2.3]
+          float_value: [-2.5, 0.5]
+          int64_value: [-100000000000, 20]
+          uint64_value: [102020202020, 10]
+          int32_value: [1020, 718]
+          bool_value: [true, false]
+          string_value: [u"a string\u044f", u"another string"]
+          bytes_value: [b"a bytes\xff\xfe", b"another bytes"]
+          enum_value: [OptionalMessage.SimpleEnum.VAL2,
+                       OptionalMessage.SimpleEnum.VAL 1]
+          >
+
+      encoded_nested:
+        <HasNestedMessage
+          nested: <NestedMessage
+            a_value: "a string"
+            >
+          >
+
+      encoded_repeated_nested:
+        <HasNestedMessage
+          repeated_nested: [
+              <NestedMessage a_value: "a string">,
+              <NestedMessage a_value: "another string">
+            ]
+          >
+
+      unexpected_tag_message:
+        An encoded message that has an undefined tag or number in the stream.
+
+      encoded_default_assigned:
+        <HasDefault
+          a_value: "a default"
+          >
+
+      encoded_nested_empty:
+        <HasOptionalNestedMessage
+          nested: <OptionalMessage>
+          >
+
+      encoded_invalid_enum:
+        <OptionalMessage
+          enum_value: (invalid value for serialization type)
+          >
+    """
+
+    encoded_empty_message = ''
+
+    def testEncodeInvalidMessage(self):
+        message = NestedMessage()
+        self.assertRaises(messages.ValidationError,
+                          self.PROTOLIB.encode_message, message)
+
+    def CompareEncoded(self, expected_encoded, actual_encoded):
+        """Compare two encoded protocol values.
+
+        Can be overridden by sub-classes to special case comparison.
+        For example, to eliminate white space from output that is not
+        relevant to encoding.
+
+        Args:
+          expected_encoded: Expected string encoded value.
+          actual_encoded: Actual string encoded value.
+        """
+        self.assertEquals(expected_encoded, actual_encoded)
+
+    def EncodeDecode(self, encoded, expected_message):
+        message = self.PROTOLIB.decode_message(type(expected_message), encoded)
+        self.assertEquals(expected_message, message)
+        self.CompareEncoded(encoded, self.PROTOLIB.encode_message(message))
+
+    def testEmptyMessage(self):
+        self.EncodeDecode(self.encoded_empty_message, OptionalMessage())
+
+    def testPartial(self):
+        """Test message with a few values set."""
+        message = OptionalMessage()
+        message.double_value = 1.23
+        message.int64_value = -100000000000
+        message.int32_value = 1020
+        message.string_value = u'a string'
+        message.enum_value = OptionalMessage.SimpleEnum.VAL2
+
+        self.EncodeDecode(self.encoded_partial, message)
+
+    def testFull(self):
+        """Test all types."""
+        message = OptionalMessage()
+        message.double_value = 1.23
+        message.float_value = -2.5
+        message.int64_value = -100000000000
+        message.uint64_value = 102020202020
+        message.int32_value = 1020
+        message.bool_value = True
+        message.string_value = u'a string\u044f'
+        message.bytes_value = b'a bytes\xff\xfe'
+        message.enum_value = OptionalMessage.SimpleEnum.VAL2
+
+        self.EncodeDecode(self.encoded_full, message)
+
+    def testRepeated(self):
+        """Test repeated fields."""
+        message = RepeatedMessage()
+        message.double_value = [1.23, 2.3]
+        message.float_value = [-2.5, 0.5]
+        message.int64_value = [-100000000000, 20]
+        message.uint64_value = [102020202020, 10]
+        message.int32_value = [1020, 718]
+        message.bool_value = [True, False]
+        message.string_value = [u'a string\u044f', u'another string']
+        message.bytes_value = [b'a bytes\xff\xfe', b'another bytes']
+        message.enum_value = [RepeatedMessage.SimpleEnum.VAL2,
+                              RepeatedMessage.SimpleEnum.VAL1]
+
+        self.EncodeDecode(self.encoded_repeated, message)
+
+    def testNested(self):
+        """Test nested messages."""
+        nested_message = NestedMessage()
+        nested_message.a_value = u'a string'
+
+        message = HasNestedMessage()
+        message.nested = nested_message
+
+        self.EncodeDecode(self.encoded_nested, message)
+
+    def testRepeatedNested(self):
+        """Test repeated nested messages."""
+        nested_message1 = NestedMessage()
+        nested_message1.a_value = u'a string'
+        nested_message2 = NestedMessage()
+        nested_message2.a_value = u'another string'
+
+        message = HasNestedMessage()
+        message.repeated_nested = [nested_message1, nested_message2]
+
+        self.EncodeDecode(self.encoded_repeated_nested, message)
+
+    def testStringTypes(self):
+        """Test that encoding str on StringField works."""
+        message = OptionalMessage()
+        message.string_value = 'Latin'
+        self.EncodeDecode(self.encoded_string_types, message)
+
+    def testEncodeUninitialized(self):
+        """Test that cannot encode uninitialized message."""
+        required = NestedMessage()
+        self.assertRaisesWithRegexpMatch(messages.ValidationError,
+                                         "Message NestedMessage is missing "
+                                         "required field a_value",
+                                         self.PROTOLIB.encode_message,
+                                         required)
+
+    def testUnexpectedField(self):
+        """Test decoding and encoding unexpected fields."""
+        loaded_message = self.PROTOLIB.decode_message(
+            OptionalMessage, self.unexpected_tag_message)
+        # Message should be equal to an empty message, since unknown
+        # values aren't included in equality.
+        self.assertEquals(OptionalMessage(), loaded_message)
+        # Verify that the encoded message matches the source, including the
+        # unknown value.
+        self.assertEquals(self.unexpected_tag_message,
+                          self.PROTOLIB.encode_message(loaded_message))
+
+    def testDoNotSendDefault(self):
+        """Test that default is not sent when nothing is assigned."""
+        self.EncodeDecode(self.encoded_empty_message, HasDefault())
+
+    def testSendDefaultExplicitlyAssigned(self):
+        """Test that default is sent when explcitly assigned."""
+        message = HasDefault()
+
+        message.a_value = HasDefault.a_value.default
+
+        self.EncodeDecode(self.encoded_default_assigned, message)
+
+    def testEncodingNestedEmptyMessage(self):
+        """Test encoding a nested empty message."""
+        message = HasOptionalNestedMessage()
+        message.nested = OptionalMessage()
+
+        self.EncodeDecode(self.encoded_nested_empty, message)
+
+    def testEncodingRepeatedNestedEmptyMessage(self):
+        """Test encoding a nested empty message."""
+        message = HasOptionalNestedMessage()
+        message.repeated_nested = [OptionalMessage(), OptionalMessage()]
+
+        self.EncodeDecode(self.encoded_repeated_nested_empty, message)
+
+    def testContentType(self):
+        self.assertTrue(isinstance(self.PROTOLIB.CONTENT_TYPE, str))
+
+    def testDecodeInvalidEnumType(self):
+        self.assertRaisesWithRegexpMatch(messages.DecodeError,
+                                         'Invalid enum value ',
+                                         self.PROTOLIB.decode_message,
+                                         OptionalMessage,
+                                         self.encoded_invalid_enum)
+
+    def testDateTimeNoTimeZone(self):
+        """Test that DateTimeFields are encoded/decoded correctly."""
+
+        class MyMessage(messages.Message):
+            value = message_types.DateTimeField(1)
+
+        value = datetime.datetime(2013, 1, 3, 11, 36, 30, 123000)
+        message = MyMessage(value=value)
+        decoded = self.PROTOLIB.decode_message(
+            MyMessage, self.PROTOLIB.encode_message(message))
+        self.assertEquals(decoded.value, value)
+
+    def testDateTimeWithTimeZone(self):
+        """Test DateTimeFields with time zones."""
+
+        class MyMessage(messages.Message):
+            value = message_types.DateTimeField(1)
+
+        value = datetime.datetime(2013, 1, 3, 11, 36, 30, 123000,
+                                  util.TimeZoneOffset(8 * 60))
+        message = MyMessage(value=value)
+        decoded = self.PROTOLIB.decode_message(
+            MyMessage, self.PROTOLIB.encode_message(message))
+        self.assertEquals(decoded.value, value)
+
+
+def pick_unused_port():
+    """Find an unused port to use in tests.
+
+      Derived from Damon Kohlers example:
+
+        http://code.activestate.com/recipes/531822-pick-unused-port
+    """
+    temp = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    try:
+        temp.bind(('localhost', 0))
+        port = temp.getsockname()[1]
+    finally:
+        temp.close()
+    return port
+
+
+def get_module_name(module_attribute):
+    """Get the module name.
+
+    Args:
+      module_attribute: An attribute of the module.
+
+    Returns:
+      The fully qualified module name or simple module name where
+      'module_attribute' is defined if the module name is "__main__".
+    """
+    if module_attribute.__module__ == '__main__':
+        module_file = inspect.getfile(module_attribute)
+        default = os.path.basename(module_file).split('.')[0]
+        return default
+    else:
+        return module_attribute.__module__