130 lines
4.4 KiB
Go
130 lines
4.4 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/golang/mock/gomock"
|
|
"github.com/segmentio/kafka-go"
|
|
"github.com/tinode/chat/server/store"
|
|
"github.com/tinode/chat/server/store/mock_store"
|
|
"github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
func TestKafkaTopicsKeepOnlyMessageInternalAsyncTopics(t *testing.T) {
|
|
topics := kafkaTopics(nil)
|
|
if topics["message_persist"] != "im.message.persist" {
|
|
t.Fatalf("unexpected private message persist topic: %q", topics["message_persist"])
|
|
}
|
|
if topics["group_message_persist"] != "im.group.message.persist" {
|
|
t.Fatalf("unexpected group message persist topic: %q", topics["group_message_persist"])
|
|
}
|
|
if topics["audit_event"] != "im.audit.event" {
|
|
t.Fatalf("unexpected audit event topic: %q", topics["audit_event"])
|
|
}
|
|
for _, removed := range []string{"file_lifecycle"} {
|
|
if _, ok := topics[removed]; ok {
|
|
t.Fatalf("internal sync topic %q must not be configured", removed)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestNormalizeKafkaConfigUsesLowLatencyBatchTimeout(t *testing.T) {
|
|
cfg := kafkaConfig{}
|
|
normalizeKafkaConfig(&cfg)
|
|
if cfg.BatchTimeoutMS != 10 {
|
|
t.Fatalf("unexpected kafka batch timeout: %d", cfg.BatchTimeoutMS)
|
|
}
|
|
if cfg.MaxRetryCount != 3 {
|
|
t.Fatalf("unexpected kafka max retry count: %d", cfg.MaxRetryCount)
|
|
}
|
|
if cfg.RequiredAcks != int(kafka.RequireAll) {
|
|
t.Fatalf("unexpected kafka required acks: %d", cfg.RequiredAcks)
|
|
}
|
|
}
|
|
|
|
func TestEnqueueKafkaAuditEventPublishesCrossSystemEvent(t *testing.T) {
|
|
oldKafkaBus := kafkaBus
|
|
oldPublisher := publishKafkaAuditEventEnvelope
|
|
defer func() {
|
|
kafkaBus = oldKafkaBus
|
|
publishKafkaAuditEventEnvelope = oldPublisher
|
|
}()
|
|
|
|
kafkaBus = &kafkaRuntime{}
|
|
var publishedTopic string
|
|
var publishedKey string
|
|
var publishedEnvelope kafkaEnvelope
|
|
publishKafkaAuditEventEnvelope = func(_ context.Context, topicName, key string, envelope kafkaEnvelope) error {
|
|
publishedTopic = topicName
|
|
publishedKey = key
|
|
publishedEnvelope = envelope
|
|
return nil
|
|
}
|
|
|
|
err := enqueueKafkaAuditEvent(context.Background(), 7, "admin", types.Uid(42),
|
|
"group.update", "group", "grp-test", "succeeded", "", map[string]any{"topic": "grp-test"})
|
|
if err != nil {
|
|
t.Fatalf("enqueueKafkaAuditEvent returned error: %v", err)
|
|
}
|
|
if publishedTopic != "audit_event" {
|
|
t.Fatalf("unexpected audit publish topic: %q", publishedTopic)
|
|
}
|
|
if publishedKey != "group:grp-test" {
|
|
t.Fatalf("unexpected audit publish key: %q", publishedKey)
|
|
}
|
|
if publishedEnvelope.TargetService != "hotgo-admin" || publishedEnvelope.EventType != "audit.group.update" {
|
|
t.Fatalf("unexpected audit envelope: %#v", publishedEnvelope)
|
|
}
|
|
if publishedEnvelope.Actor == nil || publishedEnvelope.Actor.Type != "admin" || publishedEnvelope.Actor.ID != types.Uid(42).UserId() {
|
|
t.Fatalf("unexpected audit actor: %#v", publishedEnvelope.Actor)
|
|
}
|
|
var payload kafkaAuditPayload
|
|
if err = json.Unmarshal(publishedEnvelope.Payload, &payload); err != nil {
|
|
t.Fatalf("invalid audit payload: %v", err)
|
|
}
|
|
if payload.Resource != "group" || payload.ResourceID != "grp-test" || payload.Result != "succeeded" {
|
|
t.Fatalf("unexpected audit payload: %#v", payload)
|
|
}
|
|
}
|
|
|
|
func TestEnqueueKafkaFileAttachmentsBindRunsSynchronously(t *testing.T) {
|
|
ctrl := gomock.NewController(t)
|
|
defer ctrl.Finish()
|
|
var uidGenerator types.UidGenerator
|
|
if err := uidGenerator.Init(1, []byte("0123456789abcdef")); err != nil {
|
|
t.Fatalf("initialize UID generator: %v", err)
|
|
}
|
|
store.SetTestUidGenerator(uidGenerator)
|
|
defer store.SetTestUidGenerator(types.UidGenerator{})
|
|
|
|
oldKafkaBus := kafkaBus
|
|
oldFiles := store.Files
|
|
defer func() {
|
|
kafkaBus = oldKafkaBus
|
|
store.Files = oldFiles
|
|
}()
|
|
|
|
kafkaBus = &kafkaRuntime{}
|
|
files := mock_store.NewMockFilePersistenceInterface(ctrl)
|
|
store.Files = files
|
|
databaseMsgID := types.Uid(101)
|
|
msgID := store.EncodeUid(int64(databaseMsgID))
|
|
msg := &types.Message{
|
|
ObjHeader: types.ObjHeader{Id: msgID.String(), CreatedAt: time.Date(2026, 9, 1, 10, 0, 0, 0, time.UTC)},
|
|
TenantID: 7,
|
|
Topic: "usr42",
|
|
SeqId: 3,
|
|
From: types.Uid(42).String(),
|
|
Head: types.KVMap{"x-im-type": "image"},
|
|
}
|
|
attachments := []string{"https://api.118im.com/v1/im/files/file_001"}
|
|
files.EXPECT().LinkAttachments(types.TenantID(7), "", databaseMsgID, attachments).Return(nil)
|
|
|
|
if err := enqueueKafkaFileAttachmentsBind(context.Background(), 7, msg, attachments); err != nil {
|
|
t.Fatalf("enqueueKafkaFileAttachmentsBind returned error: %v", err)
|
|
}
|
|
}
|