Sync from Piper @480194141
PROTOBUF_SYNC_PIPER
diff --git a/python/BUILD.bazel b/python/BUILD.bazel
index ffaf988..56d8c8a 100644
--- a/python/BUILD.bazel
+++ b/python/BUILD.bazel
@@ -276,6 +276,13 @@
)
py_test(
+ name = "field_mask_test",
+ srcs = ["google/protobuf/internal/field_mask_test.py"],
+ imports = ["."],
+ deps = [":python_test_lib"],
+)
+
+py_test(
name = "generator_test",
srcs = ["google/protobuf/internal/generator_test.py"],
imports = ["."],
@@ -453,12 +460,10 @@
"google/protobuf/pyext/README",
"google/protobuf/python_protobuf.h",
"internal.bzl",
- "mox.py",
"python_version.py",
"release.sh",
"setup.cfg",
"setup.py",
- "stubout.py",
"tox.ini",
],
strip_prefix = strip_prefix.from_root(""),
diff --git a/python/google/protobuf/internal/field_mask.py b/python/google/protobuf/internal/field_mask.py
new file mode 100644
index 0000000..4897699
--- /dev/null
+++ b/python/google/protobuf/internal/field_mask.py
@@ -0,0 +1,333 @@
+# Protocol Buffers - Google's data interchange format
+# Copyright 2008 Google Inc. All rights reserved.
+# https://developers.google.com/protocol-buffers/
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+# * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+# * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+# * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+"""Contains FieldMask class."""
+
+from google.protobuf.descriptor import FieldDescriptor
+
+
+class FieldMask(object):
+ """Class for FieldMask message type."""
+
+ __slots__ = ()
+
+ def ToJsonString(self):
+ """Converts FieldMask to string according to proto3 JSON spec."""
+ camelcase_paths = []
+ for path in self.paths:
+ camelcase_paths.append(_SnakeCaseToCamelCase(path))
+ return ','.join(camelcase_paths)
+
+ def FromJsonString(self, value):
+ """Converts string to FieldMask according to proto3 JSON spec."""
+ if not isinstance(value, str):
+ raise ValueError('FieldMask JSON value not a string: {!r}'.format(value))
+ self.Clear()
+ if value:
+ for path in value.split(','):
+ self.paths.append(_CamelCaseToSnakeCase(path))
+
+ def IsValidForDescriptor(self, message_descriptor):
+ """Checks whether the FieldMask is valid for Message Descriptor."""
+ for path in self.paths:
+ if not _IsValidPath(message_descriptor, path):
+ return False
+ return True
+
+ def AllFieldsFromDescriptor(self, message_descriptor):
+ """Gets all direct fields of Message Descriptor to FieldMask."""
+ self.Clear()
+ for field in message_descriptor.fields:
+ self.paths.append(field.name)
+
+ def CanonicalFormFromMask(self, mask):
+ """Converts a FieldMask to the canonical form.
+
+ Removes paths that are covered by another path. For example,
+ "foo.bar" is covered by "foo" and will be removed if "foo"
+ is also in the FieldMask. Then sorts all paths in alphabetical order.
+
+ Args:
+ mask: The original FieldMask to be converted.
+ """
+ tree = _FieldMaskTree(mask)
+ tree.ToFieldMask(self)
+
+ def Union(self, mask1, mask2):
+ """Merges mask1 and mask2 into this FieldMask."""
+ _CheckFieldMaskMessage(mask1)
+ _CheckFieldMaskMessage(mask2)
+ tree = _FieldMaskTree(mask1)
+ tree.MergeFromFieldMask(mask2)
+ tree.ToFieldMask(self)
+
+ def Intersect(self, mask1, mask2):
+ """Intersects mask1 and mask2 into this FieldMask."""
+ _CheckFieldMaskMessage(mask1)
+ _CheckFieldMaskMessage(mask2)
+ tree = _FieldMaskTree(mask1)
+ intersection = _FieldMaskTree()
+ for path in mask2.paths:
+ tree.IntersectPath(path, intersection)
+ intersection.ToFieldMask(self)
+
+ def MergeMessage(
+ self, source, destination,
+ replace_message_field=False, replace_repeated_field=False):
+ """Merges fields specified in FieldMask from source to destination.
+
+ Args:
+ source: Source message.
+ destination: The destination message to be merged into.
+ replace_message_field: Replace message field if True. Merge message
+ field if False.
+ replace_repeated_field: Replace repeated field if True. Append
+ elements of repeated field if False.
+ """
+ tree = _FieldMaskTree(self)
+ tree.MergeMessage(
+ source, destination, replace_message_field, replace_repeated_field)
+
+
+def _IsValidPath(message_descriptor, path):
+ """Checks whether the path is valid for Message Descriptor."""
+ parts = path.split('.')
+ last = parts.pop()
+ for name in parts:
+ field = message_descriptor.fields_by_name.get(name)
+ if (field is None or
+ field.label == FieldDescriptor.LABEL_REPEATED or
+ field.type != FieldDescriptor.TYPE_MESSAGE):
+ return False
+ message_descriptor = field.message_type
+ return last in message_descriptor.fields_by_name
+
+
+def _CheckFieldMaskMessage(message):
+ """Raises ValueError if message is not a FieldMask."""
+ message_descriptor = message.DESCRIPTOR
+ if (message_descriptor.name != 'FieldMask' or
+ message_descriptor.file.name != 'google/protobuf/field_mask.proto'):
+ raise ValueError('Message {0} is not a FieldMask.'.format(
+ message_descriptor.full_name))
+
+
+def _SnakeCaseToCamelCase(path_name):
+ """Converts a path name from snake_case to camelCase."""
+ result = []
+ after_underscore = False
+ for c in path_name:
+ if c.isupper():
+ raise ValueError(
+ 'Fail to print FieldMask to Json string: Path name '
+ '{0} must not contain uppercase letters.'.format(path_name))
+ if after_underscore:
+ if c.islower():
+ result.append(c.upper())
+ after_underscore = False
+ else:
+ raise ValueError(
+ 'Fail to print FieldMask to Json string: The '
+ 'character after a "_" must be a lowercase letter '
+ 'in path name {0}.'.format(path_name))
+ elif c == '_':
+ after_underscore = True
+ else:
+ result += c
+
+ if after_underscore:
+ raise ValueError('Fail to print FieldMask to Json string: Trailing "_" '
+ 'in path name {0}.'.format(path_name))
+ return ''.join(result)
+
+
+def _CamelCaseToSnakeCase(path_name):
+ """Converts a field name from camelCase to snake_case."""
+ result = []
+ for c in path_name:
+ if c == '_':
+ raise ValueError('Fail to parse FieldMask: Path name '
+ '{0} must not contain "_"s.'.format(path_name))
+ if c.isupper():
+ result += '_'
+ result += c.lower()
+ else:
+ result += c
+ return ''.join(result)
+
+
+class _FieldMaskTree(object):
+ """Represents a FieldMask in a tree structure.
+
+ For example, given a FieldMask "foo.bar,foo.baz,bar.baz",
+ the FieldMaskTree will be:
+ [_root] -+- foo -+- bar
+ | |
+ | +- baz
+ |
+ +- bar --- baz
+ In the tree, each leaf node represents a field path.
+ """
+
+ __slots__ = ('_root',)
+
+ def __init__(self, field_mask=None):
+ """Initializes the tree by FieldMask."""
+ self._root = {}
+ if field_mask:
+ self.MergeFromFieldMask(field_mask)
+
+ def MergeFromFieldMask(self, field_mask):
+ """Merges a FieldMask to the tree."""
+ for path in field_mask.paths:
+ self.AddPath(path)
+
+ def AddPath(self, path):
+ """Adds a field path into the tree.
+
+ If the field path to add is a sub-path of an existing field path
+ in the tree (i.e., a leaf node), it means the tree already matches
+ the given path so nothing will be added to the tree. If the path
+ matches an existing non-leaf node in the tree, that non-leaf node
+ will be turned into a leaf node with all its children removed because
+ the path matches all the node's children. Otherwise, a new path will
+ be added.
+
+ Args:
+ path: The field path to add.
+ """
+ node = self._root
+ for name in path.split('.'):
+ if name not in node:
+ node[name] = {}
+ elif not node[name]:
+ # Pre-existing empty node implies we already have this entire tree.
+ return
+ node = node[name]
+ # Remove any sub-trees we might have had.
+ node.clear()
+
+ def ToFieldMask(self, field_mask):
+ """Converts the tree to a FieldMask."""
+ field_mask.Clear()
+ _AddFieldPaths(self._root, '', field_mask)
+
+ def IntersectPath(self, path, intersection):
+ """Calculates the intersection part of a field path with this tree.
+
+ Args:
+ path: The field path to calculates.
+ intersection: The out tree to record the intersection part.
+ """
+ node = self._root
+ for name in path.split('.'):
+ if name not in node:
+ return
+ elif not node[name]:
+ intersection.AddPath(path)
+ return
+ node = node[name]
+ intersection.AddLeafNodes(path, node)
+
+ def AddLeafNodes(self, prefix, node):
+ """Adds leaf nodes begin with prefix to this tree."""
+ if not node:
+ self.AddPath(prefix)
+ for name in node:
+ child_path = prefix + '.' + name
+ self.AddLeafNodes(child_path, node[name])
+
+ def MergeMessage(
+ self, source, destination,
+ replace_message, replace_repeated):
+ """Merge all fields specified by this tree from source to destination."""
+ _MergeMessage(
+ self._root, source, destination, replace_message, replace_repeated)
+
+
+def _StrConvert(value):
+ """Converts value to str if it is not."""
+ # This file is imported by c extension and some methods like ClearField
+ # requires string for the field name. py2/py3 has different text
+ # type and may use unicode.
+ if not isinstance(value, str):
+ return value.encode('utf-8')
+ return value
+
+
+def _MergeMessage(
+ node, source, destination, replace_message, replace_repeated):
+ """Merge all fields specified by a sub-tree from source to destination."""
+ source_descriptor = source.DESCRIPTOR
+ for name in node:
+ child = node[name]
+ field = source_descriptor.fields_by_name[name]
+ if field is None:
+ raise ValueError('Error: Can\'t find field {0} in message {1}.'.format(
+ name, source_descriptor.full_name))
+ if child:
+ # Sub-paths are only allowed for singular message fields.
+ if (field.label == FieldDescriptor.LABEL_REPEATED or
+ field.cpp_type != FieldDescriptor.CPPTYPE_MESSAGE):
+ raise ValueError('Error: Field {0} in message {1} is not a singular '
+ 'message field and cannot have sub-fields.'.format(
+ name, source_descriptor.full_name))
+ if source.HasField(name):
+ _MergeMessage(
+ child, getattr(source, name), getattr(destination, name),
+ replace_message, replace_repeated)
+ continue
+ if field.label == FieldDescriptor.LABEL_REPEATED:
+ if replace_repeated:
+ destination.ClearField(_StrConvert(name))
+ repeated_source = getattr(source, name)
+ repeated_destination = getattr(destination, name)
+ repeated_destination.MergeFrom(repeated_source)
+ else:
+ if field.cpp_type == FieldDescriptor.CPPTYPE_MESSAGE:
+ if replace_message:
+ destination.ClearField(_StrConvert(name))
+ if source.HasField(name):
+ getattr(destination, name).MergeFrom(getattr(source, name))
+ else:
+ setattr(destination, name, getattr(source, name))
+
+
+def _AddFieldPaths(node, prefix, field_mask):
+ """Adds the field paths descended from node to field_mask."""
+ if not node and prefix:
+ field_mask.paths.append(prefix)
+ return
+ for name in sorted(node):
+ if prefix:
+ child_path = prefix + '.' + name
+ else:
+ child_path = name
+ _AddFieldPaths(node[name], child_path, field_mask)
diff --git a/python/google/protobuf/internal/field_mask_test.py b/python/google/protobuf/internal/field_mask_test.py
new file mode 100644
index 0000000..8266418
--- /dev/null
+++ b/python/google/protobuf/internal/field_mask_test.py
@@ -0,0 +1,400 @@
+# Protocol Buffers - Google's data interchange format
+# Copyright 2008 Google Inc. All rights reserved.
+# https://developers.google.com/protocol-buffers/
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+# * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+# * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+# * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+"""Test for google.protobuf.internal.well_known_types."""
+
+import unittest
+
+from google.protobuf import field_mask_pb2
+from google.protobuf import map_unittest_pb2
+from google.protobuf import unittest_pb2
+from google.protobuf.internal import field_mask
+from google.protobuf.internal import test_util
+from google.protobuf import descriptor
+
+
+class FieldMaskTest(unittest.TestCase):
+
+ def testStringFormat(self):
+ mask = field_mask_pb2.FieldMask()
+ self.assertEqual('', mask.ToJsonString())
+ mask.paths.append('foo')
+ self.assertEqual('foo', mask.ToJsonString())
+ mask.paths.append('bar')
+ self.assertEqual('foo,bar', mask.ToJsonString())
+
+ mask.FromJsonString('')
+ self.assertEqual('', mask.ToJsonString())
+ mask.FromJsonString('foo')
+ self.assertEqual(['foo'], mask.paths)
+ mask.FromJsonString('foo,bar')
+ self.assertEqual(['foo', 'bar'], mask.paths)
+
+ # Test camel case
+ mask.Clear()
+ mask.paths.append('foo_bar')
+ self.assertEqual('fooBar', mask.ToJsonString())
+ mask.paths.append('bar_quz')
+ self.assertEqual('fooBar,barQuz', mask.ToJsonString())
+
+ mask.FromJsonString('')
+ self.assertEqual('', mask.ToJsonString())
+ self.assertEqual([], mask.paths)
+ mask.FromJsonString('fooBar')
+ self.assertEqual(['foo_bar'], mask.paths)
+ mask.FromJsonString('fooBar,barQuz')
+ self.assertEqual(['foo_bar', 'bar_quz'], mask.paths)
+
+ def testDescriptorToFieldMask(self):
+ mask = field_mask_pb2.FieldMask()
+ msg_descriptor = unittest_pb2.TestAllTypes.DESCRIPTOR
+ mask.AllFieldsFromDescriptor(msg_descriptor)
+ self.assertEqual(76, len(mask.paths))
+ self.assertTrue(mask.IsValidForDescriptor(msg_descriptor))
+ for field in msg_descriptor.fields:
+ self.assertTrue(field.name in mask.paths)
+
+ def testIsValidForDescriptor(self):
+ msg_descriptor = unittest_pb2.TestAllTypes.DESCRIPTOR
+ # Empty mask
+ mask = field_mask_pb2.FieldMask()
+ self.assertTrue(mask.IsValidForDescriptor(msg_descriptor))
+ # All fields from descriptor
+ mask.AllFieldsFromDescriptor(msg_descriptor)
+ self.assertTrue(mask.IsValidForDescriptor(msg_descriptor))
+ # Child under optional message
+ mask.paths.append('optional_nested_message.bb')
+ self.assertTrue(mask.IsValidForDescriptor(msg_descriptor))
+ # Repeated field is only allowed in the last position of path
+ mask.paths.append('repeated_nested_message.bb')
+ self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
+ # Invalid top level field
+ mask = field_mask_pb2.FieldMask()
+ mask.paths.append('xxx')
+ self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
+ # Invalid field in root
+ mask = field_mask_pb2.FieldMask()
+ mask.paths.append('xxx.zzz')
+ self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
+ # Invalid field in internal node
+ mask = field_mask_pb2.FieldMask()
+ mask.paths.append('optional_nested_message.xxx.zzz')
+ self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
+ # Invalid field in leaf
+ mask = field_mask_pb2.FieldMask()
+ mask.paths.append('optional_nested_message.xxx')
+ self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
+
+ def testCanonicalFrom(self):
+ mask = field_mask_pb2.FieldMask()
+ out_mask = field_mask_pb2.FieldMask()
+ # Paths will be sorted.
+ mask.FromJsonString('baz.quz,bar,foo')
+ out_mask.CanonicalFormFromMask(mask)
+ self.assertEqual('bar,baz.quz,foo', out_mask.ToJsonString())
+ # Duplicated paths will be removed.
+ mask.FromJsonString('foo,bar,foo')
+ out_mask.CanonicalFormFromMask(mask)
+ self.assertEqual('bar,foo', out_mask.ToJsonString())
+ # Sub-paths of other paths will be removed.
+ mask.FromJsonString('foo.b1,bar.b1,foo.b2,bar')
+ out_mask.CanonicalFormFromMask(mask)
+ self.assertEqual('bar,foo.b1,foo.b2', out_mask.ToJsonString())
+
+ # Test more deeply nested cases.
+ mask.FromJsonString(
+ 'foo.bar.baz1,foo.bar.baz2.quz,foo.bar.baz2')
+ out_mask.CanonicalFormFromMask(mask)
+ self.assertEqual('foo.bar.baz1,foo.bar.baz2',
+ out_mask.ToJsonString())
+ mask.FromJsonString(
+ 'foo.bar.baz1,foo.bar.baz2,foo.bar.baz2.quz')
+ out_mask.CanonicalFormFromMask(mask)
+ self.assertEqual('foo.bar.baz1,foo.bar.baz2',
+ out_mask.ToJsonString())
+ mask.FromJsonString(
+ 'foo.bar.baz1,foo.bar.baz2,foo.bar.baz2.quz,foo.bar')
+ out_mask.CanonicalFormFromMask(mask)
+ self.assertEqual('foo.bar', out_mask.ToJsonString())
+ mask.FromJsonString(
+ 'foo.bar.baz1,foo.bar.baz2,foo.bar.baz2.quz,foo')
+ out_mask.CanonicalFormFromMask(mask)
+ self.assertEqual('foo', out_mask.ToJsonString())
+
+ def testUnion(self):
+ mask1 = field_mask_pb2.FieldMask()
+ mask2 = field_mask_pb2.FieldMask()
+ out_mask = field_mask_pb2.FieldMask()
+ mask1.FromJsonString('foo,baz')
+ mask2.FromJsonString('bar,quz')
+ out_mask.Union(mask1, mask2)
+ self.assertEqual('bar,baz,foo,quz', out_mask.ToJsonString())
+ # Overlap with duplicated paths.
+ mask1.FromJsonString('foo,baz.bb')
+ mask2.FromJsonString('baz.bb,quz')
+ out_mask.Union(mask1, mask2)
+ self.assertEqual('baz.bb,foo,quz', out_mask.ToJsonString())
+ # Overlap with paths covering some other paths.
+ mask1.FromJsonString('foo.bar.baz,quz')
+ mask2.FromJsonString('foo.bar,bar')
+ out_mask.Union(mask1, mask2)
+ self.assertEqual('bar,foo.bar,quz', out_mask.ToJsonString())
+ src = unittest_pb2.TestAllTypes()
+ with self.assertRaises(ValueError):
+ out_mask.Union(src, mask2)
+
+ def testIntersect(self):
+ mask1 = field_mask_pb2.FieldMask()
+ mask2 = field_mask_pb2.FieldMask()
+ out_mask = field_mask_pb2.FieldMask()
+ # Test cases without overlapping.
+ mask1.FromJsonString('foo,baz')
+ mask2.FromJsonString('bar,quz')
+ out_mask.Intersect(mask1, mask2)
+ self.assertEqual('', out_mask.ToJsonString())
+ self.assertEqual(len(out_mask.paths), 0)
+ self.assertEqual(out_mask.paths, [])
+ # Overlap with duplicated paths.
+ mask1.FromJsonString('foo,baz.bb')
+ mask2.FromJsonString('baz.bb,quz')
+ out_mask.Intersect(mask1, mask2)
+ self.assertEqual('baz.bb', out_mask.ToJsonString())
+ # Overlap with paths covering some other paths.
+ mask1.FromJsonString('foo.bar.baz,quz')
+ mask2.FromJsonString('foo.bar,bar')
+ out_mask.Intersect(mask1, mask2)
+ self.assertEqual('foo.bar.baz', out_mask.ToJsonString())
+ mask1.FromJsonString('foo.bar,bar')
+ mask2.FromJsonString('foo.bar.baz,quz')
+ out_mask.Intersect(mask1, mask2)
+ self.assertEqual('foo.bar.baz', out_mask.ToJsonString())
+ # Intersect '' with ''
+ mask1.Clear()
+ mask2.Clear()
+ mask1.paths.append('')
+ mask2.paths.append('')
+ self.assertEqual(mask1.paths, [''])
+ self.assertEqual('', mask1.ToJsonString())
+ out_mask.Intersect(mask1, mask2)
+ self.assertEqual(out_mask.paths, [])
+
+ def testMergeMessageWithoutMapFields(self):
+ # Test merge one field.
+ src = unittest_pb2.TestAllTypes()
+ test_util.SetAllFields(src)
+ for field in src.DESCRIPTOR.fields:
+ if field.containing_oneof:
+ continue
+ field_name = field.name
+ dst = unittest_pb2.TestAllTypes()
+ # Only set one path to mask.
+ mask = field_mask_pb2.FieldMask()
+ mask.paths.append(field_name)
+ mask.MergeMessage(src, dst)
+ # The expected result message.
+ msg = unittest_pb2.TestAllTypes()
+ if field.label == descriptor.FieldDescriptor.LABEL_REPEATED:
+ repeated_src = getattr(src, field_name)
+ repeated_msg = getattr(msg, field_name)
+ if field.cpp_type == descriptor.FieldDescriptor.CPPTYPE_MESSAGE:
+ for item in repeated_src:
+ repeated_msg.add().CopyFrom(item)
+ else:
+ repeated_msg.extend(repeated_src)
+ elif field.cpp_type == descriptor.FieldDescriptor.CPPTYPE_MESSAGE:
+ getattr(msg, field_name).CopyFrom(getattr(src, field_name))
+ else:
+ setattr(msg, field_name, getattr(src, field_name))
+ # Only field specified in mask is merged.
+ self.assertEqual(msg, dst)
+
+ # Test merge nested fields.
+ nested_src = unittest_pb2.NestedTestAllTypes()
+ nested_dst = unittest_pb2.NestedTestAllTypes()
+ nested_src.child.payload.optional_int32 = 1234
+ nested_src.child.child.payload.optional_int32 = 5678
+ mask = field_mask_pb2.FieldMask()
+ mask.FromJsonString('child.payload')
+ mask.MergeMessage(nested_src, nested_dst)
+ self.assertEqual(1234, nested_dst.child.payload.optional_int32)
+ self.assertEqual(0, nested_dst.child.child.payload.optional_int32)
+
+ mask.FromJsonString('child.child.payload')
+ mask.MergeMessage(nested_src, nested_dst)
+ self.assertEqual(1234, nested_dst.child.payload.optional_int32)
+ self.assertEqual(5678, nested_dst.child.child.payload.optional_int32)
+
+ nested_dst.Clear()
+ mask.FromJsonString('child.child.payload')
+ mask.MergeMessage(nested_src, nested_dst)
+ self.assertEqual(0, nested_dst.child.payload.optional_int32)
+ self.assertEqual(5678, nested_dst.child.child.payload.optional_int32)
+
+ nested_dst.Clear()
+ mask.FromJsonString('child')
+ mask.MergeMessage(nested_src, nested_dst)
+ self.assertEqual(1234, nested_dst.child.payload.optional_int32)
+ self.assertEqual(5678, nested_dst.child.child.payload.optional_int32)
+
+ # Test MergeOptions.
+ nested_dst.Clear()
+ nested_dst.child.payload.optional_int64 = 4321
+ # Message fields will be merged by default.
+ mask.FromJsonString('child.payload')
+ mask.MergeMessage(nested_src, nested_dst)
+ self.assertEqual(1234, nested_dst.child.payload.optional_int32)
+ self.assertEqual(4321, nested_dst.child.payload.optional_int64)
+ # Change the behavior to replace message fields.
+ mask.FromJsonString('child.payload')
+ mask.MergeMessage(nested_src, nested_dst, True, False)
+ self.assertEqual(1234, nested_dst.child.payload.optional_int32)
+ self.assertEqual(0, nested_dst.child.payload.optional_int64)
+
+ # By default, fields missing in source are not cleared in destination.
+ nested_dst.payload.optional_int32 = 1234
+ self.assertTrue(nested_dst.HasField('payload'))
+ mask.FromJsonString('payload')
+ mask.MergeMessage(nested_src, nested_dst)
+ self.assertTrue(nested_dst.HasField('payload'))
+ # But they are cleared when replacing message fields.
+ nested_dst.Clear()
+ nested_dst.payload.optional_int32 = 1234
+ mask.FromJsonString('payload')
+ mask.MergeMessage(nested_src, nested_dst, True, False)
+ self.assertFalse(nested_dst.HasField('payload'))
+
+ nested_src.payload.repeated_int32.append(1234)
+ nested_dst.payload.repeated_int32.append(5678)
+ # Repeated fields will be appended by default.
+ mask.FromJsonString('payload.repeatedInt32')
+ mask.MergeMessage(nested_src, nested_dst)
+ self.assertEqual(2, len(nested_dst.payload.repeated_int32))
+ self.assertEqual(5678, nested_dst.payload.repeated_int32[0])
+ self.assertEqual(1234, nested_dst.payload.repeated_int32[1])
+ # Change the behavior to replace repeated fields.
+ mask.FromJsonString('payload.repeatedInt32')
+ mask.MergeMessage(nested_src, nested_dst, False, True)
+ self.assertEqual(1, len(nested_dst.payload.repeated_int32))
+ self.assertEqual(1234, nested_dst.payload.repeated_int32[0])
+
+ # Test Merge oneof field.
+ new_msg = unittest_pb2.TestOneof2()
+ dst = unittest_pb2.TestOneof2()
+ dst.foo_message.moo_int = 1
+ mask = field_mask_pb2.FieldMask()
+ mask.FromJsonString('fooMessage,fooLazyMessage.mooInt')
+ mask.MergeMessage(new_msg, dst)
+ self.assertTrue(dst.HasField('foo_message'))
+ self.assertFalse(dst.HasField('foo_lazy_message'))
+
+ def testMergeMessageWithMapField(self):
+ empty_map = map_unittest_pb2.TestRecursiveMapMessage()
+ src_level_2 = map_unittest_pb2.TestRecursiveMapMessage()
+ src_level_2.a['src level 2'].CopyFrom(empty_map)
+ src = map_unittest_pb2.TestRecursiveMapMessage()
+ src.a['common key'].CopyFrom(src_level_2)
+ src.a['src level 1'].CopyFrom(src_level_2)
+
+ dst_level_2 = map_unittest_pb2.TestRecursiveMapMessage()
+ dst_level_2.a['dst level 2'].CopyFrom(empty_map)
+ dst = map_unittest_pb2.TestRecursiveMapMessage()
+ dst.a['common key'].CopyFrom(dst_level_2)
+ dst.a['dst level 1'].CopyFrom(empty_map)
+
+ mask = field_mask_pb2.FieldMask()
+ mask.FromJsonString('a')
+ mask.MergeMessage(src, dst)
+
+ # map from dst is replaced with map from src.
+ self.assertEqual(dst.a['common key'], src_level_2)
+ self.assertEqual(dst.a['src level 1'], src_level_2)
+ self.assertEqual(dst.a['dst level 1'], empty_map)
+
+ def testMergeErrors(self):
+ src = unittest_pb2.TestAllTypes()
+ dst = unittest_pb2.TestAllTypes()
+ mask = field_mask_pb2.FieldMask()
+ test_util.SetAllFields(src)
+ mask.FromJsonString('optionalInt32.field')
+ with self.assertRaises(ValueError) as e:
+ mask.MergeMessage(src, dst)
+ self.assertEqual('Error: Field optional_int32 in message '
+ 'protobuf_unittest.TestAllTypes is not a singular '
+ 'message field and cannot have sub-fields.',
+ str(e.exception))
+
+ def testSnakeCaseToCamelCase(self):
+ self.assertEqual('fooBar',
+ field_mask._SnakeCaseToCamelCase('foo_bar'))
+ self.assertEqual('FooBar',
+ field_mask._SnakeCaseToCamelCase('_foo_bar'))
+ self.assertEqual('foo3Bar',
+ field_mask._SnakeCaseToCamelCase('foo3_bar'))
+
+ # No uppercase letter is allowed.
+ self.assertRaisesRegex(
+ ValueError,
+ 'Fail to print FieldMask to Json string: Path name Foo must '
+ 'not contain uppercase letters.',
+ field_mask._SnakeCaseToCamelCase, 'Foo')
+ # Any character after a "_" must be a lowercase letter.
+ # 1. "_" cannot be followed by another "_".
+ # 2. "_" cannot be followed by a digit.
+ # 3. "_" cannot appear as the last character.
+ self.assertRaisesRegex(
+ ValueError,
+ 'Fail to print FieldMask to Json string: The character after a '
+ '"_" must be a lowercase letter in path name foo__bar.',
+ field_mask._SnakeCaseToCamelCase, 'foo__bar')
+ self.assertRaisesRegex(
+ ValueError,
+ 'Fail to print FieldMask to Json string: The character after a '
+ '"_" must be a lowercase letter in path name foo_3bar.',
+ field_mask._SnakeCaseToCamelCase, 'foo_3bar')
+ self.assertRaisesRegex(
+ ValueError,
+ 'Fail to print FieldMask to Json string: Trailing "_" in path '
+ 'name foo_bar_.', field_mask._SnakeCaseToCamelCase, 'foo_bar_')
+
+ def testCamelCaseToSnakeCase(self):
+ self.assertEqual('foo_bar',
+ field_mask._CamelCaseToSnakeCase('fooBar'))
+ self.assertEqual('_foo_bar',
+ field_mask._CamelCaseToSnakeCase('FooBar'))
+ self.assertEqual('foo3_bar',
+ field_mask._CamelCaseToSnakeCase('foo3Bar'))
+ self.assertRaisesRegex(
+ ValueError,
+ 'Fail to parse FieldMask: Path name foo_bar must not contain "_"s.',
+ field_mask._CamelCaseToSnakeCase, 'foo_bar')
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/python/google/protobuf/internal/missing_enum_values.proto b/python/google/protobuf/internal/missing_enum_values.proto
index 37baca7..96de1e4 100644
--- a/python/google/protobuf/internal/missing_enum_values.proto
+++ b/python/google/protobuf/internal/missing_enum_values.proto
@@ -30,7 +30,6 @@
syntax = "proto2";
-
package google.protobuf.python.internal;
message TestEnumValues {
@@ -53,4 +52,3 @@
message JustString {
required string dummy = 1;
}
-
diff --git a/python/google/protobuf/internal/well_known_types.py b/python/google/protobuf/internal/well_known_types.py
index 8881d75..e340f90 100644
--- a/python/google/protobuf/internal/well_known_types.py
+++ b/python/google/protobuf/internal/well_known_types.py
@@ -44,7 +44,9 @@
import collections.abc
import datetime
-from google.protobuf.descriptor import FieldDescriptor
+from google.protobuf.internal import field_mask
+
+FieldMask = field_mask.FieldMask
_TIMESTAMPFOMAT = '%Y-%m-%dT%H:%M:%S'
_NANOS_PER_SECOND = 1000000000
@@ -430,306 +432,6 @@
return result
-class FieldMask(object):
- """Class for FieldMask message type."""
-
- __slots__ = ()
-
- def ToJsonString(self):
- """Converts FieldMask to string according to proto3 JSON spec."""
- camelcase_paths = []
- for path in self.paths:
- camelcase_paths.append(_SnakeCaseToCamelCase(path))
- return ','.join(camelcase_paths)
-
- def FromJsonString(self, value):
- """Converts string to FieldMask according to proto3 JSON spec."""
- if not isinstance(value, str):
- raise ValueError('FieldMask JSON value not a string: {!r}'.format(value))
- self.Clear()
- if value:
- for path in value.split(','):
- self.paths.append(_CamelCaseToSnakeCase(path))
-
- def IsValidForDescriptor(self, message_descriptor):
- """Checks whether the FieldMask is valid for Message Descriptor."""
- for path in self.paths:
- if not _IsValidPath(message_descriptor, path):
- return False
- return True
-
- def AllFieldsFromDescriptor(self, message_descriptor):
- """Gets all direct fields of Message Descriptor to FieldMask."""
- self.Clear()
- for field in message_descriptor.fields:
- self.paths.append(field.name)
-
- def CanonicalFormFromMask(self, mask):
- """Converts a FieldMask to the canonical form.
-
- Removes paths that are covered by another path. For example,
- "foo.bar" is covered by "foo" and will be removed if "foo"
- is also in the FieldMask. Then sorts all paths in alphabetical order.
-
- Args:
- mask: The original FieldMask to be converted.
- """
- tree = _FieldMaskTree(mask)
- tree.ToFieldMask(self)
-
- def Union(self, mask1, mask2):
- """Merges mask1 and mask2 into this FieldMask."""
- _CheckFieldMaskMessage(mask1)
- _CheckFieldMaskMessage(mask2)
- tree = _FieldMaskTree(mask1)
- tree.MergeFromFieldMask(mask2)
- tree.ToFieldMask(self)
-
- def Intersect(self, mask1, mask2):
- """Intersects mask1 and mask2 into this FieldMask."""
- _CheckFieldMaskMessage(mask1)
- _CheckFieldMaskMessage(mask2)
- tree = _FieldMaskTree(mask1)
- intersection = _FieldMaskTree()
- for path in mask2.paths:
- tree.IntersectPath(path, intersection)
- intersection.ToFieldMask(self)
-
- def MergeMessage(
- self, source, destination,
- replace_message_field=False, replace_repeated_field=False):
- """Merges fields specified in FieldMask from source to destination.
-
- Args:
- source: Source message.
- destination: The destination message to be merged into.
- replace_message_field: Replace message field if True. Merge message
- field if False.
- replace_repeated_field: Replace repeated field if True. Append
- elements of repeated field if False.
- """
- tree = _FieldMaskTree(self)
- tree.MergeMessage(
- source, destination, replace_message_field, replace_repeated_field)
-
-
-def _IsValidPath(message_descriptor, path):
- """Checks whether the path is valid for Message Descriptor."""
- parts = path.split('.')
- last = parts.pop()
- for name in parts:
- field = message_descriptor.fields_by_name.get(name)
- if (field is None or
- field.label == FieldDescriptor.LABEL_REPEATED or
- field.type != FieldDescriptor.TYPE_MESSAGE):
- return False
- message_descriptor = field.message_type
- return last in message_descriptor.fields_by_name
-
-
-def _CheckFieldMaskMessage(message):
- """Raises ValueError if message is not a FieldMask."""
- message_descriptor = message.DESCRIPTOR
- if (message_descriptor.name != 'FieldMask' or
- message_descriptor.file.name != 'google/protobuf/field_mask.proto'):
- raise ValueError('Message {0} is not a FieldMask.'.format(
- message_descriptor.full_name))
-
-
-def _SnakeCaseToCamelCase(path_name):
- """Converts a path name from snake_case to camelCase."""
- result = []
- after_underscore = False
- for c in path_name:
- if c.isupper():
- raise ValueError(
- 'Fail to print FieldMask to Json string: Path name '
- '{0} must not contain uppercase letters.'.format(path_name))
- if after_underscore:
- if c.islower():
- result.append(c.upper())
- after_underscore = False
- else:
- raise ValueError(
- 'Fail to print FieldMask to Json string: The '
- 'character after a "_" must be a lowercase letter '
- 'in path name {0}.'.format(path_name))
- elif c == '_':
- after_underscore = True
- else:
- result += c
-
- if after_underscore:
- raise ValueError('Fail to print FieldMask to Json string: Trailing "_" '
- 'in path name {0}.'.format(path_name))
- return ''.join(result)
-
-
-def _CamelCaseToSnakeCase(path_name):
- """Converts a field name from camelCase to snake_case."""
- result = []
- for c in path_name:
- if c == '_':
- raise ValueError('Fail to parse FieldMask: Path name '
- '{0} must not contain "_"s.'.format(path_name))
- if c.isupper():
- result += '_'
- result += c.lower()
- else:
- result += c
- return ''.join(result)
-
-
-class _FieldMaskTree(object):
- """Represents a FieldMask in a tree structure.
-
- For example, given a FieldMask "foo.bar,foo.baz,bar.baz",
- the FieldMaskTree will be:
- [_root] -+- foo -+- bar
- | |
- | +- baz
- |
- +- bar --- baz
- In the tree, each leaf node represents a field path.
- """
-
- __slots__ = ('_root',)
-
- def __init__(self, field_mask=None):
- """Initializes the tree by FieldMask."""
- self._root = {}
- if field_mask:
- self.MergeFromFieldMask(field_mask)
-
- def MergeFromFieldMask(self, field_mask):
- """Merges a FieldMask to the tree."""
- for path in field_mask.paths:
- self.AddPath(path)
-
- def AddPath(self, path):
- """Adds a field path into the tree.
-
- If the field path to add is a sub-path of an existing field path
- in the tree (i.e., a leaf node), it means the tree already matches
- the given path so nothing will be added to the tree. If the path
- matches an existing non-leaf node in the tree, that non-leaf node
- will be turned into a leaf node with all its children removed because
- the path matches all the node's children. Otherwise, a new path will
- be added.
-
- Args:
- path: The field path to add.
- """
- node = self._root
- for name in path.split('.'):
- if name not in node:
- node[name] = {}
- elif not node[name]:
- # Pre-existing empty node implies we already have this entire tree.
- return
- node = node[name]
- # Remove any sub-trees we might have had.
- node.clear()
-
- def ToFieldMask(self, field_mask):
- """Converts the tree to a FieldMask."""
- field_mask.Clear()
- _AddFieldPaths(self._root, '', field_mask)
-
- def IntersectPath(self, path, intersection):
- """Calculates the intersection part of a field path with this tree.
-
- Args:
- path: The field path to calculates.
- intersection: The out tree to record the intersection part.
- """
- node = self._root
- for name in path.split('.'):
- if name not in node:
- return
- elif not node[name]:
- intersection.AddPath(path)
- return
- node = node[name]
- intersection.AddLeafNodes(path, node)
-
- def AddLeafNodes(self, prefix, node):
- """Adds leaf nodes begin with prefix to this tree."""
- if not node:
- self.AddPath(prefix)
- for name in node:
- child_path = prefix + '.' + name
- self.AddLeafNodes(child_path, node[name])
-
- def MergeMessage(
- self, source, destination,
- replace_message, replace_repeated):
- """Merge all fields specified by this tree from source to destination."""
- _MergeMessage(
- self._root, source, destination, replace_message, replace_repeated)
-
-
-def _StrConvert(value):
- """Converts value to str if it is not."""
- # This file is imported by c extension and some methods like ClearField
- # requires string for the field name. py2/py3 has different text
- # type and may use unicode.
- if not isinstance(value, str):
- return value.encode('utf-8')
- return value
-
-
-def _MergeMessage(
- node, source, destination, replace_message, replace_repeated):
- """Merge all fields specified by a sub-tree from source to destination."""
- source_descriptor = source.DESCRIPTOR
- for name in node:
- child = node[name]
- field = source_descriptor.fields_by_name[name]
- if field is None:
- raise ValueError('Error: Can\'t find field {0} in message {1}.'.format(
- name, source_descriptor.full_name))
- if child:
- # Sub-paths are only allowed for singular message fields.
- if (field.label == FieldDescriptor.LABEL_REPEATED or
- field.cpp_type != FieldDescriptor.CPPTYPE_MESSAGE):
- raise ValueError('Error: Field {0} in message {1} is not a singular '
- 'message field and cannot have sub-fields.'.format(
- name, source_descriptor.full_name))
- if source.HasField(name):
- _MergeMessage(
- child, getattr(source, name), getattr(destination, name),
- replace_message, replace_repeated)
- continue
- if field.label == FieldDescriptor.LABEL_REPEATED:
- if replace_repeated:
- destination.ClearField(_StrConvert(name))
- repeated_source = getattr(source, name)
- repeated_destination = getattr(destination, name)
- repeated_destination.MergeFrom(repeated_source)
- else:
- if field.cpp_type == FieldDescriptor.CPPTYPE_MESSAGE:
- if replace_message:
- destination.ClearField(_StrConvert(name))
- if source.HasField(name):
- getattr(destination, name).MergeFrom(getattr(source, name))
- else:
- setattr(destination, name, getattr(source, name))
-
-
-def _AddFieldPaths(node, prefix, field_mask):
- """Adds the field paths descended from node to field_mask."""
- if not node and prefix:
- field_mask.paths.append(prefix)
- return
- for name in sorted(node):
- if prefix:
- child_path = prefix + '.' + name
- else:
- child_path = name
- _AddFieldPaths(node[name], child_path, field_mask)
-
-
def _SetStructValue(struct_value, value):
if value is None:
struct_value.null_value = 0
diff --git a/python/google/protobuf/internal/well_known_types_test.py b/python/google/protobuf/internal/well_known_types_test.py
index a32459a..601ae41 100644
--- a/python/google/protobuf/internal/well_known_types_test.py
+++ b/python/google/protobuf/internal/well_known_types_test.py
@@ -38,15 +38,11 @@
from google.protobuf import any_pb2
from google.protobuf import duration_pb2
-from google.protobuf import field_mask_pb2
from google.protobuf import struct_pb2
from google.protobuf import timestamp_pb2
-from google.protobuf import map_unittest_pb2
from google.protobuf import unittest_pb2
from google.protobuf.internal import any_test_pb2
-from google.protobuf.internal import test_util
from google.protobuf.internal import well_known_types
-from google.protobuf import descriptor
from google.protobuf import text_format
from google.protobuf.internal import _parameterized
@@ -390,362 +386,6 @@
message.ToJsonString)
-class FieldMaskTest(unittest.TestCase):
-
- def testStringFormat(self):
- mask = field_mask_pb2.FieldMask()
- self.assertEqual('', mask.ToJsonString())
- mask.paths.append('foo')
- self.assertEqual('foo', mask.ToJsonString())
- mask.paths.append('bar')
- self.assertEqual('foo,bar', mask.ToJsonString())
-
- mask.FromJsonString('')
- self.assertEqual('', mask.ToJsonString())
- mask.FromJsonString('foo')
- self.assertEqual(['foo'], mask.paths)
- mask.FromJsonString('foo,bar')
- self.assertEqual(['foo', 'bar'], mask.paths)
-
- # Test camel case
- mask.Clear()
- mask.paths.append('foo_bar')
- self.assertEqual('fooBar', mask.ToJsonString())
- mask.paths.append('bar_quz')
- self.assertEqual('fooBar,barQuz', mask.ToJsonString())
-
- mask.FromJsonString('')
- self.assertEqual('', mask.ToJsonString())
- self.assertEqual([], mask.paths)
- mask.FromJsonString('fooBar')
- self.assertEqual(['foo_bar'], mask.paths)
- mask.FromJsonString('fooBar,barQuz')
- self.assertEqual(['foo_bar', 'bar_quz'], mask.paths)
-
- def testDescriptorToFieldMask(self):
- mask = field_mask_pb2.FieldMask()
- msg_descriptor = unittest_pb2.TestAllTypes.DESCRIPTOR
- mask.AllFieldsFromDescriptor(msg_descriptor)
- self.assertEqual(76, len(mask.paths))
- self.assertTrue(mask.IsValidForDescriptor(msg_descriptor))
- for field in msg_descriptor.fields:
- self.assertTrue(field.name in mask.paths)
-
- def testIsValidForDescriptor(self):
- msg_descriptor = unittest_pb2.TestAllTypes.DESCRIPTOR
- # Empty mask
- mask = field_mask_pb2.FieldMask()
- self.assertTrue(mask.IsValidForDescriptor(msg_descriptor))
- # All fields from descriptor
- mask.AllFieldsFromDescriptor(msg_descriptor)
- self.assertTrue(mask.IsValidForDescriptor(msg_descriptor))
- # Child under optional message
- mask.paths.append('optional_nested_message.bb')
- self.assertTrue(mask.IsValidForDescriptor(msg_descriptor))
- # Repeated field is only allowed in the last position of path
- mask.paths.append('repeated_nested_message.bb')
- self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
- # Invalid top level field
- mask = field_mask_pb2.FieldMask()
- mask.paths.append('xxx')
- self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
- # Invalid field in root
- mask = field_mask_pb2.FieldMask()
- mask.paths.append('xxx.zzz')
- self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
- # Invalid field in internal node
- mask = field_mask_pb2.FieldMask()
- mask.paths.append('optional_nested_message.xxx.zzz')
- self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
- # Invalid field in leaf
- mask = field_mask_pb2.FieldMask()
- mask.paths.append('optional_nested_message.xxx')
- self.assertFalse(mask.IsValidForDescriptor(msg_descriptor))
-
- def testCanonicalFrom(self):
- mask = field_mask_pb2.FieldMask()
- out_mask = field_mask_pb2.FieldMask()
- # Paths will be sorted.
- mask.FromJsonString('baz.quz,bar,foo')
- out_mask.CanonicalFormFromMask(mask)
- self.assertEqual('bar,baz.quz,foo', out_mask.ToJsonString())
- # Duplicated paths will be removed.
- mask.FromJsonString('foo,bar,foo')
- out_mask.CanonicalFormFromMask(mask)
- self.assertEqual('bar,foo', out_mask.ToJsonString())
- # Sub-paths of other paths will be removed.
- mask.FromJsonString('foo.b1,bar.b1,foo.b2,bar')
- out_mask.CanonicalFormFromMask(mask)
- self.assertEqual('bar,foo.b1,foo.b2', out_mask.ToJsonString())
-
- # Test more deeply nested cases.
- mask.FromJsonString(
- 'foo.bar.baz1,foo.bar.baz2.quz,foo.bar.baz2')
- out_mask.CanonicalFormFromMask(mask)
- self.assertEqual('foo.bar.baz1,foo.bar.baz2',
- out_mask.ToJsonString())
- mask.FromJsonString(
- 'foo.bar.baz1,foo.bar.baz2,foo.bar.baz2.quz')
- out_mask.CanonicalFormFromMask(mask)
- self.assertEqual('foo.bar.baz1,foo.bar.baz2',
- out_mask.ToJsonString())
- mask.FromJsonString(
- 'foo.bar.baz1,foo.bar.baz2,foo.bar.baz2.quz,foo.bar')
- out_mask.CanonicalFormFromMask(mask)
- self.assertEqual('foo.bar', out_mask.ToJsonString())
- mask.FromJsonString(
- 'foo.bar.baz1,foo.bar.baz2,foo.bar.baz2.quz,foo')
- out_mask.CanonicalFormFromMask(mask)
- self.assertEqual('foo', out_mask.ToJsonString())
-
- def testUnion(self):
- mask1 = field_mask_pb2.FieldMask()
- mask2 = field_mask_pb2.FieldMask()
- out_mask = field_mask_pb2.FieldMask()
- mask1.FromJsonString('foo,baz')
- mask2.FromJsonString('bar,quz')
- out_mask.Union(mask1, mask2)
- self.assertEqual('bar,baz,foo,quz', out_mask.ToJsonString())
- # Overlap with duplicated paths.
- mask1.FromJsonString('foo,baz.bb')
- mask2.FromJsonString('baz.bb,quz')
- out_mask.Union(mask1, mask2)
- self.assertEqual('baz.bb,foo,quz', out_mask.ToJsonString())
- # Overlap with paths covering some other paths.
- mask1.FromJsonString('foo.bar.baz,quz')
- mask2.FromJsonString('foo.bar,bar')
- out_mask.Union(mask1, mask2)
- self.assertEqual('bar,foo.bar,quz', out_mask.ToJsonString())
- src = unittest_pb2.TestAllTypes()
- with self.assertRaises(ValueError):
- out_mask.Union(src, mask2)
-
- def testIntersect(self):
- mask1 = field_mask_pb2.FieldMask()
- mask2 = field_mask_pb2.FieldMask()
- out_mask = field_mask_pb2.FieldMask()
- # Test cases without overlapping.
- mask1.FromJsonString('foo,baz')
- mask2.FromJsonString('bar,quz')
- out_mask.Intersect(mask1, mask2)
- self.assertEqual('', out_mask.ToJsonString())
- self.assertEqual(len(out_mask.paths), 0)
- self.assertEqual(out_mask.paths, [])
- # Overlap with duplicated paths.
- mask1.FromJsonString('foo,baz.bb')
- mask2.FromJsonString('baz.bb,quz')
- out_mask.Intersect(mask1, mask2)
- self.assertEqual('baz.bb', out_mask.ToJsonString())
- # Overlap with paths covering some other paths.
- mask1.FromJsonString('foo.bar.baz,quz')
- mask2.FromJsonString('foo.bar,bar')
- out_mask.Intersect(mask1, mask2)
- self.assertEqual('foo.bar.baz', out_mask.ToJsonString())
- mask1.FromJsonString('foo.bar,bar')
- mask2.FromJsonString('foo.bar.baz,quz')
- out_mask.Intersect(mask1, mask2)
- self.assertEqual('foo.bar.baz', out_mask.ToJsonString())
- # Intersect '' with ''
- mask1.Clear()
- mask2.Clear()
- mask1.paths.append('')
- mask2.paths.append('')
- self.assertEqual(mask1.paths, [''])
- self.assertEqual('', mask1.ToJsonString())
- out_mask.Intersect(mask1, mask2)
- self.assertEqual(out_mask.paths, [])
-
- def testMergeMessageWithoutMapFields(self):
- # Test merge one field.
- src = unittest_pb2.TestAllTypes()
- test_util.SetAllFields(src)
- for field in src.DESCRIPTOR.fields:
- if field.containing_oneof:
- continue
- field_name = field.name
- dst = unittest_pb2.TestAllTypes()
- # Only set one path to mask.
- mask = field_mask_pb2.FieldMask()
- mask.paths.append(field_name)
- mask.MergeMessage(src, dst)
- # The expected result message.
- msg = unittest_pb2.TestAllTypes()
- if field.label == descriptor.FieldDescriptor.LABEL_REPEATED:
- repeated_src = getattr(src, field_name)
- repeated_msg = getattr(msg, field_name)
- if field.cpp_type == descriptor.FieldDescriptor.CPPTYPE_MESSAGE:
- for item in repeated_src:
- repeated_msg.add().CopyFrom(item)
- else:
- repeated_msg.extend(repeated_src)
- elif field.cpp_type == descriptor.FieldDescriptor.CPPTYPE_MESSAGE:
- getattr(msg, field_name).CopyFrom(getattr(src, field_name))
- else:
- setattr(msg, field_name, getattr(src, field_name))
- # Only field specified in mask is merged.
- self.assertEqual(msg, dst)
-
- # Test merge nested fields.
- nested_src = unittest_pb2.NestedTestAllTypes()
- nested_dst = unittest_pb2.NestedTestAllTypes()
- nested_src.child.payload.optional_int32 = 1234
- nested_src.child.child.payload.optional_int32 = 5678
- mask = field_mask_pb2.FieldMask()
- mask.FromJsonString('child.payload')
- mask.MergeMessage(nested_src, nested_dst)
- self.assertEqual(1234, nested_dst.child.payload.optional_int32)
- self.assertEqual(0, nested_dst.child.child.payload.optional_int32)
-
- mask.FromJsonString('child.child.payload')
- mask.MergeMessage(nested_src, nested_dst)
- self.assertEqual(1234, nested_dst.child.payload.optional_int32)
- self.assertEqual(5678, nested_dst.child.child.payload.optional_int32)
-
- nested_dst.Clear()
- mask.FromJsonString('child.child.payload')
- mask.MergeMessage(nested_src, nested_dst)
- self.assertEqual(0, nested_dst.child.payload.optional_int32)
- self.assertEqual(5678, nested_dst.child.child.payload.optional_int32)
-
- nested_dst.Clear()
- mask.FromJsonString('child')
- mask.MergeMessage(nested_src, nested_dst)
- self.assertEqual(1234, nested_dst.child.payload.optional_int32)
- self.assertEqual(5678, nested_dst.child.child.payload.optional_int32)
-
- # Test MergeOptions.
- nested_dst.Clear()
- nested_dst.child.payload.optional_int64 = 4321
- # Message fields will be merged by default.
- mask.FromJsonString('child.payload')
- mask.MergeMessage(nested_src, nested_dst)
- self.assertEqual(1234, nested_dst.child.payload.optional_int32)
- self.assertEqual(4321, nested_dst.child.payload.optional_int64)
- # Change the behavior to replace message fields.
- mask.FromJsonString('child.payload')
- mask.MergeMessage(nested_src, nested_dst, True, False)
- self.assertEqual(1234, nested_dst.child.payload.optional_int32)
- self.assertEqual(0, nested_dst.child.payload.optional_int64)
-
- # By default, fields missing in source are not cleared in destination.
- nested_dst.payload.optional_int32 = 1234
- self.assertTrue(nested_dst.HasField('payload'))
- mask.FromJsonString('payload')
- mask.MergeMessage(nested_src, nested_dst)
- self.assertTrue(nested_dst.HasField('payload'))
- # But they are cleared when replacing message fields.
- nested_dst.Clear()
- nested_dst.payload.optional_int32 = 1234
- mask.FromJsonString('payload')
- mask.MergeMessage(nested_src, nested_dst, True, False)
- self.assertFalse(nested_dst.HasField('payload'))
-
- nested_src.payload.repeated_int32.append(1234)
- nested_dst.payload.repeated_int32.append(5678)
- # Repeated fields will be appended by default.
- mask.FromJsonString('payload.repeatedInt32')
- mask.MergeMessage(nested_src, nested_dst)
- self.assertEqual(2, len(nested_dst.payload.repeated_int32))
- self.assertEqual(5678, nested_dst.payload.repeated_int32[0])
- self.assertEqual(1234, nested_dst.payload.repeated_int32[1])
- # Change the behavior to replace repeated fields.
- mask.FromJsonString('payload.repeatedInt32')
- mask.MergeMessage(nested_src, nested_dst, False, True)
- self.assertEqual(1, len(nested_dst.payload.repeated_int32))
- self.assertEqual(1234, nested_dst.payload.repeated_int32[0])
-
- # Test Merge oneof field.
- new_msg = unittest_pb2.TestOneof2()
- dst = unittest_pb2.TestOneof2()
- dst.foo_message.moo_int = 1
- mask = field_mask_pb2.FieldMask()
- mask.FromJsonString('fooMessage,fooLazyMessage.mooInt')
- mask.MergeMessage(new_msg, dst)
- self.assertTrue(dst.HasField('foo_message'))
- self.assertFalse(dst.HasField('foo_lazy_message'))
-
- def testMergeMessageWithMapField(self):
- empty_map = map_unittest_pb2.TestRecursiveMapMessage()
- src_level_2 = map_unittest_pb2.TestRecursiveMapMessage()
- src_level_2.a['src level 2'].CopyFrom(empty_map)
- src = map_unittest_pb2.TestRecursiveMapMessage()
- src.a['common key'].CopyFrom(src_level_2)
- src.a['src level 1'].CopyFrom(src_level_2)
-
- dst_level_2 = map_unittest_pb2.TestRecursiveMapMessage()
- dst_level_2.a['dst level 2'].CopyFrom(empty_map)
- dst = map_unittest_pb2.TestRecursiveMapMessage()
- dst.a['common key'].CopyFrom(dst_level_2)
- dst.a['dst level 1'].CopyFrom(empty_map)
-
- mask = field_mask_pb2.FieldMask()
- mask.FromJsonString('a')
- mask.MergeMessage(src, dst)
-
- # map from dst is replaced with map from src.
- self.assertEqual(dst.a['common key'], src_level_2)
- self.assertEqual(dst.a['src level 1'], src_level_2)
- self.assertEqual(dst.a['dst level 1'], empty_map)
-
- def testMergeErrors(self):
- src = unittest_pb2.TestAllTypes()
- dst = unittest_pb2.TestAllTypes()
- mask = field_mask_pb2.FieldMask()
- test_util.SetAllFields(src)
- mask.FromJsonString('optionalInt32.field')
- with self.assertRaises(ValueError) as e:
- mask.MergeMessage(src, dst)
- self.assertEqual('Error: Field optional_int32 in message '
- 'protobuf_unittest.TestAllTypes is not a singular '
- 'message field and cannot have sub-fields.',
- str(e.exception))
-
- def testSnakeCaseToCamelCase(self):
- self.assertEqual('fooBar',
- well_known_types._SnakeCaseToCamelCase('foo_bar'))
- self.assertEqual('FooBar',
- well_known_types._SnakeCaseToCamelCase('_foo_bar'))
- self.assertEqual('foo3Bar',
- well_known_types._SnakeCaseToCamelCase('foo3_bar'))
-
- # No uppercase letter is allowed.
- self.assertRaisesRegex(
- ValueError,
- 'Fail to print FieldMask to Json string: Path name Foo must '
- 'not contain uppercase letters.',
- well_known_types._SnakeCaseToCamelCase, 'Foo')
- # Any character after a "_" must be a lowercase letter.
- # 1. "_" cannot be followed by another "_".
- # 2. "_" cannot be followed by a digit.
- # 3. "_" cannot appear as the last character.
- self.assertRaisesRegex(
- ValueError,
- 'Fail to print FieldMask to Json string: The character after a '
- '"_" must be a lowercase letter in path name foo__bar.',
- well_known_types._SnakeCaseToCamelCase, 'foo__bar')
- self.assertRaisesRegex(
- ValueError,
- 'Fail to print FieldMask to Json string: The character after a '
- '"_" must be a lowercase letter in path name foo_3bar.',
- well_known_types._SnakeCaseToCamelCase, 'foo_3bar')
- self.assertRaisesRegex(
- ValueError,
- 'Fail to print FieldMask to Json string: Trailing "_" in path '
- 'name foo_bar_.', well_known_types._SnakeCaseToCamelCase, 'foo_bar_')
-
- def testCamelCaseToSnakeCase(self):
- self.assertEqual('foo_bar',
- well_known_types._CamelCaseToSnakeCase('fooBar'))
- self.assertEqual('_foo_bar',
- well_known_types._CamelCaseToSnakeCase('FooBar'))
- self.assertEqual('foo3_bar',
- well_known_types._CamelCaseToSnakeCase('foo3Bar'))
- self.assertRaisesRegex(
- ValueError,
- 'Fail to parse FieldMask: Path name foo_bar must not contain "_"s.',
- well_known_types._CamelCaseToSnakeCase, 'foo_bar')
-
-
class StructTest(unittest.TestCase):
def testStruct(self):
diff --git a/python/mox.py b/python/mox.py
deleted file mode 100755
index 9dd8ac7..0000000
--- a/python/mox.py
+++ /dev/null
@@ -1,1401 +0,0 @@
-#!/usr/bin/python2.4
-#
-# Copyright 2008 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.
-
-# This file is used for testing. The original is at:
-# http://code.google.com/p/pymox/
-
-"""Mox, an object-mocking framework for Python.
-
-Mox works in the record-replay-verify paradigm. When you first create
-a mock object, it is in record mode. You then programmatically set
-the expected behavior of the mock object (what methods are to be
-called on it, with what parameters, what they should return, and in
-what order).
-
-Once you have set up the expected mock behavior, you put it in replay
-mode. Now the mock responds to method calls just as you told it to.
-If an unexpected method (or an expected method with unexpected
-parameters) is called, then an exception will be raised.
-
-Once you are done interacting with the mock, you need to verify that
-all the expected interactions occurred. (Maybe your code exited
-prematurely without calling some cleanup method!) The verify phase
-ensures that every expected method was called; otherwise, an exception
-will be raised.
-
-Suggested usage / workflow:
-
- # Create Mox factory
- my_mox = Mox()
-
- # Create a mock data access object
- mock_dao = my_mox.CreateMock(DAOClass)
-
- # Set up expected behavior
- mock_dao.RetrievePersonWithIdentifier('1').AndReturn(person)
- mock_dao.DeletePerson(person)
-
- # Put mocks in replay mode
- my_mox.ReplayAll()
-
- # Inject mock object and run test
- controller.SetDao(mock_dao)
- controller.DeletePersonById('1')
-
- # Verify all methods were called as expected
- my_mox.VerifyAll()
-"""
-
-from collections import deque
-import re
-import types
-import unittest
-
-import stubout
-
-class Error(AssertionError):
- """Base exception for this module."""
-
- pass
-
-
-class ExpectedMethodCallsError(Error):
- """Raised when Verify() is called before all expected methods have been called
- """
-
- def __init__(self, expected_methods):
- """Init exception.
-
- Args:
- # expected_methods: A sequence of MockMethod objects that should have been
- # called.
- expected_methods: [MockMethod]
-
- Raises:
- ValueError: if expected_methods contains no methods.
- """
-
- if not expected_methods:
- raise ValueError("There must be at least one expected method")
- Error.__init__(self)
- self._expected_methods = expected_methods
-
- def __str__(self):
- calls = "\n".join(["%3d. %s" % (i, m)
- for i, m in enumerate(self._expected_methods)])
- return "Verify: Expected methods never called:\n%s" % (calls,)
-
-
-class UnexpectedMethodCallError(Error):
- """Raised when an unexpected method is called.
-
- This can occur if a method is called with incorrect parameters, or out of the
- specified order.
- """
-
- def __init__(self, unexpected_method, expected):
- """Init exception.
-
- Args:
- # unexpected_method: MockMethod that was called but was not at the head of
- # the expected_method queue.
- # expected: MockMethod or UnorderedGroup the method should have
- # been in.
- unexpected_method: MockMethod
- expected: MockMethod or UnorderedGroup
- """
-
- Error.__init__(self)
- self._unexpected_method = unexpected_method
- self._expected = expected
-
- def __str__(self):
- return "Unexpected method call: %s. Expecting: %s" % \
- (self._unexpected_method, self._expected)
-
-
-class UnknownMethodCallError(Error):
- """Raised if an unknown method is requested of the mock object."""
-
- def __init__(self, unknown_method_name):
- """Init exception.
-
- Args:
- # unknown_method_name: Method call that is not part of the mocked class's
- # public interface.
- unknown_method_name: str
- """
-
- Error.__init__(self)
- self._unknown_method_name = unknown_method_name
-
- def __str__(self):
- return "Method called is not a member of the object: %s" % \
- self._unknown_method_name
-
-
-class Mox(object):
- """Mox: a factory for creating mock objects."""
-
- # A list of types that should be stubbed out with MockObjects (as
- # opposed to MockAnythings).
- _USE_MOCK_OBJECT = [types.ClassType, types.InstanceType, types.ModuleType,
- types.ObjectType, types.TypeType]
-
- def __init__(self):
- """Initialize a new Mox."""
-
- self._mock_objects = []
- self.stubs = stubout.StubOutForTesting()
-
- def CreateMock(self, class_to_mock):
- """Create a new mock object.
-
- Args:
- # class_to_mock: the class to be mocked
- class_to_mock: class
-
- Returns:
- MockObject that can be used as the class_to_mock would be.
- """
-
- new_mock = MockObject(class_to_mock)
- self._mock_objects.append(new_mock)
- return new_mock
-
- def CreateMockAnything(self):
- """Create a mock that will accept any method calls.
-
- This does not enforce an interface.
- """
-
- new_mock = MockAnything()
- self._mock_objects.append(new_mock)
- return new_mock
-
- def ReplayAll(self):
- """Set all mock objects to replay mode."""
-
- for mock_obj in self._mock_objects:
- mock_obj._Replay()
-
-
- def VerifyAll(self):
- """Call verify on all mock objects created."""
-
- for mock_obj in self._mock_objects:
- mock_obj._Verify()
-
- def ResetAll(self):
- """Call reset on all mock objects. This does not unset stubs."""
-
- for mock_obj in self._mock_objects:
- mock_obj._Reset()
-
- def StubOutWithMock(self, obj, attr_name, use_mock_anything=False):
- """Replace a method, attribute, etc. with a Mock.
-
- This will replace a class or module with a MockObject, and everything else
- (method, function, etc) with a MockAnything. This can be overridden to
- always use a MockAnything by setting use_mock_anything to True.
-
- Args:
- obj: A Python object (class, module, instance, callable).
- attr_name: str. The name of the attribute to replace with a mock.
- use_mock_anything: bool. True if a MockAnything should be used regardless
- of the type of attribute.
- """
-
- attr_to_replace = getattr(obj, attr_name)
- if type(attr_to_replace) in self._USE_MOCK_OBJECT and not use_mock_anything:
- stub = self.CreateMock(attr_to_replace)
- else:
- stub = self.CreateMockAnything()
-
- self.stubs.Set(obj, attr_name, stub)
-
- def UnsetStubs(self):
- """Restore stubs to their original state."""
-
- self.stubs.UnsetAll()
-
-def Replay(*args):
- """Put mocks into Replay mode.
-
- Args:
- # args is any number of mocks to put into replay mode.
- """
-
- for mock in args:
- mock._Replay()
-
-
-def Verify(*args):
- """Verify mocks.
-
- Args:
- # args is any number of mocks to be verified.
- """
-
- for mock in args:
- mock._Verify()
-
-
-def Reset(*args):
- """Reset mocks.
-
- Args:
- # args is any number of mocks to be reset.
- """
-
- for mock in args:
- mock._Reset()
-
-
-class MockAnything:
- """A mock that can be used to mock anything.
-
- This is helpful for mocking classes that do not provide a public interface.
- """
-
- def __init__(self):
- """ """
- self._Reset()
-
- def __getattr__(self, method_name):
- """Intercept method calls on this object.
-
- A new MockMethod is returned that is aware of the MockAnything's
- state (record or replay). The call will be recorded or replayed
- by the MockMethod's __call__.
-
- Args:
- # method name: the name of the method being called.
- method_name: str
-
- Returns:
- A new MockMethod aware of MockAnything's state (record or replay).
- """
-
- return self._CreateMockMethod(method_name)
-
- def _CreateMockMethod(self, method_name):
- """Create a new mock method call and return it.
-
- Args:
- # method name: the name of the method being called.
- method_name: str
-
- Returns:
- A new MockMethod aware of MockAnything's state (record or replay).
- """
-
- return MockMethod(method_name, self._expected_calls_queue,
- self._replay_mode)
-
- def __nonzero__(self):
- """Return 1 for nonzero so the mock can be used as a conditional."""
-
- return 1
-
- def __eq__(self, rhs):
- """Provide custom logic to compare objects."""
-
- return (isinstance(rhs, MockAnything) and
- self._replay_mode == rhs._replay_mode and
- self._expected_calls_queue == rhs._expected_calls_queue)
-
- def __ne__(self, rhs):
- """Provide custom logic to compare objects."""
-
- return not self == rhs
-
- def _Replay(self):
- """Start replaying expected method calls."""
-
- self._replay_mode = True
-
- def _Verify(self):
- """Verify that all of the expected calls have been made.
-
- Raises:
- ExpectedMethodCallsError: if there are still more method calls in the
- expected queue.
- """
-
- # If the list of expected calls is not empty, raise an exception
- if self._expected_calls_queue:
- # The last MultipleTimesGroup is not popped from the queue.
- if (len(self._expected_calls_queue) == 1 and
- isinstance(self._expected_calls_queue[0], MultipleTimesGroup) and
- self._expected_calls_queue[0].IsSatisfied()):
- pass
- else:
- raise ExpectedMethodCallsError(self._expected_calls_queue)
-
- def _Reset(self):
- """Reset the state of this mock to record mode with an empty queue."""
-
- # Maintain a list of method calls we are expecting
- self._expected_calls_queue = deque()
-
- # Make sure we are in setup mode, not replay mode
- self._replay_mode = False
-
-
-class MockObject(MockAnything, object):
- """A mock object that simulates the public/protected interface of a class."""
-
- def __init__(self, class_to_mock):
- """Initialize a mock object.
-
- This determines the methods and properties of the class and stores them.
-
- Args:
- # class_to_mock: class to be mocked
- class_to_mock: class
- """
-
- # This is used to hack around the mixin/inheritance of MockAnything, which
- # is not a proper object (it can be anything. :-)
- MockAnything.__dict__['__init__'](self)
-
- # Get a list of all the public and special methods we should mock.
- self._known_methods = set()
- self._known_vars = set()
- self._class_to_mock = class_to_mock
- for method in dir(class_to_mock):
- if callable(getattr(class_to_mock, method)):
- self._known_methods.add(method)
- else:
- self._known_vars.add(method)
-
- def __getattr__(self, name):
- """Intercept attribute request on this object.
-
- If the attribute is a public class variable, it will be returned and not
- recorded as a call.
-
- If the attribute is not a variable, it is handled like a method
- call. The method name is checked against the set of mockable
- methods, and a new MockMethod is returned that is aware of the
- MockObject's state (record or replay). The call will be recorded
- or replayed by the MockMethod's __call__.
-
- Args:
- # name: the name of the attribute being requested.
- name: str
-
- Returns:
- Either a class variable or a new MockMethod that is aware of the state
- of the mock (record or replay).
-
- Raises:
- UnknownMethodCallError if the MockObject does not mock the requested
- method.
- """
-
- if name in self._known_vars:
- return getattr(self._class_to_mock, name)
-
- if name in self._known_methods:
- return self._CreateMockMethod(name)
-
- raise UnknownMethodCallError(name)
-
- def __eq__(self, rhs):
- """Provide custom logic to compare objects."""
-
- return (isinstance(rhs, MockObject) and
- self._class_to_mock == rhs._class_to_mock and
- self._replay_mode == rhs._replay_mode and
- self._expected_calls_queue == rhs._expected_calls_queue)
-
- def __setitem__(self, key, value):
- """Provide custom logic for mocking classes that support item assignment.
-
- Args:
- key: Key to set the value for.
- value: Value to set.
-
- Returns:
- Expected return value in replay mode. A MockMethod object for the
- __setitem__ method that has already been called if not in replay mode.
-
- Raises:
- TypeError if the underlying class does not support item assignment.
- UnexpectedMethodCallError if the object does not expect the call to
- __setitem__.
-
- """
- setitem = self._class_to_mock.__dict__.get('__setitem__', None)
-
- # Verify the class supports item assignment.
- if setitem is None:
- raise TypeError('object does not support item assignment')
-
- # If we are in replay mode then simply call the mock __setitem__ method.
- if self._replay_mode:
- return MockMethod('__setitem__', self._expected_calls_queue,
- self._replay_mode)(key, value)
-
-
- # Otherwise, create a mock method __setitem__.
- return self._CreateMockMethod('__setitem__')(key, value)
-
- def __getitem__(self, key):
- """Provide custom logic for mocking classes that are subscriptable.
-
- Args:
- key: Key to return the value for.
-
- Returns:
- Expected return value in replay mode. A MockMethod object for the
- __getitem__ method that has already been called if not in replay mode.
-
- Raises:
- TypeError if the underlying class is not subscriptable.
- UnexpectedMethodCallError if the object does not expect the call to
- __setitem__.
-
- """
- getitem = self._class_to_mock.__dict__.get('__getitem__', None)
-
- # Verify the class supports item assignment.
- if getitem is None:
- raise TypeError('unsubscriptable object')
-
- # If we are in replay mode then simply call the mock __getitem__ method.
- if self._replay_mode:
- return MockMethod('__getitem__', self._expected_calls_queue,
- self._replay_mode)(key)
-
-
- # Otherwise, create a mock method __getitem__.
- return self._CreateMockMethod('__getitem__')(key)
-
- def __call__(self, *params, **named_params):
- """Provide custom logic for mocking classes that are callable."""
-
- # Verify the class we are mocking is callable
- callable = self._class_to_mock.__dict__.get('__call__', None)
- if callable is None:
- raise TypeError('Not callable')
-
- # Because the call is happening directly on this object instead of a method,
- # the call on the mock method is made right here
- mock_method = self._CreateMockMethod('__call__')
- return mock_method(*params, **named_params)
-
- @property
- def __class__(self):
- """Return the class that is being mocked."""
-
- return self._class_to_mock
-
-
-class MockMethod(object):
- """Callable mock method.
-
- A MockMethod should act exactly like the method it mocks, accepting parameters
- and returning a value, or throwing an exception (as specified). When this
- method is called, it can optionally verify whether the called method (name and
- signature) matches the expected method.
- """
-
- def __init__(self, method_name, call_queue, replay_mode):
- """Construct a new mock method.
-
- Args:
- # method_name: the name of the method
- # call_queue: deque of calls, verify this call against the head, or add
- # this call to the queue.
- # replay_mode: False if we are recording, True if we are verifying calls
- # against the call queue.
- method_name: str
- call_queue: list or deque
- replay_mode: bool
- """
-
- self._name = method_name
- self._call_queue = call_queue
- if not isinstance(call_queue, deque):
- self._call_queue = deque(self._call_queue)
- self._replay_mode = replay_mode
-
- self._params = None
- self._named_params = None
- self._return_value = None
- self._exception = None
- self._side_effects = None
-
- def __call__(self, *params, **named_params):
- """Log parameters and return the specified return value.
-
- If the Mock(Anything/Object) associated with this call is in record mode,
- this MockMethod will be pushed onto the expected call queue. If the mock
- is in replay mode, this will pop a MockMethod off the top of the queue and
- verify this call is equal to the expected call.
-
- Raises:
- UnexpectedMethodCall if this call is supposed to match an expected method
- call and it does not.
- """
-
- self._params = params
- self._named_params = named_params
-
- if not self._replay_mode:
- self._call_queue.append(self)
- return self
-
- expected_method = self._VerifyMethodCall()
-
- if expected_method._side_effects:
- expected_method._side_effects(*params, **named_params)
-
- if expected_method._exception:
- raise expected_method._exception
-
- return expected_method._return_value
-
- def __getattr__(self, name):
- """Raise an AttributeError with a helpful message."""
-
- raise AttributeError('MockMethod has no attribute "%s". '
- 'Did you remember to put your mocks in replay mode?' % name)
-
- def _PopNextMethod(self):
- """Pop the next method from our call queue."""
- try:
- return self._call_queue.popleft()
- except IndexError:
- raise UnexpectedMethodCallError(self, None)
-
- def _VerifyMethodCall(self):
- """Verify the called method is expected.
-
- This can be an ordered method, or part of an unordered set.
-
- Returns:
- The expected mock method.
-
- Raises:
- UnexpectedMethodCall if the method called was not expected.
- """
-
- expected = self._PopNextMethod()
-
- # Loop here, because we might have a MethodGroup followed by another
- # group.
- while isinstance(expected, MethodGroup):
- expected, method = expected.MethodCalled(self)
- if method is not None:
- return method
-
- # This is a mock method, so just check equality.
- if expected != self:
- raise UnexpectedMethodCallError(self, expected)
-
- return expected
-
- def __str__(self):
- params = ', '.join(
- [repr(p) for p in self._params or []] +
- ['%s=%r' % x for x in sorted((self._named_params or {}).items())])
- desc = "%s(%s) -> %r" % (self._name, params, self._return_value)
- return desc
-
- def __eq__(self, rhs):
- """Test whether this MockMethod is equivalent to another MockMethod.
-
- Args:
- # rhs: the right hand side of the test
- rhs: MockMethod
- """
-
- return (isinstance(rhs, MockMethod) and
- self._name == rhs._name and
- self._params == rhs._params and
- self._named_params == rhs._named_params)
-
- def __ne__(self, rhs):
- """Test whether this MockMethod is not equivalent to another MockMethod.
-
- Args:
- # rhs: the right hand side of the test
- rhs: MockMethod
- """
-
- return not self == rhs
-
- def GetPossibleGroup(self):
- """Returns a possible group from the end of the call queue or None if no
- other methods are on the stack.
- """
-
- # Remove this method from the tail of the queue so we can add it to a group.
- this_method = self._call_queue.pop()
- assert this_method == self
-
- # Determine if the tail of the queue is a group, or just a regular ordered
- # mock method.
- group = None
- try:
- group = self._call_queue[-1]
- except IndexError:
- pass
-
- return group
-
- def _CheckAndCreateNewGroup(self, group_name, group_class):
- """Checks if the last method (a possible group) is an instance of our
- group_class. Adds the current method to this group or creates a new one.
-
- Args:
-
- group_name: the name of the group.
- group_class: the class used to create instance of this new group
- """
- group = self.GetPossibleGroup()
-
- # If this is a group, and it is the correct group, add the method.
- if isinstance(group, group_class) and group.group_name() == group_name:
- group.AddMethod(self)
- return self
-
- # Create a new group and add the method.
- new_group = group_class(group_name)
- new_group.AddMethod(self)
- self._call_queue.append(new_group)
- return self
-
- def InAnyOrder(self, group_name="default"):
- """Move this method into a group of unordered calls.
-
- A group of unordered calls must be defined together, and must be executed
- in full before the next expected method can be called. There can be
- multiple groups that are expected serially, if they are given
- different group names. The same group name can be reused if there is a
- standard method call, or a group with a different name, spliced between
- usages.
-
- Args:
- group_name: the name of the unordered group.
-
- Returns:
- self
- """
- return self._CheckAndCreateNewGroup(group_name, UnorderedGroup)
-
- def MultipleTimes(self, group_name="default"):
- """Move this method into group of calls which may be called multiple times.
-
- A group of repeating calls must be defined together, and must be executed in
- full before the next expected method can be called.
-
- Args:
- group_name: the name of the unordered group.
-
- Returns:
- self
- """
- return self._CheckAndCreateNewGroup(group_name, MultipleTimesGroup)
-
- def AndReturn(self, return_value):
- """Set the value to return when this method is called.
-
- Args:
- # return_value can be anything.
- """
-
- self._return_value = return_value
- return return_value
-
- def AndRaise(self, exception):
- """Set the exception to raise when this method is called.
-
- Args:
- # exception: the exception to raise when this method is called.
- exception: Exception
- """
-
- self._exception = exception
-
- def WithSideEffects(self, side_effects):
- """Set the side effects that are simulated when this method is called.
-
- Args:
- side_effects: A callable which modifies the parameters or other relevant
- state which a given test case depends on.
-
- Returns:
- Self for chaining with AndReturn and AndRaise.
- """
- self._side_effects = side_effects
- return self
-
-class Comparator:
- """Base class for all Mox comparators.
-
- A Comparator can be used as a parameter to a mocked method when the exact
- value is not known. For example, the code you are testing might build up a
- long SQL string that is passed to your mock DAO. You're only interested that
- the IN clause contains the proper primary keys, so you can set your mock
- up as follows:
-
- mock_dao.RunQuery(StrContains('IN (1, 2, 4, 5)')).AndReturn(mock_result)
-
- Now whatever query is passed in must contain the string 'IN (1, 2, 4, 5)'.
-
- A Comparator may replace one or more parameters, for example:
- # return at most 10 rows
- mock_dao.RunQuery(StrContains('SELECT'), 10)
-
- or
-
- # Return some non-deterministic number of rows
- mock_dao.RunQuery(StrContains('SELECT'), IsA(int))
- """
-
- def equals(self, rhs):
- """Special equals method that all comparators must implement.
-
- Args:
- rhs: any python object
- """
-
- raise NotImplementedError('method must be implemented by a subclass.')
-
- def __eq__(self, rhs):
- return self.equals(rhs)
-
- def __ne__(self, rhs):
- return not self.equals(rhs)
-
-
-class IsA(Comparator):
- """This class wraps a basic Python type or class. It is used to verify
- that a parameter is of the given type or class.
-
- Example:
- mock_dao.Connect(IsA(DbConnectInfo))
- """
-
- def __init__(self, class_name):
- """Initialize IsA
-
- Args:
- class_name: basic python type or a class
- """
-
- self._class_name = class_name
-
- def equals(self, rhs):
- """Check to see if the RHS is an instance of class_name.
-
- Args:
- # rhs: the right hand side of the test
- rhs: object
-
- Returns:
- bool
- """
-
- try:
- return isinstance(rhs, self._class_name)
- except TypeError:
- # Check raw types if there was a type error. This is helpful for
- # things like cStringIO.StringIO.
- return type(rhs) == type(self._class_name)
-
- def __repr__(self):
- return str(self._class_name)
-
-class IsAlmost(Comparator):
- """Comparison class used to check whether a parameter is nearly equal
- to a given value. Generally useful for floating point numbers.
-
- Example mock_dao.SetTimeout((IsAlmost(3.9)))
- """
-
- def __init__(self, float_value, places=7):
- """Initialize IsAlmost.
-
- Args:
- float_value: The value for making the comparison.
- places: The number of decimal places to round to.
- """
-
- self._float_value = float_value
- self._places = places
-
- def equals(self, rhs):
- """Check to see if RHS is almost equal to float_value
-
- Args:
- rhs: the value to compare to float_value
-
- Returns:
- bool
- """
-
- try:
- return round(rhs-self._float_value, self._places) == 0
- except TypeError:
- # This is probably because either float_value or rhs is not a number.
- return False
-
- def __repr__(self):
- return str(self._float_value)
-
-class StrContains(Comparator):
- """Comparison class used to check whether a substring exists in a
- string parameter. This can be useful in mocking a database with SQL
- passed in as a string parameter, for example.
-
- Example:
- mock_dao.RunQuery(StrContains('IN (1, 2, 4, 5)')).AndReturn(mock_result)
- """
-
- def __init__(self, search_string):
- """Initialize.
-
- Args:
- # search_string: the string you are searching for
- search_string: str
- """
-
- self._search_string = search_string
-
- def equals(self, rhs):
- """Check to see if the search_string is contained in the rhs string.
-
- Args:
- # rhs: the right hand side of the test
- rhs: object
-
- Returns:
- bool
- """
-
- try:
- return rhs.find(self._search_string) > -1
- except Exception:
- return False
-
- def __repr__(self):
- return '<str containing \'%s\'>' % self._search_string
-
-
-class Regex(Comparator):
- """Checks if a string matches a regular expression.
-
- This uses a given regular expression to determine equality.
- """
-
- def __init__(self, pattern, flags=0):
- """Initialize.
-
- Args:
- # pattern is the regular expression to search for
- pattern: str
- # flags passed to re.compile function as the second argument
- flags: int
- """
-
- self.regex = re.compile(pattern, flags=flags)
-
- def equals(self, rhs):
- """Check to see if rhs matches regular expression pattern.
-
- Returns:
- bool
- """
-
- return self.regex.search(rhs) is not None
-
- def __repr__(self):
- s = '<regular expression \'%s\'' % self.regex.pattern
- if self.regex.flags:
- s += ', flags=%d' % self.regex.flags
- s += '>'
- return s
-
-
-class In(Comparator):
- """Checks whether an item (or key) is in a list (or dict) parameter.
-
- Example:
- mock_dao.GetUsersInfo(In('expectedUserName')).AndReturn(mock_result)
- """
-
- def __init__(self, key):
- """Initialize.
-
- Args:
- # key is any thing that could be in a list or a key in a dict
- """
-
- self._key = key
-
- def equals(self, rhs):
- """Check to see whether key is in rhs.
-
- Args:
- rhs: dict
-
- Returns:
- bool
- """
-
- return self._key in rhs
-
- def __repr__(self):
- return '<sequence or map containing \'%s\'>' % self._key
-
-
-class ContainsKeyValue(Comparator):
- """Checks whether a key/value pair is in a dict parameter.
-
- Example:
- mock_dao.UpdateUsers(ContainsKeyValue('stevepm', stevepm_user_info))
- """
-
- def __init__(self, key, value):
- """Initialize.
-
- Args:
- # key: a key in a dict
- # value: the corresponding value
- """
-
- self._key = key
- self._value = value
-
- def equals(self, rhs):
- """Check whether the given key/value pair is in the rhs dict.
-
- Returns:
- bool
- """
-
- try:
- return rhs[self._key] == self._value
- except Exception:
- return False
-
- def __repr__(self):
- return '<map containing the entry \'%s: %s\'>' % (self._key, self._value)
-
-
-class SameElementsAs(Comparator):
- """Checks whether iterables contain the same elements (ignoring order).
-
- Example:
- mock_dao.ProcessUsers(SameElementsAs('stevepm', 'salomaki'))
- """
-
- def __init__(self, expected_seq):
- """Initialize.
-
- Args:
- expected_seq: a sequence
- """
-
- self._expected_seq = expected_seq
-
- def equals(self, actual_seq):
- """Check to see whether actual_seq has same elements as expected_seq.
-
- Args:
- actual_seq: sequence
-
- Returns:
- bool
- """
-
- try:
- expected = dict([(element, None) for element in self._expected_seq])
- actual = dict([(element, None) for element in actual_seq])
- except TypeError:
- # Fall back to slower list-compare if any of the objects are unhashable.
- expected = list(self._expected_seq)
- actual = list(actual_seq)
- expected.sort()
- actual.sort()
- return expected == actual
-
- def __repr__(self):
- return '<sequence with same elements as \'%s\'>' % self._expected_seq
-
-
-class And(Comparator):
- """Evaluates one or more Comparators on RHS and returns an AND of the results.
- """
-
- def __init__(self, *args):
- """Initialize.
-
- Args:
- *args: One or more Comparator
- """
-
- self._comparators = args
-
- def equals(self, rhs):
- """Checks whether all Comparators are equal to rhs.
-
- Args:
- # rhs: can be anything
-
- Returns:
- bool
- """
-
- for comparator in self._comparators:
- if not comparator.equals(rhs):
- return False
-
- return True
-
- def __repr__(self):
- return '<AND %s>' % str(self._comparators)
-
-
-class Or(Comparator):
- """Evaluates one or more Comparators on RHS and returns an OR of the results.
- """
-
- def __init__(self, *args):
- """Initialize.
-
- Args:
- *args: One or more Mox comparators
- """
-
- self._comparators = args
-
- def equals(self, rhs):
- """Checks whether any Comparator is equal to rhs.
-
- Args:
- # rhs: can be anything
-
- Returns:
- bool
- """
-
- for comparator in self._comparators:
- if comparator.equals(rhs):
- return True
-
- return False
-
- def __repr__(self):
- return '<OR %s>' % str(self._comparators)
-
-
-class Func(Comparator):
- """Call a function that should verify the parameter passed in is correct.
-
- You may need the ability to perform more advanced operations on the parameter
- in order to validate it. You can use this to have a callable validate any
- parameter. The callable should return either True or False.
-
-
- Example:
-
- def myParamValidator(param):
- # Advanced logic here
- return True
-
- mock_dao.DoSomething(Func(myParamValidator), true)
- """
-
- def __init__(self, func):
- """Initialize.
-
- Args:
- func: callable that takes one parameter and returns a bool
- """
-
- self._func = func
-
- def equals(self, rhs):
- """Test whether rhs passes the function test.
-
- rhs is passed into func.
-
- Args:
- rhs: any python object
-
- Returns:
- the result of func(rhs)
- """
-
- return self._func(rhs)
-
- def __repr__(self):
- return str(self._func)
-
-
-class IgnoreArg(Comparator):
- """Ignore an argument.
-
- This can be used when we don't care about an argument of a method call.
-
- Example:
- # Check if CastMagic is called with 3 as first arg and 'disappear' as third.
- mymock.CastMagic(3, IgnoreArg(), 'disappear')
- """
-
- def equals(self, unused_rhs):
- """Ignores arguments and returns True.
-
- Args:
- unused_rhs: any python object
-
- Returns:
- always returns True
- """
-
- return True
-
- def __repr__(self):
- return '<IgnoreArg>'
-
-
-class MethodGroup(object):
- """Base class containing common behaviour for MethodGroups."""
-
- def __init__(self, group_name):
- self._group_name = group_name
-
- def group_name(self):
- return self._group_name
-
- def __str__(self):
- return '<%s "%s">' % (self.__class__.__name__, self._group_name)
-
- def AddMethod(self, mock_method):
- raise NotImplementedError
-
- def MethodCalled(self, mock_method):
- raise NotImplementedError
-
- def IsSatisfied(self):
- raise NotImplementedError
-
-class UnorderedGroup(MethodGroup):
- """UnorderedGroup holds a set of method calls that may occur in any order.
-
- This construct is helpful for non-deterministic events, such as iterating
- over the keys of a dict.
- """
-
- def __init__(self, group_name):
- super(UnorderedGroup, self).__init__(group_name)
- self._methods = []
-
- def AddMethod(self, mock_method):
- """Add a method to this group.
-
- Args:
- mock_method: A mock method to be added to this group.
- """
-
- self._methods.append(mock_method)
-
- def MethodCalled(self, mock_method):
- """Remove a method call from the group.
-
- If the method is not in the set, an UnexpectedMethodCallError will be
- raised.
-
- Args:
- mock_method: a mock method that should be equal to a method in the group.
-
- Returns:
- The mock method from the group
-
- Raises:
- UnexpectedMethodCallError if the mock_method was not in the group.
- """
-
- # Check to see if this method exists, and if so, remove it from the set
- # and return it.
- for method in self._methods:
- if method == mock_method:
- # Remove the called mock_method instead of the method in the group.
- # The called method will match any comparators when equality is checked
- # during removal. The method in the group could pass a comparator to
- # another comparator during the equality check.
- self._methods.remove(mock_method)
-
- # If this group is not empty, put it back at the head of the queue.
- if not self.IsSatisfied():
- mock_method._call_queue.appendleft(self)
-
- return self, method
-
- raise UnexpectedMethodCallError(mock_method, self)
-
- def IsSatisfied(self):
- """Return True if there are not any methods in this group."""
-
- return len(self._methods) == 0
-
-
-class MultipleTimesGroup(MethodGroup):
- """MultipleTimesGroup holds methods that may be called any number of times.
-
- Note: Each method must be called at least once.
-
- This is helpful, if you don't know or care how many times a method is called.
- """
-
- def __init__(self, group_name):
- super(MultipleTimesGroup, self).__init__(group_name)
- self._methods = set()
- self._methods_called = set()
-
- def AddMethod(self, mock_method):
- """Add a method to this group.
-
- Args:
- mock_method: A mock method to be added to this group.
- """
-
- self._methods.add(mock_method)
-
- def MethodCalled(self, mock_method):
- """Remove a method call from the group.
-
- If the method is not in the set, an UnexpectedMethodCallError will be
- raised.
-
- Args:
- mock_method: a mock method that should be equal to a method in the group.
-
- Returns:
- The mock method from the group
-
- Raises:
- UnexpectedMethodCallError if the mock_method was not in the group.
- """
-
- # Check to see if this method exists, and if so add it to the set of
- # called methods.
-
- for method in self._methods:
- if method == mock_method:
- self._methods_called.add(mock_method)
- # Always put this group back on top of the queue, because we don't know
- # when we are done.
- mock_method._call_queue.appendleft(self)
- return self, method
-
- if self.IsSatisfied():
- next_method = mock_method._PopNextMethod();
- return next_method, None
- else:
- raise UnexpectedMethodCallError(mock_method, self)
-
- def IsSatisfied(self):
- """Return True if all methods in this group are called at least once."""
- # NOTE(psycho): We can't use the simple set difference here because we want
- # to match different parameters which are considered the same e.g. IsA(str)
- # and some string. This solution is O(n^2) but n should be small.
- tmp = self._methods.copy()
- for called in self._methods_called:
- for expected in tmp:
- if called == expected:
- tmp.remove(expected)
- if not tmp:
- return True
- break
- return False
-
-
-class MoxMetaTestBase(type):
- """Metaclass to add mox cleanup and verification to every test.
-
- As the mox unit testing class is being constructed (MoxTestBase or a
- subclass), this metaclass will modify all test functions to call the
- CleanUpMox method of the test class after they finish. This means that
- unstubbing and verifying will happen for every test with no additional code,
- and any failures will result in test failures as opposed to errors.
- """
-
- def __init__(cls, name, bases, d):
- type.__init__(cls, name, bases, d)
-
- # also get all the attributes from the base classes to account
- # for a case when test class is not the immediate child of MoxTestBase
- for base in bases:
- for attr_name in dir(base):
- d[attr_name] = getattr(base, attr_name)
-
- for func_name, func in d.items():
- if func_name.startswith('test') and callable(func):
- setattr(cls, func_name, MoxMetaTestBase.CleanUpTest(cls, func))
-
- @staticmethod
- def CleanUpTest(cls, func):
- """Adds Mox cleanup code to any MoxTestBase method.
-
- Always unsets stubs after a test. Will verify all mocks for tests that
- otherwise pass.
-
- Args:
- cls: MoxTestBase or subclass; the class whose test method we are altering.
- func: method; the method of the MoxTestBase test class we wish to alter.
-
- Returns:
- The modified method.
- """
- def new_method(self, *args, **kwargs):
- mox_obj = getattr(self, 'mox', None)
- cleanup_mox = False
- if mox_obj and isinstance(mox_obj, Mox):
- cleanup_mox = True
- try:
- func(self, *args, **kwargs)
- finally:
- if cleanup_mox:
- mox_obj.UnsetStubs()
- if cleanup_mox:
- mox_obj.VerifyAll()
- new_method.__name__ = func.__name__
- new_method.__doc__ = func.__doc__
- new_method.__module__ = func.__module__
- return new_method
-
-
-class MoxTestBase(unittest.TestCase):
- """Convenience test class to make stubbing easier.
-
- Sets up a "mox" attribute which is an instance of Mox - any mox tests will
- want this. Also automatically unsets any stubs and verifies that all mock
- methods have been called at the end of each test, eliminating boilerplate
- code.
- """
-
- __metaclass__ = MoxMetaTestBase
-
- def setUp(self):
- self.mox = Mox()
diff --git a/python/stubout.py b/python/stubout.py
deleted file mode 100755
index ba39104..0000000
--- a/python/stubout.py
+++ /dev/null
@@ -1,143 +0,0 @@
-#!/usr/bin/python2.4
-#
-# Copyright 2008 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.
-
-# This file is used for testing. The original is at:
-# http://code.google.com/p/pymox/
-
-import inspect
-
-
-class StubOutForTesting:
- """Sample Usage:
- You want os.path.exists() to always return true during testing.
-
- stubs = StubOutForTesting()
- stubs.Set(os.path, 'exists', lambda x: 1)
- ...
- stubs.UnsetAll()
-
- The above changes os.path.exists into a lambda that returns 1. Once
- the ... part of the code finishes, the UnsetAll() looks up the old value
- of os.path.exists and restores it.
-
- """
- def __init__(self):
- self.cache = []
- self.stubs = []
-
- def __del__(self):
- self.SmartUnsetAll()
- self.UnsetAll()
-
- def SmartSet(self, obj, attr_name, new_attr):
- """Replace obj.attr_name with new_attr. This method is smart and works
- at the module, class, and instance level while preserving proper
- inheritance. It will not stub out C types however unless that has been
- explicitly allowed by the type.
-
- This method supports the case where attr_name is a staticmethod or a
- classmethod of obj.
-
- Notes:
- - If obj is an instance, then it is its class that will actually be
- stubbed. Note that the method Set() does not do that: if obj is
- an instance, it (and not its class) will be stubbed.
- - The stubbing is using the builtin getattr and setattr. So, the __get__
- and __set__ will be called when stubbing (TODO: A better idea would
- probably be to manipulate obj.__dict__ instead of getattr() and
- setattr()).
-
- Raises AttributeError if the attribute cannot be found.
- """
- if (inspect.ismodule(obj) or
- (not inspect.isclass(obj) and obj.__dict__.has_key(attr_name))):
- orig_obj = obj
- orig_attr = getattr(obj, attr_name)
-
- else:
- if not inspect.isclass(obj):
- mro = list(inspect.getmro(obj.__class__))
- else:
- mro = list(inspect.getmro(obj))
-
- mro.reverse()
-
- orig_attr = None
-
- for cls in mro:
- try:
- orig_obj = cls
- orig_attr = getattr(obj, attr_name)
- except AttributeError:
- continue
-
- if orig_attr is None:
- raise AttributeError("Attribute not found.")
-
- # Calling getattr() on a staticmethod transforms it to a 'normal' function.
- # We need to ensure that we put it back as a staticmethod.
- old_attribute = obj.__dict__.get(attr_name)
- if old_attribute is not None and isinstance(old_attribute, staticmethod):
- orig_attr = staticmethod(orig_attr)
-
- self.stubs.append((orig_obj, attr_name, orig_attr))
- setattr(orig_obj, attr_name, new_attr)
-
- def SmartUnsetAll(self):
- """Reverses all the SmartSet() calls, restoring things to their original
- definition. Its okay to call SmartUnsetAll() repeatedly, as later calls
- have no effect if no SmartSet() calls have been made.
-
- """
- self.stubs.reverse()
-
- for args in self.stubs:
- setattr(*args)
-
- self.stubs = []
-
- def Set(self, parent, child_name, new_child):
- """Replace child_name's old definition with new_child, in the context
- of the given parent. The parent could be a module when the child is a
- function at module scope. Or the parent could be a class when a class'
- method is being replaced. The named child is set to new_child, while
- the prior definition is saved away for later, when UnsetAll() is called.
-
- This method supports the case where child_name is a staticmethod or a
- classmethod of parent.
- """
- old_child = getattr(parent, child_name)
-
- old_attribute = parent.__dict__.get(child_name)
- if old_attribute is not None and isinstance(old_attribute, staticmethod):
- old_child = staticmethod(old_child)
-
- self.cache.append((parent, old_child, child_name))
- setattr(parent, child_name, new_child)
-
- def UnsetAll(self):
- """Reverses all the Set() calls, restoring things to their original
- definition. Its okay to call UnsetAll() repeatedly, as later calls have
- no effect if no Set() calls have been made.
-
- """
- # Undo calls to Set() in reverse order, in case Set() was called on the
- # same arguments repeatedly (want the original call to be last one undone)
- self.cache.reverse()
-
- for (parent, old_child, child_name) in self.cache:
- setattr(parent, child_name, old_child)
- self.cache = []
