209 lines
11 KiB
C#
209 lines
11 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>
|
|
// 使用真实 Agent/QUIC/TCP 会话的端到端验证放在慢测集合，避免和大量 API/单元测试争抢本机套接字。
|
|
[Trait("Category", "Slow")]
|
|
public sealed class AgentEndToEndTests : IClassFixture<BackendFixture>
|
|
{
|
|
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.QuicPorts[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>());
|
|
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["runtime_status"]!.GetValue<string>() == "running");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Agent_OrderlyShutdownIsReportedAsStoppingRatherThanOffline()
|
|
{
|
|
AgentHarness agent = await AgentHarness.StartAsync(_backend);
|
|
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["runtime_status"]!.GetValue<string>() == "running");
|
|
|
|
await agent.DisposeAsync();
|
|
|
|
await BackendFixture.EventuallyAsync(async () => !(await AgentViewAsync(agent))["online"]!.GetValue<bool>());
|
|
JsonNode final = await AgentViewAsync(agent);
|
|
Assert.Equal("stopping", final["runtime_status"]!.GetValue<string>());
|
|
}
|
|
|
|
[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"));
|
|
}
|
|
}
|
|
|
|
[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.QuicPorts[0], agent.NodeId);
|
|
LineTaskSet set = await second.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.QuicPorts[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.QuicPorts[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.QuicPorts[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 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.QuicPorts[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);
|
|
}
|
|
}
|