test: ≥95% coverage — daemon 96.9% (go), web 95.5-99% (vitest 142 tests); store.ts hooks-order crash fix
This commit is contained in:
Raphael Westphal
@@ -0,0 +1,493 @@
|
||||
package main
|
||||
|
||||
// hub_extra_test.go — session_info, spawn_status push, send/overflow
|
||||
// branches, upgrade failures, closed-store resilience.
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
|
||||
func TestHubSessionInfoUpdatesSession(t *testing.T) {
|
||||
ts, store := newTestServer(t)
|
||||
ws := dialAgent(t, ts)
|
||||
_ = ws.WriteJSON(helloFrame("s1"))
|
||||
_ = readFrame(t, ws)
|
||||
|
||||
// malformed session_info (no session payload) is ignored, conn stays up
|
||||
_ = ws.WriteJSON(map[string]any{"v": 1, "type": evSessionInfo, "sessionId": "s1", "seq": 1, "ts": 2})
|
||||
renamed := "renamed"
|
||||
// envelope sessionId missing → falls back to session.id
|
||||
info := map[string]any{"id": "s1", "name": renamed, "cwd": "/w", "model": "glm-4", "provider": "p", "startedAt": 99}
|
||||
_ = ws.WriteJSON(map[string]any{"v": 1, "type": evSessionInfo, "seq": 2, "ts": 3, "session": info})
|
||||
|
||||
waitFor(t, 2*time.Second, func() bool {
|
||||
rows, err := store.Sessions()
|
||||
if err != nil || len(rows) != 1 {
|
||||
return false
|
||||
}
|
||||
return rows[0].Info.Model == "glm-4"
|
||||
})
|
||||
rows, _ := store.Sessions()
|
||||
if rows[0].Info.ID != "s1" || rows[0].Info.Name == nil || *rows[0].Info.Name != renamed {
|
||||
t.Fatalf("session after session_info = %+v", rows[0].Info)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHubHelloFallbacksAndBadPayload(t *testing.T) {
|
||||
ts, _ := newTestServer(t)
|
||||
ws := dialAgent(t, ts)
|
||||
|
||||
// hello without session payload → logged, ignored, conn lives on
|
||||
_ = ws.WriteJSON(map[string]any{"v": 1, "type": evHello, "sessionId": "s1", "seq": 0, "ts": 1})
|
||||
// hello with session.id but no envelope sessionId → registers under session.id
|
||||
frame := helloFrame("ignored")
|
||||
frame["sessionId"] = nil
|
||||
_ = ws.WriteJSON(frame)
|
||||
welcome := readFrame(t, ws)
|
||||
if welcome["type"] != evWelcome || welcome["sessionId"] != "ignored" {
|
||||
t.Fatalf("welcome = %v, want fallback to session.id", welcome)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHubIsOnline(t *testing.T) {
|
||||
hub := NewHub(openTestStore(t))
|
||||
if hub.IsOnline("nope") {
|
||||
t.Fatal("unknown session cannot be online")
|
||||
}
|
||||
ac := &agentConn{hub: hub, conn: throwawayConn(t), send: make(chan []byte, 1), done: make(chan struct{})}
|
||||
hub.agents["s1"] = ac
|
||||
if !hub.IsOnline("s1") {
|
||||
t.Fatal("registered conn must report online")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHubBroadcastSpawnStatus(t *testing.T) {
|
||||
ts, _, hub := newTestServerHub(t)
|
||||
hub.SpawnStatus = func() []SpawnJob {
|
||||
return []SpawnJob{{Repo: "group/proj", State: stateRunning, ContainerID: "cid-1"}}
|
||||
}
|
||||
web := dialWeb(t, ts)
|
||||
if first := readFrame(t, web); first["type"] != frameSessionList {
|
||||
t.Fatalf("first frame = %v", first)
|
||||
}
|
||||
hub.BroadcastSpawnStatus()
|
||||
_ = web.SetReadDeadline(time.Now().Add(2 * time.Second))
|
||||
var got map[string]any
|
||||
for got == nil {
|
||||
if m := readFrame(t, web); m["type"] == frameSpawnStatus {
|
||||
got = m
|
||||
}
|
||||
}
|
||||
jobs := got["jobs"].([]any)
|
||||
if len(jobs) != 1 || jobs[0].(map[string]any)["repo"] != "group/proj" {
|
||||
t.Fatalf("spawn_status frame = %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// newTestServerHub is newTestServer but also hands back the hub so tests can
|
||||
// wire SpawnStatus themselves (NewServer normally does).
|
||||
func newTestServerHub(t *testing.T) (*httptest.Server, *Store, *Hub) {
|
||||
t.Helper()
|
||||
store := openTestStore(t)
|
||||
daemonToken = testToken
|
||||
hub := NewHub(store)
|
||||
srv := &Server{store: store, hub: hub, gitlab: NewGitLab(store, "https://gitlab.example")}
|
||||
ts := httptest.NewServer(srv.Routes(""))
|
||||
t.Cleanup(ts.Close)
|
||||
return ts, store, hub
|
||||
}
|
||||
|
||||
// throwawayConn is a client-side websocket whose only job is being closeable
|
||||
// (agentConn.drop touches conn).
|
||||
func throwawayConn(t *testing.T) *websocket.Conn {
|
||||
t.Helper()
|
||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
_ = c.Close()
|
||||
}))
|
||||
t.Cleanup(up.Close)
|
||||
conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(up.URL, "http"), nil)
|
||||
if err != nil {
|
||||
t.Fatalf("dial throwaway: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = conn.Close() })
|
||||
return conn
|
||||
}
|
||||
|
||||
func dialWeb(t *testing.T, ts *httptest.Server) *websocket.Conn {
|
||||
t.Helper()
|
||||
wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/ws?token=" + testToken
|
||||
web, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("dial web ws: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = web.Close() })
|
||||
return web
|
||||
}
|
||||
|
||||
func TestHubUpgradeFailures(t *testing.T) {
|
||||
ts, _ := newTestServer(t)
|
||||
|
||||
get := func(path string) int {
|
||||
req, _ := http.NewRequest(http.MethodGet, ts.URL+path, nil)
|
||||
req.Header.Set("Authorization", "Bearer "+testToken)
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("get %s: %v", path, err)
|
||||
}
|
||||
resp.Body.Close()
|
||||
return resp.StatusCode
|
||||
}
|
||||
// plain HTTP GETs (no upgrade headers) must not 500
|
||||
if code := get("/agent/ws"); code != http.StatusBadRequest {
|
||||
t.Fatalf("agent ws non-upgrade = %d, want 400", code)
|
||||
}
|
||||
if code := get("/ws?token=" + testToken); code != http.StatusBadRequest {
|
||||
t.Fatalf("web ws non-upgrade = %d, want 400", code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendToAgentBranches(t *testing.T) {
|
||||
hub := NewHub(openTestStore(t))
|
||||
|
||||
// dropped conn: done closed → ErrOffline via done branch (send kept full
|
||||
// so the send case cannot win the select race)
|
||||
dead := &agentConn{hub: hub, conn: throwawayConn(t), send: make(chan []byte, 1), done: make(chan struct{})}
|
||||
dead.send <- []byte("fill")
|
||||
dead.drop()
|
||||
hub.agents["gone"] = dead
|
||||
if err := hub.sendToAgent("gone", map[string]any{"type": evAbort}); err != ErrOffline {
|
||||
t.Fatalf("sendToAgent dropped conn = %v, want ErrOffline", err)
|
||||
}
|
||||
|
||||
// live conn but nobody drains send: timeout branch (3s)
|
||||
stuck := &agentConn{hub: hub, conn: throwawayConn(t), send: make(chan []byte, 1), done: make(chan struct{})}
|
||||
stuck.send <- []byte("fill")
|
||||
hub.agents["stuck"] = stuck
|
||||
start := time.Now()
|
||||
if err := hub.sendToAgent("stuck", map[string]any{"type": evAbort}); err != ErrOffline {
|
||||
t.Fatalf("sendToAgent stuck conn = %v, want ErrOffline", err)
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed < promptSendTimeout-100*time.Millisecond {
|
||||
t.Fatalf("sendToAgent returned after %v, want full timeout", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebClientOverflowDrops(t *testing.T) {
|
||||
// real ws conn so drop() has something to close
|
||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
_ = c.Close()
|
||||
}))
|
||||
t.Cleanup(up.Close)
|
||||
wsURL := "ws" + strings.TrimPrefix(up.URL, "http")
|
||||
clientConn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("dial: %v", err)
|
||||
}
|
||||
defer clientConn.Close()
|
||||
|
||||
hub := NewHub(openTestStore(t))
|
||||
c := &webClient{hub: hub, conn: clientConn, done: make(chan struct{})}
|
||||
for i := 0; i < webMaxPending; i++ {
|
||||
c.deliverEvent(pendingEvent{sessionID: "s", seq: int64(i), raw: []byte("{}")})
|
||||
}
|
||||
c.deliverEvent(pendingEvent{sessionID: "s", seq: 1, raw: []byte("{}")}) // overflow → dropped
|
||||
select {
|
||||
case <-c.done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("overflowed web client was not dropped")
|
||||
}
|
||||
if !c.dropped {
|
||||
t.Fatal("client must be marked dropped")
|
||||
}
|
||||
// deliveries to a dropped client are no-ops
|
||||
c.deliverControl([]byte("{}"))
|
||||
if len(c.control) != 0 {
|
||||
t.Fatal("dropped client must not queue control frames")
|
||||
}
|
||||
|
||||
// control-frame overflow path on a fresh client
|
||||
c2 := &webClient{hub: hub, conn: clientConn, done: make(chan struct{})}
|
||||
for i := 0; i < webMaxPending; i++ {
|
||||
c2.deliverControl([]byte("{}"))
|
||||
}
|
||||
c2.deliverControl([]byte("{}"))
|
||||
select {
|
||||
case <-c2.done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("control-overflowed web client was not dropped")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentWritePumpDropsOnWriteError(t *testing.T) {
|
||||
serverConn := make(chan *websocket.Conn, 1)
|
||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
serverConn <- c
|
||||
for { // keep server side alive until client vanishes
|
||||
if _, _, err := c.ReadMessage(); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}))
|
||||
t.Cleanup(up.Close)
|
||||
clientConn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(up.URL, "http"), nil)
|
||||
if err != nil {
|
||||
t.Fatalf("dial: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = clientConn.Close() })
|
||||
var server *websocket.Conn
|
||||
select {
|
||||
case server = <-serverConn:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("server conn never upgraded")
|
||||
}
|
||||
|
||||
hub := NewHub(openTestStore(t))
|
||||
ac := &agentConn{hub: hub, conn: server, send: make(chan []byte, 4), done: make(chan struct{})}
|
||||
go ac.writePump()
|
||||
// healthy write first
|
||||
ac.send <- []byte(`{"type":"welcome"}`)
|
||||
_ = clientConn.SetReadDeadline(time.Now().Add(2 * time.Second))
|
||||
if _, _, err := clientConn.ReadMessage(); err != nil {
|
||||
t.Fatalf("first write lost: %v", err)
|
||||
}
|
||||
// kill the TCP conn underneath gorilla, then keep writing
|
||||
if err := clientConn.UnderlyingConn().Close(); err != nil {
|
||||
t.Fatalf("close tcp: %v", err)
|
||||
}
|
||||
deadline := time.After(3 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case ac.send <- []byte(`{"type":"abort"}`):
|
||||
case <-ac.done:
|
||||
return // writePump hit the write error and dropped
|
||||
case <-deadline:
|
||||
t.Fatal("writePump did not drop after write failure")
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHubClosedStoreKeepsServing(t *testing.T) {
|
||||
ts, store := newTestServer(t)
|
||||
if err := store.Close(); err != nil {
|
||||
t.Fatalf("close store: %v", err)
|
||||
}
|
||||
|
||||
ws := dialAgent(t, ts)
|
||||
_ = ws.WriteJSON(helloFrame("s1"))
|
||||
// welcome still arrives with lastSeq 0 despite store failures
|
||||
if welcome := readFrame(t, ws); welcome["type"] != evWelcome || welcome["lastSeq"].(float64) != 0 {
|
||||
t.Fatalf("welcome with broken store = %v", welcome)
|
||||
}
|
||||
// events are not persisted but fan-out still happens; conn must survive
|
||||
_ = ws.WriteJSON(map[string]any{"v": 1, "type": evAgentSettled, "sessionId": "s1", "seq": 1, "ts": 5})
|
||||
|
||||
hub := NewHub(store) // store already closed
|
||||
if views := hub.SessionsView(); len(views) != 0 {
|
||||
t.Fatalf("SessionsView with broken store = %v, want empty", views)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHubBroadcastSpawnStatusNilProvider(t *testing.T) {
|
||||
hub := NewHub(openTestStore(t))
|
||||
hub.BroadcastSpawnStatus() // SpawnStatus nil → no-op, must not panic
|
||||
hub.SpawnStatus = func() []SpawnJob { return nil }
|
||||
hub.BroadcastSpawnStatus() // provider returning nil jobs → empty frame
|
||||
}
|
||||
|
||||
func TestHubByeClosesConnAndEventWithoutSessionIgnored(t *testing.T) {
|
||||
ts, _ := newTestServer(t)
|
||||
ws := dialAgent(t, ts)
|
||||
_ = ws.WriteJSON(helloFrame("s1"))
|
||||
_ = readFrame(t, ws)
|
||||
|
||||
// event frames without sessionId are dropped silently
|
||||
_ = ws.WriteJSON(map[string]any{"v": 1, "type": evMessageEnd, "seq": 9, "ts": 9})
|
||||
// bye closes the conn from the server side
|
||||
_ = ws.WriteJSON(map[string]any{"v": 1, "type": evBye, "sessionId": "s1", "seq": 10, "ts": 10})
|
||||
_ = ws.SetReadDeadline(time.Now().Add(2 * time.Second))
|
||||
if _, _, err := ws.ReadMessage(); err == nil {
|
||||
t.Fatal("conn must close after bye")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHubHelloWelcomeSendDropWhenQueueFull(t *testing.T) {
|
||||
hub := NewHub(openTestStore(t))
|
||||
ac := &agentConn{hub: hub, conn: throwawayConn(t), send: make(chan []byte, 1), done: make(chan struct{})}
|
||||
ac.send <- []byte("full") // nobody drains → welcome select takes default → drop
|
||||
hello := helloFrame("s1")
|
||||
raw, _ := json.Marshal(hello)
|
||||
hub.handleHello(ac, frame{typ: evHello, sessionID: "s1", raw: raw})
|
||||
select {
|
||||
case <-ac.done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("agent with full send queue must be dropped at hello")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHubWebReadWritePumpEdges(t *testing.T) {
|
||||
ts, _, hub := newTestServerHub(t)
|
||||
web := dialWeb(t, ts)
|
||||
if first := readFrame(t, web); first["type"] != frameSessionList {
|
||||
t.Fatalf("first frame = %v", first)
|
||||
}
|
||||
// garbage frames must not kill the read pump
|
||||
if err := web.WriteMessage(websocket.TextMessage, []byte("not json")); err != nil {
|
||||
t.Fatalf("write garbage: %v", err)
|
||||
}
|
||||
if err := web.WriteJSON(map[string]any{"type": frameSubscribe, "sessionId": "s1"}); err != nil {
|
||||
t.Fatalf("subscribe: %v", err)
|
||||
}
|
||||
waitFor(t, 2*time.Second, func() bool {
|
||||
hub.mu.Lock()
|
||||
defer hub.mu.Unlock()
|
||||
for c := range hub.webs {
|
||||
return c.subscription() == "s1"
|
||||
}
|
||||
return false
|
||||
})
|
||||
if err := web.WriteJSON(map[string]any{"type": frameUnsubscribe}); err != nil {
|
||||
t.Fatalf("unsubscribe: %v", err)
|
||||
}
|
||||
waitFor(t, 2*time.Second, func() bool {
|
||||
hub.mu.Lock()
|
||||
defer hub.mu.Unlock()
|
||||
for c := range hub.webs {
|
||||
return c.subscription() == ""
|
||||
}
|
||||
return false
|
||||
})
|
||||
// client disconnects → readPump exits, client unregistered
|
||||
_ = web.Close()
|
||||
waitFor(t, 2*time.Second, func() bool {
|
||||
hub.mu.Lock()
|
||||
defer hub.mu.Unlock()
|
||||
return len(hub.webs) == 0
|
||||
})
|
||||
}
|
||||
|
||||
func TestHubDropWebIdempotent(t *testing.T) {
|
||||
hub := NewHub(openTestStore(t))
|
||||
c := &webClient{hub: hub, conn: throwawayConn(t), done: make(chan struct{})}
|
||||
hub.mu.Lock()
|
||||
hub.webs[c] = struct{}{}
|
||||
hub.mu.Unlock()
|
||||
hub.dropWeb(c)
|
||||
hub.dropWeb(c) // second drop hits the already-removed early return
|
||||
select {
|
||||
case <-c.done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("web client was not dropped")
|
||||
}
|
||||
hub.mu.Lock()
|
||||
left := len(hub.webs)
|
||||
hub.mu.Unlock()
|
||||
if left != 0 {
|
||||
t.Fatalf("webs after drop = %d, want 0", left)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebWritePumpDropsOnWriteError(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
ctrl bool // deliver control frames (vs events only)
|
||||
}{
|
||||
{"control-write", true},
|
||||
{"events-write", false},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
serverConn := make(chan *websocket.Conn, 1)
|
||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
serverConn <- c
|
||||
for {
|
||||
if _, _, err := c.ReadMessage(); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}))
|
||||
t.Cleanup(up.Close)
|
||||
clientConn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(up.URL, "http"), nil)
|
||||
if err != nil {
|
||||
t.Fatalf("dial: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = clientConn.Close() })
|
||||
var server *websocket.Conn
|
||||
select {
|
||||
case server = <-serverConn:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("server conn never upgraded")
|
||||
}
|
||||
|
||||
hub := NewHub(openTestStore(t))
|
||||
c := &webClient{hub: hub, conn: server, done: make(chan struct{})}
|
||||
hub.mu.Lock()
|
||||
hub.webs[c] = struct{}{} // writePump errors route through dropWeb
|
||||
hub.mu.Unlock()
|
||||
go c.writePump()
|
||||
if tc.ctrl {
|
||||
c.deliverControl([]byte(`{"type":"session_list"}`))
|
||||
} else {
|
||||
c.deliverEvent(pendingEvent{sessionID: "s", seq: 1, raw: []byte("{}")})
|
||||
}
|
||||
_ = clientConn.SetReadDeadline(time.Now().Add(2 * time.Second))
|
||||
if _, _, err := clientConn.ReadMessage(); err != nil {
|
||||
t.Fatalf("first frame lost: %v", err)
|
||||
}
|
||||
if err := clientConn.UnderlyingConn().Close(); err != nil {
|
||||
t.Fatalf("close tcp: %v", err)
|
||||
}
|
||||
// flush cycle hits the dead conn → writePump drops the client
|
||||
deadline := time.After(3 * time.Second)
|
||||
for {
|
||||
if tc.ctrl {
|
||||
c.deliverControl([]byte(`{"type":"session_list"}`))
|
||||
} else {
|
||||
c.deliverEvent(pendingEvent{sessionID: "s", seq: 2, raw: []byte("{}")})
|
||||
}
|
||||
select {
|
||||
case <-c.done:
|
||||
return
|
||||
case <-deadline:
|
||||
t.Fatal("writePump did not drop after write failure")
|
||||
default:
|
||||
}
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebClientDeliverToDropped(t *testing.T) {
|
||||
hub := NewHub(openTestStore(t))
|
||||
c := &webClient{hub: hub, conn: throwawayConn(t), done: make(chan struct{})}
|
||||
for i := 0; i < webMaxPending+1; i++ { // overflow sets the dropped flag
|
||||
c.deliverEvent(pendingEvent{sessionID: "s", seq: int64(i), raw: []byte("{}")})
|
||||
}
|
||||
c.deliverEvent(pendingEvent{sessionID: "s", seq: 1, raw: []byte("{}")})
|
||||
if len(c.events) != webMaxPending {
|
||||
t.Fatalf("dropped client queued %d events, want %d", len(c.events), webMaxPending)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user