Merge branch 'public-master' into snap-seek2

core/google/cloud/_helpers.py

@@ -439,16 +439,9 @@ def _timedelta_to_duration_pb(timedelta_val):
     :rtype: :class:`google.protobuf.duration_pb2.Duration`
     :returns: A duration object equivalent to the time delta.
     """
-    seconds_decimal = timedelta_val.total_seconds()
-    # Truncate the parts other than the integer.
-    seconds = int(seconds_decimal)
-    if seconds_decimal < 0:
-        signed_micros = timedelta_val.microseconds - 10**6
-    else:
-        signed_micros = timedelta_val.microseconds
-    # Convert nanoseconds to microseconds.
-    nanos = 1000 * signed_micros
-    return duration_pb2.Duration(seconds=seconds, nanos=nanos)
+    duration_pb = duration_pb2.Duration()
+    duration_pb.FromTimedelta(timedelta_val)
+    return duration_pb
 
 
 def _duration_pb_to_timedelta(duration_pb):

core/google/cloud/iam.py

@@ -71,7 +71,7 @@ def __getitem__(self, key):
         return self._bindings[key]
 
     def __setitem__(self, key, value):
-        self._bindings[key] = frozenset(value)
+        self._bindings[key] = set(value)
 
     def __delitem__(self, key):
         del self._bindings[key]
@@ -91,7 +91,7 @@ def owners(self, value):
         warnings.warn(
             _ASSIGNMENT_DEPRECATED_MSG.format('owners', OWNER_ROLE),
             DeprecationWarning)
-        self._bindings[OWNER_ROLE] = list(value)
+        self[OWNER_ROLE] = value
 
     @property
     def editors(self):
@@ -108,7 +108,7 @@ def editors(self, value):
         warnings.warn(
             _ASSIGNMENT_DEPRECATED_MSG.format('editors', EDITOR_ROLE),
             DeprecationWarning)
-        self._bindings[EDITOR_ROLE] = list(value)
+        self[EDITOR_ROLE] = value
 
     @property
     def viewers(self):
@@ -125,7 +125,7 @@ def viewers(self, value):
         warnings.warn(
             _ASSIGNMENT_DEPRECATED_MSG.format('viewers', VIEWER_ROLE),
             DeprecationWarning)
-        self._bindings[VIEWER_ROLE] = list(value)
+        self[VIEWER_ROLE] = value
 
     @staticmethod
     def user(email):
@@ -209,7 +209,7 @@ def from_api_repr(cls, resource):
         for binding in resource.get('bindings', ()):
             role = binding['role']
             members = sorted(binding['members'])
-            policy._bindings[role] = members
+            policy[role] = members
         return policy
 
     def to_api_repr(self):

core/tests/unit/test_iam.py

@@ -27,27 +27,26 @@ def _make_one(self, *args, **kw):
         return self._get_target_class()(*args, **kw)
 
     def test_ctor_defaults(self):
+        empty = frozenset()
         policy = self._make_one()
         self.assertIsNone(policy.etag)
         self.assertIsNone(policy.version)
-        self.assertIsInstance(policy.owners, frozenset)
-        self.assertEqual(list(policy.owners), [])
-        self.assertIsInstance(policy.editors, frozenset)
-        self.assertEqual(list(policy.editors), [])
-        self.assertIsInstance(policy.viewers, frozenset)
-        self.assertEqual(list(policy.viewers), [])
+        self.assertEqual(policy.owners, empty)
+        self.assertEqual(policy.editors, empty)
+        self.assertEqual(policy.viewers, empty)
         self.assertEqual(len(policy), 0)
         self.assertEqual(dict(policy), {})
 
     def test_ctor_explicit(self):
         VERSION = 17
         ETAG = 'ETAG'
+        empty = frozenset()
         policy = self._make_one(ETAG, VERSION)
         self.assertEqual(policy.etag, ETAG)
         self.assertEqual(policy.version, VERSION)
-        self.assertEqual(list(policy.owners), [])
-        self.assertEqual(list(policy.editors), [])
-        self.assertEqual(list(policy.viewers), [])
+        self.assertEqual(policy.owners, empty)
+        self.assertEqual(policy.editors, empty)
+        self.assertEqual(policy.viewers, empty)
         self.assertEqual(len(policy), 0)
         self.assertEqual(dict(policy), {})
 
@@ -58,7 +57,7 @@ def test___getitem___miss(self):
 
     def test___setitem__(self):
         USER = 'user:[email protected]'
-        PRINCIPALS = frozenset([USER])
+        PRINCIPALS = set([USER])
         policy = self._make_one()
         policy['rolename'] = [USER]
         self.assertEqual(policy['rolename'], PRINCIPALS)
@@ -80,54 +79,59 @@ def test___delitem___miss(self):
     def test_owners_getter(self):
         from google.cloud.iam import OWNER_ROLE
         MEMBER = 'user:[email protected]'
+        expected = frozenset([MEMBER])
         policy = self._make_one()
         policy[OWNER_ROLE] = [MEMBER]
-        self.assertIsInstance(policy.owners, frozenset)
-        self.assertEqual(list(policy.owners), [MEMBER])
+        self.assertEqual(policy.owners, expected)
 
     def test_owners_setter(self):
         import warnings
         from google.cloud.iam import OWNER_ROLE
         MEMBER = 'user:[email protected]'
+        expected = set([MEMBER])
         policy = self._make_one()
         with warnings.catch_warnings():
+            warnings.simplefilter('always')
             policy.owners = [MEMBER]
-        self.assertEqual(list(policy[OWNER_ROLE]), [MEMBER])
+        self.assertEqual(policy[OWNER_ROLE], expected)
 
     def test_editors_getter(self):
         from google.cloud.iam import EDITOR_ROLE
         MEMBER = 'user:[email protected]'
+        expected = frozenset([MEMBER])
         policy = self._make_one()
         policy[EDITOR_ROLE] = [MEMBER]
-        self.assertIsInstance(policy.editors, frozenset)
-        self.assertEqual(list(policy.editors), [MEMBER])
+        self.assertEqual(policy.editors, expected)
 
     def test_editors_setter(self):
         import warnings
         from google.cloud.iam import EDITOR_ROLE
         MEMBER = 'user:[email protected]'
+        expected = set([MEMBER])
         policy = self._make_one()
         with warnings.catch_warnings():
+            warnings.simplefilter('always')
             policy.editors = [MEMBER]
-        self.assertEqual(list(policy[EDITOR_ROLE]), [MEMBER])
+        self.assertEqual(policy[EDITOR_ROLE], expected)
 
     def test_viewers_getter(self):
         from google.cloud.iam import VIEWER_ROLE
         MEMBER = 'user:[email protected]'
+        expected = frozenset([MEMBER])
         policy = self._make_one()
         policy[VIEWER_ROLE] = [MEMBER]
-        self.assertIsInstance(policy.viewers, frozenset)
-        self.assertEqual(list(policy.viewers), [MEMBER])
+        self.assertEqual(policy.viewers, expected)
 
     def test_viewers_setter(self):
         import warnings
         from google.cloud.iam import VIEWER_ROLE
         MEMBER = 'user:[email protected]'
+        expected = set([MEMBER])
         policy = self._make_one()
         with warnings.catch_warnings():
             warnings.simplefilter('always')
             policy.viewers = [MEMBER]
-        self.assertEqual(list(policy[VIEWER_ROLE]), [MEMBER])
+        self.assertEqual(policy[VIEWER_ROLE], expected)
 
     def test_user(self):
         EMAIL = '[email protected]'
@@ -162,16 +166,17 @@ def test_authenticated_users(self):
         self.assertEqual(policy.authenticated_users(), 'allAuthenticatedUsers')
 
     def test_from_api_repr_only_etag(self):
+        empty = frozenset()
         RESOURCE = {
             'etag': 'ACAB',
         }
         klass = self._get_target_class()
         policy = klass.from_api_repr(RESOURCE)
         self.assertEqual(policy.etag, 'ACAB')
         self.assertIsNone(policy.version)
-        self.assertEqual(list(policy.owners), [])
-        self.assertEqual(list(policy.editors), [])
-        self.assertEqual(list(policy.viewers), [])
+        self.assertEqual(policy.owners, empty)
+        self.assertEqual(policy.editors, empty)
+        self.assertEqual(policy.viewers, empty)
         self.assertEqual(dict(policy), {})
 
     def test_from_api_repr_complete(self):
@@ -196,18 +201,19 @@ def test_from_api_repr_complete(self):
                 {'role': VIEWER_ROLE, 'members': [VIEWER1, VIEWER2]},
             ],
         }
+        empty = frozenset()
         klass = self._get_target_class()
         policy = klass.from_api_repr(RESOURCE)
         self.assertEqual(policy.etag, 'DEADBEEF')
         self.assertEqual(policy.version, 17)
-        self.assertEqual(sorted(policy.owners), [OWNER1, OWNER2])
-        self.assertEqual(sorted(policy.editors), [EDITOR1, EDITOR2])
-        self.assertEqual(sorted(policy.viewers), [VIEWER1, VIEWER2])
+        self.assertEqual(policy.owners, frozenset([OWNER1, OWNER2]))
+        self.assertEqual(policy.editors, frozenset([EDITOR1, EDITOR2]))
+        self.assertEqual(policy.viewers, frozenset([VIEWER1, VIEWER2]))
         self.assertEqual(
             dict(policy), {
-                OWNER_ROLE: [OWNER1, OWNER2],
-                EDITOR_ROLE: [EDITOR1, EDITOR2],
-                VIEWER_ROLE: [VIEWER1, VIEWER2],
+                OWNER_ROLE: set([OWNER1, OWNER2]),
+                EDITOR_ROLE: set([EDITOR1, EDITOR2]),
+                VIEWER_ROLE: set([VIEWER1, VIEWER2]),
             })
 
     def test_from_api_repr_unknown_role(self):
@@ -224,7 +230,7 @@ def test_from_api_repr_unknown_role(self):
         policy = klass.from_api_repr(RESOURCE)
         self.assertEqual(policy.etag, 'DEADBEEF')
         self.assertEqual(policy.version, 17)
-        self.assertEqual(dict(policy), {'unknown': [GROUP, USER]})
+        self.assertEqual(dict(policy), {'unknown': set([GROUP, USER])})
 
     def test_to_api_repr_defaults(self):
         policy = self._make_one()

pubsub/google/cloud/pubsub/_gax.py

@@ -30,6 +30,7 @@
 
 from google.cloud._helpers import _to_bytes
 from google.cloud._helpers import _pb_timestamp_to_rfc3339
+from google.cloud._helpers import _timedelta_to_duration_pb
 from google.cloud._helpers import make_secure_channel
 from google.cloud._http import DEFAULT_USER_AGENT
 from google.cloud.exceptions import Conflict
@@ -277,7 +278,9 @@ def list_subscriptions(self, project, page_size=0, page_token=None):
         return GAXIterator(self._client, page_iter, item_to_value)
 
     def subscription_create(self, subscription_path, topic_path,
-                            ack_deadline=None, push_endpoint=None):
+                            ack_deadline=None, push_endpoint=None,
+                            retain_acked_messages=None,
+                            message_retention_duration=None):
         """API call:  create a subscription
 
         See:
@@ -303,6 +306,18 @@ def subscription_create(self, subscription_path, topic_path,
             (Optional) URL to which messages will be pushed by the back-end.
             If not set, the application must pull messages.
 
+        :type retain_acked_messages: bool
+        :param retain_acked_messages:
+            (Optional) Whether to retain acked messages. If set, acked messages
+            are retained in the subscription's backlog for a duration indicated
+            by `message_retention_duration`.
+
+        :type message_retention_duration: :class:`datetime.timedelta`
+        :param message_retention_duration:
+            (Optional) Whether to retain acked messages. If set, acked messages
+            are retained in the subscription's backlog for a duration indicated
+            by `message_retention_duration`. If unset, defaults to 7 days.
+
         :rtype: dict
         :returns: ``Subscription`` resource returned from the API.
         """
@@ -311,13 +326,16 @@ def subscription_create(self, subscription_path, topic_path,
         else:
             push_config = None
 
-        if ack_deadline is None:
-            ack_deadline = 0
+        if message_retention_duration is not None:
+            message_retention_duration = _timedelta_to_duration_pb(
+                message_retention_duration)
 
         try:
             sub_pb = self._gax_api.create_subscription(
                 subscription_path, topic_path,
-                push_config=push_config, ack_deadline_seconds=ack_deadline)
+                push_config=push_config, ack_deadline_seconds=ack_deadline,
+                retain_acked_messages=retain_acked_messages,
+                message_retention_duration=message_retention_duration)
         except GaxError as exc:
             if exc_to_code(exc.cause) == StatusCode.FAILED_PRECONDITION:
                 raise Conflict(topic_path)

pubsub/google/cloud/pubsub/_http.py

@@ -20,6 +20,7 @@
 import os
 
 from google.cloud import _http
+from google.cloud._helpers import _timedelta_to_duration_pb
 from google.cloud.environment_vars import PUBSUB_EMULATOR
 from google.cloud.iterator import HTTPIterator
 
@@ -296,7 +297,9 @@ def list_subscriptions(self, project, page_size=None, page_token=None):
             extra_params=extra_params)
 
     def subscription_create(self, subscription_path, topic_path,
-                            ack_deadline=None, push_endpoint=None):
+                            ack_deadline=None, push_endpoint=None,
+                            retain_acked_messages=None,
+                            message_retention_duration=None):
         """API call:  create a subscription
 
         See:
@@ -322,6 +325,18 @@ def subscription_create(self, subscription_path, topic_path,
             (Optional) URL to which messages will be pushed by the back-end.
             If not set, the application must pull messages.
 
+        :type retain_acked_messages: bool
+        :param retain_acked_messages:
+            (Optional) Whether to retain acked messages. If set, acked messages
+            are retained in the subscription's backlog for a duration indicated
+            by `message_retention_duration`.
+
+        :type message_retention_duration: :class:`datetime.timedelta`
+        :param message_retention_duration:
+            (Optional) Whether to retain acked messages. If set, acked messages
+            are retained in the subscription's backlog for a duration indicated
+            by `message_retention_duration`. If unset, defaults to 7 days.
+
         :rtype: dict
         :returns: ``Subscription`` resource returned from the API.
         """
@@ -334,6 +349,16 @@ def subscription_create(self, subscription_path, topic_path,
         if push_endpoint is not None:
             resource['pushConfig'] = {'pushEndpoint': push_endpoint}
 
+        if retain_acked_messages is not None:
+            resource['retainAckedMessages'] = retain_acked_messages
+
+        if message_retention_duration is not None:
+            pb = _timedelta_to_duration_pb(message_retention_duration)
+            resource['messageRetentionDuration'] = {
+                'seconds': pb.seconds,
+                'nanos': pb.nanos
+            }
+
         return self.api_request(method='PUT', path=path, data=resource)
 
     def subscription_get(self, subscription_path):

pubsub/google/cloud/pubsub/iam.py

@@ -121,7 +121,7 @@ def publishers(self, value):
             _ASSIGNMENT_DEPRECATED_MSG.format(
                 'publishers', PUBSUB_PUBLISHER_ROLE),
             DeprecationWarning)
-        self._bindings[PUBSUB_PUBLISHER_ROLE] = list(value)
+        self[PUBSUB_PUBLISHER_ROLE] = value
 
     @property
     def subscribers(self):
@@ -135,4 +135,4 @@ def subscribers(self, value):
             _ASSIGNMENT_DEPRECATED_MSG.format(
                 'subscribers', PUBSUB_SUBSCRIBER_ROLE),
             DeprecationWarning)
-        self._bindings[PUBSUB_SUBSCRIBER_ROLE] = list(value)
+        self[PUBSUB_SUBSCRIBER_ROLE] = value