googleapis · efevans · Aug 18, 2026 · Sep 3, 2026 · Sep 8, 2026 · Sep 8, 2026
@@ -0,0 +1,65 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using Google.Cloud.Spanner.V1.Internal.Logging;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Google.Cloud.Spanner.Data.CommonTesting
+{
+    /// <summary>
+    /// Base classes for test fixtures for queues.
+    /// </summary>
+    public abstract class SpannerQueueFixture : SpannerFixtureBase, IAsyncLifetime
+    {
+        public string QueueName { get; }
+
+        public SpannerQueueFixture(string queueName)
+        {
+            QueueName = queueName;
+        }
+
+        /// <summary>
+        /// Creates the queue. This method is only called when a new database has been created.
+        /// </summary>
+        protected abstract Task CreateQueue();
+
+        protected async Task ExecuteDdl(string ddl)
+        {
+            using var connection = GetConnection();
+            _ = await connection.CreateDdlCommand(ddl).ExecuteNonQueryAsync();
+        }
+
+        public override void Dispose()
+        {
+            base.Dispose();
+            RetryHelpers.MaybeLogStats($"Disposal of fixture for {QueueName}");
+        }
+
+        public async Task InitializeAsync()
+        {
+            if (Database.Fresh)
+            {
+                Logger.DefaultLogger.Debug($"Creating queue {QueueName}");
+                await CreateQueue();
+            }
+            RetryHelpers.ResetStats();
+            Logger.DefaultLogger.Debug($"Ready to run tests");
+            RetryHelpers.MaybeLogStats($"Population of {QueueName}");
+            RetryHelpers.ResetStats();
+        }
+
+        public Task DisposeAsync() => Task.CompletedTask;
+    }
+}
@@ -0,0 +1,36 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using Google.Cloud.Spanner.Data.CommonTesting;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Google.Cloud.Spanner.Data.IntegrationTests
+{
+    [CollectionDefinition(nameof(MutationsQueueFixture))]
+    public class MutationsQueueFixture : SpannerQueueFixture, ICollectionFixture<MutationsQueueFixture>
+    {
+        public MutationsQueueFixture() : base("DmlTest")
+        {
+        }
+
+        protected override async Task CreateQueue() => await ExecuteDdl(
+            $@"CREATE QUEUE {QueueName} (
+                UserId        STRING(100) NOT NULL,
+                MessageId     STRING(100) NOT NULL,
+                Payload       BYTES(MAX) NOT NULL,
+                ) PRIMARY KEY(UserId, MessageId),
+                OPTIONS(receive_mode= ""PULL"")");
+    }
+}
@@ -0,0 +1,158 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License"):
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using Google.Cloud.ClientTesting;
+using Google.Cloud.Spanner.Data.CommonTesting;
+using System;
+using System.Diagnostics;
+using System.Text;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Google.Cloud.Spanner.Data.IntegrationTests;
+
+[Collection(nameof(MutationsQueueFixture))]
+public class QueueTests
+{
+    private readonly MutationsQueueFixture _queueFixture;
+
+    public QueueTests(MutationsQueueFixture dmlQueueFixture) => _queueFixture = dmlQueueFixture;
+
+    private readonly byte[] _payloadBytes = Encoding.UTF8.GetBytes("Hello, World");
+    private static int DeliveryDelay => 10;
+    private static Func<int, DateTime> GetTimeFromNow => (int delay) => DateTime.UtcNow.AddSeconds(delay);
+    private static SendOptions DeliverAfterDelayFromNow => new() { DeliverAt = GetTimeFromNow(DeliveryDelay) };
+
+    [Trait(Constants.SupportedOnEmulator, Constants.No)]
+    [Fact]
+    public async Task QueueIsLeftWithNoMessagesAfterAck_Basic()
+    {
+        using var connection = _queueFixture.GetConnection();
+        (string userId, string messageId) = (IdGenerator.FromGuid(), IdGenerator.FromGuid());
+
+        // Send Message
+        using var sendCommand = connection.CreateSendCommand(_queueFixture.QueueName, ParametersForKeyAndPayload(userId, messageId, _payloadBytes));
+        await sendCommand.ExecuteNonQueryAsync();
+
+        // Ack messages
+        var ackCommand = connection.CreateAckCommand(_queueFixture.QueueName, ParametersForKey(userId, messageId));
+        await ackCommand.ExecuteNonQueryAsync();
+
+        // Queue is left with no messages after Ack
+        var selectCommand = connection.CreateSelectCommand(_queueFixture.QueueName);
+        selectCommand.CommandText = $"SELECT COUNT(*) FROM {_queueFixture.QueueName};";
+        var count = await selectCommand.ExecuteScalarAsync();
+
+        Assert.Equal(0L, (long) count);
+    }
+
+    [Trait(Constants.SupportedOnEmulator, Constants.No)]
+    [Fact]
+    public async Task QueueIsLeftWithNoMessagesAfterAck_DeliveryTimeSpecified_Streamed()
+    {
+        using var connection = _queueFixture.GetConnection();
+        (string userId, string messageId) = (IdGenerator.FromGuid(), IdGenerator.FromGuid());
+
+        // Send Message
+        using var sendCommand = connection.CreateSendCommand(_queueFixture.QueueName, ParametersForKeyAndPayload(userId, messageId, _payloadBytes));
+        sendCommand.SendOptions = DeliverAfterDelayFromNow;
+        await sendCommand.ExecuteNonQueryAsync();
+        Stopwatch sw = Stopwatch.StartNew();
+
+        using var receiveCommand = connection.CreateSelectCommand($"SELECT UserId, MessageId FROM RECEIVE_{_queueFixture.QueueName}(max_duration => '15s')");
+        using var reader = await receiveCommand.ExecuteReaderAsync();
+
+        Assert.True(await reader.ReadAsync());
+        // Add a buffer to compensate for the stopwatch starting after we get the response back
+        int adjustedDeliveryDelay = DeliveryDelay - 1;
+        Assert.True(sw.Elapsed.TotalSeconds > adjustedDeliveryDelay, $"Expected to receive message after {adjustedDeliveryDelay} seconds, instead was {sw.Elapsed.TotalSeconds}");
+
+        // Clean up the lingering message
+        var ackCommand = connection.CreateAckCommand(_queueFixture.QueueName, ParametersForKey(userId, messageId));
+        await ackCommand.ExecuteNonQueryAsync();
+
+    }
+
+    [Trait(Constants.SupportedOnEmulator, Constants.No)]
+    [Fact]
+    public async Task QueueIsLeftWithNoMessagesAfterAck_Streaming()
+    {
+        using var connection = _queueFixture.GetConnection();
+
+        // Send Messages
+        for (long i = 0; i < 10; i++)
+        {
+            string userId = IdGenerator.FromGuid();
+            string messageId = IdGenerator.FromGuid();
+            using var sendCommand = connection.CreateSendCommand(_queueFixture.QueueName, ParametersForKeyAndPayload(userId, messageId, _payloadBytes));
+            await sendCommand.ExecuteNonQueryAsync();
+        }
+
+        using var receiveCommand = connection.CreateSelectCommand($"SELECT UserId, MessageId FROM RECEIVE_{_queueFixture.QueueName}(max_duration => '10s')");
+        using (var reader = await receiveCommand.ExecuteReaderAsync())
+        {
+            while (await reader.ReadAsync())
+            {
+                string userId = reader.GetFieldValue<string>("UserId");
+                string messageId = reader.GetFieldValue<string>("MessageId");
+
+                var ackCommand = connection.CreateAckCommand(_queueFixture.QueueName, ParametersForKey(userId, messageId));
+                await ackCommand.ExecuteNonQueryAsync();
+            }
+        }
+
+        // Queue is left with no messages after Ack
+        var selectCommand = connection.CreateSelectCommand(_queueFixture.QueueName);
+        selectCommand.CommandText = $"SELECT COUNT(*) FROM {_queueFixture.QueueName};";
+        var count = await selectCommand.ExecuteScalarAsync();
+
+        Assert.Equal(0L, (long) count);
+    }
+
+    [Trait(Constants.SupportedOnEmulator, Constants.No)]
+    [Fact]
+    public async Task AckAMissingMessage_IgnoreNotFound_False_ThrowsException()
+    {
+        using var connection = _queueFixture.GetConnection();
+
+        // Ack missing messages
+        var ackCommand = connection.CreateAckCommand(_queueFixture.QueueName, ParametersForKey("Roger", "Federer"));
+        ackCommand.AckOptions = new() { IgnoreNotFound = false };
+        await Assert.ThrowsAsync<SpannerException>(ackCommand.ExecuteNonQueryAsync);
+    }
+
+    [Trait(Constants.SupportedOnEmulator, Constants.No)]
+    [Fact]
+    public async Task AckAMissingMessage_IgnoreFound_True_Ok()
+    {
+        using var connection = _queueFixture.GetConnection();
+
+        // Ack missing messages
+        var ackCommand = connection.CreateAckCommand(_queueFixture.QueueName, ParametersForKey("Rafael", "Nadal"));
+        ackCommand.AckOptions = new() { IgnoreNotFound = true };
+        await ackCommand.ExecuteNonQueryAsync();
+    }
+
+    private static SpannerParameterCollection ParametersForKey(string str1, string str2)
+        => new([
+            new SpannerParameter("UserId", SpannerDbType.String, value: str1),
+            new SpannerParameter("MessageId", SpannerDbType.String, value: str2),
+        ]);
+
+    private static SpannerParameter PayloadParameterForBytes(byte[] bytes)
+        => new("Payload", SpannerDbType.Bytes, bytes);
+
+    private static SpannerParameterCollection ParametersForKeyAndPayload(string str1, string str2, byte[] bytes)
+        => [.. ParametersForKey(str1, str2), PayloadParameterForBytes(bytes)];
+}