using IM.Application.Chat.DTOs; using IM.Domain.Common; using IM.Domain.Enums; using IM.Infrastructure.Persistence; using Xunit; using Microsoft.AspNetCore.SignalR.Client; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; namespace BuildFlow.IM.Integration.Tests; /// /// ChatHub 消息收发端到端用例。覆盖发送 → 广播 → 落库 → 未读累加的核心闭环。 /// [Collection(ImServerCollection.Name)] public class ChatHubMessagingTests : IAsyncLifetime { private readonly ImServerFixture _fixture; private readonly List _connections = new(); public ChatHubMessagingTests(ImServerFixture fixture) => _fixture = fixture; public Task InitializeAsync() => _fixture.CleanRedisAsync(); public async Task DisposeAsync() { foreach (var conn in _connections) { try { await conn.DisposeAsync(); } catch { /* 忽略清理异常 */ } } } /// 构造一个已连接的 Hub 连接,并在用例结束时自动回收。 private async Task ConnectAsync(Guid userId, string userName) { var conn = HubConnectionFactory.Create(_fixture, userId, userName); await conn.StartAsync(); _connections.Add(conn); return conn; } [Fact] public async Task SendMessage_broadcasts_to_both_participants_and_persists_to_mongo() { // Arrange:私聊会话 Alice ↔ Bob,两者均在连接中 _fixture.UserInfoService.SetUser(ImServerFixture.AliceId, "Alice"); _fixture.UserInfoService.SetUser(ImServerFixture.BobId, "Bob"); var convId = await ImTestSeeder.SeedPrivateConversationAsync(_fixture, ImServerFixture.AliceId, ImServerFixture.BobId); var alice = await ConnectAsync(ImServerFixture.AliceId, "Alice"); var bob = await ConnectAsync(ImServerFixture.BobId, "Bob"); MessageDto? aliceReceived = null; MessageDto? bobReceived = null; alice.On("OnMessageReceived", m => aliceReceived = m); bob.On("OnMessageReceived", m => bobReceived = m); // Act:Alice 发一条文本消息 await alice.InvokeAsync("SendMessage", new SendMessageRequest( convId, MessageContentType.Text, "hello integration", null, null)); // 广播为异步,轮询等待两端均收到 await WaitForAsync(() => aliceReceived is not null && bobReceived is not null); // Assert ①:两参与者都收到了同一消息,正文/类型正确 Assert.NotNull(aliceReceived); Assert.NotNull(bobReceived); Assert.Equal("hello integration", aliceReceived!.Content); Assert.Equal(convId, bobReceived!.ConversationId); Assert.Equal(ImServerFixture.AliceId, aliceReceived.SenderId); // Assert ②:消息正文已落 MongoDB,可按会话查到 await using (var scope = _fixture.Services.CreateAsyncScope()) { var mongo = scope.ServiceProvider.GetRequiredService(); var docs = await mongo.Messages.Find(m => m.ConversationId == convId).ToListAsync(); var persisted = Assert.Single(docs); Assert.Equal("hello integration", persisted.Content); Assert.False(persisted.IsRecalled); } // Assert ③:会话列表摘要已更新(PostgreSQL 冷数据) await using (var scope2 = _fixture.Services.CreateAsyncScope()) { var db = scope2.ServiceProvider.GetRequiredService(); var conv = await db.Conversations.AsNoTracking().FirstAsync(c => c.Id == convId); Assert.Equal("hello integration", conv.LastMessageContent); Assert.NotNull(conv.LastMessageId); Assert.NotNull(conv.LastMessageAt); } } [Fact] public async Task SendMessage_increments_redis_unread_for_other_participant_only() { // Arrange:Alice、Bob、Carol 三人群聊。发消息时只给除发送者外的两人累加未读 _fixture.UserInfoService.SetUser(ImServerFixture.AliceId, "Alice"); var (convId, _) = await ImTestSeeder.SeedGroupConversationAsync( _fixture, "三人测试群", ImServerFixture.AliceId, (ImServerFixture.AliceId, GroupMemberRole.Owner), (ImServerFixture.BobId, GroupMemberRole.Member), (ImServerFixture.CarolId, GroupMemberRole.Member)); var alice = await ConnectAsync(ImServerFixture.AliceId, "Alice"); // Act:Alice 发一条消息 await alice.InvokeAsync("SendMessage", new SendMessageRequest( convId, MessageContentType.Text, "group hi", null, null)); // Assert:Redis 未读计数 Bob=1、Carol=1,发送者 Alice 无未读键 await using var scope = _fixture.Services.CreateAsyncScope(); var unread = scope.ServiceProvider.GetRequiredService(); var bobUnread = await unread.GetAsync(convId, ImServerFixture.BobId); var carolUnread = await unread.GetAsync(convId, ImServerFixture.CarolId); var aliceUnread = await unread.GetAsync(convId, ImServerFixture.AliceId); Assert.Equal(1, bobUnread); Assert.Equal(1, carolUnread); // 发送者不计未读:Redis 中该字段为空,GetAsync 返回 null Assert.Null(aliceUnread); } [Fact] public async Task SendMessage_triggers_offline_notification_for_offline_recipients() { // Arrange:Bob 离线(不连接)、Carol 在线(连接)。离线触达仅应发给 Bob _fixture.UserInfoService.SetUser(ImServerFixture.AliceId, "Alice"); var (convId, _) = await ImTestSeeder.SeedGroupConversationAsync( _fixture, "离线通知群", ImServerFixture.AliceId, (ImServerFixture.AliceId, GroupMemberRole.Owner), (ImServerFixture.BobId, GroupMemberRole.Member), (ImServerFixture.CarolId, GroupMemberRole.Member)); var alice = await ConnectAsync(ImServerFixture.AliceId, "Alice"); _ = await ConnectAsync(ImServerFixture.CarolId, "Carol"); // Carol 在线 _fixture.NotificationPublisher.Reset(); // Act await alice.InvokeAsync("SendMessage", new SendMessageRequest( convId, MessageContentType.Text, "ping offline bob", null, null)); // 通知发布为异步,等待 fake 收到一次调用 await WaitForAsync(() => { lock (_fixture.NotificationPublisher.Calls) return _fixture.NotificationPublisher.Calls.Count > 0; }); // Assert:仅有 Bob 收到离线通知,Carol 因在线被过滤 FakeImNotificationPublisher.Call call; lock (_fixture.NotificationPublisher.Calls) call = _fixture.NotificationPublisher.Calls[0]; Assert.Equal("im-message", call.Type); Assert.Equal(convId, call.ConversationId); var recipient = Assert.Single(call.Recipients); Assert.Equal(ImServerFixture.BobId, recipient); } [Fact] public async Task Typing_broadcasts_OnUserTyping_to_participants() { // Arrange _fixture.UserInfoService.SetUser(ImServerFixture.AliceId, "Alice"); _fixture.UserInfoService.SetUser(ImServerFixture.BobId, "Bob"); var convId = await ImTestSeeder.SeedPrivateConversationAsync(_fixture, ImServerFixture.AliceId, ImServerFixture.BobId); var alice = await ConnectAsync(ImServerFixture.AliceId, "Alice"); var bob = await ConnectAsync(ImServerFixture.BobId, "Bob"); Guid? typingUserId = null; bob.On("OnUserTyping", payload => typingUserId = payload.UserId); // Act await alice.InvokeAsync("Typing", convId); // Assert:Bob 收到 Alice 的输入状态 await WaitForAsync(() => typingUserId is not null); Assert.Equal(ImServerFixture.AliceId, typingUserId); } /// 轮询等待条件成立,避免固定 sleep。超时抛异常以便定位卡死。 private static async Task WaitForAsync(Func 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("等待条件在超时内未满足"); } /// 异步条件重载:谓词返回 Task<bool>,用于轮询 Redis 等 IO 状态。 private static async Task WaitForAsync(Func> 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("等待条件在超时内未满足"); } }