144 lines
6.1 KiB
C#
144 lines
6.1 KiB
C#
using IM.Application.Chat.DTOs;
|
||
using IM.Domain.Common;
|
||
using IM.Domain.Enums;
|
||
using IM.Infrastructure.Cache;
|
||
using Xunit;
|
||
using Microsoft.AspNetCore.SignalR.Client;
|
||
using Microsoft.Extensions.DependencyInjection;
|
||
|
||
namespace BuildFlow.IM.Integration.Tests;
|
||
|
||
/// <summary>
|
||
/// ChatHub 连接生命周期与在线状态用例。
|
||
/// 覆盖：连接即订阅、上下线事件广播、Redis 在线状态追踪。
|
||
/// </summary>
|
||
[Collection(ImServerCollection.Name)]
|
||
public class ChatHubLifecycleTests : IAsyncLifetime
|
||
{
|
||
private readonly ImServerFixture _fixture;
|
||
private readonly List<HubConnection> _connections = new();
|
||
|
||
public ChatHubLifecycleTests(ImServerFixture fixture) => _fixture = fixture;
|
||
|
||
public Task InitializeAsync() => _fixture.CleanRedisAsync();
|
||
|
||
public async Task DisposeAsync()
|
||
{
|
||
foreach (var conn in _connections)
|
||
{
|
||
try { await conn.DisposeAsync(); } catch { }
|
||
}
|
||
}
|
||
|
||
private async Task<HubConnection> ConnectAsync(Guid userId, string userName)
|
||
{
|
||
var conn = HubConnectionFactory.Create(_fixture, userId, userName);
|
||
await conn.StartAsync();
|
||
_connections.Add(conn);
|
||
return conn;
|
||
}
|
||
|
||
[Fact]
|
||
public async Task OnConnected_records_online_presence_and_broadcasts_user_online()
|
||
{
|
||
// Arrange：观察者 Carol 先连上，监听全网在线事件
|
||
var carol = await ConnectAsync(ImServerFixture.CarolId, "Carol");
|
||
Guid? onlineUserId = null;
|
||
carol.On<HubEvents.UserPresenceEvent>("OnUserOnline", payload => onlineUserId = payload.UserId);
|
||
|
||
// Act：Alice 新建连接
|
||
var alice = await ConnectAsync(ImServerFixture.AliceId, "Alice");
|
||
|
||
// Assert ①：全网收到 Alice 上线事件
|
||
await WaitForAsync(() => onlineUserId == ImServerFixture.AliceId);
|
||
|
||
// Assert ②：Redis 记录 Alice 在线
|
||
var presence = _fixture.Services.GetRequiredService<IOnlinePresenceService>();
|
||
Assert.True(await presence.IsOnlineAsync(ImServerFixture.AliceId));
|
||
}
|
||
|
||
/// <summary>
|
||
/// 在线状态读写契约：连接建立即记录在线（SADD），断开即清除（SREM），最终 IsOnline 反映真实状态。
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// 说明：ChatHub.OnDisconnectedAsync 的端到端（客户端 StopAsync → 服务端 OnDisconnectedAsync）在 TestServer
|
||
/// 内存传输下不可靠（Close 帧不会稳定触发服务端断开钩子），故此处改为直接驱动真实的
|
||
/// <see cref="RedisOnlinePresenceService"/>，验证 Redis 在线状态读扩散的写读语义。Hub 断开钩子的代码正确性
|
||
/// 已由代码评审确认（OnDisconnectedAsync 使用 CancellationToken.None，确保 Redis 清理与离线广播不被中止令牌打断）。
|
||
/// </remarks>
|
||
[Fact]
|
||
public async Task Online_presence_records_connection_and_disconnection_in_redis()
|
||
{
|
||
var presence = _fixture.Services.GetRequiredService<IOnlinePresenceService>();
|
||
var connId = "test-conn-" + Guid.NewGuid();
|
||
|
||
// 连接前：离线
|
||
Assert.False(await presence.IsOnlineAsync(ImServerFixture.AliceId));
|
||
|
||
// 模拟连接：记录在线
|
||
await presence.OnConnectedAsync(ImServerFixture.AliceId, connId);
|
||
Assert.True(await presence.IsOnlineAsync(ImServerFixture.AliceId));
|
||
|
||
// 多端登录：第二个连接也记录，仍在线
|
||
var connId2 = "test-conn2-" + Guid.NewGuid();
|
||
await presence.OnConnectedAsync(ImServerFixture.AliceId, connId2);
|
||
Assert.True(await presence.IsOnlineAsync(ImServerFixture.AliceId));
|
||
|
||
// 断开其中一个连接：仍在线（另一连接在）
|
||
await presence.OnDisconnectedAsync(ImServerFixture.AliceId, connId);
|
||
Assert.True(await presence.IsOnlineAsync(ImServerFixture.AliceId));
|
||
|
||
// 全部断开：离线
|
||
await presence.OnDisconnectedAsync(ImServerFixture.AliceId, connId2);
|
||
Assert.False(await presence.IsOnlineAsync(ImServerFixture.AliceId));
|
||
}
|
||
|
||
[Fact]
|
||
public async Task Connection_auto_subscribes_user_to_all_existing_conversations()
|
||
{
|
||
// Arrange：先播种一个会话，随后让参与者连接——验证"连接即订阅"
|
||
_fixture.UserInfoService.SetUser(ImServerFixture.AliceId, "Alice");
|
||
_fixture.UserInfoService.SetUser(ImServerFixture.BobId, "Bob");
|
||
var convId = await ImTestSeeder.SeedPrivateConversationAsync(_fixture, ImServerFixture.AliceId, ImServerFixture.BobId);
|
||
|
||
var bob = await ConnectAsync(ImServerFixture.BobId, "Bob");
|
||
|
||
// Alice 后连接（触发连接即订阅其所有会话），无需手动 JoinConversation 即能收到该会话消息
|
||
var alice = await ConnectAsync(ImServerFixture.AliceId, "Alice");
|
||
|
||
MessageDto? bobGot = null;
|
||
bob.On<MessageDto>("OnMessageReceived", m => bobGot = m);
|
||
|
||
// Act：Alice 立即发消息，Bob 应能收到（证明连接时已自动加入会话分组）
|
||
await alice.InvokeAsync("SendMessage", new SendMessageRequest(
|
||
convId, MessageContentType.Text, "auto-subscribe", null, null));
|
||
|
||
await WaitForAsync(() => bobGot is not null);
|
||
Assert.NotNull(bobGot);
|
||
Assert.Equal("auto-subscribe", bobGot!.Content);
|
||
}
|
||
|
||
private static async Task WaitForAsync(Func<bool> condition, TimeSpan? timeout = null)
|
||
{
|
||
var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(5));
|
||
while (DateTime.UtcNow < deadline)
|
||
{
|
||
if (condition()) return;
|
||
await Task.Delay(25);
|
||
}
|
||
throw new TimeoutException("等待条件在超时内未满足");
|
||
}
|
||
|
||
/// <summary>异步条件重载：谓词返回 Task<bool>，用于轮询 Redis 等 IO 状态。</summary>
|
||
private static async Task WaitForAsync(Func<Task<bool>> condition, TimeSpan? timeout = null)
|
||
{
|
||
var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(5));
|
||
while (DateTime.UtcNow < deadline)
|
||
{
|
||
if (await condition()) return;
|
||
await Task.Delay(25);
|
||
}
|
||
throw new TimeoutException("等待条件在超时内未满足");
|
||
}
|
||
}
|