tests: move session API systests to own module

Refactor to use pytest fixtures / idioms, rather than old 'Config'
setup/ teardown.

Toward #472.

tests/system/_helpers.py

@@ -12,9 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import operator
 import os
 import time
 
+import grpc
+from google.rpc import code_pb2
 from google.api_core import exceptions
 from google.cloud.spanner_v1 import instance as instance_mod
 from tests import _fixtures
@@ -48,11 +51,23 @@
     _fixtures.EMULATOR_DDL_STATEMENTS if USE_EMULATOR else _fixtures.DDL_STATEMENTS
 )
 
+retry_true = retry.RetryResult(operator.truth)
+retry_false = retry.RetryResult(operator.not_)
 
 retry_503 = retry.RetryErrors(exceptions.ServiceUnavailable)
 retry_429_503 = retry.RetryErrors(
     exceptions.TooManyRequests, exceptions.ServiceUnavailable,
 )
+retry_mabye_aborted_txn = retry.RetryErrors(exceptions.ServerError, exceptions.Aborted)
+retry_mabye_conflict = retry.RetryErrors(exceptions.ServerError, exceptions.Conflict)
+
+
+def _has_all_ddl(database):
+    # Predicate to test for EC completion.
+    return len(database.ddl_statements) == len(DDL_STATEMENTS)
+
+
+retry_has_all_dll = retry.RetryInstanceState(_has_all_ddl)
 
 
 def scrub_instance_backups(to_scrub):
@@ -93,9 +108,32 @@ def unique_id(prefix, separator="-"):
     return f"{prefix}{system.unique_resource_id(separator)}"
 
 
-def _has_all_ddl(database):
-    # Predicate to test for EC completion.
-    return len(database.ddl_statements) == len(DDL_STATEMENTS)
+class FauxCall:
+    def __init__(self, code, details="FauxCall"):
+        self._code = code
+        self._details = details
 
+    def initial_metadata(self):
+        return {}
 
-retry_has_all_dll = retry.RetryInstanceState(_has_all_ddl)
+    def trailing_metadata(self):
+        return {}
+
+    def code(self):
+        return self._code
+
+    def details(self):
+        return self._details
+
+
+def _check_batch_status(status_code, expected=code_pb2.OK):
+    if status_code != expected:
+
+        _status_code_to_grpc_status_code = {
+            member.value[0]: member for member in grpc.StatusCode
+        }
+        grpc_status_code = _status_code_to_grpc_status_code[status_code]
+        call = FauxCall(status_code)
+        raise exceptions.from_grpc_status(
+            grpc_status_code, "batch_update failed", errors=[call]
+        )

tests/system/_sample_data.py

@@ -53,18 +53,20 @@ def _assert_timestamp(value, nano_value):
         assert value.microsecond * 1000 == nano_value.nanosecond
 
 
-def _check_rows_data(rows_data, expected=ROW_DATA):
+def _check_rows_data(rows_data, expected=ROW_DATA, recurse_into_lists=True):
     assert len(rows_data) == len(expected)
 
     for row, expected in zip(rows_data, expected):
-        _check_row_data(row, expected)
+        _check_row_data(row, expected, recurse_into_lists=recurse_into_lists)
 
 
-def _check_row_data(row_data, expected):
+def _check_row_data(row_data, expected, recurse_into_lists=True):
     assert len(row_data) == len(expected)
 
     for found_cell, expected_cell in zip(row_data, expected):
-        _check_cell_data(found_cell, expected_cell)
+        _check_cell_data(
+            found_cell, expected_cell, recurse_into_lists=recurse_into_lists
+        )
 
 
 def _check_cell_data(found_cell, expected_cell, recurse_into_lists=True):