185 lines
10 KiB
C#
185 lines
10 KiB
C#
using Fleet.Core;
|
|
using System.Text.Json.Nodes;
|
|
using MoShiro.Measurement;
|
|
using MoShiro.Protocol;
|
|
using MoShiro.TestSupport;
|
|
|
|
namespace MoShiro.Backend.Tests;
|
|
|
|
/// <summary>真实的后端 + 真实的 Agent（进程内）+ 拨号侧模拟器：覆盖 Agent 与 Backend 的链路、Dialer 与 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, DialerSimulator Dialer, LineTask Udp)> LineAsync()
|
|
{
|
|
AgentHarness agent = await AgentHarness.StartAsync(_backend);
|
|
await agent.OverrideLineTasksAsync(rateHz: 30);
|
|
DialerSimulator dialer = await agent.ConnectDialerAsync();
|
|
LineTaskSet set = await dialer.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "quic_ping" && t.Port == agent.App.QuicPorts[0]));
|
|
return (agent, dialer, set.Tasks.Single(t => t.Kind == "quic_ping"));
|
|
}
|
|
|
|
private async Task<JsonNode> AgentViewAsync(AgentHarness agent) => await _backend.GetJsonAsync(_backend.Admin, "/api/admin/agents/" + agent.Id);
|
|
|
|
[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, DialerSimulator dialer, LineTask udp) = await LineAsync();
|
|
await using (agent)
|
|
await using (dialer)
|
|
{
|
|
(int sent, int received) = await dialer.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("quic_ping", 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, DialerSimulator dialer, _) = await LineAsync();
|
|
await using (agent)
|
|
await using (dialer)
|
|
{
|
|
LineTask tcp = (await dialer.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "tcp_ping" && t.Port == agent.App.TcpPorts[0]))).Tasks.Single(t => t.Kind == "tcp_ping");
|
|
(int sent, int received) = await dialer.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_ping"));
|
|
}
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ARestartedDialer_NeverInheritsTheAgentRecordsOfItsPreviousSession()
|
|
{
|
|
(AgentHarness agent, DialerSimulator 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 dialer process restarts: a new session with its own streams; sequence numbers start from 0 again
|
|
await using DialerSimulator 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 == "quic_ping" && t.Port == agent.App.QuicPorts[0]));
|
|
LineTask udp2 = set.Tasks.Single(t => t.Kind == "quic_ping");
|
|
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 SeveralDialers_AreReportedTogether_CanBeDroppedOneByOne_AndADisabledOneIsKickedOut()
|
|
{
|
|
await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
|
|
await agent.OverrideLineTasksAsync(rateHz: 30);
|
|
await using DialerSimulator a = await agent.ConnectDialerAsync();
|
|
await using DialerSimulator b = await agent.ConnectDialerAsync();
|
|
|
|
// The agent reports both sessions with version and address; the dialer view shows the agent it is connected to
|
|
JsonNode view = null!;
|
|
await BackendFixture.EventuallyAsync(
|
|
async () =>
|
|
{
|
|
view = await AgentViewAsync(agent);
|
|
return view["dialers"]!.AsArray().Count == 2;
|
|
},
|
|
because: "the agent reports every connected dialer");
|
|
Assert.All(view["dialers"]!.AsArray(), p => Assert.Equal("0.0.0-sim", p!["version"]!.GetValue<string>()));
|
|
Assert.Equal("127.0.0.1", view["dialers"]![0]!["remote"]!.GetValue<string>());
|
|
JsonNode dialerView = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/agents/" + a.DialerId);
|
|
Assert.Equal(agent.Id, dialerView["outgoing"]![0]!["listener_id"]!.GetValue<string>());
|
|
|
|
// Both get their own task set (own streams)
|
|
LineTaskSet setA = await a.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "quic_ping"));
|
|
LineTaskSet setB = await b.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "quic_ping"));
|
|
Assert.NotEqual(setA.Tasks.Single(t => t.Kind == "quic_ping").StreamId, setB.Tasks.Single(t => t.Kind == "quic_ping").StreamId);
|
|
|
|
// Drop one: the other stays
|
|
using HttpResponseMessage dropped = await _backend.Admin.PostAsync($"/api/admin/agents/{agent.Id}/dialers/{a.DialerId}/disconnect", null);
|
|
Assert.Equal(System.Net.HttpStatusCode.NoContent, dropped.StatusCode);
|
|
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["dialers"]!.AsArray().Count == 1, because: "only the dropped dialer leaves");
|
|
Assert.True(b.IsConnected);
|
|
using HttpResponseMessage again = await _backend.Admin.PostAsync($"/api/admin/agents/{agent.Id}/dialers/{a.DialerId}/disconnect", null);
|
|
Assert.Equal(System.Net.HttpStatusCode.Conflict, again.StatusCode);
|
|
|
|
// Disable the remaining Agent: the listener closes its session
|
|
using HttpResponseMessage revoked = await _backend.Admin.PostAsync($"/api/admin/agents/{b.DialerId}/disable", null);
|
|
revoked.EnsureSuccessStatusCode();
|
|
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["dialers"]!.AsArray().Count == 0, because: "a disabled Agent is disconnected");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task PeerHandshake_RejectsADialerWhoseKeyTheBackendNeverIssued()
|
|
{
|
|
await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
|
|
await agent.CreateDialerAsync();
|
|
await using var stranger = new DialerSimulator("agt_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 DialerSimulator dialer = await agent.CreateDialerAsync();
|
|
await Assert.ThrowsAnyAsync<Exception>(() => dialer.ConnectAsync("127.0.0.1", agent.App.ControlPort, Fleet.Identity.NodeIdentity.Generate().PublicKeySpki, agent.App.QuicPorts[0], agent.NodeId));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Test_IsRefusedWhenNoDialerIsConnected_AndWhenAPhaseExceedsTheLocalLimits()
|
|
{
|
|
await using AgentHarness agent = await AgentHarness.StartAsync(_backend, config => config.Limits.MaxTotalMbps = 20);
|
|
await agent.OverrideLineTasksAsync();
|
|
|
|
var phase = new TestPhase { Kind = "quic_stream", DurationSeconds = 1, Params = { ["port"] = System.Text.Json.JsonSerializer.SerializeToElement(agent.App.QuicPorts[0]) } };
|
|
var run = new TestRunParams { TestId = new TestId("tst_x"), DialerId = new NodeId("agt_nobody"), Plan = "x", Phases = [phase] };
|
|
Assert.Equal("dialer_offline", agent.App.Tests.Start(run).Error.Code.Value);
|
|
|
|
await using DialerSimulator dialer = await agent.ConnectDialerAsync();
|
|
run.DialerId = new NodeId(dialer.DialerId);
|
|
phase.Params["mbps"] = System.Text.Json.JsonSerializer.SerializeToElement(50);
|
|
Assert.Equal("budget_exceeded", agent.App.Tests.Start(run).Error.Code.Value);
|
|
}
|
|
}
|