AskaEth
95f97f5df3
- 只翻注释/文档字符串：剥掉行尾注释后，每条「-」都有逐字节相同的「+」对应行（8 组，代码零改动） - 保留协议/线上标识符原文：errcode、bot_token、context_token、getupdates、binded_redirect、PKCS#7、ECB 等 - 术语对齐既有译法：owner→主人、peer→对端、long poll→长轮询、backoff→退避、sentinel error→哨兵错误 - 已知既有欠债（未顺手修）：login_reuse_test.go 在 HEAD 上就未 gofmt 干净；login_test.go 的 TestStartLogin_ScanStatusTransition 91 行 >80（HEAD 同为 91 行）
280 lines
8.3 KiB
Go
280 lines
8.3 KiB
Go
package weixin_ilink
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
|
)
|
|
|
|
// msgCollector 在互斥锁下收集已投递的消息。
|
|
type msgCollector struct {
|
|
mu sync.Mutex
|
|
msgs []adapter.UnifiedMessage
|
|
}
|
|
|
|
func (c *msgCollector) add(m adapter.UnifiedMessage) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
c.msgs = append(c.msgs, m)
|
|
}
|
|
|
|
func (c *msgCollector) snapshot() []adapter.UnifiedMessage {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
return append([]adapter.UnifiedMessage(nil), c.msgs...)
|
|
}
|
|
|
|
func (c *msgCollector) count() int {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
return len(c.msgs)
|
|
}
|
|
|
|
func textItem(t string) map[string]any {
|
|
return map[string]any{"type": 1, "text_item": map[string]any{"text": t}}
|
|
}
|
|
|
|
func TestReadMessages_LongPoll_CursorAndContextToken(t *testing.T) {
|
|
t.Parallel()
|
|
rec := &reqRecorder{}
|
|
msg1, msg2 := longPollMessages()
|
|
|
|
updatesCalls := 0
|
|
srv := newJSONServer(t, rec, func(req recordedReq) handlerResult {
|
|
if req.path != "/ilink/bot/getupdates" {
|
|
return handlerResult{status: 404}
|
|
}
|
|
updatesCalls++
|
|
if updatesCalls == 1 {
|
|
return handlerResult{body: jsonBody(map[string]any{
|
|
"msgs": []any{msg1, msg2}, "get_updates_buf": "cursor-1",
|
|
})}
|
|
}
|
|
return handlerResult{body: jsonBody(map[string]any{
|
|
"msgs": []any{}, "get_updates_buf": "cursor-2",
|
|
})}
|
|
})
|
|
|
|
ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok"}, srv)
|
|
|
|
col := &msgCollector{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
ad.ReadMessages(ctx, col.add)
|
|
}()
|
|
|
|
waitFor(t, func() bool { return col.count() >= 2 })
|
|
// 确保第二次长轮询已执行，使游标回传可被观测。
|
|
waitFor(t, func() bool { return len(getupdatesReqs(rec)) >= 2 })
|
|
cancel()
|
|
waitDone(t, done)
|
|
|
|
assertDecodedMessages(t, col.snapshot())
|
|
assertCursorAndContext(t, ad, rec)
|
|
}
|
|
|
|
func longPollMessages() (json.RawMessage, json.RawMessage) {
|
|
msg1 := inlineJSON(map[string]any{
|
|
"client_id": "m-1", "from_user_id": "user-1", "to_user_id": "bot-1",
|
|
"message_type": 1, "message_state": 2, "context_token": "tok-a",
|
|
"item_list": []any{textItem("hi")},
|
|
})
|
|
msg2 := inlineJSON(map[string]any{
|
|
"client_id": "m-2", "from_user_id": "user-2", "to_user_id": "bot-1",
|
|
"message_type": 1, "message_state": 2, "context_token": "tok-b",
|
|
"item_list": []any{map[string]any{"type": 2, "image_item": map[string]any{"url": "https://cdn/x.jpg"}}},
|
|
})
|
|
return msg1, msg2
|
|
}
|
|
|
|
func assertDecodedMessages(t *testing.T, msgs []adapter.UnifiedMessage) {
|
|
t.Helper()
|
|
if len(msgs) != 2 {
|
|
t.Fatalf("emitted %d messages, want 2", len(msgs))
|
|
}
|
|
m := msgs[0]
|
|
if m.Platform != "weixin-ilink" || m.SenderID != "user-1" || m.ChannelID != "user-1" {
|
|
t.Fatalf("msg0 identity = platform %q sender %q channel %q", m.Platform, m.SenderID, m.ChannelID)
|
|
}
|
|
if m.ChannelType != adapter.ChannelDirect || m.ContentType != adapter.ContentText || m.Content != "hi" {
|
|
t.Fatalf("msg0 = channel %q content %q type %q", m.ChannelType, m.Content, m.ContentType)
|
|
}
|
|
m2 := msgs[1]
|
|
if m2.ContentType != adapter.ContentImage || len(m2.Media) != 1 {
|
|
t.Fatalf("msg1 = type %q media %d", m2.ContentType, len(m2.Media))
|
|
}
|
|
if m2.Media[0].URL != "https://cdn/x.jpg" || m2.Media[0].Kind != adapter.MediaImage {
|
|
t.Fatalf("msg1 media = %+v", m2.Media[0])
|
|
}
|
|
if m2.Timestamp.IsZero() {
|
|
t.Fatal("msg1 Timestamp is zero")
|
|
}
|
|
}
|
|
|
|
func assertCursorAndContext(t *testing.T, ad *Adapter, rec *reqRecorder) {
|
|
t.Helper()
|
|
if got := ad.contextFor("user-1"); got != "tok-a" {
|
|
t.Fatalf("contextFor(user-1) = %q, want tok-a", got)
|
|
}
|
|
if got := ad.contextFor("user-2"); got != "tok-b" {
|
|
t.Fatalf("contextFor(user-2) = %q, want tok-b", got)
|
|
}
|
|
ups := getupdatesReqs(rec)
|
|
if len(ups) < 2 {
|
|
t.Fatalf("getupdates requests = %d, want >=2", len(ups))
|
|
}
|
|
body2 := ups[1].bodyJSON()
|
|
if body2 == nil {
|
|
t.Fatal("second getupdates body is not JSON")
|
|
}
|
|
if got := body2["get_updates_buf"]; got != "cursor-1" {
|
|
t.Fatalf("second poll get_updates_buf = %v, want cursor-1", got)
|
|
}
|
|
if ups[0].auth != "Bearer tok" {
|
|
t.Fatalf("poll Authorization = %q, want Bearer tok", ups[0].auth)
|
|
}
|
|
}
|
|
|
|
// waitDone 等待读循环 goroutine 退出，超时则让测试失败。
|
|
func waitDone(t *testing.T, done <-chan struct{}) {
|
|
t.Helper()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("ReadMessages did not return after ctx cancel")
|
|
}
|
|
}
|
|
|
|
func TestReadMessages_SkipsNonUserMessages(t *testing.T) {
|
|
t.Parallel()
|
|
rec := &reqRecorder{}
|
|
srv := onceUpdates(t, rec, jsonBody(map[string]any{
|
|
"msgs": []any{
|
|
inlineJSON(map[string]any{"message_type": 1, "from_user_id": "u-1", "item_list": []any{textItem("yo")}}),
|
|
inlineJSON(map[string]any{"message_type": 2, "from_user_id": "bot-1", "item_list": []any{textItem("echo")}}),
|
|
},
|
|
"get_updates_buf": "cursor-1",
|
|
}))
|
|
ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok"}, srv)
|
|
|
|
col := &msgCollector{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
ad.ReadMessages(ctx, col.add)
|
|
}()
|
|
waitFor(t, func() bool { return col.count() >= 1 })
|
|
cancel()
|
|
<-done
|
|
|
|
if got := col.count(); got != 1 {
|
|
t.Fatalf("emitted %d messages, want 1 (bot echo skipped)", got)
|
|
}
|
|
if msgs := col.snapshot(); msgs[0].Content != "yo" {
|
|
t.Fatalf("content = %q, want yo", msgs[0].Content)
|
|
}
|
|
}
|
|
|
|
func TestReadMessages_GroupMessage(t *testing.T) {
|
|
t.Parallel()
|
|
srv := onceUpdates(t, nil, jsonBody(map[string]any{
|
|
"msgs": []any{inlineJSON(map[string]any{
|
|
"message_type": 1, "from_user_id": "u-1", "group_id": "g-1",
|
|
"context_token": "tok-g", "item_list": []any{textItem("group msg")},
|
|
})},
|
|
"get_updates_buf": "c1",
|
|
}))
|
|
ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok"}, srv)
|
|
|
|
col := &msgCollector{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
ad.ReadMessages(ctx, col.add)
|
|
}()
|
|
waitFor(t, func() bool { return col.count() >= 1 })
|
|
cancel()
|
|
<-done
|
|
|
|
m := col.snapshot()[0]
|
|
if m.ChannelType != adapter.ChannelGroup || m.GroupID != "g-1" || m.ChannelID != "g-1" {
|
|
t.Fatalf("group message = channel %q group %q channelid %q", m.ChannelType, m.GroupID, m.ChannelID)
|
|
}
|
|
if !m.Sender.IsGroup || m.Sender.GroupID != "g-1" {
|
|
t.Fatalf("sender group flags = %+v", m.Sender)
|
|
}
|
|
if m.IsGroup() != true {
|
|
t.Fatal("IsGroup() = false, want true")
|
|
}
|
|
}
|
|
|
|
func TestReadMessages_SessionRejected(t *testing.T) {
|
|
t.Parallel()
|
|
srv := newJSONServer(t, nil, func(req recordedReq) handlerResult {
|
|
if req.path != "/ilink/bot/getupdates" {
|
|
return handlerResult{status: 404}
|
|
}
|
|
return handlerResult{body: jsonBody(map[string]any{"errcode": -14})}
|
|
})
|
|
ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok"}, srv)
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
ad.ReadMessages(ctx, func(m adapter.UnifiedMessage) {})
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
case <-time.After(2 * time.Second):
|
|
cancel()
|
|
t.Fatal("ReadMessages did not stop on a session iLink rejected")
|
|
}
|
|
cancel()
|
|
if ad.IsConnected() {
|
|
t.Fatal("IsConnected() = true after iLink rejected the session")
|
|
}
|
|
}
|
|
|
|
func TestReadMessages_HandlerDelivery(t *testing.T) {
|
|
t.Parallel()
|
|
srv := onceUpdates(t, nil, jsonBody(map[string]any{
|
|
"msgs": []any{inlineJSON(map[string]any{
|
|
"message_type": 1, "from_user_id": "u-1", "context_token": "tok-h",
|
|
"item_list": []any{textItem("hello")},
|
|
})},
|
|
"get_updates_buf": "c1",
|
|
}))
|
|
ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok"}, srv)
|
|
|
|
handlerCalls := &msgCollector{}
|
|
ad.SetMessageHandler(func(_ context.Context, m adapter.UnifiedMessage) { handlerCalls.add(m) })
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
// emit 故意传 nil：消息必须由处理器接收。
|
|
ad.ReadMessages(ctx, nil)
|
|
}()
|
|
waitFor(t, func() bool { return handlerCalls.count() >= 1 })
|
|
cancel()
|
|
<-done
|
|
|
|
if got := handlerCalls.count(); got != 1 {
|
|
t.Fatalf("handler called %d times, want 1", got)
|
|
}
|
|
if m := handlerCalls.snapshot()[0]; m.Content != "hello" {
|
|
t.Fatalf("handler content = %q, want hello", m.Content)
|
|
}
|
|
}
|