Support passing 'uniqueWriterIdentity' to 'Sink.update'. (#4708)

Also, capture server-set read-only properties returned from 'sinks.update',
such as 'writerIdentity'.

Author: tseaver

google/cloud/logging/_gax.py

@@ -265,7 +265,8 @@ def sink_get(self, project, sink_name):
         #       so `MessageToDict`` can safely be used.
         return MessageToDict(sink_pb)
 
-    def sink_update(self, project, sink_name, filter_, destination):
+    def sink_update(self, project, sink_name, filter_, destination,
+                    unique_writer_identity=False):
         """API call:  update a sink resource.
 
         :type project: str
@@ -282,15 +283,24 @@ def sink_update(self, project, sink_name, filter_, destination):
         :param destination: destination URI for the entries exported by
                             the sink.
 
+        :type unique_writer_identity: bool
+        :param unique_writer_identity: (Optional) determines the kind of
+                                       IAM identity returned as
+                                       writer_identity in the new sink.
+
         :rtype: dict
-        :returns: The sink object returned from the API (converted from a
+        :returns: The sink resource returned from the API (converted from a
                   protobuf to a dictionary).
         """
         options = None
         path = 'projects/%s/sinks/%s' % (project, sink_name)
         sink_pb = LogSink(name=path, filter=filter_, destination=destination)
         try:
-            sink_pb = self._gax_api.update_sink(path, sink_pb, options=options)
+            sink_pb = self._gax_api.update_sink(
+                path,
+                sink_pb,
+                unique_writer_identity=unique_writer_identity,
+                options=options)
         except GaxError as exc:
             if exc_to_code(exc.cause) == StatusCode.NOT_FOUND:
                 raise NotFound(path)

google/cloud/logging/_http.py

@@ -290,7 +290,8 @@ def sink_get(self, project, sink_name):
         target = '/projects/%s/sinks/%s' % (project, sink_name)
         return self.api_request(method='GET', path=target)
 
-    def sink_update(self, project, sink_name, filter_, destination):
+    def sink_update(self, project, sink_name, filter_, destination,
+                    unique_writer_identity=False):
         """API call:  update a sink resource.
 
         See
@@ -310,6 +311,11 @@ def sink_update(self, project, sink_name, filter_, destination):
         :param destination: destination URI for the entries exported by
                             the sink.
 
+        :type unique_writer_identity: bool
+        :param unique_writer_identity: (Optional) determines the kind of
+                                       IAM identity returned as
+                                       writer_identity in the new sink.
+
         :rtype: dict
         :returns: The returned (updated) resource.
         """
@@ -319,7 +325,9 @@ def sink_update(self, project, sink_name, filter_, destination):
             'filter': filter_,
             'destination': destination,
         }
-        return self.api_request(method='PUT', path=target, data=data)
+        query_params = {'uniqueWriterIdentity': unique_writer_identity}
+        return self.api_request(
+            method='PUT', path=target, query_params=query_params, data=data)
 
     def sink_delete(self, project, sink_name):
         """API call:  delete a sink resource.

google/cloud/logging/sink.py

@@ -175,7 +175,7 @@ def reload(self, client=None):
         resource = client.sinks_api.sink_get(self.project, self.name)
         self._update_from_api_repr(resource)
 
-    def update(self, client=None):
+    def update(self, client=None, unique_writer_identity=False):
         """API call:  update sink configuration via a PUT request
 
         See
@@ -185,10 +185,18 @@ def update(self, client=None):
                       ``NoneType``
         :param client: the client to use.  If not passed, falls back to the
                        ``client`` stored on the current sink.
+
+        :type unique_writer_identity: bool
+        :param unique_writer_identity: (Optional) determines the kind of
+                                    IAM identity returned as
+                                    writer_identity in the new sink.
         """
         client = self._require_client(client)
-        client.sinks_api.sink_update(
-            self.project, self.name, self.filter_, self.destination)
+        resource = client.sinks_api.sink_update(
+            self.project, self.name, self.filter_, self.destination,
+            unique_writer_identity=unique_writer_identity,
+        )
+        self._update_from_api_repr(resource)
 
     def delete(self, client=None):
         """API call:  delete a sink via a DELETE request

tests/unit/test__gax.py

@@ -744,6 +744,12 @@ def test_sink_create_ok(self):
         from google.cloud.proto.logging.v2.logging_config_pb2 import LogSink
 
         gax_api = _GAXSinksAPI()
+        gax_api._create_sink_response = LogSink(
+            name=self.SINK_NAME,
+            destination=self.DESTINATION_URI,
+            filter=self.FILTER,
+            writer_identity=self.SINK_WRITER_IDENTITY,
+        )
         api = self._make_one(gax_api, None)
 
         returned = api.sink_create(
@@ -824,6 +830,7 @@ def test_sink_update_error(self):
 
     def test_sink_update_miss(self):
         from google.cloud.exceptions import NotFound
+        from google.cloud.proto.logging.v2.logging_config_pb2 import LogSink
 
         gax_api = _GAXSinksAPI()
         api = self._make_one(gax_api, None)
@@ -833,25 +840,50 @@ def test_sink_update_miss(self):
                 self.PROJECT, self.SINK_NAME, self.FILTER,
                 self.DESTINATION_URI)
 
+        sink_name, sink, unique_writer_identity, options = (
+            gax_api._update_sink_called_with)
+        self.assertEqual(sink_name, self.SINK_PATH)
+        self.assertIsInstance(sink, LogSink)
+        self.assertEqual(sink.name, self.SINK_PATH)
+        self.assertEqual(sink.filter, self.FILTER)
+        self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertFalse(unique_writer_identity)
+        self.assertIsNone(options)
+
     def test_sink_update_hit(self):
         from google.cloud.proto.logging.v2.logging_config_pb2 import LogSink
 
-        response = LogSink(name=self.SINK_NAME,
-                           destination=self.DESTINATION_URI,
-                           filter=self.FILTER)
+        response = LogSink(
+            name=self.SINK_NAME,
+            destination=self.DESTINATION_URI,
+            filter=self.FILTER,
+            writer_identity=Test_SinksAPI.SINK_WRITER_IDENTITY,
+        )
         gax_api = _GAXSinksAPI(_update_sink_response=response)
         api = self._make_one(gax_api, None)
 
-        api.sink_update(
-            self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI)
+        returned = api.sink_update(
+            self.PROJECT,
+            self.SINK_NAME,
+            self.FILTER,
+            self.DESTINATION_URI,
+            unique_writer_identity=True)
 
-        sink_name, sink, options = (
+        self.assertEqual(returned, {
+            'name': self.SINK_NAME,
+            'filter': self.FILTER,
+            'destination': self.DESTINATION_URI,
+            'writerIdentity': self.SINK_WRITER_IDENTITY,
+        })
+
+        sink_name, sink, unique_writer_identity, options = (
             gax_api._update_sink_called_with)
         self.assertEqual(sink_name, self.SINK_PATH)
         self.assertIsInstance(sink, LogSink)
         self.assertEqual(sink.name, self.SINK_PATH)
         self.assertEqual(sink.filter, self.FILTER)
         self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertTrue(unique_writer_identity)
         self.assertIsNone(options)
 
     def test_sink_delete_error(self):
@@ -1497,12 +1529,7 @@ def create_sink(self, parent, sink, unique_writer_identity, options):
             raise GaxError('error')
         if self._create_sink_conflict:
             raise GaxError('conflict', self._make_grpc_failed_precondition())
-        return LogSink(
-            name=sink.name,
-            destination=sink.destination,
-            filter=sink.filter,
-            writer_identity=Test_SinksAPI.SINK_WRITER_IDENTITY,
-        )
+        return self._create_sink_response
 
     def get_sink(self, sink_name, options):
         from google.gax.errors import GaxError
@@ -1515,10 +1542,11 @@ def get_sink(self, sink_name, options):
         except AttributeError:
             raise GaxError('notfound', self._make_grpc_not_found())
 
-    def update_sink(self, sink_name, sink, options=None):
+    def update_sink(self, sink_name, sink, unique_writer_identity, options):
         from google.gax.errors import GaxError
 
-        self._update_sink_called_with = sink_name, sink, options
+        self._update_sink_called_with = (
+            sink_name, sink, unique_writer_identity, options)
         if self._random_gax_error:
             raise GaxError('error')
         try:

tests/unit/test__http.py

@@ -527,7 +527,7 @@ def test_sink_get_hit(self):
     def test_sink_update_miss(self):
         from google.cloud.exceptions import NotFound
 
-        SENT = {
+        sent = {
             'name': self.SINK_NAME,
             'filter': self.FILTER,
             'destination': self.DESTINATION_URI,
@@ -541,28 +541,43 @@ def test_sink_update_miss(self):
                 self.PROJECT, self.SINK_NAME, self.FILTER,
                 self.DESTINATION_URI)
 
-        self.assertEqual(conn._called_with['method'], 'PUT')
         path = '/projects/%s/sinks/%s' % (self.PROJECT, self.SINK_NAME)
-        self.assertEqual(conn._called_with['path'], path)
-        self.assertEqual(conn._called_with['data'], SENT)
+        expected = {
+            'method': 'PUT',
+            'path': path,
+            'data': sent,
+            'query_params': {'uniqueWriterIdentity': False},
+        }
+        self.assertEqual(conn._called_with, expected)
 
     def test_sink_update_hit(self):
-        SENT = {
+        sent = {
             'name': self.SINK_NAME,
             'filter': self.FILTER,
             'destination': self.DESTINATION_URI,
         }
-        conn = _Connection({})
+        after_update = sent.copy()
+        after_update['writerIdentity'] = self.WRITER_IDENTITY
+        conn = _Connection(after_update)
         client = _Client(conn)
         api = self._make_one(client)
 
-        api.sink_update(
-            self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI)
+        returned = api.sink_update(
+            self.PROJECT,
+            self.SINK_NAME,
+            self.FILTER,
+            self.DESTINATION_URI,
+            unique_writer_identity=True)
 
-        self.assertEqual(conn._called_with['method'], 'PUT')
+        self.assertEqual(returned, after_update)
         path = '/projects/%s/sinks/%s' % (self.PROJECT, self.SINK_NAME)
-        self.assertEqual(conn._called_with['path'], path)
-        self.assertEqual(conn._called_with['data'], SENT)
+        expected = {
+            'method': 'PUT',
+            'path': path,
+            'data': sent,
+            'query_params': {'uniqueWriterIdentity': True},
+        }
+        self.assertEqual(conn._called_with, expected)
 
     def test_sink_delete_miss(self):
         from google.cloud.exceptions import NotFound

tests/unit/test_sink.py

@@ -231,29 +231,61 @@ def test_reload_w_alternate_client(self):
     def test_update_w_bound_client(self):
         client = _Client(project=self.PROJECT)
         api = client.sinks_api = _DummySinksAPI()
+        api._sink_update_response = {
+            'name': self.SINK_NAME,
+            'filter': self.FILTER,
+            'destination': self.DESTINATION_URI,
+            'writerIdentity': self.WRITER_IDENTITY,
+        }
         sink = self._make_one(self.SINK_NAME, self.FILTER,
                               self.DESTINATION_URI,
                               client=client)
 
         sink.update()
 
+        self.assertEqual(sink.name, self.SINK_NAME)
+        self.assertEqual(sink.filter_, self.FILTER)
+        self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertEqual(sink.writer_identity, self.WRITER_IDENTITY)
         self.assertEqual(
             api._sink_update_called_with,
-            (self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI))
+            (
+                self.PROJECT,
+                self.SINK_NAME,
+                self.FILTER,
+                self.DESTINATION_URI,
+                False,
+            ))
 
     def test_update_w_alternate_client(self):
         client1 = _Client(project=self.PROJECT)
         client2 = _Client(project=self.PROJECT)
         api = client2.sinks_api = _DummySinksAPI()
+        api._sink_update_response = {
+            'name': self.SINK_NAME,
+            'filter': self.FILTER,
+            'destination': self.DESTINATION_URI,
+            'writerIdentity': self.WRITER_IDENTITY,
+        }
         sink = self._make_one(self.SINK_NAME, self.FILTER,
                               self.DESTINATION_URI,
                               client=client1)
 
-        sink.update(client=client2)
+        sink.update(client=client2, unique_writer_identity=True)
 
+        self.assertEqual(sink.name, self.SINK_NAME)
+        self.assertEqual(sink.filter_, self.FILTER)
+        self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertEqual(sink.writer_identity, self.WRITER_IDENTITY)
         self.assertEqual(
             api._sink_update_called_with,
-            (self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI))
+            (
+                self.PROJECT,
+                self.SINK_NAME,
+                self.FILTER,
+                self.DESTINATION_URI,
+                True,
+            ))
 
     def test_delete_w_bound_client(self):
         client = _Client(project=self.PROJECT)
@@ -304,9 +336,11 @@ def sink_get(self, project, sink_name):
         except AttributeError:
             raise NotFound('miss')
 
-    def sink_update(self, project, sink_name, filter_, destination):
+    def sink_update(self, project, sink_name, filter_, destination,
+                    unique_writer_identity=False):
         self._sink_update_called_with = (
-            project, sink_name, filter_, destination)
+            project, sink_name, filter_, destination, unique_writer_identity)
+        return self._sink_update_response
 
     def sink_delete(self, project, sink_name):
         self._sink_delete_called_with = (project, sink_name)