dingtalk-workspace-cli/internal/event/busctl/consumer_stop_test.go

// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");

package busctl

import (
	"errors"
	"net"
	"reflect"
	"strings"
	"testing"

	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
)

func TestStopConsumersRoundTrip(t *testing.T) {
	client, server := net.Pipe()
	oldDial := consumerStopDial
	consumerStopDial = func(string) (net.Conn, error) { return client, nil }
	t.Cleanup(func() {
		consumerStopDial = oldDial
		_ = server.Close()
	})

	requestCh := make(chan transport.ConsumerStopReq, 1)
	errCh := make(chan error, 1)
	go func() {
		r := transport.NewReader(server)
		w := transport.NewWriter(server)
		var hello transport.Hello
		if err := r.ReadJSON(&hello); err != nil {
			errCh <- err
			return
		}
		if hello.Role != transport.HelloRoleConsumerStop {
			errCh <- errors.New("unexpected hello role")
			return
		}
		var req transport.ConsumerStopReq
		if err := r.ReadJSON(&req); err != nil {
			errCh <- err
			return
		}
		requestCh <- req
		errCh <- w.WriteJSON(transport.ConsumerStopResp{
			Type:     transport.FrameTypeConsumerStopResp,
			Stopped:  []string{"sub-a"},
			NotFound: []string{"sub-b"},
		})
	}()

	resp, err := StopConsumers("pipe", []string{" sub-a ", "sub-a", "", "sub-b"})
	if err != nil {
		t.Fatalf("StopConsumers() error = %v", err)
	}
	if !reflect.DeepEqual(resp.Stopped, []string{"sub-a"}) || !reflect.DeepEqual(resp.NotFound, []string{"sub-b"}) {
		t.Fatalf("response = %#v", resp)
	}
	if err := <-errCh; err != nil {
		t.Fatalf("server error = %v", err)
	}
	req := <-requestCh
	if req.Type != transport.FrameTypeConsumerStopReq || !reflect.DeepEqual(req.SubscribeIDs, []string{"sub-a", "sub-b"}) {
		t.Fatalf("request = %#v", req)
	}
}

func TestStopConsumersDetectsLegacyBus(t *testing.T) {
	client, server := net.Pipe()
	oldDial := consumerStopDial
	consumerStopDial = func(string) (net.Conn, error) { return client, nil }
	t.Cleanup(func() {
		consumerStopDial = oldDial
		_ = server.Close()
	})

	go func() {
		r := transport.NewReader(server)
		w := transport.NewWriter(server)
		var hello transport.Hello
		_ = r.ReadJSON(&hello)
		var req transport.ConsumerStopReq
		_ = r.ReadJSON(&req)
		_ = w.WriteJSON(transport.HelloAck{Type: transport.FrameTypeHelloAck, BusPID: 1})
	}()

	_, err := StopConsumers("pipe", []string{"sub-a"})
	if !errors.Is(err, ErrConsumerStopUnsupported) {
		t.Fatalf("StopConsumers() error = %v, want ErrConsumerStopUnsupported", err)
	}
}

func TestStopConsumersReturnsServerProtocolError(t *testing.T) {
	client, server := net.Pipe()
	oldDial := consumerStopDial
	consumerStopDial = func(string) (net.Conn, error) { return client, nil }
	t.Cleanup(func() {
		consumerStopDial = oldDial
		_ = server.Close()
	})

	go func() {
		r := transport.NewReader(server)
		w := transport.NewWriter(server)
		var hello transport.Hello
		_ = r.ReadJSON(&hello)
		var req transport.ConsumerStopReq
		_ = r.ReadJSON(&req)
		_ = w.WriteJSON(transport.ConsumerStopResp{
			Type:  transport.FrameTypeConsumerStopResp,
			Error: "malformed consumer stop request",
		})
	}()

	_, err := StopConsumers("pipe", []string{"sub-a"})
	if err == nil || !strings.Contains(err.Error(), "malformed consumer stop request") {
		t.Fatalf("StopConsumers() error = %v", err)
	}
}

func TestStopConsumersValidatesInputAndDialErrors(t *testing.T) {
	if _, err := StopConsumers("", []string{"sub-a"}); err == nil {
		t.Fatal("empty endpoint succeeded")
	}
	if _, err := StopConsumers("pipe", []string{"", "  "}); err == nil {
		t.Fatal("empty subscriptions succeeded")
	}

	wantErr := errors.New("dial")
	oldDial := consumerStopDial
	consumerStopDial = func(string) (net.Conn, error) { return nil, wantErr }
	t.Cleanup(func() { consumerStopDial = oldDial })
	if _, err := StopConsumers("pipe", []string{"sub-a"}); !errors.Is(err, wantErr) {
		t.Fatalf("dial error = %v", err)
	}
}

func TestCrossPlatformCoverageStopConsumersProtocolErrors(t *testing.T) {
	oldDial := consumerStopDial
	t.Cleanup(func() { consumerStopDial = oldDial })

	for _, test := range []struct {
		name   string
		failAt int
		want   string
	}{
		{name: "hello write", failAt: 1, want: "write consumer stop hello"},
		{name: "request write", failAt: 2, want: "write consumer stop request"},
		{name: "response read", failAt: 0, want: "read consumer stop response"},
	} {
		t.Run(test.name, func(t *testing.T) {
			consumerStopDial = func(string) (net.Conn, error) {
				return &queryErrorConn{failAt: test.failAt}, nil
			}
			_, err := StopConsumers("pipe", []string{"sub-a"})
			if !errors.Is(err, errBusctlInjected) || !strings.Contains(err.Error(), test.want) {
				t.Fatalf("error = %v, want %q", err, test.want)
			}
		})
	}
}