review comments and tests

google/cloud/spanner_v1/client.py

@@ -169,6 +169,7 @@ class Client(ClientWithProject):
     :type default_transaction_options: :class:`~google.cloud.spanner_v1.TransactionOptions`
         or :class:`dict`
     :param default_transaction_options: (Optional) Default options to use for all read/write transactions.
+        Any fields other than `isolation_level` will be ignored.
 
     :raises: :class:`ValueError <exceptions.ValueError>` if both ``read_only``
              and ``admin`` are :data:`True`
@@ -507,5 +508,6 @@ def default_transaction_options(self, default_transaction_options):
         :type default_transaction_options: :class:`~google.cloud.spanner_v1.TransactionOptions`
             or :class:`dict`
         :param default_transaction_options: Default options to use for all read/write transactions.
+            Any fields other than `isolation_level` will be ignored.
         """
         self._default_transaction_options = default_transaction_options

google/cloud/spanner_v1/database.py

@@ -824,16 +824,7 @@ def batch(
 
         # Set isolation level
         if isolation_level is None:
-            isolation_level = (
-                self.default_transaction_options.get(
-                    "isolation_level",
-                    TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED,
-                )
-                if isinstance(self.default_transaction_options, dict)
-                else self.default_transaction_options.isolation_level
-                if isinstance(self.default_transaction_options, TransactionOptions)
-                else TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED
-            )
+            isolation_level = self.get_default_isolation_level()
         return BatchCheckout(
             self,
             request_options,
@@ -1154,6 +1145,23 @@ def set_iam_policy(self, policy):
         response = api.set_iam_policy(request=request, metadata=metadata)
         return response
 
+    def get_default_isolation_level(self):
+        """
+        Returns the isolation level set in default transaction options when creating
+        the SpannerClient.
+        """
+        if isinstance(self.default_transaction_options, dict):
+            return self.default_transaction_options.get(
+                "isolation_level",
+                TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED,
+            )
+
+        return getattr(
+            self.default_transaction_options,
+            "isolation_level",
+            TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED,
+        )
+
     @property
     def observability_options(self):
         """

google/cloud/spanner_v1/session.py

@@ -472,18 +472,7 @@ def run_in_transaction(self, func, *args, **kw):
         isolation_level = kw.pop("isolation_level", None)
 
         if isolation_level is None:
-            isolation_level = (
-                self._database.default_transaction_options.get(
-                    "isolation_level",
-                    TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED,
-                )
-                if isinstance(self._database.default_transaction_options, dict)
-                else self._database.default_transaction_options.isolation_level
-                if isinstance(
-                    self._database.default_transaction_options, TransactionOptions
-                )
-                else TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED
-            )
+            isolation_level = self._database.get_default_isolation_level()
         attempts = 0
 
         observability_options = getattr(self._database, "observability_options", None)

tests/unit/test_session.py

@@ -94,11 +94,31 @@ def _make_database(
         database.database_role = database_role
         database._route_to_leader_enabled = True
         database.default_transaction_options = default_transaction_options
+
+        # Define side_effect function to use injected default_transaction_options
+        def get_default_isolation_level_side_effect():
+            if isinstance(database.default_transaction_options, dict):
+                return database.default_transaction_options.get(
+                    "isolation_level",
+                    TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED,
+                )
+
+            return getattr(
+                database.default_transaction_options,
+                "isolation_level",
+                TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED,
+            )
+
+        # Mock get_default_isolation_level method
+        database.get_default_isolation_level = mock.Mock(
+            side_effect=get_default_isolation_level_side_effect
+        )
+
         return database
 
     @staticmethod
     def _make_session_pb(name, labels=None, database_role=None):
-        return Session(name=name, labels=labels, creator_role=database_role)
+        return SessionRequestProto(name=name, labels=labels, creator_role=database_role)
 
     def _make_spanner_api(self):
         return mock.Mock(autospec=SpannerClient, instance=True)