Function TestIntegration_HelloAndCall

internal/nodeagent/integration_test.go:82–157

TestIntegration_HelloAndCall 验证 hello 含 config_hash + hub.Call → dispatcher 路由。

(t *testing.T)

// TestIntegration_HelloAndCall 验证 hello 含 config_hash + hub.Call → dispatcher 路由。
func TestIntegration_HelloAndCall(t *testing.T) {
	t.Parallel()
	ph := &integrationPushHandler{}
	hub, dialer, stop := startHub(t, ph)
	defer stop()

	api := newTestAPI()
	disp := NewAPIDispatcher(api, nil)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	cfg := Config{
		NodeID:        "n1",
		ServerAddr:    "passthrough:///bufnet",
		Dispatcher:    disp,
		HelloProvider: DefaultHelloProvider("n1", ConfigHasher(api)),
		Dialer: func(_ context.Context) (*grpc.ClientConn, error) {
			return grpc.NewClient("passthrough:///bufnet",
				grpc.WithContextDialer(dialer),
				grpc.WithTransportCredentials(insecure.NewCredentials()),
			)
		},
		ReconnectBackoff: []time.Duration{20 * time.Millisecond},
	}
	done := make(chan error, 1)
	go func() { done <- Run(ctx, cfg) }()
	defer func() { cancel(); <-done }()

	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if hb, _ := ph.snapshot(); len(hb) > 0 {
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	hb, _ := ph.snapshot()
	if len(hb) == 0 {
		t.Fatalf("no hello received")
	}
	var payload struct {
		NodeID     string `json:"node_id"`
		ConfigHash string `json:"config_hash"`
		Version    string `json:"version"`
	}
	if err := json.Unmarshal(hb[0], &payload); err != nil {
		t.Fatalf("decode hello: %v", err)
	}
	if payload.NodeID != "n1" {
		t.Fatalf("hello node_id=%q want n1", payload.NodeID)
	}
	// xray 未启动时 config 字符串为空，hash 也为空；这是已知行为。
	_ = payload.ConfigHash

	// 等 hub 看到这个 conn 上线。
	deadline = time.Now().Add(time.Second)
	for time.Now().Before(deadline) && !hub.IsOnline("n1") {
		time.Sleep(10 * time.Millisecond)