lux: raise tunnel coverage >90% and add a live production e2e test (spec 18)

- internal/tunnel to 90.4%: cover DefaultURL, NodeID persistence (incl. the
  no-XDG home path), writeError, the forwarder happy/unreachable/closed-
  stream paths, getJSON errors, bad-JSON discovery, runSession discover
  error, Run context-cancel exit, and heartbeatLoop cancel.
- prod_e2e_test: an opt-in real e2e (LATERE_LUX_E2E=1 + LATERE_LUX_TOKEN)
  that runs `latere lux serve` against a live Lux (production by default),
  waits for the local model to appear in the catalog, and asserts a real
  completion via /local/v1. Skips without the opt-in / token / local Ollama.

Author: changkun

internal/tunnel/extra_test.go

@@ -0,0 +1,209 @@
+package tunnel
+
+import (
+	"bufio"
+	"context"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestDefaultURL(t *testing.T) {
+	cases := map[string]string{
+		RuntimeOllama:   "http://localhost:11434",
+		RuntimeVLLM:     "http://localhost:8000",
+		RuntimeLMStudio: "http://localhost:1234",
+		RuntimeLlamaCPP: "http://localhost:8080",
+		RuntimeMLX:      "http://localhost:8080",
+		"unknown":       "http://localhost:11434",
+	}
+	for runtime, want := range cases {
+		if got := DefaultURL(runtime); got != want {
+			t.Errorf("DefaultURL(%q) = %q, want %q", runtime, got, want)
+		}
+	}
+}
+
+func TestNodeIDStableAndPersisted(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("XDG_CONFIG_HOME", dir)
+	first := NodeID()
+	if !strings.HasPrefix(first, "node-") {
+		t.Errorf("NodeID = %q, want node- prefix", first)
+	}
+	if second := NodeID(); second != first {
+		t.Errorf("NodeID not stable: %q vs %q", first, second)
+	}
+	// Persisted to disk and reused across the helper's path.
+	if got := nodeIDPath(); got == "" || !strings.Contains(got, "tunnel-node-id") {
+		t.Errorf("nodeIDPath = %q", got)
+	}
+}
+
+func TestNodeIDPathUsesHomeWhenNoXDG(t *testing.T) {
+	t.Setenv("XDG_CONFIG_HOME", "")
+	got := nodeIDPath()
+	if got == "" || !strings.Contains(got, "latere") {
+		t.Errorf("nodeIDPath without XDG = %q, want a path under the home config dir", got)
+	}
+}
+
+func TestDiscoverBadJSON(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte("not json"))
+	}))
+	defer srv.Close()
+	if _, err := discover(context.Background(), srv.Client(), RuntimeVLLM, srv.URL, nil); err == nil {
+		t.Error("discover with bad JSON = nil error, want decode error")
+	}
+}
+
+func TestHandleClosedStream(t *testing.T) {
+	f := &forwarder{ctx: context.Background(), client: &http.Client{}, upstream: "http://127.0.0.1:1"}
+	c1, c2 := net.Pipe()
+	_ = c2.Close() // peer hangs up before sending a request
+	// handle must return without panicking on the ReadRequest error.
+	done := make(chan struct{})
+	go func() { f.handle(c1); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Error("handle did not return on a closed stream")
+	}
+}
+
+func TestWriteErrorProducesBadGateway(t *testing.T) {
+	c1, c2 := net.Pipe()
+	go func() {
+		writeError(c1, io.EOF)
+		_ = c1.Close()
+	}()
+	resp, err := http.ReadResponse(bufio.NewReader(c2), nil)
+	if err != nil {
+		t.Fatalf("read response: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusBadGateway {
+		t.Errorf("status = %d, want 502", resp.StatusCode)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	if !strings.Contains(string(body), "local.unreachable") {
+		t.Errorf("body = %s, want local.unreachable", body)
+	}
+}
+
+func TestForwarderUnreachableUpstream(t *testing.T) {
+	f := &forwarder{ctx: context.Background(), client: &http.Client{Timeout: time.Second}, upstream: "http://127.0.0.1:1"}
+	c1, c2 := net.Pipe()
+	go f.handle(c1)
+
+	req, _ := http.NewRequest(http.MethodPost, "http://x/v1/chat/completions", strings.NewReader(`{}`))
+	go func() { _ = req.Write(c2) }()
+	resp, err := http.ReadResponse(bufio.NewReader(c2), req)
+	if err != nil {
+		t.Fatalf("read response: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusBadGateway {
+		t.Errorf("status = %d, want 502 for unreachable upstream", resp.StatusCode)
+	}
+}
+
+func TestForwarderForwardsToUpstream(t *testing.T) {
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte("ok:" + r.URL.Path))
+	}))
+	defer upstream.Close()
+
+	f := &forwarder{ctx: context.Background(), client: &http.Client{}, upstream: upstream.URL}
+	c1, c2 := net.Pipe()
+	go f.handle(c1)
+	req, _ := http.NewRequest(http.MethodPost, "http://x/v1/chat/completions", strings.NewReader(`{}`))
+	go func() { _ = req.Write(c2) }()
+	resp, err := http.ReadResponse(bufio.NewReader(c2), req)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+	if string(body) != "ok:/v1/chat/completions" {
+		t.Errorf("body = %q", body)
+	}
+}
+
+func TestGetJSONErrors(t *testing.T) {
+	// Non-200 surfaces an error.
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer srv.Close()
+	var dst struct{}
+	if err := getJSON(context.Background(), srv.Client(), srv.URL, &dst); err == nil {
+		t.Error("getJSON on 500 = nil error, want error")
+	}
+	// Unreachable host surfaces an error.
+	if err := getJSON(context.Background(), &http.Client{Timeout: time.Second}, "http://127.0.0.1:1/x", &dst); err == nil {
+		t.Error("getJSON on unreachable = nil error, want error")
+	}
+}
+
+func TestRunSessionDiscoverError(t *testing.T) {
+	// An unreachable upstream makes discovery fail, so runSession returns
+	// before any dial.
+	err := runSession(context.Background(), Options{
+		LuxURL:      "http://127.0.0.1:1",
+		Bearer:      func(context.Context) (string, error) { return "t", nil },
+		Runtime:     RuntimeOllama,
+		UpstreamURL: "http://127.0.0.1:1",
+		Out:         io.Discard,
+	})
+	if err == nil {
+		t.Error("runSession with unreachable upstream = nil error, want discovery error")
+	}
+}
+
+func TestRunStopsOnContextCancel(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+	// Upstream unreachable so each session fails fast; Run should keep
+	// retrying then return when the context expires.
+	err := Run(ctx, Options{
+		LuxURL:            "http://127.0.0.1:1",
+		Bearer:            func(context.Context) (string, error) { return "t", nil },
+		Runtime:           RuntimeOllama,
+		UpstreamURL:       "http://127.0.0.1:1",
+		HeartbeatInterval: time.Second,
+		Out:               io.Discard,
+	})
+	if err == nil {
+		t.Error("Run = nil error on context expiry, want ctx error")
+	}
+}
+
+func TestHeartbeatLoopStopsOnContextCancel(t *testing.T) {
+	c1, c2 := net.Pipe()
+	defer c1.Close()
+	defer c2.Close()
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		heartbeatLoop(ctx, c1, Options{
+			HeartbeatInterval: 10 * time.Millisecond,
+			Bearer:            func(context.Context) (string, error) { return "tok", nil },
+		})
+		close(done)
+	}()
+	// Drain the pipe so heartbeat writes don't block.
+	go io.Copy(io.Discard, c2)
+	time.Sleep(40 * time.Millisecond)
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Error("heartbeatLoop did not stop on context cancel")
+	}
+}

internal/tunnel/prod_e2e_test.go

@@ -0,0 +1,120 @@
+package tunnel
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+// TestProdE2EServeAndCall is a real end-to-end test of `latere lux serve`
+// against a live Lux server (production by default). It is opt-in because
+// it exposes the local model to the target Lux and needs a real credential:
+// it skips unless LATERE_LUX_E2E=1 and LATERE_LUX_TOKEN are set, and unless
+// a local Ollama serving the model is reachable.
+//
+// Run it with, e.g.:
+//
+//	LATERE_LUX_E2E=1 \
+//	LATERE_LUX_TOKEN="$(latere lux token)" \
+//	go test ./internal/tunnel/ -run TestProdE2E -v
+//
+// Override LUX_API_URL to target staging, and LATERE_LUX_E2E_MODEL to use a
+// different local model (default gemma4:latest).
+func TestProdE2EServeAndCall(t *testing.T) {
+	if os.Getenv("LATERE_LUX_E2E") != "1" {
+		t.Skip("set LATERE_LUX_E2E=1 to run the live Lux e2e (it exposes a local model to the target)")
+	}
+	token := os.Getenv("LATERE_LUX_TOKEN")
+	if token == "" {
+		t.Skip("LATERE_LUX_E2E set but LATERE_LUX_TOKEN missing; provide a bearer (e.g. `latere lux token`)")
+	}
+	luxURL := os.Getenv("LUX_API_URL")
+	if luxURL == "" {
+		luxURL = "https://lux.latere.ai"
+	}
+	luxURL = strings.TrimRight(luxURL, "/")
+	model := os.Getenv("LATERE_LUX_E2E_MODEL")
+	if model == "" {
+		model = "gemma4:latest"
+	}
+
+	// Require a reachable local Ollama serving the model.
+	hc := &http.Client{Timeout: 5 * time.Second}
+	if _, err := discover(context.Background(), hc, RuntimeOllama, DefaultURL(RuntimeOllama), []string{model}); err != nil {
+		t.Skipf("local Ollama not serving %q: %v", model, err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Start the real serve loop against the live Lux.
+	serveErr := make(chan error, 1)
+	go func() {
+		serveErr <- Run(ctx, Options{
+			LuxURL:            luxURL,
+			Bearer:            func(context.Context) (string, error) { return token, nil },
+			Runtime:           RuntimeOllama,
+			Models:            []string{model},
+			NodeID:            "e2e-prod-" + NodeID(),
+			HeartbeatInterval: 5 * time.Second,
+			Out:               io.Discard,
+		})
+	}()
+
+	// Wait until the model shows up as a live tunnel in the catalog,
+	// confirming the tunnel registered through the live server.
+	if !waitForCatalog(t, hc, luxURL, token, model, 30*time.Second) {
+		t.Fatalf("model %q never appeared as local/ in %s/lux/v1/models within 30s "+
+			"(is the tunnel feature enabled on the target and does the token carry llm.serve?)", model, luxURL)
+	}
+
+	// Call it through Lux and assert a real completion comes back.
+	body := `{"model":"` + model + `","stream":false,"max_tokens":20,` +
+		`"messages":[{"role":"user","content":"Reply with exactly: PONG"}]}`
+	req, _ := http.NewRequestWithContext(ctx, http.MethodPost, luxURL+"/local/v1/chat/completions", strings.NewReader(body))
+	req.Header.Set("Authorization", "Bearer "+token)
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := (&http.Client{Timeout: 120 * time.Second}).Do(req)
+	if err != nil {
+		t.Fatalf("call /local: %v", err)
+	}
+	defer resp.Body.Close()
+	respBody, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("/local status = %d: %s", resp.StatusCode, respBody)
+	}
+	var out struct {
+		Choices []struct {
+			Message struct{ Content string } `json:"message"`
+		} `json:"choices"`
+	}
+	if err := json.Unmarshal(respBody, &out); err != nil || len(out.Choices) == 0 || out.Choices[0].Message.Content == "" {
+		t.Fatalf("no completion via prod tunnel: err=%v body=%s", err, respBody)
+	}
+	t.Logf("prod e2e: %s replied %q via %s/local/v1", model, out.Choices[0].Message.Content, luxURL)
+}
+
+func waitForCatalog(t *testing.T, hc *http.Client, luxURL, token, model string, timeout time.Duration) bool {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		req, _ := http.NewRequest(http.MethodGet, luxURL+"/lux/v1/models", nil)
+		req.Header.Set("Authorization", "Bearer "+token)
+		resp, err := hc.Do(req)
+		if err == nil {
+			b, _ := io.ReadAll(resp.Body)
+			_ = resp.Body.Close()
+			if resp.StatusCode == http.StatusOK && bytes.Contains(b, []byte(`"local/`+model+`"`)) {
+				return true
+			}
+		}
+		time.Sleep(time.Second)
+	}
+	return false
+}