Skip to content

Try to address some test flakes #1672

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Sep 12, 2024
Merged

Try to address some test flakes #1672

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions projects/Test/Common/TestConnectionRecoveryBase.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ await WithTemporaryNonExclusiveQueueAsync(m, async (_, q) =>
await m.QueueBindAsync(q, x, rk);
await m.BasicPublishAsync(x, rk, _messageBody);

Assert.True(await TestConnectionRecoveryBase.WaitForConfirmsWithCancellationAsync(m));
Assert.True(await WaitForConfirmsWithCancellationAsync(m));
await m.ExchangeDeclarePassiveAsync(x);
});
}
Expand Down Expand Up @@ -234,11 +234,11 @@ protected static TaskCompletionSource<bool> PrepareForShutdown(IConnection conn)
return tcs;
}

protected static Task<bool> WaitForConfirmsWithCancellationAsync(IChannel m)
protected static Task<bool> WaitForConfirmsWithCancellationAsync(IChannel channel)
{
using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(4)))
{
return m.WaitForConfirmsAsync(cts.Token);
return channel.WaitForConfirmsAsync(cts.Token);
}
}

Expand Down
2 changes: 0 additions & 2 deletions projects/Test/Integration/TestConcurrentAccessWithSharedChannel.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,8 +128,6 @@ await TestConcurrentOperationsAsync(async () =>

Assert.True(await tcs.Task);
}, Iterations);

_output.WriteLine("@@@@@@@@ PUBLISH ACK COUNT: {0}", publishAckCount);
}
}
}
15 changes: 9 additions & 6 deletions projects/Test/Integration/TestConnectionRecoveryWithoutSetup.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,6 +270,12 @@ public async Task TestCreateChannelOnClosedAutorecoveringConnectionDoesNotHang()
[Fact]
public async Task TestTopologyRecoveryConsumerFilter()
{
const string exchange = "topology.recovery.exchange";
const string queueWithRecoveredConsumer = "topology.recovery.queue.1";
const string queueWithIgnoredConsumer = "topology.recovery.queue.2";
const string binding1 = "recovered.binding.1";
const string binding2 = "recovered.binding.2";

var filter = new TopologyRecoveryFilter
{
ConsumerFilter = consumer => !consumer.ConsumerTag.Contains("filtered")
Expand All @@ -280,17 +286,13 @@ public async Task TestTopologyRecoveryConsumerFilter()
using (AutorecoveringConnection conn = await CreateAutorecoveringConnectionWithTopologyRecoveryFilterAsync(filter))
{
conn.RecoverySucceeded += (source, ea) => connectionRecoveryTcs.SetResult(true);
conn.ConnectionRecoveryError += (source, ea) => connectionRecoveryTcs.SetException(ea.Exception);
conn.CallbackException += (source, ea) => connectionRecoveryTcs.SetException(ea.Exception);

using (IChannel ch = await conn.CreateChannelAsync())
{
await ch.ConfirmSelectAsync();

string exchange = "topology.recovery.exchange";
string queueWithRecoveredConsumer = "topology.recovery.queue.1";
string queueWithIgnoredConsumer = "topology.recovery.queue.2";
string binding1 = "recovered.binding.1";
string binding2 = "recovered.binding.2";

await ch.ExchangeDeclareAsync(exchange, "direct");
await ch.QueueDeclareAsync(queueWithRecoveredConsumer, false, false, false);
await ch.QueueDeclareAsync(queueWithIgnoredConsumer, false, false, false);
Expand Down Expand Up @@ -325,6 +327,7 @@ public async Task TestTopologyRecoveryConsumerFilter()
Assert.True(ch.IsOpen);
await ch.BasicPublishAsync(exchange, binding1, _encoding.GetBytes("test message"));
await ch.BasicPublishAsync(exchange, binding2, _encoding.GetBytes("test message"));
await WaitForConfirmsWithCancellationAsync(ch);

await consumerRecoveryTcs.Task.WaitAsync(TimeSpan.FromSeconds(5));
Assert.True(await consumerRecoveryTcs.Task);
Expand Down
31 changes: 16 additions & 15 deletions projects/Test/Integration/TestConnectionTopologyRecovery.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,21 +228,24 @@ public async Task TestTopologyRecoveryBindingFilter()
[Fact]
public async Task TestTopologyRecoveryDefaultFilterRecoversAllEntities()
{
const string exchange = "topology.recovery.exchange";
const string queue1 = "topology.recovery.queue.1";
const string queue2 = "topology.recovery.queue.2";
const string binding1 = "recovered.binding";
const string binding2 = "filtered.binding";

var connectionRecoveryTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
var filter = new TopologyRecoveryFilter();
AutorecoveringConnection conn = await CreateAutorecoveringConnectionWithTopologyRecoveryFilterAsync(filter);
conn.RecoverySucceeded += (source, ea) => connectionRecoveryTcs.SetResult(true);
conn.ConnectionRecoveryError += (source, ea) => connectionRecoveryTcs.SetException(ea.Exception);
conn.CallbackException += (source, ea) => connectionRecoveryTcs.SetException(ea.Exception);

IChannel ch = await conn.CreateChannelAsync();
try
{
await ch.ConfirmSelectAsync();

string exchange = "topology.recovery.exchange";
string queue1 = "topology.recovery.queue.1";
string queue2 = "topology.recovery.queue.2";
string binding1 = "recovered.binding";
string binding2 = "filtered.binding";

await ch.ExchangeDeclareAsync(exchange, "direct");
await ch.QueueDeclareAsync(queue1, false, false, false);
await ch.QueueDeclareAsync(queue2, false, false, false);
Expand Down Expand Up @@ -281,16 +284,14 @@ public async Task TestTopologyRecoveryDefaultFilterRecoversAllEntities()
await ch.QueueDeclarePassiveAsync(queue1);
await ch.QueueDeclarePassiveAsync(queue2);

await ch.BasicPublishAsync(exchange, binding1, true, _encoding.GetBytes("test message"));
// await ch.WaitForConfirmsOrDieAsync();

await ch.BasicPublishAsync(exchange, binding2, true, _encoding.GetBytes("test message"));
// await ch.WaitForConfirmsOrDieAsync();
var pt1 = ch.BasicPublishAsync(exchange, binding1, true, _encoding.GetBytes("test message"));
var pt2 = ch.BasicPublishAsync(exchange, binding2, true, _encoding.GetBytes("test message"));
await WaitForConfirmsWithCancellationAsync(ch);
await Task.WhenAll(pt1.AsTask(), pt2.AsTask()).WaitAsync(WaitSpan);

await consumerReceivedTcs1.Task.WaitAsync(TimeSpan.FromSeconds(5));
await consumerReceivedTcs2.Task.WaitAsync(TimeSpan.FromSeconds(5));
Assert.True(consumerReceivedTcs1.Task.IsCompletedSuccessfully());
Assert.True(consumerReceivedTcs2.Task.IsCompletedSuccessfully());
await Task.WhenAll(consumerReceivedTcs1.Task, consumerReceivedTcs2.Task).WaitAsync(WaitSpan);
Assert.True(await consumerReceivedTcs1.Task);
Assert.True(await consumerReceivedTcs2.Task);
}
finally
{
Expand Down