54dc3fc26e
- 移除IWorkflowTriggerService接口及其实现，改用Mediator模式直接发送命令 - 更新AiBotHandler移除workflowTrigger依赖，使用新命令模式触发工作流 - 修改ApproveTaskEndpoint和RejectTaskEndpoint返回bool类型并使用Send.OkAsync - 更新appsettings.json中AI助手的默认指令配置 - 在BuildFlowHostModule.cs中移除旧的工作流触发服务注册 - 添加SSE流式响应路径前缀配置以支持RAG聊天流 - 重构ChatAgentService使用HTTP客户端工厂和运行时配置解析 - 实现AiChatOutputSanitizer用于清洗AI输出中的模板内容 - 更新EmbeddingService使用运行时配置解析器动态创建客户端 - 调整RAG查询处理器中的系统提示指令内容 - 添加构建流按钮操作的单元测试验证发布和禁用功能 - 扩展即时消息集成测试覆盖文件消息元数据处理场景 - 更新平台权限种子数据添加按钮发布和禁用权限 - 实现AI模型运行时配置解析器支持动态模型切换
276 lines
12 KiB
C#
276 lines
12 KiB
C#
using IM.Application.Chat.DTOs;
|
||
using IM.Domain.Common;
|
||
using IM.Domain.Enums;
|
||
using IM.Infrastructure.Mongo;
|
||
using IM.Infrastructure.Persistence;
|
||
using MongoDB.Entities;
|
||
using Xunit;
|
||
using Microsoft.AspNetCore.SignalR.Client;
|
||
using Microsoft.EntityFrameworkCore;
|
||
using Microsoft.Extensions.DependencyInjection;
|
||
|
||
namespace BuildFlow.IM.Integration.Tests;
|
||
|
||
/// <summary>
|
||
/// ChatHub 消息收发端到端用例。覆盖发送 → 广播 → 落库 → 未读累加的核心闭环。
|
||
/// </summary>
|
||
[Xunit.Collection(ImServerCollection.Name)]
|
||
public class ChatHubMessagingTests : IAsyncLifetime
|
||
{
|
||
private readonly ImServerFixture _fixture;
|
||
private readonly List<HubConnection> _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 { /* 忽略清理异常 */ }
|
||
}
|
||
}
|
||
|
||
/// <summary>构造一个已连接的 Hub 连接，并在用例结束时自动回收。</summary>
|
||
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 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<MessageDto>("OnMessageReceived", m => aliceReceived = m);
|
||
bob.On<MessageDto>("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，可按会话查到（MongoDB.Entities 静态 DB.Default）
|
||
var docs = await DB.Default.Find<MessageDocument>()
|
||
.Match(m => m.ConversationId == convId)
|
||
.ExecuteAsync();
|
||
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<ImDbContext>();
|
||
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_file_message_preserves_metadata_reply_and_updates_conversation_summary()
|
||
{
|
||
// Arrange：前端文件发送链路会先完成分片上传，再通过 ChatHub 发送文件名、大小和预览 URL。
|
||
_fixture.UserInfoService.SetUser(ImServerFixture.AliceId, "Alice");
|
||
_fixture.UserInfoService.SetUser(ImServerFixture.BobId, "Bob");
|
||
var convId = await ImTestSeeder.SeedPrivateConversationAsync(
|
||
_fixture,
|
||
ImServerFixture.AliceId,
|
||
ImServerFixture.BobId);
|
||
|
||
var quoted = new MessageDocument
|
||
{
|
||
ConversationId = convId,
|
||
SenderId = ImServerFixture.BobId,
|
||
SenderName = "Bob",
|
||
ContentType = MessageContentType.Text,
|
||
Content = "请把方案文件发我",
|
||
CreatedAt = DateTime.UtcNow.AddSeconds(-10),
|
||
};
|
||
await DB.Default.SaveAsync(quoted);
|
||
|
||
var alice = await ConnectAsync(ImServerFixture.AliceId, "Alice");
|
||
var bob = await ConnectAsync(ImServerFixture.BobId, "Bob");
|
||
|
||
MessageDto? bobReceived = null;
|
||
bob.On<MessageDto>("OnMessageReceived", m => bobReceived = m);
|
||
|
||
var metadata = new MessageMetadataDto(
|
||
FileName: "BuildFlow-方案.pdf",
|
||
FileSize: 2048,
|
||
FileUrl: "/api/files/preview?objectKey=multipart%2Fhash",
|
||
ImageUrl: null,
|
||
ImageWidth: null,
|
||
ImageHeight: null,
|
||
EmojiCode: null);
|
||
|
||
// Act：Alice 发送文件消息，并引用 Bob 的原始消息。
|
||
await alice.InvokeAsync("SendMessage", new SendMessageRequest(
|
||
convId,
|
||
MessageContentType.File,
|
||
"BuildFlow-方案.pdf",
|
||
metadata,
|
||
quoted.ID));
|
||
|
||
await WaitForAsync(() => bobReceived is not null);
|
||
|
||
// Assert ①：广播 DTO 保留文件元数据和引用消息 ID，前端可以直接渲染文件卡片与引用预览。
|
||
Assert.NotNull(bobReceived);
|
||
Assert.Equal(MessageContentType.File, bobReceived!.ContentType);
|
||
Assert.Equal("BuildFlow-方案.pdf", bobReceived.Content);
|
||
Assert.Equal(quoted.ID, bobReceived.ReplyToId);
|
||
Assert.Equal("BuildFlow-方案.pdf", bobReceived.Metadata?.FileName);
|
||
Assert.Equal(2048, bobReceived.Metadata?.FileSize);
|
||
Assert.Equal("/api/files/preview?objectKey=multipart%2Fhash", bobReceived.Metadata?.FileUrl);
|
||
|
||
// Assert ②：MongoDB 持久化同样保留元数据和引用关系，支撑历史搜索/定位后的再次渲染。
|
||
var persisted = await DB.Default.Find<MessageDocument>()
|
||
.Match(m => m.ID == bobReceived.Id)
|
||
.ExecuteFirstAsync();
|
||
Assert.NotNull(persisted);
|
||
Assert.Equal(quoted.ID, persisted!.ReplyToId);
|
||
Assert.Equal("BuildFlow-方案.pdf", persisted.Metadata?.FileName);
|
||
Assert.Equal("/api/files/preview?objectKey=multipart%2Fhash", persisted.Metadata?.FileUrl);
|
||
|
||
// Assert ③：会话列表摘要使用文件占位，避免长 URL 或元数据污染列表。
|
||
await using var scope = _fixture.Services.CreateAsyncScope();
|
||
var db = scope.ServiceProvider.GetRequiredService<ImDbContext>();
|
||
var conv = await db.Conversations.AsNoTracking().FirstAsync(c => c.Id == convId);
|
||
Assert.Equal($"[文件] {metadata.FileName}", conv.LastMessageContent);
|
||
Assert.Equal(bobReceived.Id, conv.LastMessageId);
|
||
}
|
||
|
||
[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<IUnreadCounter>();
|
||
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<HubEvents.UserTypingEvent>("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);
|
||
}
|
||
|
||
/// <summary>轮询等待条件成立，避免固定 sleep。超时抛异常以便定位卡死。</summary>
|
||
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("等待条件在超时内未满足");
|
||
}
|
||
}
|