348 lines
19 KiB
C#
348 lines
19 KiB
C#
using System.Text.Json.Nodes;
|
||
using MoShiro.Measurement;
|
||
using MoShiro.Protocol;
|
||
using MoShiro.TestSupport;
|
||
|
||
namespace MoShiro.Backend.Tests;
|
||
|
||
/// <summary>真实的后端 + 真实的 Agent（进程内）+ 假 Probe：覆盖 Agent 与 Backend 的链路、Probe 与 Agent 的点对点会话和数据面。</summary>
|
||
[Collection("backend")]
|
||
public sealed class AgentEndToEndTests
|
||
{
|
||
private readonly BackendFixture _backend;
|
||
|
||
public AgentEndToEndTests(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.UdpPorts[0]));
|
||
return (agent, probe, set.Tasks.Single(t => t.Kind == "udp_echo"));
|
||
}
|
||
|
||
private async Task<JsonNode> AgentViewAsync(AgentHarness agent) => await _backend.GetJsonAsync(_backend.Admin, "/api/admin/agents/" + agent.Id);
|
||
|
||
[Fact]
|
||
public async Task Agent_ConnectsAndReportsItsRealPorts_AndTheProbeSessionIsVisibleToTheBackend()
|
||
{
|
||
await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
|
||
JsonNode view = await AgentViewAsync(agent);
|
||
Assert.Equal("active", view["status"]!.GetValue<string>());
|
||
Assert.False(view["probe_connected"]!.GetValue<bool>());
|
||
|
||
await using ProbeSimulator probe = await agent.ConnectProbeAsync();
|
||
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probe_connected"]!.GetValue<bool>());
|
||
}
|
||
|
||
[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<string>()));
|
||
|
||
// 覆盖成实际端口之后恢复
|
||
await agent.OverrideLineTasksAsync();
|
||
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == 0);
|
||
}
|
||
|
||
[Fact]
|
||
public async Task UdpEcho_EndToEnd_FragmentsAreMergedWithAgentTimestamps_AndStoredByTheBackend()
|
||
{
|
||
(AgentHarness agent, ProbeSimulator probe, LineTask udp) = await LineAsync();
|
||
await using (agent)
|
||
await using (probe)
|
||
{
|
||
(int sent, int received) = await probe.RunUdpEchoAsync(udp, 40, TimeSpan.FromMilliseconds(40));
|
||
Assert.Equal(40, sent);
|
||
Assert.Equal(40, received);
|
||
|
||
JsonNode stored = null!;
|
||
await BackendFixture.EventuallyAsync(async () =>
|
||
{
|
||
stored = await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={agent.Id}");
|
||
return stored["items"]!.AsArray().Count >= 1;
|
||
}, because: (await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/events?agent_id={agent.Id}")).ToJsonString() + $" fragments={agent.App.Counters.Fragments} discarded={agent.App.Counters.FragmentsDiscarded} outbox={agent.App.Link.Outbox.PendingCount}");
|
||
JsonNode chunk = stored["items"]![0]!;
|
||
Assert.True(chunk["kind"] is not null, stored.ToJsonString());
|
||
Assert.Equal("udp_echo", chunk["kind"]!.GetValue<string>());
|
||
Assert.True(agent.App.Counters.Fragments >= 1);
|
||
Assert.True(agent.App.Counters.PacketsIn >= 40);
|
||
}
|
||
}
|
||
|
||
[Theory]
|
||
[InlineData(false)]
|
||
[InlineData(true)]
|
||
public async Task TcpEcho_EndToEnd_AuthenticatedFirstFrame_PlainOrTls(bool tls)
|
||
{
|
||
(AgentHarness agent, ProbeSimulator probe, _) = await LineAsync();
|
||
await using (agent)
|
||
await using (probe)
|
||
{
|
||
LineTask tcp = (await probe.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "tcp_echo" && t.Port == agent.App.TcpPorts[0]))).Tasks.Single(t => t.Kind == "tcp_echo");
|
||
(int sent, int received) = await probe.RunTcpEchoAsync(tcp, agent.App.TcpPorts[0], 15, TimeSpan.FromMilliseconds(60), tls);
|
||
Assert.Equal(15, sent);
|
||
Assert.Equal(15, received);
|
||
await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={agent.Id}"))["items"]!.AsArray().Any(c => c!["kind"]!.GetValue<string>() == "tcp_echo"));
|
||
}
|
||
}
|
||
|
||
/// <summary>读到 0 字节（对端关闭）或者连接被重置，都算「没收到任何东西」；收到数据返回字节数。</summary>
|
||
internal static async Task<int> 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()
|
||
{
|
||
(AgentHarness agent, ProbeSimulator first, LineTask udp) = await LineAsync();
|
||
await using (agent)
|
||
{
|
||
await first.RunUdpEchoAsync(udp, 20, TimeSpan.FromMilliseconds(40));
|
||
RowStore records = agent.App.Streams.Records(udp.StreamId, "msg")!;
|
||
Assert.True(records.Contains(5), "the agent recorded t2/t3 for the first session's packets");
|
||
await first.DisposeAsync();
|
||
|
||
// The probe process restarts: a new session with its own streams; sequence numbers start from 0 again
|
||
await using ProbeSimulator second = first.Restarted();
|
||
await second.ConnectAsync("127.0.0.1", agent.App.ControlPort, agent.TlsPublicKey, agent.App.UdpPorts[0], agent.NodeId);
|
||
LineTaskSet set = await second.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.UdpPorts[0]));
|
||
LineTask udp2 = set.Tasks.Single(t => t.Kind == "udp_echo");
|
||
Assert.NotEqual(udp.StreamId, udp2.StreamId);
|
||
|
||
// The previous session's streams are released with it ...
|
||
await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Streams.Records(udp.StreamId, "msg") is null), because: "the previous session's streams must be released");
|
||
|
||
// ... and a packet the new session never sent has no agent record: if the old t2/t3 survived, a lost uplink packet would look like a downlink loss
|
||
RowStore fresh = agent.App.Streams.Records(udp2.StreamId, "msg")!;
|
||
Assert.False(fresh.Contains(3));
|
||
Assert.False(fresh.Contains(5));
|
||
}
|
||
}
|
||
|
||
[Fact]
|
||
public async Task SeveralProbes_AreReportedTogether_CanBeDroppedOneByOne_AndARevokedOneIsKickedOut()
|
||
{
|
||
await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
|
||
await agent.OverrideLineTasksAsync(rateHz: 30);
|
||
await using ProbeSimulator a = await agent.ConnectProbeAsync();
|
||
await using ProbeSimulator b = await agent.ConnectProbeAsync();
|
||
|
||
// The agent reports both sessions with version and address; the probe view shows the agent it is connected to
|
||
JsonNode view = null!;
|
||
await BackendFixture.EventuallyAsync(
|
||
async () =>
|
||
{
|
||
view = await AgentViewAsync(agent);
|
||
return view["probes"]!.AsArray().Count == 2;
|
||
},
|
||
because: "the agent reports every connected probe");
|
||
Assert.All(view["probes"]!.AsArray(), p => Assert.Equal("0.0.0-sim", p!["version"]!.GetValue<string>()));
|
||
Assert.Equal("127.0.0.1", view["probes"]![0]!["remote"]!.GetValue<string>());
|
||
JsonNode probeView = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/probes/" + a.ProbeId);
|
||
Assert.Equal(agent.Id, probeView["connections"]![0]!["agent_id"]!.GetValue<string>());
|
||
|
||
// Both get their own task set (own streams)
|
||
LineTaskSet setA = await a.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo"));
|
||
LineTaskSet setB = await b.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo"));
|
||
Assert.NotEqual(setA.Tasks.Single(t => t.Kind == "udp_echo").StreamId, setB.Tasks.Single(t => t.Kind == "udp_echo").StreamId);
|
||
|
||
// Drop one: the other stays
|
||
using HttpResponseMessage dropped = await _backend.Admin.PostAsync($"/api/admin/agents/{agent.Id}/probes/{a.ProbeId}/disconnect", null);
|
||
Assert.Equal(System.Net.HttpStatusCode.NoContent, dropped.StatusCode);
|
||
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probes"]!.AsArray().Count == 1, because: "only the dropped probe leaves");
|
||
Assert.True(b.IsConnected);
|
||
using HttpResponseMessage again = await _backend.Admin.PostAsync($"/api/admin/agents/{agent.Id}/probes/{a.ProbeId}/disconnect", null);
|
||
Assert.Equal(System.Net.HttpStatusCode.Conflict, again.StatusCode);
|
||
|
||
// Revoke the remaining one: the agent closes its session
|
||
using HttpResponseMessage revoked = await _backend.Admin.PostAsync($"/api/admin/probes/{b.ProbeId}/revoke", null);
|
||
revoked.EnsureSuccessStatusCode();
|
||
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probes"]!.AsArray().Count == 0, because: "a revoked probe is disconnected");
|
||
}
|
||
|
||
[Fact]
|
||
public async Task PeerHandshake_RejectsAProbeWhoseKeyTheBackendNeverIssued()
|
||
{
|
||
await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
|
||
await agent.CreateProbeAsync();
|
||
await using var stranger = new ProbeSimulator("prb_stranger", Fleet.Identity.NodeIdentity.Generate());
|
||
await Assert.ThrowsAnyAsync<Exception>(() => stranger.ConnectAsync("127.0.0.1", agent.App.ControlPort, agent.TlsPublicKey, agent.App.UdpPorts[0], agent.NodeId));
|
||
|
||
// 证书公钥不对（中间人）也不会连上
|
||
await using ProbeSimulator probe = await agent.CreateProbeAsync();
|
||
await Assert.ThrowsAnyAsync<Exception>(() => probe.ConnectAsync("127.0.0.1", agent.App.ControlPort, Fleet.Identity.NodeIdentity.Generate().PublicKeySpki, agent.App.UdpPorts[0], agent.NodeId));
|
||
}
|
||
|
||
[Fact]
|
||
public async Task ProbeDisconnect_IsReportedToTheBackend()
|
||
{
|
||
(AgentHarness agent, ProbeSimulator probe, _) = await LineAsync();
|
||
await using (agent)
|
||
{
|
||
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probe_connected"]!.GetValue<bool>());
|
||
await probe.DisposeAsync();
|
||
await BackendFixture.EventuallyAsync(async () => !(await AgentViewAsync(agent))["probe_connected"]!.GetValue<bool>());
|
||
}
|
||
}
|
||
|
||
[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<string>() == agent.Id && t["kind"]!.GetValue<string>() == "udp_echo")!;
|
||
using HttpResponseMessage disabled = await _backend.Admin.PatchAsync("/api/admin/tasks/" + mine["id"]!.GetValue<string>(), 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<bool>());
|
||
|
||
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<string>();
|
||
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<bool>(), 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<string>();
|
||
|
||
JsonNode test = null!;
|
||
await BackendFixture.EventuallyAsync(
|
||
async () =>
|
||
{
|
||
test = await _backend.GetJsonAsync(_backend.Admin, "/api/public/tests/" + testId);
|
||
return test["status"]!.GetValue<string>() is "succeeded" or "failed" or "aborted";
|
||
},
|
||
TimeSpan.FromSeconds(90));
|
||
Assert.True(test["status"]!.GetValue<string>() == "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<string>() == "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<bool>());
|
||
|
||
// 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()
|
||
{
|
||
await using AgentHarness agent = await AgentHarness.StartAsync(_backend, config => config.Limits.MaxTotalMbps = 20);
|
||
await agent.OverrideLineTasksAsync();
|
||
|
||
var phase = new TestPhase { Kind = "udp_train", DurationSeconds = 1, Params = { ["port"] = System.Text.Json.JsonSerializer.SerializeToElement(agent.App.UdpPorts[0]) } };
|
||
var run = new TestRunParams { TestId = new TestId("tst_x"), ProbeId = new ProbeId("prb_nobody"), Plan = "x", Phases = [phase] };
|
||
Assert.Equal("probe_offline", agent.App.Tests.Start(run).Error.Code.Value);
|
||
|
||
await using ProbeSimulator probe = await agent.ConnectProbeAsync();
|
||
run.ProbeId = new ProbeId(probe.ProbeId);
|
||
phase.Params["mbps"] = System.Text.Json.JsonSerializer.SerializeToElement(50);
|
||
Assert.Equal("budget_exceeded", agent.App.Tests.Start(run).Error.Code.Value);
|
||
}
|
||
}
|