feat: add encryption_info to Database (#284)

* feat: add encryption_info to Database * test: fix test name Co-authored-by: larkee <larkee@users.noreply.github.com>

Author: larkee

google/cloud/spanner_v1/database.py

@@ -142,6 +142,7 @@ def __init__(
  self._restore_info = None
  self._version_retention_period = None
  self._earliest_version_time = None
+ self._encryption_info = None
  self.log_commit_stats = False
  self._logger = logger
  self._encryption_config = encryption_config
@@ -269,6 +270,14 @@ def encryption_config(self):
  """
  return self._encryption_config
 
+ @property
+ def encryption_info(self):
+ """Encryption info for this database.
+ :rtype: a list of :class:`~google.cloud.spanner_admin_instance_v1.types.EncryptionInfo`
+ :returns: a list of objects representing encryption info for this database
+ """
+ return self._encryption_info
+
  @property
  def ddl_statements(self):
  """DDL Statements used to define database schema.
@@ -403,6 +412,7 @@ def reload(self):
  self._version_retention_period = response.version_retention_period
  self._earliest_version_time = response.earliest_version_time
  self._encryption_config = response.encryption_config
+ self._encryption_info = response.encryption_info
 
  def update_ddl(self, ddl_statements, operation_id=""):
  """Update DDL for this database.

tests/unit/test_database.py

@@ -318,6 +318,17 @@ def test_encryption_config(self):
  )
  self.assertEqual(database.encryption_config, encryption_config)
 
+ def test_encryption_info(self):
+ from google.cloud.spanner_admin_database_v1 import EncryptionInfo
+
+ instance = _Instance(self.INSTANCE_NAME)
+ pool = _Pool()
+ database = self._make_one(self.DATABASE_ID, instance, pool=pool)
+ encryption_info = database._encryption_info = [
+ mock.create_autospec(EncryptionInfo, instance=True)
+ ]
+ self.assertEqual(database.encryption_info, encryption_info)
+
  def test_spanner_api_property_w_scopeless_creds(self):
 
  client = _Client()
@@ -682,6 +693,7 @@ def test_reload_not_found(self):
  def test_reload_success(self):
  from google.cloud.spanner_admin_database_v1 import Database
  from google.cloud.spanner_admin_database_v1 import EncryptionConfig
+ from google.cloud.spanner_admin_database_v1 import EncryptionInfo
  from google.cloud.spanner_admin_database_v1 import GetDatabaseDdlResponse
  from google.cloud.spanner_admin_database_v1 import RestoreInfo
  from google.cloud._helpers import _datetime_to_pb_timestamp
@@ -693,6 +705,12 @@ def test_reload_success(self):
  client = _Client()
  ddl_pb = GetDatabaseDdlResponse(statements=DDL_STATEMENTS)
  encryption_config = EncryptionConfig(kms_key_name="kms_key")
+ encryption_info = [
+ EncryptionInfo(
+ encryption_type=EncryptionInfo.Type.CUSTOMER_MANAGED_ENCRYPTION,
+ kms_key_version="kms_key_version",
+ )
+ ]
  api = client.database_admin_api = self._make_database_admin_api()
  api.get_database_ddl.return_value = ddl_pb
  db_pb = Database(
@@ -702,6 +720,7 @@ def test_reload_success(self):
  version_retention_period="1d",
  earliest_version_time=_datetime_to_pb_timestamp(timestamp),
  encryption_config=encryption_config,
+ encryption_info=encryption_info,
  )
  api.get_database.return_value = db_pb
  instance = _Instance(self.INSTANCE_NAME, client=client)
@@ -716,6 +735,7 @@ def test_reload_success(self):
  self.assertEqual(database._earliest_version_time, timestamp)
  self.assertEqual(database._ddl_statements, tuple(DDL_STATEMENTS))
  self.assertEqual(database._encryption_config, encryption_config)
+ self.assertEqual(database._encryption_info, encryption_info)
 
  api.get_database_ddl.assert_called_once_with(
  database=self.DATABASE_NAME,