test(chat,playout): enable test suite, coverage gate, switch to pure-Go SQLite driver
This commit is contained in:
zwitschi committed
1 parent
b3a10ba3ef
commit
e8a43b8a79
16 files changed
+3257
-8
No files matched your search
@@ -0,0 +1,563 @@
|
||||
package ws
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/glebarez/sqlite"
|
||||
"github.com/gorilla/websocket"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"chat/src/db"
|
||||
)
|
||||
|
||||
// mockRedis implements db.RedisClient for testing.
|
||||
type mockRedis struct {
|
||||
mu sync.Mutex
|
||||
store map[string]string
|
||||
pubsubs []*mockPubSub
|
||||
}
|
||||
|
||||
func newMockRedis() *mockRedis {
|
||||
return &mockRedis{store: make(map[string]string)}
|
||||
}
|
||||
|
||||
func (m *mockRedis) Ping(ctx context.Context) error { return nil }
|
||||
func (m *mockRedis) Set(ctx context.Context, key string, value interface{}, expiration time.Duration) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.store[key] = value.(string)
|
||||
return nil
|
||||
}
|
||||
func (m *mockRedis) Get(ctx context.Context, key string) (string, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.store[key], nil
|
||||
}
|
||||
func (m *mockRedis) Del(ctx context.Context, keys ...string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for _, k := range keys {
|
||||
delete(m.store, k)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (m *mockRedis) Incr(ctx context.Context, key string) (int64, error) { return 0, nil }
|
||||
func (m *mockRedis) Decr(ctx context.Context, key string) (int64, error) { return 0, nil }
|
||||
func (m *mockRedis) Expire(ctx context.Context, key string, expiration time.Duration) error {
|
||||
return nil
|
||||
}
|
||||
func (m *mockRedis) Exists(ctx context.Context, keys ...string) (int64, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for _, k := range keys {
|
||||
if _, ok := m.store[k]; ok {
|
||||
return 1, nil
|
||||
}
|
||||
}
|
||||
return 0, nil
|
||||
}
|
||||
func (m *mockRedis) Publish(ctx context.Context, channel string, message interface{}) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
payload, _ := json.Marshal(message)
|
||||
for _, ps := range m.pubsubs {
|
||||
ps.messages <- &db.PubSubMessage{Channel: channel, Payload: string(payload)}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (m *mockRedis) PSubscribe(ctx context.Context, channels ...string) db.PubSub {
|
||||
ps := &mockPubSub{messages: make(chan *db.PubSubMessage, 100)}
|
||||
m.mu.Lock()
|
||||
m.pubsubs = append(m.pubsubs, ps)
|
||||
m.mu.Unlock()
|
||||
return ps
|
||||
}
|
||||
func (m *mockRedis) Scan(ctx context.Context, cursor uint64, match string, count int64) ([]string, uint64, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
var keys []string
|
||||
for k := range m.store {
|
||||
if strings.Contains(k, strings.TrimSuffix(match, "*")) {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
}
|
||||
return keys, 0, nil
|
||||
}
|
||||
func (m *mockRedis) Info(ctx context.Context, sections ...string) (string, error) { return "", nil }
|
||||
func (m *mockRedis) Close() error { return nil }
|
||||
|
||||
type mockPubSub struct {
|
||||
messages chan *db.PubSubMessage
|
||||
}
|
||||
|
||||
func (m *mockPubSub) ReceiveMessage(ctx context.Context) (*db.PubSubMessage, error) {
|
||||
select {
|
||||
case msg := <-m.messages:
|
||||
return msg, nil
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
func (m *mockPubSub) Close() error { return nil }
|
||||
|
||||
func setupTestHub(t *testing.T) (*Hub, *gorm.DB, *mockRedis) {
|
||||
t.Helper()
|
||||
gormDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to open test DB: %v", err)
|
||||
}
|
||||
gormDB.AutoMigrate(&db.Message{}, &db.ModerationLog{})
|
||||
mr := newMockRedis()
|
||||
hub := NewHub(gormDB, mr, "test-instance", 10, 10*time.Second, "admin")
|
||||
return hub, gormDB, mr
|
||||
}
|
||||
|
||||
func setupWSServer(hub *Hub) *httptest.Server {
|
||||
return httptest.NewServer(http.HandlerFunc(hub.HandleConnection))
|
||||
}
|
||||
|
||||
// wsClient wraps a websocket.Conn with a channel-based read helper.
|
||||
type wsClient struct {
|
||||
conn *websocket.Conn
|
||||
msgs chan db.Message
|
||||
errors chan error
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
func newWSClient(t *testing.T, url string) *wsClient {
|
||||
t.Helper()
|
||||
conn, _, err := websocket.DefaultDialer.Dial(url, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("Dial error: %v", err)
|
||||
}
|
||||
wc := &wsClient{
|
||||
conn: conn,
|
||||
msgs: make(chan db.Message, 100),
|
||||
errors: make(chan error, 1),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
go wc.readLoop()
|
||||
return wc
|
||||
}
|
||||
|
||||
func (wc *wsClient) readLoop() {
|
||||
defer close(wc.msgs)
|
||||
for {
|
||||
var msg db.Message
|
||||
if err := wc.conn.ReadJSON(&msg); err != nil {
|
||||
wc.errors <- err
|
||||
return
|
||||
}
|
||||
select {
|
||||
case wc.msgs <- msg:
|
||||
case <-wc.done:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (wc *wsClient) close() {
|
||||
close(wc.done)
|
||||
wc.conn.Close()
|
||||
}
|
||||
|
||||
// recvMsg reads the next message with a timeout.
|
||||
func (wc *wsClient) recvMsg(timeout time.Duration) (*db.Message, error) {
|
||||
select {
|
||||
case msg := <-wc.msgs:
|
||||
return &msg, nil
|
||||
case err := <-wc.errors:
|
||||
return nil, err
|
||||
case <-time.After(timeout):
|
||||
return nil, context.DeadlineExceeded
|
||||
}
|
||||
}
|
||||
|
||||
// skipSystem reads and discards system messages.
|
||||
func (wc *wsClient) skipSystem(timeout time.Duration) {
|
||||
for {
|
||||
msg, err := wc.recvMsg(timeout)
|
||||
if err != nil || msg.Type != "system" {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (wc *wsClient) writeJSON(v interface{}) error {
|
||||
return wc.conn.WriteJSON(v)
|
||||
}
|
||||
|
||||
func TestHub_Broadcast(t *testing.T) {
|
||||
hub, _, _ := setupTestHub(t)
|
||||
|
||||
msg := db.Message{
|
||||
Room: "main",
|
||||
Username: "user1",
|
||||
Text: "hello",
|
||||
Timestamp: time.Now(),
|
||||
Type: "public",
|
||||
}
|
||||
hub.Broadcast(msg)
|
||||
|
||||
var count int64
|
||||
hub.DB.Model(&db.Message{}).Count(&count)
|
||||
if count != 1 {
|
||||
t.Errorf("message count = %d, want 1", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_ConnectedClients(t *testing.T) {
|
||||
hub, _, _ := setupTestHub(t)
|
||||
if hub.ConnectedClients() != 0 {
|
||||
t.Errorf("ConnectedClients() = %d, want 0", hub.ConnectedClients())
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_WebSocketConnection(t *testing.T) {
|
||||
hub, _, _ := setupTestHub(t)
|
||||
srv := setupWSServer(hub)
|
||||
defer srv.Close()
|
||||
|
||||
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?username=testuser&room=main"
|
||||
wc := newWSClient(t, wsURL)
|
||||
defer wc.close()
|
||||
|
||||
msg, err := wc.recvMsg(2 * time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("recvMsg error: %v", err)
|
||||
}
|
||||
if !strings.Contains(msg.Text, "joined") {
|
||||
t.Errorf("expected join message, got text=%s", msg.Text)
|
||||
}
|
||||
|
||||
if hub.ConnectedClients() != 1 {
|
||||
t.Errorf("ConnectedClients() = %d, want 1", hub.ConnectedClients())
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_SendMessage(t *testing.T) {
|
||||
hub, _, _ := setupTestHub(t)
|
||||
srv := setupWSServer(hub)
|
||||
defer srv.Close()
|
||||
|
||||
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?username=testuser&room=main"
|
||||
wc := newWSClient(t, wsURL)
|
||||
defer wc.close()
|
||||
|
||||
wc.skipSystem(2 * time.Second)
|
||||
|
||||
if err := wc.writeJSON(db.Message{Room: "main", Text: "hello world"}); err != nil {
|
||||
t.Fatalf("writeJSON error: %v", err)
|
||||
}
|
||||
|
||||
recvMsg, err := wc.recvMsg(2 * time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("recvMsg error: %v", err)
|
||||
}
|
||||
if recvMsg.Text != "hello world" {
|
||||
t.Errorf("recvMsg.Text = %q, want hello world", recvMsg.Text)
|
||||
}
|
||||
if recvMsg.Username != "testuser" {
|
||||
t.Errorf("recvMsg.Username = %q, want testuser", recvMsg.Username)
|
||||
}
|
||||
if recvMsg.Type != "public" {
|
||||
t.Errorf("recvMsg.Type = %q, want public", recvMsg.Type)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_Disconnect(t *testing.T) {
|
||||
hub, _, _ := setupTestHub(t)
|
||||
srv := setupWSServer(hub)
|
||||
defer srv.Close()
|
||||
|
||||
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?username=testuser&room=main"
|
||||
wc := newWSClient(t, wsURL)
|
||||
wc.skipSystem(2 * time.Second)
|
||||
|
||||
if hub.ConnectedClients() != 1 {
|
||||
t.Errorf("ConnectedClients() = %d, want 1", hub.ConnectedClients())
|
||||
}
|
||||
|
||||
wc.close()
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
|
||||
if hub.ConnectedClients() != 0 {
|
||||
t.Errorf("ConnectedClients() after close = %d, want 0", hub.ConnectedClients())
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_BannedUser(t *testing.T) {
|
||||
hub, _, mr := setupTestHub(t)
|
||||
srv := setupWSServer(hub)
|
||||
defer srv.Close()
|
||||
|
||||
mr.Set(context.Background(), "banned:testuser", "1", 0)
|
||||
|
||||
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?username=testuser&room=main"
|
||||
wc := newWSClient(t, wsURL)
|
||||
defer wc.close()
|
||||
|
||||
wc.skipSystem(2 * time.Second)
|
||||
wc.writeJSON(db.Message{Room: "main", Text: "hello"})
|
||||
|
||||
recvMsg, err := wc.recvMsg(2 * time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("recvMsg error: %v", err)
|
||||
}
|
||||
if !strings.Contains(recvMsg.Text, "banned") {
|
||||
t.Errorf("expected banned message, got %q", recvMsg.Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_MutedUser(t *testing.T) {
|
||||
hub, _, mr := setupTestHub(t)
|
||||
srv := setupWSServer(hub)
|
||||
defer srv.Close()
|
||||
|
||||
mr.Set(context.Background(), "muted:testuser", "1", 0)
|
||||
|
||||
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?username=testuser&room=main"
|
||||
wc := newWSClient(t, wsURL)
|
||||
defer wc.close()
|
||||
|
||||
wc.skipSystem(2 * time.Second)
|
||||
wc.writeJSON(db.Message{Room: "main", Text: "hello"})
|
||||
|
||||
recvMsg, err := wc.recvMsg(2 * time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("recvMsg error: %v", err)
|
||||
}
|
||||
if !strings.Contains(recvMsg.Text, "muted") {
|
||||
t.Errorf("expected muted message, got %q", recvMsg.Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_ModerationCommand(t *testing.T) {
|
||||
hub, _, _ := setupTestHub(t)
|
||||
srv := setupWSServer(hub)
|
||||
defer srv.Close()
|
||||
|
||||
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?username=admin&room=main"
|
||||
wc := newWSClient(t, wsURL)
|
||||
defer wc.close()
|
||||
|
||||
wc.skipSystem(2 * time.Second)
|
||||
wc.writeJSON(db.Message{Room: "main", Text: "/mute baduser 5 spamming"})
|
||||
|
||||
recvMsg, err := wc.recvMsg(2 * time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("recvMsg error: %v", err)
|
||||
}
|
||||
if !strings.Contains(recvMsg.Text, "mute") {
|
||||
t.Errorf("expected mute broadcast, got text=%s", recvMsg.Text)
|
||||
}
|
||||
|
||||
var count int64
|
||||
hub.DB.Model(&db.ModerationLog{}).Count(&count)
|
||||
if count != 1 {
|
||||
t.Errorf("moderation log count = %d, want 1", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_NonAdminModerationCommand(t *testing.T) {
|
||||
hub, _, _ := setupTestHub(t)
|
||||
srv := setupWSServer(hub)
|
||||
defer srv.Close()
|
||||
|
||||
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?username=regularuser&room=main"
|
||||
wc := newWSClient(t, wsURL)
|
||||
defer wc.close()
|
||||
|
||||
wc.skipSystem(2 * time.Second)
|
||||
wc.writeJSON(db.Message{Room: "main", Text: "/mute baduser 5 spamming"})
|
||||
|
||||
recvMsg, err := wc.recvMsg(2 * time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("recvMsg error: %v", err)
|
||||
}
|
||||
if !strings.Contains(recvMsg.Text, "Only moderators") {
|
||||
t.Errorf("expected moderator-only message, got %q", recvMsg.Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_PrivateDM(t *testing.T) {
|
||||
hub, _, _ := setupTestHub(t)
|
||||
srv := setupWSServer(hub)
|
||||
defer srv.Close()
|
||||
|
||||
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http")
|
||||
wc1 := newWSClient(t, wsURL+"?username=alice&room=main")
|
||||
defer wc1.close()
|
||||
wc2 := newWSClient(t, wsURL+"?username=bob&room=main")
|
||||
defer wc2.close()
|
||||
|
||||
wc1.skipSystem(2 * time.Second)
|
||||
wc2.skipSystem(2 * time.Second)
|
||||
|
||||
wc1.writeJSON(db.Message{Room: "dm:bob", Text: "hey bob"})
|
||||
|
||||
recvMsg, err := wc2.recvMsg(2 * time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("bob recvMsg error: %v", err)
|
||||
}
|
||||
if recvMsg.Type != "private" {
|
||||
t.Errorf("expected private type, got %s", recvMsg.Type)
|
||||
}
|
||||
if recvMsg.Text != "hey bob" {
|
||||
t.Errorf("expected 'hey bob', got %q", recvMsg.Text)
|
||||
}
|
||||
|
||||
recvMsg2, err := wc1.recvMsg(2 * time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("alice recvMsg error: %v", err)
|
||||
}
|
||||
if recvMsg2.Type != "private" {
|
||||
t.Errorf("expected private confirmation, got %s", recvMsg2.Type)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_PushChatHistory(t *testing.T) {
|
||||
hub, gormDB, _ := setupTestHub(t)
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
gormDB.Create(&db.Message{
|
||||
Room: "main",
|
||||
Username: "user1",
|
||||
Text: "msg",
|
||||
Timestamp: time.Now().Add(time.Duration(i) * time.Second),
|
||||
Type: "public",
|
||||
})
|
||||
}
|
||||
|
||||
srv := setupWSServer(hub)
|
||||
defer srv.Close()
|
||||
|
||||
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?username=newuser&room=main"
|
||||
wc := newWSClient(t, wsURL)
|
||||
defer wc.close()
|
||||
|
||||
count := 0
|
||||
for i := 0; i < 10; i++ {
|
||||
msg, err := wc.recvMsg(2 * time.Second)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
if msg.Type == "public" {
|
||||
count++
|
||||
}
|
||||
}
|
||||
if count != 5 {
|
||||
t.Errorf("history message count = %d, want 5", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_MuteUser_UnmuteUser(t *testing.T) {
|
||||
hub, _, mr := setupTestHub(t)
|
||||
ctx := context.Background()
|
||||
|
||||
err := hub.MuteUser(ctx, "user1", 10*time.Minute)
|
||||
if err != nil {
|
||||
t.Fatalf("MuteUser error = %v", err)
|
||||
}
|
||||
|
||||
exists, _ := mr.Exists(ctx, "muted:user1")
|
||||
if exists != 1 {
|
||||
t.Error("muted key not found after MuteUser")
|
||||
}
|
||||
|
||||
err = hub.UnmuteUser(ctx, "user1")
|
||||
if err != nil {
|
||||
t.Fatalf("UnmuteUser error = %v", err)
|
||||
}
|
||||
|
||||
exists, _ = mr.Exists(ctx, "muted:user1")
|
||||
if exists != 0 {
|
||||
t.Error("muted key still exists after UnmuteUser")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_BanUser_UnbanUser(t *testing.T) {
|
||||
hub, _, mr := setupTestHub(t)
|
||||
ctx := context.Background()
|
||||
|
||||
err := hub.BanUser(ctx, "user1")
|
||||
if err != nil {
|
||||
t.Fatalf("BanUser error = %v", err)
|
||||
}
|
||||
|
||||
exists, _ := mr.Exists(ctx, "banned:user1")
|
||||
if exists != 1 {
|
||||
t.Error("banned key not found after BanUser")
|
||||
}
|
||||
|
||||
err = hub.UnbanUser(ctx, "user1")
|
||||
if err != nil {
|
||||
t.Fatalf("UnbanUser error = %v", err)
|
||||
}
|
||||
|
||||
exists, _ = mr.Exists(ctx, "banned:user1")
|
||||
if exists != 0 {
|
||||
t.Error("banned key still exists after UnbanUser")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_ScanPresence(t *testing.T) {
|
||||
hub, _, mr := setupTestHub(t)
|
||||
ctx := context.Background()
|
||||
|
||||
mr.Set(ctx, "presence:main:user1", "online", 0)
|
||||
mr.Set(ctx, "presence:main:user2", "online", 0)
|
||||
mr.Set(ctx, "presence:lobby:user3", "online", 0)
|
||||
|
||||
presences, err := hub.ScanPresence(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("ScanPresence error = %v", err)
|
||||
}
|
||||
if len(presences) != 3 {
|
||||
t.Errorf("ScanPresence len = %d, want 3", len(presences))
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_SubscribeRoomMessages(t *testing.T) {
|
||||
hub, _, mr := setupTestHub(t)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
go func() {
|
||||
hub.SubscribeRoomMessages()
|
||||
}()
|
||||
|
||||
payload, _ := json.Marshal(map[string]any{
|
||||
"instance": "other-instance",
|
||||
"msg": db.Message{
|
||||
Room: "main",
|
||||
Username: "remoteuser",
|
||||
Text: "remote hello",
|
||||
Timestamp: time.Now(),
|
||||
Type: "public",
|
||||
},
|
||||
})
|
||||
mr.Publish(ctx, "room:main:messages", string(payload))
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
ownPayload, _ := json.Marshal(map[string]any{
|
||||
"instance": "test-instance",
|
||||
"msg": db.Message{
|
||||
Room: "main",
|
||||
Username: "localuser",
|
||||
Text: "local hello",
|
||||
Timestamp: time.Now(),
|
||||
Type: "public",
|
||||
},
|
||||
})
|
||||
mr.Publish(ctx, "room:main:messages", string(ownPayload))
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
}
|
||||
Reference in new issue
Block a user