using BuildFlow.Application.Chat.Commands; using BuildFlow.Application.Chat.DTOs; using BuildFlow.Application.Chat.Handlers; using BuildFlow.Application.Chat.Services; using BuildFlow.Domain.Entities.ChatMessage; using BuildFlow.Domain.Entities.Conversation; using BuildFlow.Domain.Enums; using BuildFlow.Infrastructure.Persistence; using BuildFlow.Infrastructure.Cache; using BuildFlow.Rag.Tests.Stub; using BuildFlow.Shared.Domain.Errors; using BuildFlow.Shared.Domain.Exceptions; using BuildFlow.Shared.Infrastructure; using FluentAssertions; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Options; using Xunit; namespace BuildFlow.Rag.Tests; /// /// RAG 流式发送和编辑重生成的对象所有权与事件契约测试。 /// public class ChatStreamOwnershipTests { [Fact] public async Task StreamMessageForOwnerShouldYieldChunksThenDoneAndPersistMessages() { var userContext = new TestCurrentUserContext(); await using var db = CreateDb(userContext); var conversationId = await SeedConversationAsync( db, Guid.Parse(userContext.UserId)); var cache = new StubChatMessageCache(); var handler = new StreamMessageCommandHandler( db, new StubAIChatAgent { StreamTokens = ["你", "好", "!"] }, cache, CreateWindow(db), userContext); var events = await CollectAsync(handler.Handle( new StreamMessageCommand(conversationId, "你好"), CancellationToken.None)); events.Take(3).Select(x => x.Content).Should().Equal("你", "好", "!"); events.Should().HaveCount(4); events[^1].MessageId.Should().NotBeNull(); events[^1].MessageId!.Value.Should().NotBeEmpty(); db.ChatMessages.Count(x => x.Role == ChatRoleEnum.User).Should().Be(1); db.ChatMessages.Count(x => x.Role == ChatRoleEnum.Assistant).Should().Be(1); cache.Count(conversationId).Should().Be(2); } [Fact] public async Task StreamMessageForNonOwnerShouldThrowNotFoundBeforeWriting() { var userContext = new TestCurrentUserContext(); await using var db = CreateDb(userContext); var conversationId = await SeedConversationAsync(db, Guid.NewGuid()); var cache = new StubChatMessageCache(); var handler = new StreamMessageCommandHandler( db, new StubAIChatAgent(), cache, CreateWindow(db), userContext); var act = () => CollectAsync(handler.Handle( new StreamMessageCommand(conversationId, "越权消息"), CancellationToken.None)); await act.Should().ThrowAsync() .Where(x => x.Code == ErrorCode.NotFound && x.Message == "会话不存在"); db.ChatMessages.Should().BeEmpty(); cache.Count(conversationId).Should().Be(0); } [Fact] public async Task EditStreamForNonOwnerShouldThrowNotFoundBeforeEditing() { var userContext = new TestCurrentUserContext(); await using var db = CreateDb(userContext); var conversationId = await SeedConversationAsync(db, Guid.NewGuid()); var target = new ChatMessageEntity { ConversationId = conversationId, Role = ChatRoleEnum.User, Content = "原始问题", }; var later = new ChatMessageEntity { ConversationId = conversationId, Role = ChatRoleEnum.Assistant, Content = "原始回答", }; db.ChatMessages.AddRange(target, later); await db.SaveChangesAsync(); var handler = new EditMessageStreamCommandHandler( db, new StubAIChatAgent(), new StubChatMessageCache(), CreateWindow(db), userContext); var act = () => CollectAsync(handler.Handle( new EditMessageStreamCommand(conversationId, target.Id, "越权编辑"), CancellationToken.None)); await act.Should().ThrowAsync() .Where(x => x.Code == ErrorCode.NotFound && x.Message == "会话不存在"); target.Content.Should().Be("原始问题"); db.ChatMessages.Should().HaveCount(2); } [Fact] public async Task EditStreamForOwnerShouldPreserveChunkAndDoneEventContract() { var userContext = new TestCurrentUserContext(); await using var db = CreateDb(userContext); var conversationId = await SeedConversationAsync( db, Guid.Parse(userContext.UserId)); var target = new ChatMessageEntity { ConversationId = conversationId, Role = ChatRoleEnum.User, Content = "原始问题", }; db.ChatMessages.Add(target); await db.SaveChangesAsync(); await Task.Delay(5); db.ChatMessages.Add(new ChatMessageEntity { ConversationId = conversationId, Role = ChatRoleEnum.Assistant, Content = "旧回答", }); await db.SaveChangesAsync(); var cache = new StubChatMessageCache(); var handler = new EditMessageStreamCommandHandler( db, new StubAIChatAgent { StreamTokens = ["新", "回", "答"] }, cache, CreateWindow(db), userContext); var events = await CollectAsync(handler.Handle( new EditMessageStreamCommand(conversationId, target.Id, "新问题"), CancellationToken.None)); events.Take(3).Select(x => x.Content).Should().Equal("新", "回", "答"); events.Should().HaveCount(4); events[^1].MessageId.Should().NotBeNull(); events[^1].MessageId!.Value.Should().NotBeEmpty(); target.Content.Should().Be("新问题"); db.ChatMessages.Should().HaveCount(2); db.ChatMessages.Single(x => x.Role == ChatRoleEnum.Assistant) .Content.Should().Be("新回答"); cache.Count(conversationId).Should().Be(2); } private static RagDbContext CreateDb(TestCurrentUserContext userContext) { var options = new DbContextOptionsBuilder() .UseInMemoryDatabase($"chat-stream-{Guid.NewGuid():N}") .AddInterceptors(new AuditInterceptor(userContext)) .Options; return new RagDbContext(options); } private static async Task SeedConversationAsync(RagDbContext db, Guid userId) { var conversation = new ConversationEntity { Title = "测试会话", UserId = userId, }; db.Conversations.Add(conversation); await db.SaveChangesAsync(); return conversation.Id; } private static ChatContextWindowBuilder CreateWindow(RagDbContext db) => new(db, Options.Create(new ChatOptions())); private static async Task> CollectAsync(IAsyncEnumerable source) { var result = new List(); await foreach (var item in source) result.Add(item); return result; } }