test(server): 刀2 — v2ray 真实采集后端测试(缺口②)

V2RayUsageSource 是「线上按租户记账」链路的源头(agent 读 sing-box v2ray_api StatsService per-user 计数器 → 聚合 → ReportUsage),此前零测试覆盖 (复查发现之前以为有单测是错的)。本刀按用户定的范围只补 CI 可跑的后端采集: - TestParseUserStat:表驱动,正常 uplink/downlink + 非 user/段数异常应 ok=false。 - TestV2RayCollect:起假 StatsService(loopback gRPC,ServiceName 与 client.go 的规范全名一致),让真 Collect() 跑真链路,断言:按 dp_uuid 聚合 up/down 正确、方向不串、全 0 用户与非 user 计数器被丢弃、SessionMinutes=1、且请求以 Reset_=true + pattern "user>>>" 发出(窗口取值清零语义)。纯 go test、无 docker / 无真 sing-box;自动进 go-server job + 计入覆盖率闸。真 sing-box 真出网 e2e 仍按计划留文档手测。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-25 23:01:27 +08:00
parent 60daa6b399
commit 75cdbf583f
1 changed files with 116 additions and 0 deletions
@@ -0,0 +1,116 @@
+package agentd
+
+// 问题② 真实流量采集后端测试(CI 可跑,无需 docker / 真 sing-box)。
+//
+// V2RayUsageSource 是「线上按租户记账」链路的源头:agent 读 sing-box v2ray_api
+// StatsService 的 per-user 计数器 → 聚合 → ReportUsage。此前零测试覆盖。
+// 这里:① parseUserStat 纯函数表驱动;② 起一个假 StatsService(loopback gRPC,
+// 注册与 client.go 一致的规范全名),让真 Collect() 跑真解析/聚合/清零语义。
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/wangjia/pangolin/server/internal/agentd/v2rayapi"
+	"google.golang.org/grpc"
+)
+
+func TestParseUserStat(t *testing.T) {
+	cases := []struct {
+		name     string
+		wantUUID string
+		wantDir  string
+		wantOK   bool
+	}{
+		{"user>>>abc>>>traffic>>>uplink", "abc", "uplink", true},
+		{"user>>>abc>>>traffic>>>downlink", "abc", "downlink", true},
+		{"inbound>>>mixed>>>traffic>>>uplink", "", "", false},    // 非 user 计数器
+		{"user>>>abc>>>other>>>uplink", "", "", false},           // 第三段非 traffic
+		{"user>>>abc>>>traffic", "", "", false},                  // 段数不足
+		{"user>>>abc>>>traffic>>>uplink>>>extra", "", "", false}, // 段数过多
+		{"", "", "", false},
+	}
+	for _, c := range cases {
+		uuid, dir, ok := parseUserStat(c.name)
+		if ok != c.wantOK || uuid != c.wantUUID || dir != c.wantDir {
+			t.Errorf("parseUserStat(%q) = (%q,%q,%v), want (%q,%q,%v)",
+				c.name, uuid, dir, ok, c.wantUUID, c.wantDir, c.wantOK)
+		}
+	}
+}
+
+// fakeStats 实现 v2ray StatsService.QueryStats:返回预设计数器,记录收到的请求。
+type fakeStats struct {
+	resp    *v2rayapi.QueryStatsResponse
+	lastReq *v2rayapi.QueryStatsRequest
+}
+
+func (f *fakeStats) queryStats(_ any, _ context.Context, dec func(any) error, _ grpc.UnaryServerInterceptor) (any, error) {
+	in := new(v2rayapi.QueryStatsRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	f.lastReq = in
+	return f.resp, nil
+}
+
+// startFakeStats 在 loopback 起假 StatsService,返回地址 + 取最近请求的闭包。
+// ServiceName 与 client.go 硬编码的规范全名一致,否则 raw Invoke 命中不到。
+func startFakeStats(t *testing.T, resp *v2rayapi.QueryStatsResponse) (string, func() *v2rayapi.QueryStatsRequest) {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	f := &fakeStats{resp: resp}
+	srv := grpc.NewServer()
+	srv.RegisterService(&grpc.ServiceDesc{
+		ServiceName: "v2ray.core.app.stats.command.StatsService",
+		HandlerType: (*any)(nil),
+		Methods:     []grpc.MethodDesc{{MethodName: "QueryStats", Handler: f.queryStats}},
+	}, f)
+	go func() { _ = srv.Serve(lis) }()
+	t.Cleanup(srv.Stop)
+	return lis.Addr().String(), func() *v2rayapi.QueryStatsRequest { return f.lastReq }
+}
+
+func TestV2RayCollect(t *testing.T) {
+	resp := &v2rayapi.QueryStatsResponse{Stat: []*v2rayapi.Stat{
+		{Name: "user>>>uuid-a>>>traffic>>>uplink", Value: 1000},
+		{Name: "user>>>uuid-a>>>traffic>>>downlink", Value: 2000},
+		{Name: "user>>>uuid-b>>>traffic>>>uplink", Value: 500}, // b 只有上行
+		{Name: "user>>>uuid-b>>>traffic>>>downlink", Value: 0},
+		{Name: "user>>>uuid-c>>>traffic>>>uplink", Value: 0}, // c 全 0 → 应丢弃
+		{Name: "user>>>uuid-c>>>traffic>>>downlink", Value: 0},
+		{Name: "inbound>>>mixed>>>traffic>>>uplink", Value: 9999}, // 非 user → 应丢弃
+	}}
+	addr, lastReq := startFakeStats(t, resp)
+
+	s := &V2RayUsageSource{addr: addr}
+	entries := s.Collect()
+
+	// 窗口语义:必须以 reset=true + pattern "user>>>" 拉取(取值并清零)。
+	if r := lastReq(); r == nil || !r.Reset_ || r.Pattern != "user>>>" {
+		t.Fatalf("QueryStats 请求 = %+v, 期望 Reset_=true Pattern=\"user>>>\"", r)
+	}
+
+	byUUID := make(map[string]struct{ up, down, mins int64 })
+	for _, e := range entries {
+		byUUID[e.DpUUID] = struct{ up, down, mins int64 }{e.BytesUp, e.BytesDown, e.SessionMinutes}
+	}
+
+	// uuid-c(全 0)与非 user 计数器都不应入账。
+	if len(entries) != 2 {
+		t.Fatalf("entries=%d (%+v), 期望 2(a/b;c 全0 与 inbound 丢弃)", len(entries), byUUID)
+	}
+	if a := byUUID["uuid-a"]; a.up != 1000 || a.down != 2000 || a.mins != 1 {
+		t.Errorf("uuid-a = %+v, 期望 up=1000 down=2000 mins=1", a)
+	}
+	if b := byUUID["uuid-b"]; b.up != 500 || b.down != 0 || b.mins != 1 {
+		t.Errorf("uuid-b = %+v, 期望 up=500 down=0 mins=1", b)
+	}
+	if _, ok := byUUID["uuid-c"]; ok {
+		t.Errorf("uuid-c 全 0 不应入账")
+	}
+}