From 792354959c53c2e6fdcd5fe064df93522bbf2f32 Mon Sep 17 00:00:00 2001 From: andig Date: Mon, 13 Jul 2026 17:29:35 +0200 Subject: [PATCH] chore: test remote tunnel connection retry (#31760) --- server/remote/tunnel_test.go | 172 +++++++++++++++++++++++++++++++++++ 1 file changed, 172 insertions(+) create mode 100644 server/remote/tunnel_test.go diff --git a/server/remote/tunnel_test.go b/server/remote/tunnel_test.go new file mode 100644 index 000000000..3bad1eb48 --- /dev/null +++ b/server/remote/tunnel_test.go @@ -0,0 +1,172 @@ +package remote + +import ( + "bufio" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/evcc-io/evcc/util" + "github.com/hashicorp/yamux" + "github.com/stretchr/testify/require" +) + +// serveSession upgrades the request to a websocket, wraps it in a yamux server +// session, hands it to the sessions channel and blocks until the session closes. +func serveSession(w http.ResponseWriter, r *http.Request, sessions chan<- *yamux.Session) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + netConn := websocket.NetConn(r.Context(), conn, websocket.MessageBinary) + + session, err := yamux.Server(netConn, yamux.DefaultConfig()) + if err != nil { + netConn.Close() + return + } + + sessions <- session + <-session.CloseChan() // keep the handler (and websocket) alive until closed +} + +// tunnelTestServer accepts websocket connections and hands each yamux server +// session to the sessions channel. Simulates the cloud proxy. +func tunnelTestServer(t *testing.T, sessions chan<- *yamux.Session) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + serveSession(w, r, sessions) + })) +} + +// requireReachable opens a stream through the server session and issues an +// authenticated request, asserting the client's handler answers. +func requireReachable(t *testing.T, session *yamux.Session) { + t.Helper() + + stream, err := session.Open() + require.NoError(t, err) + defer stream.Close() + + req, err := http.NewRequest(http.MethodGet, "http://tunnel/", nil) + require.NoError(t, err) + req.SetBasicAuth("user", "pass") + require.NoError(t, req.Write(stream)) + + resp, err := http.ReadResponse(bufio.NewReader(stream), req) + require.NoError(t, err) + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, http.StatusOK, resp.StatusCode) + require.Equal(t, "pong", string(body)) +} + +// TestTunnelReconnect verifies the client becomes reachable again after the +// server closes the tunnel connection and later accepts a new one. +func TestTunnelReconnect(t *testing.T) { + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("pong")) + }) + authenticate := func(user, pass string) bool { return true } + + sessions := make(chan *yamux.Session, 4) + srv := tunnelTestServer(t, sessions) + defer srv.Close() + + wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + + tun := NewTunnel(wsURL, "token", handler, authenticate, nil, util.NewLogger("test"), nil) + go tun.run() + defer tun.Close() + + // first connection: client reachable through the tunnel + session := waitSession(t, sessions) + requireReachable(t, session) + require.True(t, tun.IsConnected()) + + // server drops the tunnel connection + session.Close() + + // client reconnects (backoff ~1s) and is reachable again + session = waitSession(t, sessions) + requireReachable(t, session) + require.True(t, tun.IsConnected()) +} + +// TestTunnelRejectedCredentialsStops verifies the client does not retry when +// the proxy rejects credentials (401/403); a new token requires a restart. +func TestTunnelRejectedCredentialsStops(t *testing.T) { + var attempts atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + attempts.Add(1) + w.WriteHeader(http.StatusUnauthorized) + })) + defer srv.Close() + + wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + + tun := NewTunnel(wsURL, "token", nil, nil, nil, util.NewLogger("test"), nil) + defer tun.Close() + + done := make(chan struct{}) + go func() { tun.run(); close(done) }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("run did not return after credential rejection") + } + + require.Equal(t, int32(1), attempts.Load(), "must not retry after credential rejection") + require.False(t, tun.IsConnected()) +} + +// TestTunnelReconnectsAfterTransientError verifies the client keeps retrying +// through a transient proxy failure and connects once the proxy recovers. +func TestTunnelReconnectsAfterTransientError(t *testing.T) { + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("pong")) + }) + authenticate := func(user, pass string) bool { return true } + + var attempts atomic.Int32 + sessions := make(chan *yamux.Session, 4) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // first attempt fails transiently, later ones succeed + if attempts.Add(1) == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + serveSession(w, r, sessions) + })) + defer srv.Close() + + wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + + tun := NewTunnel(wsURL, "token", handler, authenticate, nil, util.NewLogger("test"), nil) + go tun.run() + defer tun.Close() + + session := waitSession(t, sessions) + requireReachable(t, session) + require.True(t, tun.IsConnected()) + require.GreaterOrEqual(t, attempts.Load(), int32(2), "must retry after transient failure") +} + +func waitSession(t *testing.T, sessions <-chan *yamux.Session) *yamux.Session { + t.Helper() + select { + case s := <-sessions: + return s + case <-time.After(10 * time.Second): + t.Fatal("timeout waiting for tunnel connection") + return nil + } +}