Cyrene_v2/plugins/channels/qq_obv11/adapter_test.go

package qq_obv11

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
)

func TestAdapterNameAndConfig(t *testing.T) {
	t.Parallel()

	a := NewAdapter(Config{
		Mode:                  "client",
		RemoteURL:             "ws://onebot.example:6700",
		AccessToken:           "secret",
		MaxProactivePerWindow: 3,
		ProactiveWindow:       5 * time.Second,
	})
	if a.Name() != platformName {
		t.Errorf("Name() = %q, want %q", a.Name(), platformName)
	}
	cfg := a.Config()
	if cfg["mode"] != "client" {
		t.Errorf("Config() mode = %v, want client", cfg["mode"])
	}
	if cfg["remoteURL"] != "ws://onebot.example:6700" {
		t.Errorf("Config() remoteURL = %v", cfg["remoteURL"])
	}
	if cfg["accessToken"] != "secret" {
		t.Errorf("Config() accessToken = %v", cfg["accessToken"])
	}
	if cfg["maxProactivePerWindow"] != 3 {
		t.Errorf("Config() maxProactivePerWindow = %v, want 3", cfg["maxProactivePerWindow"])
	}
	if got := cfg["proactiveWindowSeconds"].(float64); got != 5 {
		t.Errorf("Config() proactiveWindowSeconds = %v, want 5", got)
	}
}

func TestAdapterDefaultModeIsServer(t *testing.T) {
	t.Parallel()

	a := NewAdapter(Config{})
	if a.Config()["mode"] != "server" {
		t.Errorf("default mode = %v, want server", a.Config()["mode"])
	}
}

func TestAdapterCapabilities(t *testing.T) {
	t.Parallel()

	caps := NewAdapter(Config{}).Capabilities().Normalize()
	if caps.MaxMessageLength != defaultMaxMessageLength {
		t.Errorf("MaxMessageLength = %d, want %d", caps.MaxMessageLength, defaultMaxMessageLength)
	}
	if caps.RecommendBurstMax != defaultRecommendBurstMax {
		t.Errorf("RecommendBurstMax = %d, want %d", caps.RecommendBurstMax, defaultRecommendBurstMax)
	}
	if !caps.Markdown || !caps.Image || !caps.Voice || !caps.Proactive {
		t.Errorf("expected markdown/image/voice/proactive on, got %+v", caps)
	}
	if caps.Reaction || caps.TypingHint {
		t.Errorf("expected reaction/typing off, got %+v", caps)
	}
}

func TestAdapterSetMessageHandlerAndDispatch(t *testing.T) {
	t.Parallel()

	a := NewAdapter(Config{})
	got := make(chan adapter.UnifiedMessage, 1)
	a.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) {
		got <- m
	})

	want := adapter.UnifiedMessage{Platform: platformName, ChannelType: adapter.ChannelDirect, ContentType: adapter.ContentText, Content: "hi"}
	a.dispatch(context.Background(), want)

	select {
	case m := <-got:
		if m.Content != "hi" {
			t.Errorf("handler got content = %q, want hi", m.Content)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("handler not invoked")
	}
}

func TestAdapterHealthCheckDisconnected(t *testing.T) {
	t.Parallel()

	a := NewAdapter(Config{Mode: "server", ListenAddr: "127.0.0.1:0"})
	if a.IsConnected() {
		t.Error("IsConnected() = true before Connect, want false")
	}
	if err := a.HealthCheck(context.Background()); !errors.Is(err, adapter.ErrNotConnected) {
		t.Errorf("HealthCheck() error = %v, want ErrNotConnected", err)
	}
}

func TestAdapterSendWhenDisconnected(t *testing.T) {
	t.Parallel()

	a := NewAdapter(Config{})
	err := a.SendMessage(context.Background(), adapter.UnifiedResponse{
		Messages: []adapter.UnifiedMessage{{ChannelType: adapter.ChannelDirect, ChannelID: "100", Content: "x"}},
	})
	if !errors.Is(err, adapter.ErrNotConnected) {
		t.Errorf("SendMessage() error = %v, want ErrNotConnected", err)
	}
}

func TestAdapterDisconnectIdempotent(t *testing.T) {
	t.Parallel()

	a := NewAdapter(Config{})
	ctx := context.Background()
	if err := a.Disconnect(ctx); err != nil {
		t.Fatalf("first Disconnect error: %v", err)
	}
	if err := a.Disconnect(ctx); err != nil {
		t.Fatalf("second Disconnect error: %v", err)
	}
}

// TestAdapterServerModeRoundTrip spins up the adapter as a server, connects a
// fake OneBot client, and verifies the message event is normalized and that a
// reply is emitted as a send_msg frame back to the client.
func TestAdapterServerModeRoundTrip(t *testing.T) {
	t.Parallel()

	a := NewAdapter(Config{Mode: "server", ListenAddr: "127.0.0.1:0"})
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}
	defer func() { _ = a.Disconnect(context.Background()) }()

	addr := a.Addr()
	if addr == "" {
		t.Fatal("Addr() empty after Connect")
	}
	if a.IsConnected() {
		t.Error("server mode should not be connected before a client connects")
	}

	msgCh := make(chan adapter.UnifiedMessage, 1)
	a.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) { msgCh <- m })

	client, err := dialWS(context.Background(), nil, "ws://"+addr+"/", "")
	if err != nil {
		t.Fatalf("client dial error: %v", err)
	}
	defer client.Close()

	if !waitUntil(3*time.Second, a.IsConnected) {
		t.Fatal("server never reported connected after client dialed")
	}

	evt := `{"post_type":"message","message_type":"group","time":1700000000,"self_id":100,"message_id":9,"user_id":42,"group_id":123,"group_name":"Team","message":[{"type":"text","data":{"text":"hello"}}],"sender":{"user_id":42,"nickname":"Bob","card":"Bobby","role":"member"}}`
	if err := client.WriteMessage(opText, []byte(evt)); err != nil {
		t.Fatalf("client write error: %v", err)
	}

	select {
	case m := <-msgCh:
		if m.Platform != platformName || m.ChannelType != adapter.ChannelGroup || m.GroupID != "123" {
			t.Errorf("unexpected message: platform=%q type=%q group=%q", m.Platform, m.ChannelType, m.GroupID)
		}
		if m.GroupName != "Team" || m.Content != "hello" {
			t.Errorf("group message = %+v, want Team/hello", m)
		}
		if m.SenderName != "Bobby" || m.Sender.Role != adapter.RoleMember {
			t.Errorf("sender = %+v, want Bobby/member", m.Sender)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("server handler never received a message")
	}

	// Reply.
	reply := adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{
		{ChannelType: adapter.ChannelGroup, GroupID: "123", Content: "pong"},
	}}
	if err := a.SendMessage(context.Background(), reply); err != nil {
		t.Fatalf("SendMessage error: %v", err)
	}

	seen := make(chan []byte, 1)
	go func() {
		_, payload, err := client.ReadMessage()
		if err != nil {
			seen <- nil
			return
		}
		seen <- payload
	}()
	select {
	case payload := <-seen:
		if payload == nil {
			t.Fatal("client never received a send_msg")
		}
		assertSendMsg(t, payload)
	case <-time.After(3 * time.Second):
		t.Fatal("client never received a send_msg frame")
	}
}

// TestAdapterClientModeReceiveAndSend spins up a fake OneBot server that pushes
// a group event, then verifies the adapter (client) normalizes it and replies.
func TestAdapterClientModeReceiveAndSend(t *testing.T) {
	t.Parallel()

	gotCh := make(chan []byte, 4)
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := upgradeWS(w, r)
		if err != nil {
			t.Errorf("server upgrade error: %v", err)
			return
		}
		defer conn.Close()
		evt := `{"post_type":"message","message_type":"group","time":1700000001,"self_id":100,"message_id":10,"user_id":42,"group_id":123,"message":[{"type":"text","data":{"text":"hi"}}],"sender":{"user_id":42,"nickname":"Bob","role":"member"}}`
		_ = conn.WriteMessage(opText, []byte(evt))
		for {
			_, payload, err := conn.ReadMessage()
			if err != nil {
				return
			}
			gotCh <- payload
		}
	}))
	defer srv.Close()

	url := "ws://" + strings.TrimPrefix(srv.URL, "http://")
	a := NewAdapter(Config{Mode: "client", RemoteURL: url})
	msgCh := make(chan adapter.UnifiedMessage, 1)
	a.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) { msgCh <- m })

	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}
	defer func() { _ = a.Disconnect(context.Background()) }()

	if !a.IsConnected() {
		t.Error("client mode IsConnected() = false after Connect, want true")
	}
	if err := a.HealthCheck(context.Background()); err != nil {
		t.Errorf("HealthCheck() error = %v, want nil", err)
	}

	select {
	case m := <-msgCh:
		if m.ChannelType != adapter.ChannelGroup || m.GroupID != "123" || m.Content != "hi" {
			t.Errorf("received message = %+v, want group/123/hi", m)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("adapter never received the pushed event")
	}

	if err := a.SendMessage(context.Background(), adapter.UnifiedResponse{
		Messages: []adapter.UnifiedMessage{{ChannelType: adapter.ChannelGroup, GroupID: "123", Content: "pong"}},
	}); err != nil {
		t.Fatalf("SendMessage error: %v", err)
	}

	select {
	case payload := <-gotCh:
		assertSendMsg(t, payload)
	case <-time.After(3 * time.Second):
		t.Fatal("fake OneBot server never received a send_msg")
	}
}

// TestSendProactiveRateLimited verifies proactive sends are rejected after the
// configured per-window quota is exceeded.
func TestSendProactiveRateLimited(t *testing.T) {
	t.Parallel()

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := upgradeWS(w, r)
		if err != nil {
			return
		}
		defer conn.Close()
		for {
			if _, _, err := conn.ReadMessage(); err != nil {
				return
			}
		}
	}))
	defer srv.Close()

	url := "ws://" + strings.TrimPrefix(srv.URL, "http://")
	a := NewAdapter(Config{
		Mode:                  "client",
		RemoteURL:             url,
		MaxProactivePerWindow: 2,
		ProactiveWindow:       10 * time.Second,
	})
	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}
	defer func() { _ = a.Disconnect(context.Background()) }()

	msg := adapter.UnifiedMessage{ChannelType: adapter.ChannelDirect, ChannelID: "100", Content: "x"}
	if err := a.SendProactive(context.Background(), msg); err != nil {
		t.Fatalf("first SendProactive error: %v", err)
	}
	if err := a.SendProactive(context.Background(), msg); err != nil {
		t.Fatalf("second SendProactive error: %v", err)
	}
	if err := a.SendProactive(context.Background(), msg); !errors.Is(err, ErrRateLimited) {
		t.Errorf("third SendProactive error = %v, want ErrRateLimited", err)
	}
}

// TestReadMessagesPullMode verifies the pull-based ReadMessages sink receives
// normalized messages when no push handler is installed.
func TestReadMessagesPullMode(t *testing.T) {
	t.Parallel()

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := upgradeWS(w, r)
		if err != nil {
			t.Errorf("server upgrade error: %v", err)
			return
		}
		defer conn.Close()
		evt := `{"post_type":"message","message_type":"private","time":1700000000,"self_id":100,"message_id":3,"user_id":200,"message":[{"type":"text","data":{"text":"hi"}}],"sender":{"user_id":200,"nickname":"Alice","role":""}}`
		_ = conn.WriteMessage(opText, []byte(evt))
		for {
			if _, _, err := conn.ReadMessage(); err != nil {
				return
			}
		}
	}))
	defer srv.Close()

	url := "ws://" + strings.TrimPrefix(srv.URL, "http://")
	a := NewAdapter(Config{Mode: "client", RemoteURL: url})

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	emitCh := make(chan adapter.UnifiedMessage, 1)
	go a.ReadMessages(ctx, func(m adapter.UnifiedMessage) { emitCh <- m })
	// Wait until the sink is registered before connecting, so the pushed event
	// is not dropped before ReadMessages installs its emit callback.
	if !waitUntil(3*time.Second, func() bool {
		a.emitMu.Lock()
		defer a.emitMu.Unlock()
		return a.emit != nil
	}) {
		t.Fatal("ReadMessages never registered its emit sink")
	}

	if err := a.Connect(context.Background()); err != nil {
		t.Fatalf("Connect error: %v", err)
	}
	defer func() { _ = a.Disconnect(context.Background()) }()

	select {
	case m := <-emitCh:
		if m.ChannelType != adapter.ChannelDirect || m.ChannelID != "200" || m.Content != "hi" {
			t.Errorf("pull-mode message = %+v, want direct/200/hi", m)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("ReadMessages never received the pushed event")
	}
}

func assertSendMsg(t *testing.T, payload []byte) {
	t.Helper()
	var req struct {
		Action string `json:"action"`
		Params struct {
			MessageType string           `json:"message_type"`
			UserID      int64            `json:"user_id"`
			GroupID     int64            `json:"group_id"`
			Message     []map[string]any `json:"message"`
		} `json:"params"`
	}
	if err := json.Unmarshal(payload, &req); err != nil {
		t.Fatalf("send_msg unmarshal error: %v (payload=%s)", err, payload)
	}
	if req.Action != "send_msg" {
		t.Errorf("action = %q, want send_msg", req.Action)
	}
	if req.Params.MessageType != "group" {
		t.Errorf("message_type = %q, want group", req.Params.MessageType)
	}
	if req.Params.GroupID != 123 {
		t.Errorf("group_id = %d, want 123", req.Params.GroupID)
	}
	if len(req.Params.Message) != 1 {
		t.Fatalf("message segments len = %d, want 1", len(req.Params.Message))
	}
	first := req.Params.Message[0]
	if first["type"] != "text" || first["data"].(map[string]any)["text"] != "pong" {
		t.Errorf("first segment = %+v, want text pong", first)
	}
}

func waitUntil(timeout time.Duration, cond func() bool) bool {
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if cond() {
			return true
		}
		time.Sleep(20 * time.Millisecond)
	}
	return false
}