pangolin/server/test/e2e/smoke_test.go

//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} }