//go:build e2e // Package e2e 是 L4 进程级端到端冒烟的 driver(build tag e2e,不进常规 go test ./...)。 // 由 scripts/e2e-smoke.sh 起好真 server 进程后,经 E2E_* env 指向真 HTTP/gRPC 地址驱动。 // // 5a:HTTP 段(造用户 → 登录 → /v1/me → /v1/usage)。5b 追加 gRPC 全链路。 package e2e import ( "bytes" "context" "crypto/tls" "crypto/x509" "encoding/json" "io" "net/http" "os" "testing" "time" "github.com/wangjia/pangolin/server/internal/agentd" "github.com/wangjia/pangolin/server/internal/auth" "github.com/wangjia/pangolin/server/internal/db" agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" "google.golang.org/grpc" "google.golang.org/grpc/credentials" ) const ( e2eEmail = "e2e@example.com" e2ePass = "e2e-pass-12345" e2eDPUUID = "e2e-dp-uuid-00000001" ) func TestE2ESmoke(t *testing.T) { base := os.Getenv("E2E_HTTP") dsn := os.Getenv("E2E_DB_DSN") if base == "" || dsn == "" { t.Fatal("E2E_HTTP / E2E_DB_DSN 未设——请经 scripts/e2e-smoke.sh 运行") } seedUser(t, dsn) // ── HTTP 用户旅程 ────────────────────────────────────────────── token := login(t, base, e2eEmail, e2ePass) if token == "" { t.Fatal("登录未返回 access_token") } me := getJSON(t, base, "/v1/me", token) if got := me["dp_uuid"]; got != e2eDPUUID { t.Errorf("/v1/me dp_uuid = %v, want %s", got, e2eDPUUID) } if got := me["plan"]; got != "free" { t.Errorf("/v1/me plan = %v, want free", got) } if got := me["email"]; got != e2eEmail { t.Errorf("/v1/me email = %v, want %s", got, e2eEmail) } // /v1/usage 鉴权通过 + 可解析(初始无流量;具体 bytes 断言见 gRPC 段注入用量后)。 assertStatus200(t, base, "/v1/usage", token) // 注:/v1/usage/devices(按设备用量)是 stats-overhaul WIP、committed 代码未注册该 // 路由,断言它会 404。待 stats-overhaul 合并后再加回。 // 无 token → 401(鉴权真生效)。 assertUnauthorized(t, base, "/v1/me") t.Log("✓ HTTP 段通过:登录 → /v1/me(dp_uuid/plan/email) → /v1/usage(鉴权)") // ── gRPC 全链路:enroll → ReportUsage → /v1/usage bytes 断言 ────── if os.Getenv("E2E_GRPC_ADDR") != "" { runGRPCChain(t, base, token) } else { t.Log("E2E_GRPC_ADDR 未设,跳过 gRPC 全链路段") } } // runGRPCChain 镜像真 agent:enroll(mTLS,bootstrap token)→ 用 enroll 到的 client // cert 调 ReportUsage 注入用量 → 轮询 HTTP /v1/usage 断言 bytes 真入库真读出。 func runGRPCChain(t *testing.T, base, token string) { t.Helper() grpcAddr := os.Getenv("E2E_GRPC_ADDR") caPEM, err := os.ReadFile(os.Getenv("E2E_CA_CERT")) if err != nil { t.Fatalf("读 CA cert: %v", err) } ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) defer cancel() // 1. enroll(复用生产 agent 的 EnsureEnrolled,真 TCP enroll dialer) cfg := agentd.Config{ BootstrapToken: os.Getenv("E2E_BOOTSTRAP_TOKEN"), StateDir: t.TempDir(), ServerName: "localhost", AgentVersion: "e2e-smoke", } nodeUUID, err := agentd.EnsureEnrolled(ctx, cfg, enrollDialer(grpcAddr, caPEM)) if err != nil { t.Fatalf("enroll: %v", err) } if nodeUUID == "" { t.Fatal("enroll 返回空 node uuid") } t.Logf("✓ enroll 成功 node=%s", nodeUUID) // 2. 用 enroll 到的 client cert 建 mTLS 连接 → ReportUsage cert, err := tls.LoadX509KeyPair(cfg.CertPath(), cfg.KeyPath()) if err != nil { t.Fatalf("加载 node 证书: %v", err) } pool := x509.NewCertPool() if !pool.AppendCertsFromPEM(caPEM) { t.Fatal("追加 CA 失败") } conn, err := grpc.NewClient(grpcAddr, grpc.WithTransportCredentials( credentials.NewTLS(&tls.Config{ Certificates: []tls.Certificate{cert}, RootCAs: pool, ServerName: "localhost", }), )) if err != nil { t.Fatalf("mTLS 拨号: %v", err) } defer conn.Close() const wantUp, wantDown = int64(1) << 30, int64(100) << 20 // 1GiB / 100MiB now := time.Now().Unix() if _, err := agentv1.NewAgentServiceClient(conn).ReportUsage(ctx, &agentv1.UsageReport{ NodeUUID: nodeUUID, WindowStartUnix: now - 60, WindowEndUnix: now, Entries: []*agentv1.UsageEntry{{ DpUUID: e2eDPUUID, BytesUp: wantUp, BytesDown: wantDown, SessionMinutes: 1, }}, }); err != nil { t.Fatalf("ReportUsage: %v", err) } t.Log("✓ ReportUsage 上报 1GiB/100MiB") // 3. 轮询 /v1/usage 直到用量入库可读(聚合是同步写,留少量重试容错) var up, down int64 for i := 0; i < 30; i++ { up, down = sumUsage(getJSON(t, base, "/v1/usage", token)) if up == wantUp && down == wantDown { t.Logf("✓ /v1/usage 读回 bytes_up=%d bytes_down=%d —— gRPC 全链路通", up, down) return } time.Sleep(150 * time.Millisecond) } t.Fatalf("/v1/usage 最终 bytes_up=%d(want %d) bytes_down=%d(want %d)", up, wantUp, down, wantDown) } // enrollDialer 返回未认证(无 client cert,bootstrap token 鉴权)的 enroll 连接, // 仅信任 server CA。enroll 拿到 client cert 后,后续 RPC 才走 mTLS。 func enrollDialer(addr string, caPEM []byte) agentd.EnrollDialer { return func(_ context.Context) (*grpc.ClientConn, error) { pool := x509.NewCertPool() pool.AppendCertsFromPEM(caPEM) return grpc.NewClient(addr, grpc.WithTransportCredentials( credentials.NewTLS(&tls.Config{RootCAs: pool, ServerName: "localhost"}), )) } } func sumUsage(m map[string]any) (up, down int64) { pts, _ := m["points"].([]any) for _, p := range pts { pm, _ := p.(map[string]any) up += toInt64(pm["bytes_up"]) down += toInt64(pm["bytes_down"]) } return up, down } func toInt64(v any) int64 { f, _ := v.(float64); return int64(f) } // seedUser 直接写 sqlite 造一个 active 用户(复用 argon2id HashPassword), // 比走注册验证码稳。server 刚起基本空闲,一次性 INSERT 锁风险极低。 func seedUser(t *testing.T, dsn string) { t.Helper() conn, err := db.Open(dsn) if err != nil { t.Fatalf("db.Open: %v", err) } defer conn.Close() hash, err := auth.HashPassword(e2ePass) if err != nil { t.Fatalf("HashPassword: %v", err) } if _, err := conn.Exec( `INSERT INTO users (uuid, email, pw_hash, dp_uuid, status) VALUES (?, ?, ?, ?, 'active')`, "e2e-user-uuid-0001", e2eEmail, hash, e2eDPUUID, ); err != nil { t.Fatalf("seed user: %v", err) } } func login(t *testing.T, base, email, pass string) string { t.Helper() body, _ := json.Marshal(map[string]string{"email": email, "password": pass}) resp, err := httpClient().Post(base+"/v1/auth/login", "application/json", bytes.NewReader(body)) if err != nil { t.Fatalf("login POST: %v", err) } defer resp.Body.Close() raw, _ := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK { t.Fatalf("login status = %d, body = %s", resp.StatusCode, raw) } var tp struct { AccessToken string `json:"access_token"` } if err := json.Unmarshal(raw, &tp); err != nil { t.Fatalf("login decode: %v (body=%s)", err, raw) } return tp.AccessToken } func getJSON(t *testing.T, base, path, token string) map[string]any { t.Helper() raw, status := doGet(t, base, path, token) if status != http.StatusOK { t.Fatalf("GET %s status = %d, body = %s", path, status, raw) } var m map[string]any if err := json.Unmarshal(raw, &m); err != nil { t.Fatalf("GET %s decode: %v (body=%s)", path, err, raw) } return m } func assertStatus200(t *testing.T, base, path, token string) { t.Helper() raw, status := doGet(t, base, path, token) if status != http.StatusOK { t.Errorf("GET %s status = %d, want 200, body = %s", path, status, raw) } } func assertUnauthorized(t *testing.T, base, path string) { t.Helper() _, status := doGet(t, base, path, "") if status != http.StatusUnauthorized { t.Errorf("GET %s (无 token) status = %d, want 401", path, status) } } func doGet(t *testing.T, base, path, token string) ([]byte, int) { t.Helper() req, _ := http.NewRequest(http.MethodGet, base+path, nil) if token != "" { req.Header.Set("Authorization", "Bearer "+token) } resp, err := httpClient().Do(req) if err != nil { t.Fatalf("GET %s: %v", path, err) } defer resp.Body.Close() raw, _ := io.ReadAll(resp.Body) return raw, resp.StatusCode } func httpClient() *http.Client { return &http.Client{Timeout: 10 * time.Second} }