From 0d08e90dab6a1dfd3b4a092220f28a1110dd0d72 Mon Sep 17 00:00:00 2001 From: JustMe Date: Sat, 10 Oct 2026 19:33:36 +0800 Subject: [PATCH] perf(tests): isolate network E2E and await QUIC cleanup --- backend/agent/src/MoShiro.Agent/AgentApp.cs | 19 ++- backend/agent/src/MoShiro.Agent/Program.cs | 2 +- .../src/MoShiro.Probe/Link/AgentConnection.cs | 2 +- .../Net/QuicDatagramTransport.cs | 12 +- .../AgentDataPlaneTests.cs | 78 +++++++++ .../AgentEndToEndTests.cs | 156 +----------------- .../MoShiro.Backend.Tests/AgentHarness.cs | 6 +- .../MoShiro.Backend.Tests/AgentLimitsTests.cs | 31 ++++ .../AgentTaskSetEndToEndTests.cs | 36 ++++ .../MoShiro.Backend.Tests/AotSmokeTests.cs | 6 +- .../MoShiro.Backend.Tests/BackendFixture.cs | 4 +- .../EnrollmentAndTaskSetTests.cs | 2 +- .../tests/MoShiro.Backend.Tests/HostTests.cs | 1 + .../MoShiro.Backend.Tests/NodeInfoTests.cs | 1 + .../MoShiro.Backend.Tests/PasskeyTests.cs | 2 +- .../MoShiro.Backend.Tests/ProbeHarness.cs | 2 +- .../MoShiro.Backend.Tests/ReleaseFeedTests.cs | 2 +- .../tests/MoShiro.Backend.Tests/SetupTests.cs | 4 +- .../tests/MoShiro.Backend.Tests/TrustTests.cs | 4 +- .../MoShiro.TestSupport/ProbeSimulator.cs | 42 ++++- 20 files changed, 231 insertions(+), 181 deletions(-) create mode 100644 backend/tests/MoShiro.Backend.Tests/AgentDataPlaneTests.cs create mode 100644 backend/tests/MoShiro.Backend.Tests/AgentLimitsTests.cs create mode 100644 backend/tests/MoShiro.Backend.Tests/AgentTaskSetEndToEndTests.cs diff --git a/backend/agent/src/MoShiro.Agent/AgentApp.cs b/backend/agent/src/MoShiro.Agent/AgentApp.cs index 7e87230..3f531c3 100644 --- a/backend/agent/src/MoShiro.Agent/AgentApp.cs +++ b/backend/agent/src/MoShiro.Agent/AgentApp.cs @@ -11,7 +11,7 @@ using MoShiro.Protocol; namespace MoShiro.Agent; /// Agent 的组装点：数据面、点对点监听、与 Backend 的链路都在这里接起来。没有全局状态，测试可以在同一进程里启动多个。 -public sealed class AgentApp : IDisposable +public sealed class AgentApp : IDisposable, IAsyncDisposable { public const string Version = MoShiro.Protocol.BuildInfo.Version; @@ -40,6 +40,7 @@ public sealed class AgentApp : IDisposable private readonly TaskManager _tasks; private readonly TestRunner _tests; private readonly FleetLink _link; + private int _disposed; /// 配置目录（容器里的 /etc/moshiro）：宿主机更新器和节点在这里交换清单、请求和状态。 public AgentApp(AgentConfig config, LoadedNodeIdentity identity, string dataDirectory, TimeProvider time, INodeLog log, RestartController? restart = null) @@ -113,10 +114,26 @@ public sealed class AgentApp : IDisposable public void Dispose() { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + _datagrams.Dispose(); _ownedRestart?.Dispose(); } + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + await _datagrams.DisposeAsync().ConfigureAwait(false); + _ownedRestart?.Dispose(); + } + internal StreamRegistry Streams => _streams; public IReadOnlyList QuicPorts => _datagrams.BoundPorts; diff --git a/backend/agent/src/MoShiro.Agent/Program.cs b/backend/agent/src/MoShiro.Agent/Program.cs index 11fe862..b2072ec 100644 --- a/backend/agent/src/MoShiro.Agent/Program.cs +++ b/backend/agent/src/MoShiro.Agent/Program.cs @@ -43,7 +43,7 @@ catch (Exception ex) when (ex is IOException or InvalidDataException or InvalidO using var restartScope = restart; using (identity) { - using var app = new AgentApp(config, identity, Path.GetDirectoryName(Path.GetFullPath(configPath))!, time, log, restart); + await using var app = new AgentApp(config, identity, Path.GetDirectoryName(Path.GetFullPath(configPath))!, time, log, restart); await ServiceShell.RunAsync(app.RunAsync, log, restart, time); } diff --git a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs index 717b3d7..8028c52 100644 --- a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs +++ b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs @@ -196,7 +196,7 @@ public sealed class AgentConnection : IDisposable var remote = (IPEndPoint)info.Remote!; IReadOnlyList quicPorts = Endpoint.EffectiveQuicPorts; int quicPort = quicPorts.Count == 0 ? 7443 : quicPorts[0]; - using var datagrams = new QuicDatagramTransport( + await using var datagrams = new QuicDatagramTransport( data, _env.Clock, new DataPlaneCounters(), diff --git a/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs b/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs index a36876c..f403e0b 100644 --- a/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs +++ b/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs @@ -10,7 +10,7 @@ namespace MoShiro.Measurement; /// open one QUIC connection from a newly selected ephemeral source port. Measurement frames are sent /// only with MsQuic DATAGRAM calls; the existing TLS peer connection continues to carry session control. /// -public sealed class QuicDatagramTransport : IDisposable +public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable { private readonly MeasurementDatagramRouter _router; private readonly INodeLog _log; @@ -166,6 +166,16 @@ public sealed class QuicDatagramTransport : IDisposable _ = DisposeNativeResourcesAsync(); } + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + await DisposeNativeResourcesAsync().ConfigureAwait(false); + } + private async Task DisposeNativeResourcesAsync() { try diff --git a/backend/tests/MoShiro.Backend.Tests/AgentDataPlaneTests.cs b/backend/tests/MoShiro.Backend.Tests/AgentDataPlaneTests.cs new file mode 100644 index 0000000..540c6c7 --- /dev/null +++ b/backend/tests/MoShiro.Backend.Tests/AgentDataPlaneTests.cs @@ -0,0 +1,78 @@ +using System.Net; +using System.Net.Sockets; +using MoShiro.Measurement; +using MoShiro.Protocol; +using MoShiro.TestSupport; + +namespace MoShiro.Backend.Tests; + +/// 数据面协议拒绝错误首帧、伪造包、重放包和垃圾数据。 +[Trait("Category", "Slow")] +public sealed class AgentDataPlaneTests : IClassFixture +{ + private readonly BackendFixture _backend; + + public AgentDataPlaneTests(BackendFixture backend) + { + _backend = backend; + } + + private async Task<(AgentHarness Agent, ProbeSimulator Probe, LineTask Udp)> LineAsync() + { + AgentHarness agent = await AgentHarness.StartAsync(_backend); + await agent.OverrideLineTasksAsync(rateHz: 30); + ProbeSimulator probe = await agent.ConnectProbeAsync(); + LineTaskSet set = await probe.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.QuicPorts[0])); + return (agent, probe, set.Tasks.Single(t => t.Kind == "udp_echo")); + } + + [Fact] + public async Task TcpDataPlane_ClosesConnectionsWithoutAValidFirstFrame() + { + (AgentHarness agent, ProbeSimulator probe, _) = await LineAsync(); + await using (agent) + await using (probe) + { + using var client = new TcpClient(); + await client.ConnectAsync(IPAddress.Loopback, agent.App.TcpPorts[0]); + await client.GetStream().WriteAsync(new byte[TcpWire.AuthBytes]); // 全零：magic 不对 + byte[] buffer = new byte[16]; + Assert.Equal(0, await ReadOrResetAsync(client.GetStream(), buffer)); // 连接被关闭：没有任何字节（Windows 上可能是 RST） + await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid >= 1)); + } + } + + [Fact] + public async Task DataPlane_DropsForgedReplayedAndGarbagePackets_WithoutAnswering() + { + (AgentHarness agent, ProbeSimulator probe, LineTask udp) = await LineAsync(); + await using (agent) + await using (probe) + { + long before = agent.App.Counters.DroppedInvalid; + await probe.SendRawUdpAsync([1, 2, 3]); + await probe.SendRawUdpAsync(new byte[64]); + await probe.SendForgedEchoAsync(udp, 1); + await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid - before >= 3), because: "garbage and forged packets are counted"); + + // 重放：同一个序号发两次，只有第一次被接受 + await probe.SendEchoAsync(udp, 500); + await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.PacketsOut >= 1)); + long dropped = agent.App.Counters.DroppedInvalid; + await probe.SendEchoAsync(udp, 500); + await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid > dropped), because: "a replayed packet is dropped"); + } + } + + private static async Task ReadOrResetAsync(Stream stream, byte[] buffer) + { + try + { + return await stream.ReadAsync(buffer).AsTask().WaitAsync(TimeSpan.FromSeconds(10)); + } + catch (IOException) + { + return 0; + } + } +} diff --git a/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs b/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs index 34d1c45..3f29da2 100644 --- a/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs @@ -6,6 +6,8 @@ using MoShiro.TestSupport; namespace MoShiro.Backend.Tests; /// 真实的后端 + 真实的 Agent（进程内）+ 假 Probe：覆盖 Agent 与 Backend 的链路、Probe 与 Agent 的点对点会话和数据面。 +// 使用真实 Agent/QUIC/TCP 会话的端到端验证放在慢测集合，避免和大量 API/单元测试争抢本机套接字。 +[Trait("Category", "Slow")] public sealed class AgentEndToEndTests : IClassFixture { private readonly BackendFixture _backend; @@ -52,20 +54,6 @@ public sealed class AgentEndToEndTests : IClassFixture Assert.Equal("stopping", final["runtime_status"]!.GetValue()); } - [Fact] - public async Task LocalLimits_RejectTasksOutsideTheAllowedPorts_AndTheReasonReachesTheAdmin() - { - // 默认模板的端口（7443/7444）不在这个 Agent 实际监听的端口里：任务被拒绝，原因是 policy_denied - await using AgentHarness agent = await AgentHarness.StartAsync(_backend); - await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == Defaults.TaskCount); - JsonNode rejected = (await AgentViewAsync(agent))["rejected"]!; - Assert.All(rejected.AsArray(), item => Assert.Equal("policy_denied", item!["reason"]!.GetValue())); - - // 覆盖成实际端口之后恢复 - await agent.OverrideLineTasksAsync(); - await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == 0); - } - [Fact] public async Task UdpEcho_EndToEnd_FragmentsAreMergedWithAgentTimestamps_AndStoredByTheBackend() { @@ -108,57 +96,6 @@ public sealed class AgentEndToEndTests : IClassFixture } } - /// 读到 0 字节（对端关闭）或者连接被重置，都算「没收到任何东西」；收到数据返回字节数。 - internal static async Task ReadOrResetAsync(Stream stream, byte[] buffer) - { - try - { - return await stream.ReadAsync(buffer).AsTask().WaitAsync(TimeSpan.FromSeconds(10)); - } - catch (IOException) - { - return 0; - } - } - - [Fact] - public async Task TcpDataPlane_ClosesConnectionsWithoutAValidFirstFrame() - { - (AgentHarness agent, ProbeSimulator probe, _) = await LineAsync(); - await using (agent) - await using (probe) - { - using var client = new System.Net.Sockets.TcpClient(); - await client.ConnectAsync(System.Net.IPAddress.Loopback, agent.App.TcpPorts[0]); - await client.GetStream().WriteAsync(new byte[TcpWire.AuthBytes]); // 全零：magic 不对 - byte[] buffer = new byte[16]; - Assert.Equal(0, await ReadOrResetAsync(client.GetStream(), buffer)); // 连接被关闭：没有任何字节（Windows 上可能是 RST） - await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid >= 1)); - } - } - - [Fact] - public async Task DataPlane_DropsForgedReplayedAndGarbagePackets_WithoutAnswering() - { - (AgentHarness agent, ProbeSimulator probe, LineTask udp) = await LineAsync(); - await using (agent) - await using (probe) - { - long before = agent.App.Counters.DroppedInvalid; - await probe.SendRawUdpAsync([1, 2, 3]); - await probe.SendRawUdpAsync(new byte[64]); - await probe.SendForgedEchoAsync(udp, 1); - await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid - before >= 3), because: "garbage and forged packets are counted"); - - // 重放：同一个序号发两次，只有第一次被接受 - await probe.SendEchoAsync(udp, 500); - await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.PacketsOut >= 1)); - long dropped = agent.App.Counters.DroppedInvalid; - await probe.SendEchoAsync(udp, 500); - await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.DroppedInvalid > dropped), because: "a replayed packet is dropped"); - } - } - [Fact] public async Task ARestartedProbe_NeverInheritsTheAgentRecordsOfItsPreviousSession() { @@ -253,95 +190,6 @@ public sealed class AgentEndToEndTests : IClassFixture } } - [Fact] - public async Task LineTaskSet_FollowsTheDesiredState_AndIsPushedToTheProbe() - { - (AgentHarness agent, ProbeSimulator probe, LineTask udp) = await LineAsync(); - await using (agent) - await using (probe) - { - JsonNode tasks = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/tasks"); - JsonNode mine = tasks["items"]!.AsArray().Single(t => t!["agent_id"]?.GetValue() == agent.Id && t["kind"]!.GetValue() == "udp_echo")!; - using HttpResponseMessage disabled = await _backend.Admin.PatchAsync("/api/admin/tasks/" + mine["id"]!.GetValue(), BackendFixture.Json(new JsonObject { ["enabled"] = false }, "application/merge-patch+json")); - disabled.EnsureSuccessStatusCode(); - - LineTaskSet next = await probe.WaitForTasksAsync(s => s.Tasks.All(t => t.Kind != "udp_echo") && s.Tasks.Count >= 0 && s.Version > 0); - Assert.DoesNotContain(next.Tasks, t => t.Id == udp.Id); - } - } - - // 慢的端到端测试（真实时间的阶梯、要等测量块出来）：CI 里单独一批跑（Category=Slow），不拦着出镜像，见 ci/verify-backend.sh - [Fact] - [Trait("Category", "Slow")] - public async Task Hy2Ladder_EndToEnd_AgentSendsTheTrain_ProbeReportsReceipts_AndTheTestSucceeds() - { - await using AgentHarness agent = await AgentHarness.StartAsync(_backend); - await agent.OverrideLineTasksAsync(); - await using ProbeSimulator probe = await agent.ConnectProbeAsync(); - await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probe_connected"]!.GetValue()); - - HttpResponseMessage? accepted = null; - string? startFailure = null; - for (int attempt = 0; attempt < 5; attempt++) - { - accepted = await _backend.Admin.PostAsync($"/api/admin/lines/{agent.Id}~{probe.ProbeId}/tests", BackendFixture.Json(new JsonObject { ["plan"] = "hy2_ladder", ["params"] = new JsonObject { ["size"] = 1000, ["duration_s"] = 5 } })); - if (accepted.StatusCode == System.Net.HttpStatusCode.Accepted) - { - break; - } - - startFailure = await accepted.Content.ReadAsStringAsync(); - string? code = JsonNode.Parse(startFailure)?["code"]?.GetValue(); - accepted.Dispose(); - accepted = null; - if (code != "probe_offline" || attempt == 4) - { - Assert.Fail($"The test request was rejected: {startFailure}"); - } - - await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probe_connected"]!.GetValue(), TimeSpan.FromSeconds(15), "the probe should reconnect after a transient disconnect"); - } - - HttpResponseMessage acceptedResponse = accepted ?? throw new InvalidOperationException("The test request was not accepted."); - using (acceptedResponse) - { - string testId = JsonNode.Parse(await acceptedResponse.Content.ReadAsStringAsync())!["id"]!.GetValue(); - - JsonNode test = null!; - await BackendFixture.EventuallyAsync( - async () => - { - test = await _backend.GetJsonAsync(_backend.Admin, "/api/public/tests/" + testId); - return test["status"]!.GetValue() is "succeeded" or "failed" or "aborted"; - }, - TimeSpan.FromSeconds(90)); - Assert.True(test["status"]!.GetValue() == "succeeded", test.ToJsonString()); - Assert.Equal(4, probe.Phases.Count); - Assert.All(probe.Phases, phase => Assert.Equal("udp_train", phase.Kind)); - - // 单块最多 50000 条记录：50 和 100 Mbit/s 阶段（5 s、1000 B 的包）里 100 Mbit/s 的那一段有 6 万多条记录，被 Probe 切成两个列块 - Assert.True(agent.App.Counters.Fragments > 4); - await BackendFixture.EventuallyAsync( - async () => (await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={agent.Id}"))["items"]!.AsArray().Count(c => c!["kind"]!.GetValue() == "udp_train") == agent.App.Counters.Fragments, - because: (await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/events?agent_id={agent.Id}")).ToJsonString() + $" fragments={agent.App.Counters.Fragments} discarded={agent.App.Counters.FragmentsDiscarded} pending={agent.App.Link.Outbox.PendingCount}/{agent.App.Link.Outbox.PendingBytes} dropped={agent.App.Link.Outbox.Dropped} link={agent.App.Link.IsConnected}"); - } - } - - [Fact] - public async Task Plan_UsingTcpBulk_IsAcceptedByARealAgentAndStartsItsFirstPhase() - { - await using AgentHarness agent = await AgentHarness.StartAsync(_backend); - await using ProbeSimulator probe = await agent.ConnectProbeAsync(); - await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probe_connected"]!.GetValue()); - - // reality_session 需要 connect、tcp_bulk、hold：真实的 Agent 都实现了 - using HttpResponseMessage response = await _backend.Admin.PostAsync($"/api/admin/lines/{agent.Id}~{probe.ProbeId}/tests", BackendFixture.Json(new JsonObject { ["plan"] = "reality_session" })); - Assert.Equal(System.Net.HttpStatusCode.Accepted, response.StatusCode); - await BackendFixture.EventuallyAsync(() => Task.FromResult(!probe.Phases.IsEmpty), because: "the first phase should start"); - Assert.True(probe.Phases.TryPeek(out var first)); - Assert.Equal("connect", first.Kind); - } - [Fact] public async Task Test_IsRefusedWhenNoProbeIsConnected_AndWhenAPhaseExceedsTheLocalLimits() { diff --git a/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs b/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs index 0fd5fbf..f6fd72e 100644 --- a/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs +++ b/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs @@ -46,7 +46,7 @@ public sealed class AgentHarness : IAsyncDisposable { string directory = Path.Combine(Path.GetTempPath(), "moshiro-agent-" + Guid.NewGuid().ToString("N")); string token = pin?.Code ?? await backend.CreateJoinTokenAsync(); - using var http = new HttpClient(); + using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) }; Result enrolled = await EnrollmentClient.EnrollAsync( http, backend.BaseAddress, token, name ?? backend.UniqueName("agent"), directory, AgentApp.Version, null, TimeProvider.System, CancellationToken.None, pin?.Genesis, pin?.Controllers); Assert.True(enrolled.IsSuccess, enrolled.IsSuccess ? "" : enrolled.Error.ToString()); @@ -62,7 +62,7 @@ public sealed class AgentHarness : IAsyncDisposable }; configure?.Invoke(config); LoadedNodeIdentity identity = LoadedNodeIdentity.Load(directory); - var app = new AgentApp(config, identity, directory, TimeProvider.System, new ConsoleNodeLog(TimeProvider.System, Console.Out)); + var app = new AgentApp(config, identity, directory, TimeProvider.System, NullNodeLog.Instance); var harness = new AgentHarness(backend, directory, identity, app); await app.Ready.WaitAsync(TimeSpan.FromSeconds(10)); await BackendFixture.EventuallyAsync(() => Task.FromResult(app.Link.IsConnected), because: "the agent should connect to the backend"); @@ -137,7 +137,7 @@ public sealed class AgentHarness : IAsyncDisposable // 停止中 } - App.Dispose(); + await App.DisposeAsync(); _identity.Dispose(); _stop.Dispose(); try diff --git a/backend/tests/MoShiro.Backend.Tests/AgentLimitsTests.cs b/backend/tests/MoShiro.Backend.Tests/AgentLimitsTests.cs new file mode 100644 index 0000000..5c286b5 --- /dev/null +++ b/backend/tests/MoShiro.Backend.Tests/AgentLimitsTests.cs @@ -0,0 +1,31 @@ +using System.Text.Json.Nodes; + +namespace MoShiro.Backend.Tests; + +/// Agent local policy validation has its own Backend fixture so task overrides from line-session tests cannot leak into it. +[Trait("Category", "Slow")] +public sealed class AgentLimitsTests : IClassFixture +{ + private readonly BackendFixture _backend; + + public AgentLimitsTests(BackendFixture backend) + { + _backend = backend; + } + + private async Task AgentViewAsync(AgentHarness agent) => await _backend.GetJsonAsync(_backend.Admin, "/api/admin/agents/" + agent.Id); + + [Fact] + public async Task LocalLimits_RejectTasksOutsideTheAllowedPorts_AndTheReasonReachesTheAdmin() + { + // The default task ports (7443/7444) do not match this Agent's actual listeners, so policy rejects them. + await using AgentHarness agent = await AgentHarness.StartAsync(_backend); + await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == Defaults.TaskCount); + JsonNode rejected = (await AgentViewAsync(agent))["rejected"]!; + Assert.All(rejected.AsArray(), item => Assert.Equal("policy_denied", item!["reason"]!.GetValue())); + + // Override the task ports to match the actual listeners and verify recovery. + await agent.OverrideLineTasksAsync(); + await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == 0); + } +} diff --git a/backend/tests/MoShiro.Backend.Tests/AgentTaskSetEndToEndTests.cs b/backend/tests/MoShiro.Backend.Tests/AgentTaskSetEndToEndTests.cs new file mode 100644 index 0000000..028ca61 --- /dev/null +++ b/backend/tests/MoShiro.Backend.Tests/AgentTaskSetEndToEndTests.cs @@ -0,0 +1,36 @@ +using System.Text.Json.Nodes; +using MoShiro.Measurement; +using MoShiro.Protocol; +using MoShiro.TestSupport; + +namespace MoShiro.Backend.Tests; + +/// Exercises the complete Backend → Agent → Probe desired task update with an isolated Backend fixture. +[Trait("Category", "Slow")] +public sealed class AgentTaskSetEndToEndTests : IClassFixture +{ + private readonly BackendFixture _backend; + + public AgentTaskSetEndToEndTests(BackendFixture backend) + { + _backend = backend; + } + + [Fact] + public async Task LineTaskSet_FollowsTheDesiredState_AndIsPushedToTheProbe() + { + await using AgentHarness agent = await AgentHarness.StartAsync(_backend); + await agent.OverrideLineTasksAsync(rateHz: 30); + await using ProbeSimulator probe = await agent.ConnectProbeAsync(); + LineTask udp = (await probe.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.QuicPorts[0]))) + .Tasks.Single(t => t.Kind == "udp_echo"); + + JsonNode tasks = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/tasks"); + JsonNode mine = tasks["items"]!.AsArray().Single(t => t!["agent_id"]?.GetValue() == agent.Id && t["kind"]!.GetValue() == "udp_echo")!; + using HttpResponseMessage disabled = await _backend.Admin.PatchAsync("/api/admin/tasks/" + mine["id"]!.GetValue(), BackendFixture.Json(new JsonObject { ["enabled"] = false }, "application/merge-patch+json")); + disabled.EnsureSuccessStatusCode(); + + LineTaskSet next = await probe.WaitForTasksAsync(s => s.Tasks.All(t => t.Kind != "udp_echo") && s.Version > 0); + Assert.DoesNotContain(next.Tasks, t => t.Id == udp.Id); + } +} diff --git a/backend/tests/MoShiro.Backend.Tests/AotSmokeTests.cs b/backend/tests/MoShiro.Backend.Tests/AotSmokeTests.cs index d07c39c..e1d4d05 100644 --- a/backend/tests/MoShiro.Backend.Tests/AotSmokeTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/AotSmokeTests.cs @@ -20,7 +20,7 @@ public sealed class AotSmokeTests { Assert.SkipUnless(Url is not null, "set MOSHIRO_AOT_URL to the address of a running NativeAOT build to run this test"); var baseAddress = new Uri(Url!); - using var http = new HttpClient { BaseAddress = baseAddress }; + using var http = new HttpClient { BaseAddress = baseAddress, Timeout = TimeSpan.FromSeconds(15) }; // 管理员：登录 → 令牌 using HttpResponseMessage login = await http.PostAsync("/api/admin/auth/login", BackendFixture.Json(new JsonObject { ["username"] = BackendFixture.AdminUser, ["password"] = BackendFixture.AdminPassword })); @@ -41,7 +41,7 @@ public sealed class AotSmokeTests using HttpResponseMessage probeJoin = await http.PostAsync("/api/admin/join-tokens", BackendFixture.Json(new JsonObject { ["role"] = "probe", ["max_uses"] = 1 })); probeJoin.EnsureSuccessStatusCode(); string probeToken = JsonNode.Parse(await probeJoin.Content.ReadAsStringAsync())!["plaintext"]!.GetValue(); - using var probeHttp = new HttpClient(); + using var probeHttp = new HttpClient { Timeout = TimeSpan.FromSeconds(15) }; Fleet.Core.Result probeEnrolled = await Fleet.Enrollment.Client.EnrollmentClient.EnrollAsync( probeHttp, baseAddress, probeToken, "aot-probe-" + Guid.NewGuid().ToString("N")[..6], Path.Combine(Path.GetTempPath(), "moshiro-aot-probe-" + Guid.NewGuid().ToString("N")), "0.0.0", null, TimeProvider.System, default); Assert.True(probeEnrolled.IsSuccess, probeEnrolled.IsSuccess ? "" : probeEnrolled.Error.ToString()); @@ -51,7 +51,7 @@ public sealed class AotSmokeTests using HttpResponseMessage joinResponse = await http.PostAsync("/api/admin/join-tokens", BackendFixture.Json(new JsonObject { ["max_uses"] = 1 })); joinResponse.EnsureSuccessStatusCode(); string joinToken = JsonNode.Parse(await joinResponse.Content.ReadAsStringAsync())!["plaintext"]!.GetValue(); - using var anonymous = new HttpClient { BaseAddress = baseAddress }; + using var anonymous = new HttpClient { BaseAddress = baseAddress, Timeout = TimeSpan.FromSeconds(15) }; await using AgentSimulator agent = await AgentSimulator.EnrollAsync(anonymous, joinToken, "aot-" + Guid.NewGuid().ToString("N")[..6]); agent.ProbeForChunks = new ProbeId(probeId); string lineId = $"{agent.NodeId.Value}~{probeId}"; diff --git a/backend/tests/MoShiro.Backend.Tests/BackendFixture.cs b/backend/tests/MoShiro.Backend.Tests/BackendFixture.cs index d4efcae..f330c1e 100644 --- a/backend/tests/MoShiro.Backend.Tests/BackendFixture.cs +++ b/backend/tests/MoShiro.Backend.Tests/BackendFixture.cs @@ -65,7 +65,7 @@ public class BackendFixture : IAsyncLifetime { string token = await CreateJoinTokenAsync(role: "probe"); string directory = Path.Combine(Path.GetTempPath(), "moshiro-probe-id-" + Guid.NewGuid().ToString("N")); - using var http = new HttpClient(); + using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) }; Fleet.Core.Result enrolled = await Fleet.Enrollment.Client.EnrollmentClient.EnrollAsync( http, BaseAddress, token, name ?? UniqueName("probe"), directory, "0.0.0-test", null, TimeProvider.System, default); Assert.True(enrolled.IsSuccess, enrolled.IsSuccess ? "" : enrolled.Error.ToString()); @@ -179,7 +179,7 @@ public class BackendFixture : IAsyncLifetime return (client, passkey, name); } - public HttpClient NewClient() => new() { BaseAddress = BaseAddress }; + public HttpClient NewClient() => new() { BaseAddress = BaseAddress, Timeout = TimeSpan.FromSeconds(15) }; public string UniqueName(string prefix) => $"{prefix}-{Interlocked.Increment(ref _counter)}"; diff --git a/backend/tests/MoShiro.Backend.Tests/EnrollmentAndTaskSetTests.cs b/backend/tests/MoShiro.Backend.Tests/EnrollmentAndTaskSetTests.cs index 37b9210..fd80370 100644 --- a/backend/tests/MoShiro.Backend.Tests/EnrollmentAndTaskSetTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/EnrollmentAndTaskSetTests.cs @@ -376,7 +376,7 @@ public sealed class EnrollmentAndTaskSetTests : IClassFixture JsonNode listed = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/probes"); Assert.Contains(listed["items"]!.AsArray(), p => p!["id"]!.GetValue() == id && !p["revoked"]!.GetValue()); string duplicateToken = await _backend.CreateJoinTokenAsync(role: "probe"); - using var http = new HttpClient(); + using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) }; Fleet.Core.Result duplicate = await Fleet.Enrollment.Client.EnrollmentClient.EnrollAsync( http, _backend.BaseAddress, duplicateToken, name, Path.Combine(Path.GetTempPath(), "moshiro-dup-" + Guid.NewGuid().ToString("N")), "0.0.0", null, TimeProvider.System, default); Assert.False(duplicate.IsSuccess); diff --git a/backend/tests/MoShiro.Backend.Tests/HostTests.cs b/backend/tests/MoShiro.Backend.Tests/HostTests.cs index 70036aa..575d38b 100644 --- a/backend/tests/MoShiro.Backend.Tests/HostTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/HostTests.cs @@ -184,6 +184,7 @@ public sealed class HostStatusTests : IClassFixture } [Fact] + [Trait("Category", "Slow")] public async Task ARealProbeReportsItsHostStatus_LikeAnAgent_NeverCarriesModelsOrAddresses_AndDisappearsWhenRevoked() { await _backend.SiteMustBePublicAsync(); diff --git a/backend/tests/MoShiro.Backend.Tests/NodeInfoTests.cs b/backend/tests/MoShiro.Backend.Tests/NodeInfoTests.cs index 5ee8fbb..63d40aa 100644 --- a/backend/tests/MoShiro.Backend.Tests/NodeInfoTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/NodeInfoTests.cs @@ -174,6 +174,7 @@ public sealed partial class NodeInfoTests /// 访客能拿到的东西里不能有 IP 等敏感信息：公开接口的整个响应里不出现 IP 形状的字符串，也没有 ip、city、asn、经纬度这些字段。 /// Agent 和 Probe 这里是真的连上了（Probe 的来源地址是 127.0.0.1），所以如果哪里把地址带出来了，这里一定能发现。 /// +[Trait("Category", "Slow")] public sealed partial class NodePrivacyTests : IClassFixture { private static readonly string[] s_forbiddenKeys = ["ip", "city", "asn", "latitude", "longitude", "cidr", "remote", "public_host", "geo_host", "host_name", "hostname"]; diff --git a/backend/tests/MoShiro.Backend.Tests/PasskeyTests.cs b/backend/tests/MoShiro.Backend.Tests/PasskeyTests.cs index 027bb85..245913b 100644 --- a/backend/tests/MoShiro.Backend.Tests/PasskeyTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/PasskeyTests.cs @@ -169,7 +169,7 @@ public sealed class PasskeyTests : IClassFixture Assert.Equal("no-store", login.Headers.CacheControl?.ToString()); // 拿这个会话令牌就是登录状态，能力和密码登录一样 - using var signedIn = new HttpClient { BaseAddress = _backend.BaseAddress }; + using var signedIn = new HttpClient { BaseAddress = _backend.BaseAddress, Timeout = TimeSpan.FromSeconds(15) }; signedIn.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", body["session_token"]!.GetValue()); JsonNode me = await JsonOf(await signedIn.GetAsync("/api/admin/auth/me")); Assert.Equal(_user, me["username"]!.GetValue()); diff --git a/backend/tests/MoShiro.Backend.Tests/ProbeHarness.cs b/backend/tests/MoShiro.Backend.Tests/ProbeHarness.cs index 40e55ad..8c160fb 100644 --- a/backend/tests/MoShiro.Backend.Tests/ProbeHarness.cs +++ b/backend/tests/MoShiro.Backend.Tests/ProbeHarness.cs @@ -38,7 +38,7 @@ public sealed class ProbeHarness : IAsyncDisposable ProbeFile file = Fleet.Hosting.NodeConfig.Load(path, ProbeConfigContext.Default.ProbeFile); configure?.Invoke(file); - var app = new ProbeApp(file, enrolled.Identity, directory, TimeProvider.System, new Fleet.Hosting.ConsoleNodeLog(TimeProvider.System, Console.Out)); + var app = new ProbeApp(file, enrolled.Identity, directory, TimeProvider.System, NullNodeLog.Instance); return new ProbeHarness(directory, app, enrolled) { ProbeId = enrolled.ProbeId }; } diff --git a/backend/tests/MoShiro.Backend.Tests/ReleaseFeedTests.cs b/backend/tests/MoShiro.Backend.Tests/ReleaseFeedTests.cs index 04e91e9..a71e9d4 100644 --- a/backend/tests/MoShiro.Backend.Tests/ReleaseFeedTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/ReleaseFeedTests.cs @@ -149,7 +149,7 @@ public sealed class ReleaseFeedTests : IAsyncLifetime await panel.StartAsync(); try { - using var http = new HttpClient { BaseAddress = new Uri(panel.Services.GetRequiredService().Features.Get()!.Addresses.First()) }; + using var http = new HttpClient { BaseAddress = new Uri(panel.Services.GetRequiredService().Features.Get()!.Addresses.First()), Timeout = TimeSpan.FromSeconds(15) }; using HttpResponseMessage setup = await http.PostAsync("/api/admin/auth/setup", BackendFixture.Json(new JsonObject { ["claim_code"] = "FEED-TEST-CODE", ["username"] = "owner", ["password"] = "correct horse battery", ["return_session"] = true })); Assert.Equal(HttpStatusCode.Created, setup.StatusCode); http.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", JsonNode.Parse(await setup.Content.ReadAsStringAsync())!["session_token"]!.GetValue()); diff --git a/backend/tests/MoShiro.Backend.Tests/SetupTests.cs b/backend/tests/MoShiro.Backend.Tests/SetupTests.cs index 22e87d4..c314c53 100644 --- a/backend/tests/MoShiro.Backend.Tests/SetupTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/SetupTests.cs @@ -35,7 +35,7 @@ public sealed class SetupTests : IAsyncLifetime options.GeoLookupUrl = string.Empty; }); await _app.StartAsync(); - _http = new HttpClient { BaseAddress = new Uri(_app.Services.GetRequiredService().Features.Get()!.Addresses.First()) }; + _http = new HttpClient { BaseAddress = new Uri(_app.Services.GetRequiredService().Features.Get()!.Addresses.First()), Timeout = TimeSpan.FromSeconds(15) }; } public async ValueTask DisposeAsync() @@ -142,7 +142,7 @@ public sealed class SetupTests : IAsyncLifetime await generated.StartAsync(); try { - using var http = new HttpClient { BaseAddress = new Uri(generated.Services.GetRequiredService().Features.Get()!.Addresses.First()) }; + using var http = new HttpClient { BaseAddress = new Uri(generated.Services.GetRequiredService().Features.Get()!.Addresses.First()), Timeout = TimeSpan.FromSeconds(15) }; foreach (string guess in new[] { Code, "0000-0000-0000-0000", "admin", "TEST-CLAIM-CODE-0001" }) { using HttpResponseMessage response = await http.PostAsync("/api/admin/auth/setup", BackendFixture.Json(new JsonObject { ["claim_code"] = guess, ["username"] = "owner", ["password"] = "correct horse battery" })); diff --git a/backend/tests/MoShiro.Backend.Tests/TrustTests.cs b/backend/tests/MoShiro.Backend.Tests/TrustTests.cs index be968b9..7b08685 100644 --- a/backend/tests/MoShiro.Backend.Tests/TrustTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/TrustTests.cs @@ -49,7 +49,7 @@ public sealed class TrustTests : IClassFixture private async Task> EnrollAsync(string code, string? genesis, string[]? controllers, string? directory = null) { directory ??= Path.Combine(Path.GetTempPath(), "moshiro-trust-id-" + Guid.NewGuid().ToString("N")); - using var http = new HttpClient(); + using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) }; return await EnrollmentClient.EnrollAsync(http, _backend.BaseAddress, code, _backend.UniqueName("node"), directory, "0.0.0-test", null, TimeProvider.System, default, genesis, controllers); } @@ -128,6 +128,7 @@ public sealed class TrustTests : IClassFixture } [Fact] + [Trait("Category", "Slow")] public async Task APinnedAgent_FollowsTheOperatorSet_WhenThePasskeysChange_AndOnlyWhenTheOperatorSigns() { Operator op = await OperatorAsync(); @@ -215,6 +216,7 @@ public sealed class TrustTests : IClassFixture } [Fact] + [Trait("Category", "Slow")] public async Task APinnedAgent_ReportsItsOwner_AndThePanelFindsTheAccountAndPasskey() { // 宿主（属主）要求在节点这一侧：节点把安装时固定的属主报上来，面板顺着凭证 id 反查出它绑定在哪个账户的哪把 passkey 上 diff --git a/backend/tests/MoShiro.TestSupport/ProbeSimulator.cs b/backend/tests/MoShiro.TestSupport/ProbeSimulator.cs index 5d02f70..5d723db 100644 --- a/backend/tests/MoShiro.TestSupport/ProbeSimulator.cs +++ b/backend/tests/MoShiro.TestSupport/ProbeSimulator.cs @@ -70,8 +70,10 @@ public sealed class ProbeSimulator : IAsyncDisposable public async Task ConnectAsync(string host, int controlPort, byte[] agentPublicKey, int udpPort, NodeId agentId, TimeProvider? time = null, CancellationToken ct = default) { + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); + timeout.CancelAfter(TimeSpan.FromSeconds(10)); _agentId = agentId; - (PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(host, controlPort, agentPublicKey, _identity, ProbeId, "0.0.0-sim", 8 << 20, time ?? TimeProvider.System, ct); + (PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(host, controlPort, agentPublicKey, _identity, ProbeId, "0.0.0-sim", 8 << 20, time ?? TimeProvider.System, timeout.Token); _info = info; _agentKey = agentPublicKey; _dataSession = new DataSession(info.Credentials.SessionIdValue, info.Credentials.Key); @@ -88,8 +90,8 @@ public sealed class ProbeSimulator : IAsyncDisposable _rpc.OnBinary = (_, _) => ValueTask.CompletedTask; _run = _rpc.RunAsync(_stop.Token); _receiver = _quic.RunAsync(_stop.Token); - await SyncClockAsync(); - await _quic.ConnectAsync(ct); + await SyncClockAsync(ct: timeout.Token); + await _quic.ConnectAsync(timeout.Token); } public async Task WaitForTasksAsync(Func predicate, TimeSpan? timeout = null) @@ -111,16 +113,27 @@ public sealed class ProbeSimulator : IAsyncDisposable } /// NTP 式的时钟同步：偏移 = Agent 时钟 − (往返中点的) Probe 时钟。取最小往返的一次。 - public async Task SyncClockAsync(int rounds = 5) + public async Task SyncClockAsync(int rounds = 3, CancellationToken ct = default) { long bestRoundTrip = long.MaxValue; for (int i = 0; i < rounds; i++) { long t0 = NowUs(); - Result reply = await _rpc!.RequestAsync(MoShiroMethods.Clock, new ClockParams { ProbeMonoUs = t0 }, MoShiroJsonContext.Default.ClockParams, MoShiroJsonContext.Default.ClockResult, CancellationToken.None); + Result reply = await _rpc!.RequestAsync( + MoShiroMethods.Clock, + new ClockParams { ProbeMonoUs = t0 }, + MoShiroJsonContext.Default.ClockParams, + MoShiroJsonContext.Default.ClockResult, + ct, + TimeSpan.FromSeconds(2)); long t1 = NowUs(); + if (!reply.IsSuccess) + { + continue; + } + long roundTrip = t1 - t0; - if (reply.IsSuccess && roundTrip < bestRoundTrip) + if (roundTrip < bestRoundTrip) { bestRoundTrip = roundTrip; ClockOffsetUs = reply.Value.AgentMonoUs - ((t0 + t1) / 2); @@ -164,6 +177,9 @@ public sealed class ProbeSimulator : IAsyncDisposable /// 对一个 tcp_echo 线路任务建立一条 TCP（可选 TLS）连接，发 条回显消息，然后交出一个列块。 public async Task<(int Sent, int Received)> RunTcpEchoAsync(LineTask task, int tcpPort, int count, TimeSpan interval, bool tls, CancellationToken ct = default) { + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); + timeout.CancelAfter(TimeSpan.FromSeconds(15)); + ct = timeout.Token; using var client = new System.Net.Sockets.TcpClient(); await client.ConnectAsync(IPAddress.Loopback, tcpPort, ct); Stream stream = client.GetStream(); @@ -246,10 +262,20 @@ public sealed class ProbeSimulator : IAsyncDisposable public async ValueTask DisposeAsync() { await _stop.CancelAsync(); - _quic?.Dispose(); + if (_quic is not null) + { + await _quic.DisposeAsync(); + } if (_rpc is not null) { - await _rpc.DisposeAsync(); + try + { + await _rpc.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(2)); + } + catch (Exception ex) when (ex is TimeoutException or OperationCanceledException or IOException or ObjectDisposedException) + { + // 测试连接关闭属于尽力清理，不能让一条卡住的流拖住整个测试程序集。 + } } foreach (Task? task in new[] { _run, _receiver })