text: #!/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.
  #
  
  """Tests for protorpc.stub."""
  
  __author__ = 'rafek@google.com (Rafe Kaplan)'
  
  
  import StringIO
  import sys
  import types
  import unittest
  
  from protorpc import definition
  from protorpc import descriptor
  from protorpc import message_types
  from protorpc import messages
  from protorpc import protobuf
  from protorpc import remote
  from protorpc import test_util
  
  import mox
  
  
  class ModuleInterfaceTest(test_util.ModuleInterfaceTest,
                            test_util.TestCase):
  
    MODULE = definition
  
  
  class DefineEnumTest(test_util.TestCase):
    """Test for define_enum."""
  
    def testDefineEnum_Empty(self):
      """Test defining an empty enum."""
      enum_descriptor = descriptor.EnumDescriptor()
      enum_descriptor.name = 'Empty'
  
      enum_class = definition.define_enum(enum_descriptor, 'whatever')
  
      self.assertEquals('Empty', enum_class.__name__)
      self.assertEquals('whatever', enum_class.__module__)
  
      self.assertEquals(enum_descriptor, descriptor.describe_enum(enum_class))
  
    def testDefineEnum(self):
      """Test defining an enum."""
      red = descriptor.EnumValueDescriptor()
      green = descriptor.EnumValueDescriptor()
      blue = descriptor.EnumValueDescriptor()
  
      red.name = 'RED'
      red.number = 1
      green.name = 'GREEN'
      green.number = 2
      blue.name = 'BLUE'
      blue.number = 3
  
      enum_descriptor = descriptor.EnumDescriptor()
      enum_descriptor.name = 'Colors'
      enum_descriptor.values = [red, green, blue]
  
      enum_class = definition.define_enum(enum_descriptor, 'whatever')
  
      self.assertEquals('Colors', enum_class.__name__)
      self.assertEquals('whatever', enum_class.__module__)
  
      self.assertEquals(enum_descriptor, descriptor.describe_enum(enum_class))
  
  
  class DefineFieldTest(test_util.TestCase):
    """Test for define_field."""
  
    def testDefineField_Optional(self):
      """Test defining an optional field instance from a method descriptor."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_field'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.INT32
      field_descriptor.label = descriptor.FieldDescriptor.Label.OPTIONAL
  
      field = definition.define_field(field_descriptor)
  
      # Name will not be set from the original descriptor.
      self.assertFalse(hasattr(field, 'name'))
  
      self.assertTrue(isinstance(field, messages.IntegerField))
      self.assertEquals(1, field.number)
      self.assertEquals(descriptor.FieldDescriptor.Variant.INT32, field.variant)
      self.assertFalse(field.required)
      self.assertFalse(field.repeated)
  
    def testDefineField_Required(self):
      """Test defining a required field instance from a method descriptor."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_field'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.STRING
      field_descriptor.label = descriptor.FieldDescriptor.Label.REQUIRED
  
      field = definition.define_field(field_descriptor)
  
      # Name will not be set from the original descriptor.
      self.assertFalse(hasattr(field, 'name'))
  
      self.assertTrue(isinstance(field, messages.StringField))
      self.assertEquals(1, field.number)
      self.assertEquals(descriptor.FieldDescriptor.Variant.STRING, field.variant)
      self.assertTrue(field.required)
      self.assertFalse(field.repeated)
  
    def testDefineField_Repeated(self):
      """Test defining a repeated field instance from a method descriptor."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_field'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.DOUBLE
      field_descriptor.label = descriptor.FieldDescriptor.Label.REPEATED
  
      field = definition.define_field(field_descriptor)
  
      # Name will not be set from the original descriptor.
      self.assertFalse(hasattr(field, 'name'))
  
      self.assertTrue(isinstance(field, messages.FloatField))
      self.assertEquals(1, field.number)
      self.assertEquals(descriptor.FieldDescriptor.Variant.DOUBLE, field.variant)
      self.assertFalse(field.required)
      self.assertTrue(field.repeated)
  
    def testDefineField_Message(self):
      """Test defining a message field."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_field'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.MESSAGE
      field_descriptor.type_name = 'something.yet.to.be.Defined'
      field_descriptor.label = descriptor.FieldDescriptor.Label.REPEATED
  
      field = definition.define_field(field_descriptor)
  
      # Name will not be set from the original descriptor.
      self.assertFalse(hasattr(field, 'name'))
  
      self.assertTrue(isinstance(field, messages.MessageField))
      self.assertEquals(1, field.number)
      self.assertEquals(descriptor.FieldDescriptor.Variant.MESSAGE, field.variant)
      self.assertFalse(field.required)
      self.assertTrue(field.repeated)
      self.assertRaisesWithRegexpMatch(messages.DefinitionNotFoundError,
                                       'Could not find definition for '
                                       'something.yet.to.be.Defined',
                                       getattr, field, 'type')
  
    def testDefineField_DateTime(self):
      """Test defining a date time field."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_timestamp'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.MESSAGE
      field_descriptor.type_name = 'protorpc.message_types.DateTimeMessage'
      field_descriptor.label = descriptor.FieldDescriptor.Label.REPEATED
  
      field = definition.define_field(field_descriptor)
  
      # Name will not be set from the original descriptor.
      self.assertFalse(hasattr(field, 'name'))
  
      self.assertTrue(isinstance(field, message_types.DateTimeField))
      self.assertEquals(1, field.number)
      self.assertEquals(descriptor.FieldDescriptor.Variant.MESSAGE, field.variant)
      self.assertFalse(field.required)
      self.assertTrue(field.repeated)
  
    def testDefineField_Enum(self):
      """Test defining an enum field."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_field'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.ENUM
      field_descriptor.type_name = 'something.yet.to.be.Defined'
      field_descriptor.label = descriptor.FieldDescriptor.Label.REPEATED
  
      field = definition.define_field(field_descriptor)
  
      # Name will not be set from the original descriptor.
      self.assertFalse(hasattr(field, 'name'))
  
      self.assertTrue(isinstance(field, messages.EnumField))
      self.assertEquals(1, field.number)
      self.assertEquals(descriptor.FieldDescriptor.Variant.ENUM, field.variant)
      self.assertFalse(field.required)
      self.assertTrue(field.repeated)
      self.assertRaisesWithRegexpMatch(messages.DefinitionNotFoundError,
                                       'Could not find definition for '
                                       'something.yet.to.be.Defined',
                                       getattr, field, 'type')
  
    def testDefineField_Default_Bool(self):
      """Test defining a default value for a bool."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_field'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.BOOL
      field_descriptor.default_value = u'true'
  
      field = definition.define_field(field_descriptor)
  
      # Name will not be set from the original descriptor.
      self.assertFalse(hasattr(field, 'name'))
  
      self.assertTrue(isinstance(field, messages.BooleanField))
      self.assertEquals(1, field.number)
      self.assertEquals(descriptor.FieldDescriptor.Variant.BOOL, field.variant)
      self.assertFalse(field.required)
      self.assertFalse(field.repeated)
      self.assertEqual(field.default, True)
  
      field_descriptor.default_value = u'false'
  
      field = definition.define_field(field_descriptor)
  
      self.assertEqual(field.default, False)
  
    def testDefineField_Default_Float(self):
      """Test defining a default value for a float."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_field'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.FLOAT
      field_descriptor.default_value = u'34.567'
  
      field = definition.define_field(field_descriptor)
  
      # Name will not be set from the original descriptor.
      self.assertFalse(hasattr(field, 'name'))
  
      self.assertTrue(isinstance(field, messages.FloatField))
      self.assertEquals(1, field.number)
      self.assertEquals(descriptor.FieldDescriptor.Variant.FLOAT, field.variant)
      self.assertFalse(field.required)
      self.assertFalse(field.repeated)
      self.assertEqual(field.default, 34.567)
  
    def testDefineField_Default_Int(self):
      """Test defining a default value for an int."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_field'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.INT64
      field_descriptor.default_value = u'34'
  
      field = definition.define_field(field_descriptor)
  
      # Name will not be set from the original descriptor.
      self.assertFalse(hasattr(field, 'name'))
  
      self.assertTrue(isinstance(field, messages.IntegerField))
      self.assertEquals(1, field.number)
      self.assertEquals(descriptor.FieldDescriptor.Variant.INT64, field.variant)
      self.assertFalse(field.required)
      self.assertFalse(field.repeated)
      self.assertEqual(field.default, 34)
  
    def testDefineField_Default_Str(self):
      """Test defining a default value for a str."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_field'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.STRING
      field_descriptor.default_value = u'Test'
  
      field = definition.define_field(field_descriptor)
  
      # Name will not be set from the original descriptor.
      self.assertFalse(hasattr(field, 'name'))
  
      self.assertTrue(isinstance(field, messages.StringField))
      self.assertEquals(1, field.number)
      self.assertEquals(descriptor.FieldDescriptor.Variant.STRING, field.variant)
      self.assertFalse(field.required)
      self.assertFalse(field.repeated)
      self.assertEqual(field.default, u'Test')
  
    def testDefineField_Default_Invalid(self):
      """Test defining a default value that is not valid."""
      field_descriptor = descriptor.FieldDescriptor()
  
      field_descriptor.name = 'a_field'
      field_descriptor.number = 1
      field_descriptor.variant = descriptor.FieldDescriptor.Variant.INT64
      field_descriptor.default_value = u'Test'
  
      # Verify that the string is passed to the Constructor.
      mock = mox.Mox()
      mock.StubOutWithMock(messages.IntegerField, '__init__')
      messages.IntegerField.__init__(
          default=u'Test',
          number=1,
          variant=messages.Variant.INT64
          ).AndRaise(messages.InvalidDefaultError)
  
      mock.ReplayAll()
      self.assertRaises(messages.InvalidDefaultError,
                        definition.define_field, field_descriptor)
      mock.VerifyAll()
  
      mock.ResetAll()
      mock.UnsetStubs()
  
  
  class DefineMessageTest(test_util.TestCase):
    """Test for define_message."""
  
    def testDefineMessageEmpty(self):
      """Test definition a message with no fields or enums."""
  
      class AMessage(messages.Message):
        pass
  
      message_descriptor = descriptor.describe_message(AMessage)
  
      message_class = definition.define_message(message_descriptor, '__main__')
  
      self.assertEquals('AMessage', message_class.__name__)
      self.assertEquals('__main__', message_class.__module__)
  
      self.assertEquals(message_descriptor,
                        descriptor.describe_message(message_class))
  
    def testDefineMessageEnumOnly(self):
      """Test definition a message with only enums."""
  
      class AMessage(messages.Message):
        class NestedEnum(messages.Enum):
          pass
  
      message_descriptor = descriptor.describe_message(AMessage)
  
      message_class = definition.define_message(message_descriptor, '__main__')
  
      self.assertEquals('AMessage', message_class.__name__)
      self.assertEquals('__main__', message_class.__module__)
  
      self.assertEquals(message_descriptor,
                        descriptor.describe_message(message_class))
  
    def testDefineMessageFieldsOnly(self):
      """Test definition a message with only fields."""
  
      class AMessage(messages.Message):
  
        field1 = messages.IntegerField(1)
        field2 = messages.StringField(2)
  
      message_descriptor = descriptor.describe_message(AMessage)
  
      message_class = definition.define_message(message_descriptor, '__main__')
  
      self.assertEquals('AMessage', message_class.__name__)
      self.assertEquals('__main__', message_class.__module__)
  
      self.assertEquals(message_descriptor,
                        descriptor.describe_message(message_class))
  
    def testDefineMessage(self):
      """Test defining Message class from descriptor."""
  
      class AMessage(messages.Message):
        class NestedEnum(messages.Enum):
          pass
  
        field1 = messages.IntegerField(1)
        field2 = messages.StringField(2)
  
      message_descriptor = descriptor.describe_message(AMessage)
  
      message_class = definition.define_message(message_descriptor, '__main__')
  
      self.assertEquals('AMessage', message_class.__name__)
      self.assertEquals('__main__', message_class.__module__)
  
      self.assertEquals(message_descriptor,
                        descriptor.describe_message(message_class))
  
  
  class DefineServiceTest(test_util.TestCase):
    """Test service proxy definition."""
  
    def setUp(self):
      """Set up mock and request classes."""
      self.module = types.ModuleType('stocks')
  
      class GetQuoteRequest(messages.Message):
        __module__ = 'stocks'
  
        symbols = messages.StringField(1, repeated=True)
  
      class GetQuoteResponse(messages.Message):
        __module__ = 'stocks'
  
        prices = messages.IntegerField(1, repeated=True)
  
      self.module.GetQuoteRequest = GetQuoteRequest
      self.module.GetQuoteResponse = GetQuoteResponse
  
    def testDefineService(self):
      """Test service definition from descriptor."""
      method_descriptor = descriptor.MethodDescriptor()
      method_descriptor.name = 'get_quote'
      method_descriptor.request_type = 'GetQuoteRequest'
      method_descriptor.response_type = 'GetQuoteResponse'
  
      service_descriptor = descriptor.ServiceDescriptor()
      service_descriptor.name = 'Stocks'
      service_descriptor.methods = [method_descriptor]
  
      StockService = definition.define_service(service_descriptor, self.module)
  
      self.assertTrue(issubclass(StockService, remote.Service))
      self.assertTrue(issubclass(StockService.Stub, remote.StubBase))
  
      request = self.module.GetQuoteRequest()
      service = StockService()
      self.assertRaises(NotImplementedError,
                        service.get_quote, request)
  
      self.assertEquals(self.module.GetQuoteRequest,
                        service.get_quote.remote.request_type)
      self.assertEquals(self.module.GetQuoteResponse,
                        service.get_quote.remote.response_type)
  
  
  class ModuleTest(test_util.TestCase):
    """Test for module creation and importation functions."""
  
    def MakeFileDescriptor(self, package):
      """Helper method to construct FileDescriptors.
  
      Creates FileDescriptor with a MessageDescriptor and an EnumDescriptor.
  
      Args:
        package: Package name to give new file descriptors.
  
      Returns:
        New FileDescriptor instance.
      """
      enum_descriptor = descriptor.EnumDescriptor()
      enum_descriptor.name = u'MyEnum'
  
      message_descriptor = descriptor.MessageDescriptor()
      message_descriptor.name = u'MyMessage'
  
      service_descriptor = descriptor.ServiceDescriptor()
      service_descriptor.name = u'MyService'
  
      file_descriptor = descriptor.FileDescriptor()
      file_descriptor.package = package
      file_descriptor.enum_types = [enum_descriptor]
      file_descriptor.message_types = [message_descriptor]
      file_descriptor.service_types = [service_descriptor]
  
      return file_descriptor
  
    def testDefineModule(self):
      """Test define_module function."""
      file_descriptor = self.MakeFileDescriptor('my.package')
  
      module = definition.define_file(file_descriptor)
  
      self.assertEquals('my.package', module.__name__)
      self.assertEquals('my.package', module.MyEnum.__module__)
      self.assertEquals('my.package', module.MyMessage.__module__)
      self.assertEquals('my.package', module.MyService.__module__)
  
      self.assertEquals(file_descriptor, descriptor.describe_file(module))
  
    def testDefineModule_ReuseModule(self):
      """Test updating module with additional definitions."""
      file_descriptor = self.MakeFileDescriptor('my.package')
  
      module = types.ModuleType('override')
      self.assertEquals(module, definition.define_file(file_descriptor, module))
  
      self.assertEquals('override', module.MyEnum.__module__)
      self.assertEquals('override', module.MyMessage.__module__)
      self.assertEquals('override', module.MyService.__module__)
  
      # One thing is different between original descriptor and new.
      file_descriptor.package = 'override'
      self.assertEquals(file_descriptor, descriptor.describe_file(module))
  
    def testImportFile(self):
      """Test importing FileDescriptor in to module space."""
      modules = {}
      file_descriptor = self.MakeFileDescriptor('standalone')
      definition.import_file(file_descriptor, modules=modules)
      self.assertEquals(file_descriptor,
                        descriptor.describe_file(modules['standalone']))
  
    def testImportFile_InToExisting(self):
      """Test importing FileDescriptor in to existing module."""
      module = types.ModuleType('standalone')
      modules = {'standalone': module}
      file_descriptor = self.MakeFileDescriptor('standalone')
      definition.import_file(file_descriptor, modules=modules)
      self.assertEquals(module, modules['standalone'])
      self.assertEquals(file_descriptor,
                        descriptor.describe_file(modules['standalone']))
  
    def testImportFile_InToGlobalModules(self):
      """Test importing FileDescriptor in to global modules."""
      original_modules = sys.modules
      try:
        sys.modules = dict(sys.modules)
        if 'standalone' in sys.modules:
          del sys.modules['standalone']
        file_descriptor = self.MakeFileDescriptor('standalone')
        definition.import_file(file_descriptor)
        self.assertEquals(file_descriptor,
                          descriptor.describe_file(sys.modules['standalone']))
      finally:
        sys.modules = original_modules
  
    def testImportFile_Nested(self):
      """Test importing FileDescriptor in to existing nested module."""
      modules = {}
      file_descriptor = self.MakeFileDescriptor('root.nested')
      definition.import_file(file_descriptor, modules=modules)
      self.assertEquals(modules['root'].nested, modules['root.nested'])
      self.assertEquals(file_descriptor,
                        descriptor.describe_file(modules['root.nested']))
  
    def testImportFile_NoPackage(self):
      """Test importing FileDescriptor with no package."""
      file_descriptor = self.MakeFileDescriptor('does not matter')
      file_descriptor.reset('package')
      self.assertRaisesWithRegexpMatch(ValueError,
                                       'File descriptor must have package name',
                                       definition.import_file,
                                       file_descriptor)
  
    def testImportFileSet(self):
      """Test importing a whole file set."""
      file_set = descriptor.FileSet()
      file_set.files = [self.MakeFileDescriptor(u'standalone'),
                        self.MakeFileDescriptor(u'root.nested'),
                        self.MakeFileDescriptor(u'root.nested.nested'),
                       ]
  
      root = types.ModuleType('root')
      nested = types.ModuleType('root.nested')
      root.nested = nested
      modules = {
          'root': root,
          'root.nested': nested,
      }
  
      definition.import_file_set(file_set, modules=modules)
  
      self.assertEquals(root, modules['root'])
      self.assertEquals(nested, modules['root.nested'])
      self.assertEquals(nested.nested, modules['root.nested.nested'])
  
      self.assertEquals(file_set,
                        descriptor.describe_file_set(
                            [modules['standalone'],
                             modules['root.nested'],
                             modules['root.nested.nested'],
                            ]))
  
    def testImportFileSetFromFile(self):
      """Test importing a whole file set from a file."""
      file_set = descriptor.FileSet()
      file_set.files = [self.MakeFileDescriptor(u'standalone'),
                        self.MakeFileDescriptor(u'root.nested'),
                        self.MakeFileDescriptor(u'root.nested.nested'),
                       ]
  
      stream = StringIO.StringIO(protobuf.encode_message(file_set))
  
      self.mox = mox.Mox()
      opener = self.mox.CreateMockAnything()
      opener('my-file.dat', 'rb').AndReturn(stream)
  
      self.mox.ReplayAll()
  
      modules = {}
      definition.import_file_set('my-file.dat', modules=modules, _open=opener)
  
      self.assertEquals(file_set,
                        descriptor.describe_file_set(
                            [modules['standalone'],
                             modules['root.nested'],
                             modules['root.nested.nested'],
                            ]))
  
    def testImportBuiltInProtorpcClasses(self):
      """Test that built in Protorpc classes are skipped."""
      file_set = descriptor.FileSet()
      file_set.files = [self.MakeFileDescriptor(u'standalone'),
                        self.MakeFileDescriptor(u'root.nested'),
                        self.MakeFileDescriptor(u'root.nested.nested'),
                        descriptor.describe_file(descriptor),
      ]
  
      root = types.ModuleType('root')
      nested = types.ModuleType('root.nested')
      root.nested = nested
      modules = {
          'root': root,
          'root.nested': nested,
          'protorpc.descriptor': descriptor,
      }
  
      definition.import_file_set(file_set, modules=modules)
  
      self.assertEquals(root, modules['root'])
      self.assertEquals(nested, modules['root.nested'])
      self.assertEquals(nested.nested, modules['root.nested.nested'])
      self.assertEquals(descriptor, modules['protorpc.descriptor'])
  
      self.assertEquals(file_set,
                        descriptor.describe_file_set(
                            [modules['standalone'],
                             modules['root.nested'],
                             modules['root.nested.nested'],
                             modules['protorpc.descriptor'],
                            ]))
  
  
  if __name__ == '__main__':
    unittest.main()
  