restic-manager/internal/server/http/tree_rpc_test.go

// tree_rpc_test.go — full round-trip test for the tree.list synchronous
// RPC (P3-X2). A fake agent reads the inbound tree.list, replies with a
// canned tree.list.result, and we assert the server's SendRPC returned
// the expected payload.
package http

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	"github.com/coder/websocket"

	"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
)

func TestSendRPCTreeListRoundTrip(t *testing.T) {
	t.Parallel()
	srv, ts, st := rawTestServer(t)
	hostID, token := enrolHostForWS(t, srv, st, "rpc-host")
	c := agentDial(t, srv, ts, hostID, token)
	sendHello(t, c, "rpc-host")
	_ = drainUntil(t, c, api.MsgScheduleSet)

	// Fake agent: read inbound envelopes, mirror tree.list with a
	// canned result. Other inbound envelopes (config.update etc) are
	// already drained above.
	done := make(chan error, 1)
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		for {
			mt, raw, err := c.Read(ctx)
			if err != nil {
				done <- err
				return
			}
			if mt != websocket.MessageText {
				continue
			}
			var env api.Envelope
			if err := json.Unmarshal(raw, &env); err != nil {
				done <- err
				return
			}
			if env.Type != api.MsgTreeList {
				continue
			}
			var req api.TreeListRequestPayload
			if err := env.UnmarshalPayload(&req); err != nil {
				done <- err
				return
			}
			result := api.TreeListResultPayload{
				SnapshotID: req.SnapshotID,
				Path:       req.Path,
				Entries: []api.TreeListEntry{
					{Name: "etc", Type: "dir"},
					{Name: "var", Type: "dir"},
				},
			}
			out, err := api.Marshal(api.MsgTreeListResult, env.ID, result)
			if err != nil {
				done <- err
				return
			}
			rawOut, _ := json.Marshal(out)
			if err := c.Write(ctx, websocket.MessageText, rawOut); err != nil {
				done <- err
				return
			}
			done <- nil
			return
		}
	}()

	// Server-side SendRPC.
	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
	defer cancel()
	reply, err := srv.deps.Hub.SendRPC(ctx, hostID, api.MsgTreeList,
		api.TreeListRequestPayload{SnapshotID: "f3a7b2c1", Path: "/"},
		3*time.Second)
	if err != nil {
		t.Fatalf("SendRPC: %v", err)
	}
	if reply.Type != api.MsgTreeListResult {
		t.Fatalf("reply type: got %q want %q", reply.Type, api.MsgTreeListResult)
	}
	var result api.TreeListResultPayload
	if err := reply.UnmarshalPayload(&result); err != nil {
		t.Fatalf("unmarshal reply: %v", err)
	}
	if result.SnapshotID != "f3a7b2c1" || result.Path != "/" {
		t.Fatalf("payload: got %+v", result)
	}
	if len(result.Entries) != 2 || result.Entries[0].Name != "etc" {
		t.Fatalf("entries: %+v", result.Entries)
	}

	// Make sure the fake agent didn't error out.
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("fake agent: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("fake agent didn't finish")
	}
}

// TestSendRPCTimeoutNoReply: SendRPC times out cleanly when the agent
// never replies; the registry entry is released so a stray late reply
// wouldn't deadlock anything.
func TestSendRPCTimeoutNoReply(t *testing.T) {
	t.Parallel()
	srv, ts, st := rawTestServer(t)
	hostID, token := enrolHostForWS(t, srv, st, "rpc-timeout-host")
	c := agentDial(t, srv, ts, hostID, token)
	sendHello(t, c, "rpc-timeout-host")
	_ = drainUntil(t, c, api.MsgScheduleSet)

	// Fake agent reads but never replies.
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()
		for {
			if _, _, err := c.Read(ctx); err != nil {
				return
			}
		}
	}()

	ctx := context.Background()
	t0 := time.Now()
	_, err := srv.deps.Hub.SendRPC(ctx, hostID, api.MsgTreeList,
		api.TreeListRequestPayload{SnapshotID: "x", Path: "/"},
		300*time.Millisecond)
	if err == nil {
		t.Fatal("expected timeout error")
	}
	elapsed := time.Since(t0)
	if elapsed < 250*time.Millisecond || elapsed > 2*time.Second {
		t.Fatalf("timeout took %s, expected ~300ms", elapsed)
	}
}