- 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 语法通过。
752 lines
24 KiB
Go
752 lines
24 KiB
Go
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)
|
||
}
|
||
}
|