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 })