monitoring/internal/fleet/gate_test.go

package fleet

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"testing"
	"time"
)

// gateFleet is the fixture: two pods at the tip, one inside MAX_LAG, one
// outside it, one whose RPC is dead, and one that is not Running. The starting
// labels are deliberately wrong in BOTH directions so every branch fires — a
// lagging pod currently labelled serving must be unlabelled, and a caught-up
// pod that is not labelled must be labelled.
var gateFleet = []struct {
	Name, IP, Phase string
	Tip             int64 // -1 = RPC dead
	Label           string
}{
	{"luxd-0", "10.0.0.10", "Running", 1000, "fresh"},
	{"luxd-1", "10.0.0.11", "Running", 999, ""},
	{"luxd-2", "10.0.0.12", "Running", 997, "fresh"},
	{"luxd-3", "10.0.0.13", "Running", -1, "fresh"},
	{"luxd-4", "10.0.0.14", "Pending", 1000, ""},
}

// fakeCluster serves the pod list and the per-IP RPC, and records every PATCH.
func fakeCluster(t *testing.T, patches *[]string) *httptest.Server {
	t.Helper()
	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch {
		case r.Method == http.MethodGet && strings.Contains(r.URL.Path, "/pods"):
			items := make([]map[string]any, 0, len(gateFleet))
			for _, p := range gateFleet {
				labels := map[string]string{}
				if p.Label != "" {
					labels["lux.network/rpc"] = p.Label
				}
				items = append(items, map[string]any{
					"metadata": map[string]any{"name": p.Name, "labels": labels},
					"status":   map[string]any{"phase": p.Phase, "podIP": p.IP},
				})
			}
			_ = json.NewEncoder(w).Encode(map[string]any{"items": items})
		case r.Method == http.MethodPatch:
			b, _ := io.ReadAll(r.Body)
			parts := strings.Split(r.URL.Path, "/")
			*patches = append(*patches, parts[len(parts)-1]+" "+string(b))
			_, _ = w.Write([]byte(`{}`))
		case r.Method == http.MethodPost:
			ip := strings.TrimPrefix(strings.Split(r.URL.Path, "/v1/")[0], "/ip/")
			for _, p := range gateFleet {
				if p.IP == ip && p.Tip >= 0 {
					_, _ = fmt.Fprintf(w, `{"jsonrpc":"2.0","id":1,"result":"%#x"}`, p.Tip)
					return
				}
			}
			w.WriteHeader(http.StatusBadGateway)
		default:
			w.WriteHeader(http.StatusNotFound)
		}
	}))
}

func gateConfig(base string) GateConfig {
	return GateConfig{
		Namespace: "lux-mainnet", Selector: "app.kubernetes.io/name=luxd",
		RPCPort: "9630", RPCPath: "/v1/chain/C/rpc",
		LabelKey: "lux.network/rpc", LabelValue: "fresh",
		MaxLag: 1, Interval: time.Second, Timeout: 5 * time.Second,
		KubeBase: base,
		NodeBase: func(ip string) string { return base + "/ip/" + ip },
	}
}

func TestGateLabelsTheCaughtUpAndUnlabelsTheRest(t *testing.T) {
	var patches []string
	srv := fakeCluster(t, &patches)
	defer srv.Close()

	g := NewGate(gateConfig(srv.URL))
	var lines []string
	g.Log = func(s string) { lines = append(lines, s) }

	if err := g.Reconcile(context.Background()); err != nil {
		t.Fatal(err)
	}

	wantLines := []string{
		"luxd-0 tip=1000 lag=0 serving=true",
		"luxd-1 tip=999 lag=1 serving=true (changed)",
		"luxd-2 tip=997 lag=3 serving=false (changed)",
		"luxd-3 tip=None lag=- serving=false (changed)",
		"luxd-4 tip=None lag=- serving=false",
	}
	if strings.Join(lines, "\n") != strings.Join(wantLines, "\n") {
		t.Fatalf("decision log drift:\n got:\n%s\nwant:\n%s",
			strings.Join(lines, "\n"), strings.Join(wantLines, "\n"))
	}

	// Exactly three writes, and each is the minimal merge-patch: the one pod
	// that earned the label, and the two that lost it (null deletes the key).
	wantPatches := []string{
		`luxd-1 {"metadata":{"labels":{"lux.network/rpc":"fresh"}}}`,
		`luxd-2 {"metadata":{"labels":{"lux.network/rpc":null}}}`,
		`luxd-3 {"metadata":{"labels":{"lux.network/rpc":null}}}`,
	}
	if strings.Join(patches, "\n") != strings.Join(wantPatches, "\n") {
		t.Fatalf("patch drift:\n got:\n%s\nwant:\n%s",
			strings.Join(patches, "\n"), strings.Join(wantPatches, "\n"))
	}
}

// The one failure mode that must never happen: losing sight of the fleet is not
// evidence the fleet is unfit. If nothing answers, the labels stand — emptying
// the pool would take the public endpoint down outright.
func TestGateFailsStaticWhenNothingAnswers(t *testing.T) {
	var patches []string
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method == http.MethodGet {
			_, _ = w.Write([]byte(`{"items":[
			  {"metadata":{"name":"luxd-0","labels":{"lux.network/rpc":"fresh"}},"status":{"phase":"Running","podIP":"10.0.0.10"}},
			  {"metadata":{"name":"luxd-1","labels":{}},"status":{"phase":"Running","podIP":"10.0.0.11"}}]}`))
			return
		}
		if r.Method == http.MethodPatch {
			patches = append(patches, r.URL.Path)
		}
		w.WriteHeader(http.StatusBadGateway)
	}))
	defer srv.Close()

	g := NewGate(gateConfig(srv.URL))
	var lines []string
	g.Log = func(s string) { lines = append(lines, s) }
	if err := g.Reconcile(context.Background()); err != nil {
		t.Fatal(err)
	}
	if len(patches) != 0 {
		t.Fatalf("labels must stand when nothing answered; got patches %v", patches)
	}
	if len(lines) != 1 || lines[0] != "no node answered; labels unchanged" {
		t.Fatalf("got %v", lines)
	}
}

// TestGateAgainstFakeCluster drives the real reconcile against the shared fake
// the Python gate is also pointed at, and writes its decision log for diffing.
func TestGateAgainstFakeCluster(t *testing.T) {
	base := os.Getenv("GATE_FAKE")
	if base == "" {
		t.Skip("set GATE_FAKE=http://host:port to run the differential")
	}
	g := NewGate(gateConfig(base))
	var lines []string
	g.Log = func(s string) { lines = append(lines, s) }
	if err := g.Reconcile(context.Background()); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(os.Getenv("GATE_OUT"), []byte(strings.Join(lines, "\n")+"\n"), 0o644); err != nil {
		t.Fatal(err)
	}
}