shangguanxuan.sgx
99e5a3cceb
The macOS/Windows coverage gates only execute tests matching ^(TestAllShortcuts|TestCrossPlatformCoverage), so the 401 refresh-retry tests added for this change were invisible to them, leaving 12 changed statements uncovered (92.73% < 100%). Rename the 12 existing tests into the TestCrossPlatformCoverage prefix and add a fetchTicketAttempt edge test covering transport failures, retryable statuses, and missing endpoint/ticket payload fields.
431 lines
16 KiB
Go
431 lines
16 KiB
Go
package source
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
|
|
"github.com/gorilla/websocket"
|
|
"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
|
|
)
|
|
|
|
// TestCrossPlatformCoveragePortalStart401RefreshRetryEndToEnd drives the full production chain
|
|
// DingtalkSource.Start → startPortalTicket → requestPortalTicket: the first
|
|
// ticket request is rejected with 401, ForceRefreshToken rotates the token,
|
|
// the in-chain retry succeeds with the fresh token and a WebSocket event is
|
|
// delivered to emit.
|
|
func TestCrossPlatformCoveragePortalStart401RefreshRetryEndToEnd(t *testing.T) {
|
|
var ticketCalls, refreshCalls atomic.Int64
|
|
var rejectedSeen atomic.Value
|
|
|
|
upgrader := websocket.Upgrader{}
|
|
var wsURL string
|
|
mux := http.NewServeMux()
|
|
mux.HandleFunc("/ticket", func(w http.ResponseWriter, r *http.Request) {
|
|
ticketCalls.Add(1)
|
|
switch r.Header.Get("x-user-access-token") {
|
|
case "fresh-token":
|
|
_ = json.NewEncoder(w).Encode(map[string]any{
|
|
"success": true,
|
|
"result": map[string]string{"endpoint": wsURL, "ticket": "ticket-1"},
|
|
})
|
|
default:
|
|
w.WriteHeader(http.StatusUnauthorized)
|
|
_, _ = io.WriteString(w, "token expired")
|
|
}
|
|
})
|
|
mux.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
|
|
conn, err := upgrader.Upgrade(w, r, nil)
|
|
if err != nil {
|
|
return
|
|
}
|
|
defer conn.Close()
|
|
df := payload.DataFrame{Type: "event", Headers: payload.DataFrameHeader{payload.DataFrameHeaderKMessageId: "msg-1"}, Data: `{}`}
|
|
_ = conn.WriteJSON(df)
|
|
_, _, _ = conn.ReadMessage()
|
|
})
|
|
srv := httptest.NewServer(mux)
|
|
defer srv.Close()
|
|
wsURL = "ws" + strings.TrimPrefix(srv.URL, "http") + "/ws"
|
|
|
|
s, err := New(Config{PortalTicket: &PortalTicketConfig{
|
|
TicketURL: srv.URL + "/ticket",
|
|
AccessTokenProvider: func(context.Context) (string, error) {
|
|
return "stale-token", nil
|
|
},
|
|
ForceRefreshToken: func(_ context.Context, rejectedToken string) (string, error) {
|
|
refreshCalls.Add(1)
|
|
rejectedSeen.Store(rejectedToken)
|
|
return "fresh-token", nil
|
|
},
|
|
SourceID: "source",
|
|
HTTPClient: srv.Client(),
|
|
}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
emitted := make(chan struct{}, 1)
|
|
done := make(chan error, 1)
|
|
go func() { done <- s.Start(ctx, func(*dwsevent.RawEvent) { emitted <- struct{}{} }) }()
|
|
select {
|
|
case <-emitted:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("portal event timeout after 401 refresh retry")
|
|
}
|
|
cancel()
|
|
select {
|
|
case err := <-done:
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("portal stop = %v", err)
|
|
}
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("portal stop timeout")
|
|
}
|
|
|
|
if got := ticketCalls.Load(); got != 2 {
|
|
t.Fatalf("ticket calls = %d, want 2", got)
|
|
}
|
|
if got := refreshCalls.Load(); got != 1 {
|
|
t.Fatalf("refresh calls = %d, want 1", got)
|
|
}
|
|
if got, _ := rejectedSeen.Load().(string); got != "stale-token" {
|
|
t.Fatalf("rejected token = %q, want %q", got, "stale-token")
|
|
}
|
|
}
|
|
|
|
// TestCrossPlatformCoverageRequestPortalTicketRetryUsesRotatedTokenDirectly asserts the in-chain
|
|
// retry sends the token returned by ForceRefreshToken instead of re-invoking
|
|
// the provider (which could still serve the stale token).
|
|
func TestCrossPlatformCoverageRequestPortalTicketRetryUsesRotatedTokenDirectly(t *testing.T) {
|
|
providerCalls := 0
|
|
var attemptTokens []string
|
|
client := &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
|
|
token := req.Header.Get("x-user-access-token")
|
|
attemptTokens = append(attemptTokens, token)
|
|
if token != "rotated" {
|
|
return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
|
|
}
|
|
return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"endpoint":"wss://x","ticket":"t"}`)), Header: make(http.Header)}, nil
|
|
})}
|
|
ticket, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
|
|
TicketURL: "https://x",
|
|
AccessTokenProvider: func(context.Context) (string, error) {
|
|
providerCalls++
|
|
return "stale", nil
|
|
},
|
|
ForceRefreshToken: func(_ context.Context, rejectedToken string) (string, error) {
|
|
if rejectedToken != "stale" {
|
|
t.Fatalf("rejected token = %q, want %q", rejectedToken, "stale")
|
|
}
|
|
return "rotated", nil
|
|
},
|
|
SourceID: "s",
|
|
HTTPClient: client,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("requestPortalTicket = %v", err)
|
|
}
|
|
if ticket.Endpoint != "wss://x" || ticket.Ticket != "t" {
|
|
t.Fatalf("ticket = %#v", ticket)
|
|
}
|
|
if providerCalls != 1 {
|
|
t.Fatalf("provider calls = %d, want 1", providerCalls)
|
|
}
|
|
if len(attemptTokens) != 2 || attemptTokens[0] != "stale" || attemptTokens[1] != "rotated" {
|
|
t.Fatalf("attempt tokens = %v", attemptTokens)
|
|
}
|
|
}
|
|
|
|
// TestCrossPlatformCoverageRequestPortalTicketRefreshFailureKeepsBothErrors asserts a failing
|
|
// refresh neither retries nor drops the refresh error or the original 401.
|
|
func TestCrossPlatformCoverageRequestPortalTicketRefreshFailureKeepsBothErrors(t *testing.T) {
|
|
refreshErr := errors.New("refresh_token exchange failed")
|
|
attempts := 0
|
|
client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
|
|
attempts++
|
|
return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
|
|
})}
|
|
_, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
|
|
TicketURL: "https://x",
|
|
AccessToken: "stale",
|
|
ForceRefreshToken: func(context.Context, string) (string, error) {
|
|
return "", refreshErr
|
|
},
|
|
SourceID: "s",
|
|
HTTPClient: client,
|
|
})
|
|
if !errors.Is(err, refreshErr) {
|
|
t.Fatalf("error should wrap refresh error, got %v", err)
|
|
}
|
|
if err == nil || !strings.Contains(err.Error(), "HTTP 401") {
|
|
t.Fatalf("error should keep original 401, got %v", err)
|
|
}
|
|
if attempts != 1 {
|
|
t.Fatalf("attempts = %d, want 1 (no retry after failed refresh)", attempts)
|
|
}
|
|
}
|
|
|
|
// TestCrossPlatformCoverageRequestPortalTicketWithoutRefreshCallback401StaysFatal covers backward
|
|
// compatibility: nil ForceRefreshToken keeps the single-attempt fatal 401.
|
|
func TestCrossPlatformCoverageRequestPortalTicketWithoutRefreshCallback401StaysFatal(t *testing.T) {
|
|
attempts := 0
|
|
client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
|
|
attempts++
|
|
return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
|
|
})}
|
|
_, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
|
|
TicketURL: "https://x", AccessToken: "stale", SourceID: "s", HTTPClient: client,
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "HTTP 401") {
|
|
t.Fatalf("fatal 401 expected, got %v", err)
|
|
}
|
|
if attempts != 1 {
|
|
t.Fatalf("attempts = %d, want 1", attempts)
|
|
}
|
|
}
|
|
|
|
// TestCrossPlatformCoverageRequestPortalTicketSecond401IsFatal guards against refresh loops: the
|
|
// controlled retry happens exactly once even if the rotated token is also
|
|
// rejected.
|
|
func TestCrossPlatformCoverageRequestPortalTicketSecond401IsFatal(t *testing.T) {
|
|
attempts := 0
|
|
refreshCalls := 0
|
|
client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
|
|
attempts++
|
|
return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
|
|
})}
|
|
_, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
|
|
TicketURL: "https://x",
|
|
AccessToken: "stale",
|
|
ForceRefreshToken: func(context.Context, string) (string, error) {
|
|
refreshCalls++
|
|
return "rotated-but-still-rejected", nil
|
|
},
|
|
SourceID: "s",
|
|
HTTPClient: client,
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "HTTP 401") {
|
|
t.Fatalf("fatal 401 expected after single retry, got %v", err)
|
|
}
|
|
if attempts != 2 || refreshCalls != 1 {
|
|
t.Fatalf("attempts = %d refreshCalls = %d, want 2/1", attempts, refreshCalls)
|
|
}
|
|
}
|
|
|
|
// TestCrossPlatformCoverageRequestPortalTicketRefreshEmptyTokenIsFatal asserts an empty rotated
|
|
// token is rejected instead of being sent to the server.
|
|
func TestCrossPlatformCoverageRequestPortalTicketRefreshEmptyTokenIsFatal(t *testing.T) {
|
|
attempts := 0
|
|
client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
|
|
attempts++
|
|
return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
|
|
})}
|
|
_, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
|
|
TicketURL: "https://x",
|
|
AccessToken: "stale",
|
|
ForceRefreshToken: func(context.Context, string) (string, error) {
|
|
return " ", nil
|
|
},
|
|
SourceID: "s",
|
|
HTTPClient: client,
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "empty token") {
|
|
t.Fatalf("empty rotated token error expected, got %v", err)
|
|
}
|
|
if attempts != 1 {
|
|
t.Fatalf("attempts = %d, want 1", attempts)
|
|
}
|
|
}
|
|
|
|
// TestCrossPlatformCoveragePersonalFetchTicket401RefreshRetry mirrors the portal behavior for the
|
|
// personal stream ticket path.
|
|
func TestCrossPlatformCoveragePersonalFetchTicket401RefreshRetry(t *testing.T) {
|
|
var attemptTokens []string
|
|
src, err := NewPersonal(PersonalConfig{
|
|
AccessTokenProvider: func(context.Context) (string, error) { return "stale", nil },
|
|
ForceRefreshToken: func(_ context.Context, rejectedToken string) (string, error) {
|
|
if rejectedToken != "stale" {
|
|
t.Fatalf("rejected token = %q, want %q", rejectedToken, "stale")
|
|
}
|
|
return "rotated", nil
|
|
},
|
|
ClientID: "client",
|
|
SourceID: "source",
|
|
TicketURL: "https://ticket.test",
|
|
HTTPClient: &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
|
|
token := req.Header.Get("x-user-access-token")
|
|
attemptTokens = append(attemptTokens, token)
|
|
if token != "rotated" {
|
|
return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
|
|
}
|
|
return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"endpoint":"wss://stream.test","ticket":"ticket"}`)), Header: make(http.Header)}, nil
|
|
})},
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ticket, err := src.fetchTicket(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("fetchTicket = %v", err)
|
|
}
|
|
if ticket.Endpoint != "wss://stream.test" || ticket.Ticket != "ticket" {
|
|
t.Fatalf("ticket = %#v", ticket)
|
|
}
|
|
if len(attemptTokens) != 2 || attemptTokens[0] != "stale" || attemptTokens[1] != "rotated" {
|
|
t.Fatalf("attempt tokens = %v", attemptTokens)
|
|
}
|
|
}
|
|
|
|
// TestCrossPlatformCoveragePersonalFetchTicket401RefreshFailureStaysFatal asserts a failed refresh
|
|
// keeps the 401 fatal (not retryable) and wraps the refresh error.
|
|
func TestCrossPlatformCoveragePersonalFetchTicket401RefreshFailureStaysFatal(t *testing.T) {
|
|
refreshErr := errors.New("refresh_token exchange failed")
|
|
src, err := NewPersonal(PersonalConfig{
|
|
AccessToken: "stale",
|
|
ForceRefreshToken: func(context.Context, string) (string, error) {
|
|
return "", refreshErr
|
|
},
|
|
ClientID: "client",
|
|
SourceID: "source",
|
|
TicketURL: "https://ticket.test",
|
|
HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
|
|
return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("expired")), Header: make(http.Header)}, nil
|
|
})},
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, err = src.fetchTicket(context.Background())
|
|
if !errors.Is(err, refreshErr) {
|
|
t.Fatalf("error should wrap refresh error, got %v", err)
|
|
}
|
|
if isRetryablePersonalError(err) {
|
|
t.Fatalf("failed refresh should stay fatal, got retryable %v", err)
|
|
}
|
|
}
|
|
|
|
// brokenBody simulates a response body that fails mid-read, e.g. the server
|
|
// closing the connection before the error payload is fully written.
|
|
type brokenBody struct{}
|
|
|
|
func (brokenBody) Read([]byte) (int, error) { return 0, io.ErrUnexpectedEOF }
|
|
func (brokenBody) Close() error { return nil }
|
|
|
|
// TestCrossPlatformCoveragePersonalFetchTicket401TruncatedBodyStaysFatal guards the single
|
|
// refresh-retry protection: a 401 whose body fails with unexpected EOF must
|
|
// be classified by status (fatal) and never wrapped as retryable, otherwise
|
|
// the outer reconnect loop would refresh again on every iteration.
|
|
func TestCrossPlatformCoveragePersonalFetchTicket401TruncatedBodyStaysFatal(t *testing.T) {
|
|
attempts := 0
|
|
refreshCalls := 0
|
|
src, err := NewPersonal(PersonalConfig{
|
|
AccessToken: "stale",
|
|
ForceRefreshToken: func(context.Context, string) (string, error) {
|
|
refreshCalls++
|
|
return "rotated", nil
|
|
},
|
|
ClientID: "client",
|
|
SourceID: "source",
|
|
TicketURL: "https://ticket.test",
|
|
HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
|
|
attempts++
|
|
return &http.Response{StatusCode: http.StatusUnauthorized, Body: brokenBody{}, Header: make(http.Header)}, nil
|
|
})},
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, err = src.fetchTicket(context.Background())
|
|
if err == nil || !strings.Contains(err.Error(), "HTTP 401") {
|
|
t.Fatalf("fatal 401 error expected, got %v", err)
|
|
}
|
|
if isRetryablePersonalError(err) {
|
|
t.Fatalf("401 with truncated body must stay fatal, got retryable %v", err)
|
|
}
|
|
if refreshCalls != 1 {
|
|
t.Fatalf("refresh calls = %d, want 1", refreshCalls)
|
|
}
|
|
if attempts != 2 {
|
|
t.Fatalf("attempts = %d, want 2", attempts)
|
|
}
|
|
}
|
|
|
|
// TestCrossPlatformCoveragePersonalFetchTicket200TruncatedBodyStaysRetryable pins the existing
|
|
// behavior for success responses: a body read failure on 2xx is a transient
|
|
// transport problem and remains retryable.
|
|
func TestCrossPlatformCoveragePersonalFetchTicket200TruncatedBodyStaysRetryable(t *testing.T) {
|
|
src, err := NewPersonal(PersonalConfig{
|
|
AccessToken: "token",
|
|
ClientID: "client",
|
|
SourceID: "source",
|
|
TicketURL: "https://ticket.test",
|
|
HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
|
|
return &http.Response{StatusCode: 200, Body: brokenBody{}, Header: make(http.Header)}, nil
|
|
})},
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, err = src.fetchTicket(context.Background())
|
|
if !errors.Is(err, io.ErrUnexpectedEOF) {
|
|
t.Fatalf("error should wrap io.ErrUnexpectedEOF, got %v", err)
|
|
}
|
|
if !isRetryablePersonalError(err) {
|
|
t.Fatalf("2xx body read failure should stay retryable, got %v", err)
|
|
}
|
|
}
|
|
|
|
// TestCrossPlatformCoveragePersonalFetchTicketAttemptTransportAndPayloadEdges covers the
|
|
// remaining fetchTicketAttempt branches: transport failures and retryable
|
|
// statuses stay retryable, while a well-formed response missing the endpoint
|
|
// or ticket fields stays fatal.
|
|
func TestCrossPlatformCoveragePersonalFetchTicketAttemptTransportAndPayloadEdges(t *testing.T) {
|
|
newSource := func(rt roundTripFunc) *PersonalSource {
|
|
src, err := NewPersonal(PersonalConfig{
|
|
AccessToken: "token",
|
|
ClientID: "client",
|
|
SourceID: "source",
|
|
TicketURL: "https://ticket.test",
|
|
HTTPClient: &http.Client{Transport: rt},
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return src
|
|
}
|
|
|
|
dialErr := errors.New("dial tcp: connection refused")
|
|
src := newSource(func(*http.Request) (*http.Response, error) { return nil, dialErr })
|
|
_, _, err := src.fetchTicketAttempt(context.Background(), "token")
|
|
if !errors.Is(err, dialErr) || !isRetryablePersonalError(err) {
|
|
t.Fatalf("transport failure should stay retryable and wrap cause, got %v", err)
|
|
}
|
|
|
|
src = newSource(func(*http.Request) (*http.Response, error) {
|
|
return &http.Response{StatusCode: http.StatusServiceUnavailable, Body: io.NopCloser(strings.NewReader("busy")), Header: make(http.Header)}, nil
|
|
})
|
|
_, status, err := src.fetchTicketAttempt(context.Background(), "token")
|
|
if status != http.StatusServiceUnavailable || !isRetryablePersonalError(err) {
|
|
t.Fatalf("503 should stay retryable, got status %d err %v", status, err)
|
|
}
|
|
|
|
src = newSource(func(*http.Request) (*http.Response, error) {
|
|
return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"endpoint":"","ticket":""}`)), Header: make(http.Header)}, nil
|
|
})
|
|
_, _, err = src.fetchTicketAttempt(context.Background(), "token")
|
|
if err == nil || isRetryablePersonalError(err) || !strings.Contains(err.Error(), "missing endpoint or ticket") {
|
|
t.Fatalf("missing ticket fields should stay fatal, got %v", err)
|
|
}
|
|
}
|