Files
Scoreboard/server/internal/realtime/hub_test.go
T
AskaEth 27bddefaad fix: 上线前快修(大屏自愈 / 奖牌口径 / 聚合作废 / 密码可发现性)
- P0 慢客户端关闭码 1008 → 4000:前端把所有 1008 当「登录失效」并永久停止重连,
  大屏被踢后再也不自愈(服务端 markSlow 改用自定义码,前端逻辑不动)
- P1 REST 与 WS 的奖牌榜补 events.scored = 1 过滤,与导出成绩册口径一致
  (否则「只排名不计分」项目只要挂着通道就会在大屏凭空出奖牌)
- P1 聚合流水作废去掉 source_id 子查询:改通道会重建通道 id,旧聚合流水
  永远匹配不上 → 再插一份 → 分数翻倍
- P1 审核页/历史页的折算系数兜底由 ?? 1 改为 >0 ? : 1:/ws 快照对非 admin
  会把 individualToTeamWeight 置 0,recorder 看到的个人折算分全是 0
- P2 删掉 signature.local.txt 后清理残留的 signature.build.txt,避免继续注入旧署名
- 初始管理员密码:banner 加纯 ASCII 锚 INITIAL ADMIN CREATED,文档改为搜它
  (原来教的 findstr "password=" 已经失效,且中文 findstr 本身不可靠)

自检:服务端 12 包全绿;前端 vue-tsc + build 通过、node --test 24/24;console 语法通过。
2026-09-09 23:13:20 +08:00

752 lines
24 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package realtime
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"path/filepath"
"reflect"
"strings"
"testing"
"time"
"github.com/coder/websocket"
_ "modernc.org/sqlite"
"git.yeij.top/AskaEth/Scoreboard/server/internal/auth"
"git.yeij.top/AskaEth/Scoreboard/server/internal/middleware"
"git.yeij.top/AskaEth/Scoreboard/server/internal/store"
)
// ---------- 测试脚手架 ----------
type testEnv struct {
hub *Hub
db *sql.DB
srv *httptest.Server
}
// newTestEnv 建一个未登录(guest)的服务端:默认身份与大屏端一致。
func newTestEnv(t *testing.T) *testEnv {
t.Helper()
return newTestEnvAs(t, &auth.User{Role: auth.RoleGuest})
}
// newTestEnvAs 注入会话身份,等价于真实链路里 Authenticate 中间件写进 context 的 User。
func newTestEnvAs(t *testing.T, u *auth.User) *testEnv {
t.Helper()
db := newTestDB(t)
h := NewHub(db, slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError})))
handler := h.Handler()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
handler.ServeHTTP(w, r.WithContext(middleware.WithUser(r.Context(), u)))
}))
t.Cleanup(func() {
srv.Close()
_ = h.Close()
_ = db.Close()
})
return &testEnv{hub: h, db: db, srv: srv}
}
func newTestDB(t *testing.T) *sql.DB {
t.Helper()
path := filepath.Join(t.TempDir(), "event.db")
db, err := sql.Open("sqlite", "file:"+path+
"?_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_pragma=busy_timeout(5000)")
if err != nil {
t.Fatalf("打开测试库: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
if err := store.MigrateEvent(db); err != nil {
t.Fatalf("迁移测试库: %v", err)
}
return db
}
func (e *testEnv) dial(t *testing.T, hello *helloPayload) *wsClient {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
url := "ws" + strings.TrimPrefix(e.srv.URL, "http") + "/ws"
conn, _, err := websocket.Dial(ctx, url, nil)
if err != nil {
cancel()
t.Fatalf("连接 /ws: %v", err)
}
c := &wsClient{t: t, ctx: ctx, cancel: cancel, conn: conn, msgs: make(chan wsMessage, 256)}
go c.readLoop()
t.Cleanup(c.close)
if hello != nil {
c.send(EventHello, *hello)
}
return c
}
type wsMessage struct {
event string
data json.RawMessage
err error
closeCode websocket.StatusCode
}
type wsClient struct {
t *testing.T
ctx context.Context
cancel context.CancelFunc
conn *websocket.Conn
msgs chan wsMessage
}
// readLoop 单独读取:测试里等待「没有消息」时不能靠取消 Read 的 context,
// 那会把整条连接关掉,后续用例就没法继续读了。
func (c *wsClient) readLoop() {
defer close(c.msgs)
for {
_, data, err := c.conn.Read(c.ctx)
if err != nil {
c.msgs <- wsMessage{err: err, closeCode: websocket.CloseStatus(err)}
return
}
var env envelope
if err := json.Unmarshal(data, &env); err != nil {
c.msgs <- wsMessage{err: err}
return
}
c.msgs <- wsMessage{event: env.Event, data: env.Data}
}
}
func (c *wsClient) close() {
_ = c.conn.CloseNow()
c.cancel()
}
func (c *wsClient) send(event string, data any) {
c.t.Helper()
msg, err := encode(event, data)
if err != nil {
c.t.Fatalf("编码 %s: %v", event, err)
}
ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
defer cancel()
if err := c.conn.Write(ctx, websocket.MessageText, msg); err != nil {
c.t.Fatalf("发送 %s: %v", event, err)
}
}
// recvWithin 在 d 内取一条消息;超时返回 ok=false(连接保持可用)。
func (c *wsClient) recvWithin(d time.Duration) (string, json.RawMessage, bool) {
c.t.Helper()
timer := time.NewTimer(d)
defer timer.Stop()
select {
case msg, ok := <-c.msgs:
if !ok || msg.err != nil {
c.t.Fatalf("连接已断开: %v", msg.err)
}
return msg.event, msg.data, true
case <-timer.C:
return "", nil, false
}
}
// recvEvent 等待指定事件,跳过心跳等无关消息。
func (c *wsClient) recvEvent(want string) json.RawMessage {
c.t.Helper()
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
ev, data, ok := c.recvWithin(time.Until(deadline))
if !ok {
break
}
if ev == want {
return data
}
}
c.t.Fatalf("等待事件 %s 超时", want)
return nil
}
// expectClosed 断言连接在 d 内被服务端关闭,且关闭前没有下发任何事件。
func (c *wsClient) expectClosed(d time.Duration) {
c.t.Helper()
timer := time.NewTimer(d)
defer timer.Stop()
for {
select {
case msg, ok := <-c.msgs:
if !ok || msg.err != nil {
return
}
c.t.Fatalf("连接关闭前仍收到事件 %s", msg.event)
case <-timer.C:
c.t.Fatalf("等待服务端关闭连接超时")
}
}
}
// expectClosedWithCode 断言连接被服务端以指定关闭码关闭,且关闭前没有下发任何事件。
func (c *wsClient) expectClosedWithCode(d time.Duration, want websocket.StatusCode) {
c.t.Helper()
timer := time.NewTimer(d)
defer timer.Stop()
for {
select {
case msg, ok := <-c.msgs:
if !ok {
c.t.Fatalf("连接已关闭但没有收到关闭帧")
}
if msg.err != nil {
if msg.closeCode != want {
c.t.Fatalf("关闭码应为 %d,得到 %d (%v)", want, msg.closeCode, msg.err)
}
return
}
c.t.Fatalf("连接关闭前仍收到事件 %s", msg.event)
case <-timer.C:
c.t.Fatalf("等待服务端关闭连接超时")
}
}
}
// expectClosedWithCodeDraining 断言连接最终被服务端以指定关闭码关闭。
// 关闭前已下发的正常事件会被丢弃:慢客户端被踢属于「先推后关」,不能要求关闭前零事件。
func (c *wsClient) expectClosedWithCodeDraining(d time.Duration, want websocket.StatusCode) {
c.t.Helper()
timer := time.NewTimer(d)
defer timer.Stop()
for {
select {
case msg, ok := <-c.msgs:
if !ok {
c.t.Fatalf("连接已关闭但没有收到关闭帧")
}
if msg.err != nil {
if msg.closeCode != want {
c.t.Fatalf("关闭码应为 %d,得到 %d (%v)", want, msg.closeCode, msg.err)
}
return
}
case <-timer.C:
c.t.Fatalf("等待服务端关闭连接超时")
}
}
}
func decode[T any](t *testing.T, raw json.RawMessage) T {
t.Helper()
var v T
if err := json.Unmarshal(raw, &v); err != nil {
t.Fatalf("解析载荷 %s: %v", raw, err)
}
return v
}
func insertChange(t *testing.T, db *sql.DB, entity string, entityID int64, action, payload string) int64 {
t.Helper()
res, err := db.Exec(
`INSERT INTO change_log(entity_type, entity_id, action, payload_json) VALUES (?, ?, ?, ?)`,
entity, entityID, action, payload)
if err != nil {
t.Fatalf("写入 change_log: %v", err)
}
id, err := res.LastInsertId()
if err != nil {
t.Fatalf("读取 change_log id: %v", err)
}
return id
}
func loadChanges(t *testing.T, h *Hub, since int64, limit int) []Change {
t.Helper()
changes, err := h.LoadChanges(context.Background(), since, limit)
if err != nil {
t.Fatalf("读取增量变更: %v", err)
}
return changes
}
func waitFor(t *testing.T, what string, cond func() bool) {
t.Helper()
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
if cond() {
return
}
time.Sleep(5 * time.Millisecond)
}
t.Fatalf("等待条件超时: %s", what)
}
func ptrInt64(v int64) *int64 { return &v }
// ---------- 客户端状态归约(模拟前端按 version 应用变更) ----------
type clientState map[string]json.RawMessage
func stateKey(ch Change) string { return fmt.Sprintf("%s/%d", ch.Entity, ch.ID) }
func applyChanges(s clientState, changes []Change) clientState {
for _, ch := range changes {
switch ch.Action {
case ActionVoid:
delete(s, stateKey(ch))
case ActionReplace:
s[ch.Entity] = ch.Data
default:
s[stateKey(ch)] = ch.Data
}
}
return s
}
// ---------- 用例 ----------
func TestHelloSnapshotThenChange(t *testing.T) {
env := newTestEnv(t)
c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))
if snap.Version != 0 {
t.Fatalf("空库快照版本应为 0,得到 %d", snap.Version)
}
v := insertChange(t, env.db, "result", 42, ActionUpsert, `{"rank":1}`)
env.hub.Broadcast(loadChanges(t, env.hub, 0, 0))
got := decode[changePayload](t, c.recvEvent(EventChange))
if got.Version != v {
t.Fatalf("change 版本应为 %d,得到 %d", v, got.Version)
}
if len(got.Changes) != 1 {
t.Fatalf("change 条数应为 1,得到 %d", len(got.Changes))
}
ch := got.Changes[0]
if ch.Entity != "result" || ch.ID != 42 || ch.Action != ActionUpsert {
t.Fatalf("change 内容不符: %+v", ch)
}
if string(ch.Data) != `{"rank":1}` {
t.Fatalf("change data 不符: %s", ch.Data)
}
if env.hub.Version() != v {
t.Fatalf("Hub 版本应推进到 %d,得到 %d", v, env.hub.Version())
}
}
// 断线期间的变更通过 catchup 补齐,且终态与全量重放的终态一致。
func TestReconnectCatchupTerminalState(t *testing.T) {
env := newTestEnv(t)
v1 := insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
v2 := insertChange(t, env.db, "standings", 0, ActionReplace, `[{"teamId":1,"points":10}]`)
c1 := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
snap := decode[Snapshot](t, c1.recvEvent(EventSnapshot))
if snap.Version != v2 || snap.Version != v1+1 {
t.Fatalf("首次快照版本应为 %d,得到 %d", v2, snap.Version)
}
c1.close()
v3 := insertChange(t, env.db, "result", 2, ActionUpsert, `{"rank":2}`)
v4 := insertChange(t, env.db, "result", 1, ActionVoid, `{}`)
v5 := insertChange(t, env.db, "standings", 0, ActionReplace, `[{"teamId":1,"points":12}]`)
if v4 != v3+1 || v5 != v4+1 {
t.Fatalf("change_log 版本应连续递增: %d %d %d", v3, v4, v5)
}
c2 := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1", LastVersion: ptrInt64(snap.Version)})
cp := decode[catchupPayload](t, c2.recvEvent(EventCatchup))
if cp.FromVersion != snap.Version {
t.Fatalf("fromVersion 应为 %d,得到 %d", snap.Version, cp.FromVersion)
}
if cp.Version != v5 {
t.Fatalf("catchup 版本应为 %d,得到 %d", v5, cp.Version)
}
// 版本号不上线路(协议只在消息层带 version),因此按内容核对补发的三条变更。
wantChanges := []struct {
entity string
id int64
action string
}{
{"result", 2, ActionUpsert},
{"result", 1, ActionVoid},
{"standings", 0, ActionReplace},
}
if len(cp.Changes) != len(wantChanges) {
t.Fatalf("catchup 条数应为 %d,得到 %d", len(wantChanges), len(cp.Changes))
}
for i, want := range wantChanges {
got := cp.Changes[i]
if got.Entity != want.entity || got.ID != want.id || got.Action != want.action {
t.Fatalf("第 %d 条变更应为 %+v,得到 %+v", i, want, got)
}
}
// 断线前已应用的变更 + catchup 的变更 == 从头全量重放,终态必须一致。
base := applyChanges(clientState{}, loadChanges(t, env.hub, 0, int(snap.Version)))
got := applyChanges(base, cp.Changes)
want := applyChanges(clientState{}, loadChanges(t, env.hub, 0, 0))
if !reflect.DeepEqual(got, want) {
t.Fatalf("catchup 终态与全量重放不一致:\n got=%v\nwant=%v", got, want)
}
// 新连接的整包快照版本必须与 catchup 的终点一致。
fresh, err := env.hub.Snapshot(context.Background())
if err != nil {
t.Fatalf("构建快照: %v", err)
}
if fresh.Version != cp.Version {
t.Fatalf("快照版本 %d 与 catchup 版本 %d 不一致", fresh.Version, cp.Version)
}
}
// 增量超过阈值时回落全量快照。
func TestCatchupFallsBackToSnapshotWhenTooOld(t *testing.T) {
env := newTestEnv(t)
env.hub.SetMaxCatchup(2)
first := insertChange(t, env.db, "result", 1, ActionUpsert, `{}`)
for i := int64(2); i <= 5; i++ {
insertChange(t, env.db, "result", i, ActionUpsert, `{}`)
}
c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1", LastVersion: ptrInt64(first - 1)})
snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))
if snap.Version != 5 {
t.Fatalf("应回落全量快照且版本为 5,得到 %d", snap.Version)
}
}
func TestDuplicateDeliveryIgnored(t *testing.T) {
env := newTestEnv(t)
c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
c.recvEvent(EventSnapshot)
v1 := insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
changes := loadChanges(t, env.hub, 0, 0)
env.hub.Broadcast(changes)
got := decode[changePayload](t, c.recvEvent(EventChange))
if got.Version != v1 {
t.Fatalf("首投版本应为 %d,得到 %d", v1, got.Version)
}
c.send(EventAck, ackPayload{Version: v1})
// 同一批变更再投一次,客户端不应再收到。
env.hub.Broadcast(changes)
if ev, _, ok := c.recvWithin(300 * time.Millisecond); ok {
t.Fatalf("重复投递未被忽略,收到事件 %s", ev)
}
// 新变更仍能正常下发。
v2 := insertChange(t, env.db, "result", 2, ActionUpsert, `{"rank":2}`)
env.hub.Broadcast(loadChanges(t, env.hub, v1, 0))
got = decode[changePayload](t, c.recvEvent(EventChange))
if got.Version != v2 {
t.Fatalf("新变更版本应为 %d,得到 %d", v2, got.Version)
}
}
// 心跳携带版本号与服务器时间;版本落后时客户端 resync 能拿到最新快照。
func TestHeartbeatAndResync(t *testing.T) {
env := newTestEnv(t)
env.hub.SetHeartbeatInterval(40 * time.Millisecond)
c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
c.recvEvent(EventSnapshot)
// 只写库不广播:心跳版本会领先于客户端,客户端据此发起 resync。
insertChange(t, env.db, "result", 1, ActionUpsert, `{}`)
insertChange(t, env.db, "result", 2, ActionUpsert, `{}`)
hb := decode[heartbeatPayload](t, c.recvEvent(EventHeartbeat))
if hb.Version != 2 {
t.Fatalf("心跳版本应为 2,得到 %d", hb.Version)
}
if _, err := time.Parse(time.RFC3339Nano, hb.ServerTime); err != nil {
t.Fatalf("serverTime 不是 RFC3339: %q (%v)", hb.ServerTime, err)
}
c.send(EventResync, struct{}{})
snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))
if snap.Version != 2 {
t.Fatalf("resync 快照版本应为 2,得到 %d", snap.Version)
}
}
// 大屏实例注册表:instanceId、在线状态、当前页、最后心跳可查。
func TestDisplayInstancesRegistry(t *testing.T) {
env := newTestEnv(t)
c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "screen-1", Page: "standings"})
c.recvEvent(EventSnapshot)
insts := env.hub.Instances()
if len(insts) != 1 {
t.Fatalf("实例数应为 1,得到 %d", len(insts))
}
if insts[0].ID != "screen-1" || !insts[0].Online || insts[0].Page != "standings" {
t.Fatalf("实例注册内容不符: %+v", insts[0])
}
if insts[0].LastHeartbeat.IsZero() {
t.Fatalf("最后心跳不应为零值: %+v", insts[0])
}
// 重复 hello 不应把实例引用计数加两次,否则断开后仍会显示在线。
c.send(EventHello, helloPayload{Role: RoleDisplay, InstanceID: "screen-1", Page: "standings"})
c.recvEvent(EventSnapshot)
if insts := env.hub.Instances(); len(insts) != 1 || !insts[0].Online {
t.Fatalf("重复 hello 后实例应仍只有一个且在线: %+v", insts)
}
gotoCmd, err := NewDisplayCommand(CmdGoto, map[string]string{"page": "medals"})
if err != nil {
t.Fatalf("构造 goto: %v", err)
}
env.hub.SendCommand(gotoCmd)
cmd := decode[DisplayCommand](t, c.recvEvent(EventDisplayCommand))
if cmd.Action != CmdGoto {
t.Fatalf("指令 action 应为 goto,得到 %s", cmd.Action)
}
if insts := env.hub.Instances(); insts[0].Page != "medals" {
t.Fatalf("goto 后当前页应为 medals,得到 %s", insts[0].Page)
}
detail, err := NewDisplayCommand(CmdShowDetail, map[string]any{"eventId": 7})
if err != nil {
t.Fatalf("构造 show-detail: %v", err)
}
env.hub.SendCommand(detail)
if got := decode[DisplayCommand](t, c.recvEvent(EventDisplayCommand)); got.Action != CmdShowDetail {
t.Fatalf("指令 action 应为 show-detail,得到 %s", got.Action)
}
c.close()
waitFor(t, "实例离线", func() bool {
insts := env.hub.Instances()
return len(insts) == 1 && !insts[0].Online
})
}
// 房间隔离:admin:<userId> 只发给对应管理端,且 userId 来自会话而非客户端自报。
func TestAdminRoomTargeting(t *testing.T) {
env := newTestEnvAs(t, &auth.User{ID: 7, Username: "boss", Role: auth.RoleAdmin, Active: true})
display := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "d1"})
display.recvEvent(EventSnapshot)
// hello 不再携带 userId:定向房间必须用会话里的 ID。
admin := env.dial(t, &helloPayload{Role: RoleAdmin})
admin.recvEvent(EventSnapshot)
insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
env.hub.BroadcastTo(AdminRoom(7), loadChanges(t, env.hub, 0, 0))
got := decode[changePayload](t, admin.recvEvent(EventChange))
if got.Version != 1 {
t.Fatalf("定向 change 版本应为 1,得到 %d", got.Version)
}
if ev, _, ok := display.recvWithin(300 * time.Millisecond); ok {
t.Fatalf("大屏不应收到 admin:<uid> 房间的消息,收到 %s", ev)
}
}
// guest 自报 role=admin 必须被拒绝:管理端房间只能由会话授予。
// 断开必须带上策略性关闭码,否则客户端会当成网络异常无限重连。
func TestGuestCannotSubscribeAdminRoom(t *testing.T) {
env := newTestEnv(t)
c := env.dial(t, &helloPayload{Role: RoleAdmin})
c.expectClosedWithCode(3*time.Second, websocket.StatusPolicyViolation)
}
// guest 只能订阅 display 房间:即使 hello 里塞了 userId 也不能进入定向房间。
func TestGuestStaysInDisplayRoom(t *testing.T) {
env := newTestEnv(t)
c := env.dial(t, nil)
c.send(EventHello, map[string]any{"role": RoleDisplay, "userId": 99})
c.recvEvent(EventSnapshot)
insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
changes := loadChanges(t, env.hub, 0, 0)
env.hub.BroadcastTo(RoomAdmin, changes)
env.hub.BroadcastTo(AdminRoom(99), changes)
if ev, _, ok := c.recvWithin(300 * time.Millisecond); ok {
t.Fatalf("guest 不应收到管理端房间的消息,收到 %s", ev)
}
// display 房间的消息仍然正常到达。
env.hub.BroadcastTo(RoomDisplay, changes)
if got := decode[changePayload](t, c.recvEvent(EventChange)); got.Version != 1 {
t.Fatalf("大屏应收到 display 房间消息,版本得到 %d", got.Version)
}
}
// recorder 会话可以进管理端房间(大屏控制/数据同步都由管理端承担)。
func TestRecorderSessionJoinsAdminRoom(t *testing.T) {
env := newTestEnvAs(t, &auth.User{ID: 3, Username: "rec", Role: auth.RoleRecorder, Active: true})
c := env.dial(t, &helloPayload{Role: RoleAdmin})
c.recvEvent(EventSnapshot)
insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
env.hub.BroadcastTo(RoomAdmin, loadChanges(t, env.hub, 0, 0))
if got := decode[changePayload](t, c.recvEvent(EventChange)); got.Version != 1 {
t.Fatalf("recorder 应进入管理端房间,版本得到 %d", got.Version)
}
}
// 慢客户端:ack 落后超过阈值即被识别并断开。
func TestSlowClientDetected(t *testing.T) {
env := newTestEnv(t)
env.hub.SetSlowLag(1)
c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "slow-1"})
c.recvEvent(EventSnapshot)
for i := int64(1); i <= 3; i++ {
insertChange(t, env.db, "result", i, ActionUpsert, `{}`)
}
env.hub.Broadcast(loadChanges(t, env.hub, 0, 0))
waitFor(t, "慢客户端被标记", func() bool {
insts := env.hub.Instances()
return len(insts) == 1 && insts[0].Slow
})
waitFor(t, "慢客户端连接被断开", func() bool {
insts := env.hub.Instances()
return len(insts) == 1 && !insts[0].Online
})
// 必须用自定义关闭码:前端把所有 1008 当登录失效并永久停重连,大屏会就此冻结。
c.expectClosedWithCodeDraining(3*time.Second, slowCloseCode)
}
// 快照对所有角色下发,但系统配置是 admin-only:guest/recorder 只能拿到大屏渲染
// 所需的 event/display/榜单开关,默认积分模板、审核开关与折算系数必须置空/清零。
func TestSnapshotSettingsRedactedForNonAdmin(t *testing.T) {
for _, tc := range []struct {
name string
user *auth.User
}{
{"guest", &auth.User{Role: auth.RoleGuest}},
{"recorder", &auth.User{ID: 3, Username: "rec", Role: auth.RoleRecorder, Active: true}},
} {
t.Run(tc.name, func(t *testing.T) {
env := newTestEnvAs(t, tc.user)
mustExec(t, env.db, `UPDATE settings SET value = '0.5' WHERE key = 'standings.individual_to_team_weight'`)
mustExec(t, env.db, `UPDATE settings SET value = 'true' WHERE key = 'results.review_required'`)
c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: tc.name + "-1"})
snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))
if snap.Settings.DefaultProfileID != nil {
t.Fatalf("非 admin 不应看到 defaultProfileId: %+v", snap.Settings.DefaultProfileID)
}
if snap.Settings.RequireApproval {
t.Fatalf("非 admin 不应看到 requireApproval")
}
if snap.Settings.Standings.IndividualToTeamWeight != 0 {
t.Fatalf("非 admin 不应看到 individualToTeamWeight: %v", snap.Settings.Standings.IndividualToTeamWeight)
}
// 大屏要用的字段必须保留,裁剪不能误伤标题/主题/榜单开关。
if snap.Settings.Event.Name == "" || snap.Settings.Display.RotateSecond == 0 {
t.Fatalf("event/display 不应被裁剪: %+v", snap.Settings)
}
if snap.Settings.Standings.RankingMode != "points" ||
!snap.Settings.Standings.ShowIndividualBoard || !snap.Settings.Standings.ShowIndividualMedals {
t.Fatalf("standings 展示开关不应被裁剪: %+v", snap.Settings.Standings)
}
})
}
}
// admin 仍拿到完整配置,管理端才能在 WS 首包里读到默认积分模板与审核开关。
func TestSnapshotSettingsFullForAdmin(t *testing.T) {
env := newTestEnvAs(t, &auth.User{ID: 1, Username: "boss", Role: auth.RoleAdmin, Active: true})
mustExec(t, env.db, `UPDATE settings SET value = '0.5' WHERE key = 'standings.individual_to_team_weight'`)
mustExec(t, env.db, `UPDATE settings SET value = 'true' WHERE key = 'results.review_required'`)
c := env.dial(t, &helloPayload{Role: RoleAdmin})
snap := decode[Snapshot](t, c.recvEvent(EventSnapshot))
if snap.Settings.DefaultProfileID == nil {
t.Fatalf("admin 应看到 defaultProfileId")
}
if !snap.Settings.RequireApproval {
t.Fatalf("admin 应看到 requireApproval")
}
if snap.Settings.Standings.IndividualToTeamWeight != 0.5 {
t.Fatalf("admin 应看到 individualToTeamWeight=0.5,得到 %v", snap.Settings.Standings.IndividualToTeamWeight)
}
}
// resync 在 hello 之前不可用:否则会绕过房间分配并提前清掉 hello 超时。
func TestResyncBeforeHelloIgnored(t *testing.T) {
env := newTestEnv(t)
insertChange(t, env.db, "result", 1, ActionUpsert, `{"rank":1}`)
c := env.dial(t, nil)
c.send(EventResync, struct{}{})
if ev, _, ok := c.recvWithin(300 * time.Millisecond); ok {
t.Fatalf("hello 之前的 resync 不应下发快照,收到事件 %s", ev)
}
c.send(EventHello, helloPayload{Role: RoleDisplay, InstanceID: "d1"})
if snap := decode[Snapshot](t, c.recvEvent(EventSnapshot)); snap.Version != 1 {
t.Fatalf("hello 后应收到版本 1 的快照,得到 %d", snap.Version)
}
c.send(EventResync, struct{}{})
if snap := decode[Snapshot](t, c.recvEvent(EventSnapshot)); snap.Version != 1 {
t.Fatalf("hello 后 resync 应拿到快照,得到 %d", snap.Version)
}
}
// 离线实例不能无限堆积:超过 TTL 后在下一次读取注册表时被清理。
func TestInstanceRegistryPrunesStaleOffline(t *testing.T) {
env := newTestEnv(t)
c := env.dial(t, &helloPayload{Role: RoleDisplay, InstanceID: "stale-1"})
c.recvEvent(EventSnapshot)
c.close()
waitFor(t, "实例离线", func() bool {
insts := env.hub.Instances()
return len(insts) == 1 && !insts[0].Online
})
env.hub.mu.Lock()
env.hub.instances["stale-1"].lastHeartbeat = time.Now().Add(-instanceTTL - time.Minute)
env.hub.mu.Unlock()
if insts := env.hub.Instances(); len(insts) != 0 {
t.Fatalf("过期离线实例应被清理,得到 %+v", insts)
}
}
// 读写 goroutine 的 panic 必须被兜住并断开该连接,而不是带走进程。
func TestGuardLoopRecoversPanic(t *testing.T) {
env := newTestEnv(t)
c := newClient(env.hub, nil, "test", RoleDisplay, 0, false)
func() {
defer c.guardLoop("读")
panic("boom")
}()
select {
case <-c.done:
default:
t.Fatal("panic 后应关闭该连接")
}
if c.closeCode != websocket.StatusInternalError {
t.Fatalf("panic 后关闭码应为 %d,得到 %d", websocket.StatusInternalError, c.closeCode)
}
}