feat: add send and ack mutations Cloud Spanner Queues

[finn-the-coder]

include unit and system tests

Fixes #17727

[gemini-code-assist]

Code Review

This pull request introduces send and ack methods to both the async and sync Batch classes to support Cloud Spanner queues, along with corresponding unit and system tests. The feedback highlights that the unit tests incorrectly compare proto-plus message fields directly to raw Python values, which should be resolved by comparing protobuf messages using helper functions. Additionally, the system tests should be optimized to use the shared_instance fixture instead of creating and deleting new Spanner instances, and they should utilize pytest.skip() to properly report skipped tests when queues are not supported.

[parthea]

==================================== ERRORS ====================================
_________________ ERROR at setup of test_db_batch_send_and_ack _________________
file /tmpfs/src/github/google-cloud-python/packages/google-cloud-spanner/tests/system/_async/test_database_api.py, line 90
  @pytest.mark.asyncio
  async def test_db_batch_send_and_ack(not_emulator, spanner_client, database_dialect, instance_config):
      import uuid
      from google.cloud.spanner_admin_instance_v1.types import spanner_instance_admin
      from google.cloud.spanner_admin_database_v1 import DatabaseDialect
      from google.api_core.exceptions import MethodNotImplemented, GoogleAPIError

      instance_id = f"test-instance-{uuid.uuid4().hex[:8]}"
      db_name = f"test-db-{uuid.uuid4().hex[:8]}"
      queue_name = f"test_queue_{uuid.uuid4().hex[:8]}"

      config_name = instance_config.name
      request = spanner_instance_admin.CreateInstanceRequest(
          parent=spanner_client.project_name,
          instance_id=instance_id,
          instance=spanner_instance_admin.Instance(
              config=config_name,
              display_name=instance_id,
              node_count=1,
              edition=spanner_instance_admin.Instance.Edition.ENTERPRISE,
          ),
      )
      print(f"instance creation request: {request}")
      operation = await spanner_client.instance_admin_api.create_instance(request=request)
      operation.result(600)

      test_instance = spanner_client.instance(instance_id, configuration_name=config_name)

      try:
          test_database = await test_instance.database(db_name, database_dialect=database_dialect)
          operation = await test_database.create()
          operation.result(300)
          print("Database created successfully!")

          try:
              # 3. Create the Queue
              if database_dialect == DatabaseDialect.POSTGRESQL:
                  queue_ddl = f"""CREATE QUEUE {queue_name} (
                      id bigint NOT NULL,
                      "Payload" varchar NOT NULL,
                      PRIMARY KEY (id)
                  )"""
              else:
                  queue_ddl = f"""CREATE QUEUE {queue_name} (
                      Id INT64 NOT NULL,
                      Payload STRING(MAX) NOT NULL
                  ) PRIMARY KEY (Id)"""

              try:
                  operation = await test_database.update_ddl([queue_ddl])
                  await operation.result(600)
              except MethodNotImplemented as e:
                  print(f"MethodNotImplemented. Skipping test because Queues are not implemented yet: {e}")
                  return
              except GoogleAPIError as e:
                  if getattr(e, 'code', None) == 501 or (getattr(e, 'grpc_status_code', None) and e.grpc_status_code.name == 'UNIMPLEMENTED') or "UNIMPLEMENTED" in str(e):
                      print(f"Skipping test because Queues are not implemented yet: {e}")
                      return
                  raise
              print("Queue created successfully.")

              # 4. Run mutations
              print("Sending message to queue...")
              async with test_database.batch() as batch:
                  batch.send(
                      queue=queue_name,
                      key=(2,),
                      payload="Hello, Queues!",
                  )
              print("Send successful.")

              print("Acking message in queue...")
              async with test_database.batch() as batch:
                  batch.ack(
                      queue=queue_name,
                      key=(2,),
                  )
              print("Ack successful.")

          finally:
              print("Dropping database...")
              await test_database.drop()
      finally:
          print("Dropping instance...")
          await test_instance.delete()
E       fixture 'not_emulator' not found
>       available fixtures: _asyncio_loop_factory, _class_scoped_runner, _function_scoped_runner, _module_scoped_runner, _package_scoped_runner, _session_scoped_runner, cache, capfd, capfdbinary, caplog, capsys, capsysbinary, capteesys, database_dialect, database_operation_timeout, databases_to_delete, doctest_namespace, event_loop_policy, instance_config, instance_configs, instance_operation_timeout, monkeypatch, not_postgres, proto_descriptor_file, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, reset_cached_apis, shared_database, shared_instance, shared_instance_id, spanner_client, subtests, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory, unused_tcp_port, unused_tcp_port_factory, unused_udp_port, unused_udp_port_factory
>       use 'pytest --fixtures [testpath]' for help on them.