feat(bigtable): support requested_policy_version for Instance IAM

bigtable/google/cloud/bigtable/instance.py

@@ -23,7 +23,7 @@
 
 from google.protobuf import field_mask_pb2
 
-from google.cloud.bigtable_admin_v2.types import instance_pb2
+from google.cloud.bigtable_admin_v2.types import instance_pb2, options_pb2
 
 from google.api_core.exceptions import NotFound
 
@@ -434,7 +434,7 @@ def delete(self):
         """
         self._client.instance_admin_client.delete_instance(name=self.name)
 
-    def get_iam_policy(self):
+    def get_iam_policy(self, requested_policy_version=None):
         """Gets the access control policy for an instance resource.
 
         For example:
@@ -443,11 +443,30 @@ def get_iam_policy(self):
             :start-after: [START bigtable_get_iam_policy]
             :end-before: [END bigtable_get_iam_policy]
 
+        :type requested_policy_version: int or ``NoneType``
+        :param requested_policy_version: Optional. The version of IAM policies to request.
+                                         If a policy with a condition is requested without
+                                         setting this, the server will return an error.
+                                         This must be set to a value of 3 to retrieve IAM
+                                         policies containing conditions. This is to prevent
+                                         client code that isn't aware of IAM conditions from
+                                         interpreting and modifying policies incorrectly.
+                                         The service might return a policy with version lower
+                                         than the one that was requested, based on the
+                                         feature syntax in the policy fetched.
+
         :rtype: :class:`google.cloud.bigtable.policy.Policy`
         :returns: The current IAM policy of this instance
         """
+        args = {"resource": self.name}
+        if requested_policy_version is not None:
+            args["options_"] = options_pb2.GetPolicyOptions(
+                requested_policy_version=requested_policy_version
+            )
+
         instance_admin_client = self._client.instance_admin_client
-        resp = instance_admin_client.get_iam_policy(resource=self.name)
+
+        resp = instance_admin_client.get_iam_policy(**args)
         return Policy.from_pb(resp)
 
     def set_iam_policy(self, policy):

bigtable/tests/unit/test_instance.py

@@ -633,6 +633,44 @@ def test_get_iam_policy(self):
         for found, expected in zip(sorted(admins), sorted(members)):
             self.assertEqual(found, expected)
 
+    def test_get_iam_policy_w_requested_policy_version(self):
+        from google.cloud.bigtable_admin_v2.gapic import bigtable_instance_admin_client
+        from google.iam.v1 import policy_pb2, options_pb2
+        from google.cloud.bigtable.policy import BIGTABLE_ADMIN_ROLE
+
+        credentials = _make_credentials()
+        client = self._make_client(
+            project=self.PROJECT, credentials=credentials, admin=True
+        )
+        instance = self._make_one(self.INSTANCE_ID, client)
+
+        version = 1
+        etag = b"etag_v1"
+        members = ["serviceAccount:service_acc1@test.com", "user:user1@test.com"]
+        bindings = [{"role": BIGTABLE_ADMIN_ROLE, "members": members}]
+        iam_policy = policy_pb2.Policy(version=version, etag=etag, bindings=bindings)
+
+        # Patch the stub used by the API method.
+        instance_api = mock.create_autospec(
+            bigtable_instance_admin_client.BigtableInstanceAdminClient
+        )
+        client._instance_admin_client = instance_api
+        instance_api.get_iam_policy.return_value = iam_policy
+
+        # Perform the method and check the result.
+        result = instance.get_iam_policy(requested_policy_version=3)
+
+        instance_api.get_iam_policy.assert_called_once_with(
+            resource=instance.name,
+            options_=options_pb2.GetPolicyOptions(requested_policy_version=3),
+        )
+        self.assertEqual(result.version, version)
+        self.assertEqual(result.etag, etag)
+        admins = result.bigtable_admins
+        self.assertEqual(len(admins), len(members))
+        for found, expected in zip(sorted(admins), sorted(members)):
+            self.assertEqual(found, expected)
+
     def test_set_iam_policy(self):
         from google.cloud.bigtable_admin_v2.gapic import bigtable_instance_admin_client
         from google.iam.v1 import policy_pb2