109 lines
6.0 KiB
C#
109 lines
6.0 KiB
C#
using System.Text.Json.Nodes;
|
||
using Fleet.Core;
|
||
using MoShiro.Backend.Application;
|
||
using MoShiro.Protocol;
|
||
using MoShiro.TestSupport;
|
||
|
||
namespace MoShiro.Backend.Tests;
|
||
|
||
public sealed class AdaptiveAuditTests : IClassFixture<BackendFixture>
|
||
{
|
||
private static readonly long s_t0 = new DateTimeOffset(2026, 10, 22, 10, 0, 0, TimeSpan.Zero).ToUnixTimeMilliseconds();
|
||
private const long Minute = 60_000;
|
||
private const long Hour = 3_600_000;
|
||
|
||
private readonly BackendFixture _backend;
|
||
|
||
public AdaptiveAuditTests(BackendFixture backend)
|
||
{
|
||
_backend = backend;
|
||
}
|
||
|
||
private static RateChange Change(long atMs, string kind, string from, string to, double hz, string reason = "quiet", double lossPct = 0) =>
|
||
new() { AtMs = atMs, Kind = kind, From = from, To = to, Hz = hz, Reason = reason, LossPct = lossPct, Sent = 1000, Lost = (int)(lossPct * 10) };
|
||
|
||
/// <summary>状态变化还原成每个时间桶里各状态的秒数：跨桶的一段要切开，区间开头的状态取第一次变化的「从」，不同方法各算各的。</summary>
|
||
[Fact]
|
||
public void Summarize_SplitsStateTimeAcrossBuckets_AndWeightsTheAverageRateByTime()
|
||
{
|
||
RateChange[] changes =
|
||
[
|
||
Change(s_t0, "quic_ping", "start", "watch", 10, "start"), // 10:00 起步：watch 10 Hz
|
||
Change(s_t0 + (30 * Minute), "quic_ping", "watch", "quiet", 2), // 10:30 起 quiet 2 Hz
|
||
Change(s_t0 + (90 * Minute), "quic_ping", "quiet", "active", 10, "sustained loss", 0.4), // 11:30 起 active 10 Hz
|
||
Change(s_t0 + (30 * Minute), "tcp_ping", "quiet", "watch", 20, "rtt above baseline"), // 这条方法 10:30 才有第一条记录
|
||
];
|
||
|
||
AdaptiveResult result = AdaptiveAudit.Summarize(changes, s_t0, s_t0 + (2 * Hour), Hour);
|
||
|
||
AdaptiveKind quic = Assert.Single(result.Kinds, k => k.Kind == "quic_ping");
|
||
Assert.Equal(2, quic.Buckets.Count);
|
||
AdaptiveBucket first = quic.Buckets[0];
|
||
Assert.Equal((1800, 1800, 0), ((int)first.WatchSeconds, (int)first.QuietSeconds, (int)first.ActiveSeconds)); // 10:00–10:30 watch，10:30–11:00 quiet
|
||
Assert.Equal(6, first.AverageHz); // (10 Hz × 30 min + 2 Hz × 30 min) / 60 min
|
||
Assert.Equal(2, first.Changes);
|
||
AdaptiveBucket second = quic.Buckets[1];
|
||
Assert.Equal((1800, 1800), ((int)second.QuietSeconds, (int)second.ActiveSeconds)); // 11:00–11:30 quiet，11:30–12:00 active
|
||
Assert.Equal(6, second.AverageHz); // (2 Hz × 30 min + 10 Hz × 30 min) / 60 min
|
||
Assert.Equal((1, 0.4), (second.Changes, second.MaxLossPct));
|
||
|
||
AdaptiveKind tcp = Assert.Single(result.Kinds, k => k.Kind == "tcp_ping");
|
||
Assert.Equal((0, 1800), ((int)tcp.Buckets[0].QuietSeconds, (int)tcp.Buckets[0].WatchSeconds)); // 10:00–10:30 没有记录，不算任何状态
|
||
Assert.Equal(3600, (int)tcp.Buckets[1].WatchSeconds);
|
||
|
||
Assert.Equal(4, result.Changes.Count);
|
||
Assert.Equal("quic_ping", result.Changes[0].Kind); // 最新的在前
|
||
}
|
||
|
||
/// <summary>变化在区间开头之前发生：那之后的状态延续到区间里。</summary>
|
||
[Fact]
|
||
public void Summarize_CarriesTheStateInFromBeforeTheRange()
|
||
{
|
||
RateChange[] changes = [Change(s_t0 - (2 * Hour), "quic_ping", "quiet", "active", 10, "sustained loss", 1)];
|
||
|
||
AdaptiveResult result = Summarize(changes);
|
||
|
||
AdaptiveBucket bucket = Assert.Single(Assert.Single(result.Kinds).Buckets);
|
||
Assert.Equal(3600, (int)bucket.ActiveSeconds);
|
||
Assert.Equal((10, 0), (bucket.AverageHz, bucket.Changes));
|
||
Assert.Empty(result.Changes); // 区间里没有变化
|
||
|
||
AdaptiveResult Summarize(RateChange[] c) => AdaptiveAudit.Summarize(c, s_t0, s_t0 + Hour, Hour);
|
||
}
|
||
|
||
/// <summary>端到端：拨号节点随心跳上报的状态变化存成事件，几天后还能通过接口按时段查到。</summary>
|
||
[Fact]
|
||
public async Task RateChangesReportedInHeartbeats_AreStored_AndServedByTheAdaptiveEndpoint()
|
||
{
|
||
await using AgentSimulator first = await _backend.ConnectAgentAsync();
|
||
await using AgentSimulator second = await _backend.ConnectAgentAsync();
|
||
(AgentSimulator dialer, AgentSimulator listener) = string.CompareOrdinal(first.NodeId.Value, second.NodeId.Value) < 0 ? (first, second) : (second, first); // 小 id 拨大 id
|
||
long now = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds();
|
||
|
||
RateChange Of(long atMs, string from, string to, double hz, string reason, double loss)
|
||
{
|
||
RateChange change = Change(atMs, "quic_ping", from, to, hz, reason, loss);
|
||
change.Listener = listener.NodeId.Value;
|
||
change.IpVersion = "ipv4";
|
||
return change;
|
||
}
|
||
|
||
await dialer.SendRateChangesAsync(Of(now - (20 * Minute), "watch", "quiet", 2, "quiet", 0), Of(now - (10 * Minute), "quiet", "active", 10, "sustained loss", 0.5));
|
||
|
||
string lineId = listener.NodeId.Value + "~" + dialer.NodeId.Value;
|
||
JsonNode? audit = null;
|
||
await BackendFixture.EventuallyAsync(async () =>
|
||
{
|
||
audit = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{lineId}/adaptive?range=1h&step=10m");
|
||
return audit["changes"]!.AsArray().Count == 2;
|
||
});
|
||
|
||
JsonNode quic = Assert.Single(audit!["kinds"]!.AsArray())!;
|
||
Assert.Equal("quic_ping", quic["kind"]!.GetValue<string>());
|
||
double active = quic["buckets"]!.AsArray().Sum(b => b!["active_seconds"]!.GetValue<double>());
|
||
Assert.InRange(active, 500, 700); // 最近 10 分钟 active
|
||
JsonNode newest = audit!["changes"]![0]!;
|
||
Assert.Equal(("quiet", "active", "sustained loss"), (newest["from"]!.GetValue<string>(), newest["to"]!.GetValue<string>(), newest["reason"]!.GetValue<string>()));
|
||
}
|
||
}
|