351 lines
12 KiB
Go
351 lines
12 KiB
Go
package control
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"cmpp-platform/gateway/internal/inbound"
|
|
"cmpp-platform/gateway/internal/queue"
|
|
)
|
|
|
|
type controlRecordingLimiter struct {
|
|
channelID string
|
|
rate int
|
|
configuredChannelID string
|
|
configuredRate int
|
|
}
|
|
|
|
func (l *controlRecordingLimiter) Wait(_ context.Context, channelID string, rate int) (time.Duration, error) {
|
|
l.channelID = channelID
|
|
l.rate = rate
|
|
return 0, nil
|
|
}
|
|
|
|
func (l *controlRecordingLimiter) Configure(_ context.Context, channelID string, rate int) error {
|
|
l.configuredChannelID = channelID
|
|
l.configuredRate = rate
|
|
return nil
|
|
}
|
|
|
|
func TestUpstreamSubmitUsesGatewayChannelLimiter(t *testing.T) {
|
|
limiter := &controlRecordingLimiter{}
|
|
handler := handlerWithServer(Server{
|
|
Limiter: limiter,
|
|
Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
|
|
return queue.SubmitResult{Envelope: command.Envelope, SubmitStatus: "accepted"}, nil
|
|
},
|
|
})
|
|
resp := httptest.NewRecorder()
|
|
req := httptest.NewRequest(http.MethodPost, "/upstream/submit", strings.NewReader(`{
|
|
"schemaVersion":"v1","messageType":"SubmitCommand","messageId":"msg-1","channelId":"channel-1",
|
|
"submitId":"submit-1","tenantId":"tenant-1","applicationId":"app-1","phoneNumber":"13800138000","content":"hello",
|
|
"route":{"channelCode":"CMPP-A","cmppAccountCode":"sp","rateLimitPerSecond":100},
|
|
"cmpp":{"serviceId":"SMS","srcId":"10690000","registeredDelivery":1,"msgFmt":8},
|
|
"upstream":{"gatewayHost":"127.0.0.1","gatewayPort":17890,"account":"sp","passwordCipher":"secret","cmppVersion":"3.0"},
|
|
"retry":{"attempt":0,"maxAttempts":1}
|
|
}`))
|
|
handler.ServeHTTP(resp, req)
|
|
if resp.Code != http.StatusOK {
|
|
t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String())
|
|
}
|
|
if limiter.channelID != "channel-1" || limiter.rate != 100 {
|
|
t.Fatalf("unexpected limiter call: %+v", limiter)
|
|
}
|
|
}
|
|
|
|
func TestConnectChannelCallbacksConnectedState(t *testing.T) {
|
|
limiter := &controlRecordingLimiter{}
|
|
handler := handlerWithServer(Server{Limiter: limiter, Connect: func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) {
|
|
return ConnectionStateCallback{
|
|
ChannelID: "channel-1",
|
|
ConnectionID: "channel-1:primary",
|
|
Status: "connected",
|
|
DesiredConnections: 2,
|
|
CurrentConnections: 2,
|
|
LastConnectedAt: time.Now().UTC().Format(time.RFC3339Nano),
|
|
LastHeartbeatAt: time.Now().UTC().Format(time.RFC3339Nano),
|
|
}, nil
|
|
}})
|
|
|
|
resp := httptest.NewRecorder()
|
|
req := httptest.NewRequest(http.MethodPost, "/connections/connect", strings.NewReader(validConnectCommand()))
|
|
handler.ServeHTTP(resp, req)
|
|
|
|
if resp.Code != http.StatusOK {
|
|
t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String())
|
|
}
|
|
var callback ConnectionStateCallback
|
|
if err := json.Unmarshal(resp.Body.Bytes(), &callback); err != nil {
|
|
t.Fatalf("decode response: %v", err)
|
|
}
|
|
if callback.Status != "connected" || callback.CurrentConnections != 2 || callback.DesiredConnections != 2 {
|
|
t.Fatalf("unexpected callback state: %+v", callback)
|
|
}
|
|
if callback.ChannelID != "channel-1" || callback.ConnectionID != "channel-1:primary" {
|
|
t.Fatalf("unexpected callback identity: %+v", callback)
|
|
}
|
|
if callback.LastConnectedAt == "" || callback.LastHeartbeatAt == "" {
|
|
t.Fatalf("expected connection timestamps: %+v", callback)
|
|
}
|
|
if limiter.configuredChannelID != "channel-1" || limiter.configuredRate != 100 {
|
|
t.Fatalf("unexpected configured limiter: %+v", limiter)
|
|
}
|
|
}
|
|
|
|
func TestConnectChannelCallbacksFailedState(t *testing.T) {
|
|
handler := handlerWithConnect(func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) {
|
|
return ConnectionStateCallback{
|
|
ChannelID: "channel-1",
|
|
ConnectionID: "channel-1:primary",
|
|
Status: "failed",
|
|
DesiredConnections: 2,
|
|
CurrentConnections: 0,
|
|
LastDisconnectedAt: time.Now().UTC().Format(time.RFC3339Nano),
|
|
LastError: errTestConnect.Error(),
|
|
}, nil
|
|
})
|
|
|
|
resp := httptest.NewRecorder()
|
|
req := httptest.NewRequest(http.MethodPost, "/connections/connect", strings.NewReader(validConnectCommand()))
|
|
handler.ServeHTTP(resp, req)
|
|
|
|
if resp.Code != http.StatusOK {
|
|
t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String())
|
|
}
|
|
var callback ConnectionStateCallback
|
|
if err := json.Unmarshal(resp.Body.Bytes(), &callback); err != nil {
|
|
t.Fatalf("decode response: %v", err)
|
|
}
|
|
if callback.Status != "failed" || callback.CurrentConnections != 0 || callback.LastError == "" {
|
|
t.Fatalf("unexpected callback state: %+v", callback)
|
|
}
|
|
}
|
|
|
|
func TestConnectChannelRejectsInvalidCommand(t *testing.T) {
|
|
handler := handlerWithConnect(func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) {
|
|
return ConnectionStateCallback{}, nil
|
|
})
|
|
|
|
resp := httptest.NewRecorder()
|
|
req := httptest.NewRequest(http.MethodPost, "/connections/connect", strings.NewReader(`{"messageType":"SubmitCommand"}`))
|
|
handler.ServeHTTP(resp, req)
|
|
|
|
if resp.Code != http.StatusBadRequest {
|
|
t.Fatalf("unexpected response status: %d", resp.Code)
|
|
}
|
|
}
|
|
|
|
func TestConnectChannelRejectsCapacityOutsideSupportedMatrix(t *testing.T) {
|
|
handler := handlerWithConnect(func(context.Context, ConnectChannelCommand) (ConnectionStateCallback, error) {
|
|
return ConnectionStateCallback{}, nil
|
|
})
|
|
for _, body := range []string{
|
|
`{"schemaVersion":"v1","messageType":"ConnectChannel","channelId":"c","connectionId":"x","desiredConnections":9,"channel":{"gatewayHost":"127.0.0.1","gatewayPort":17890,"account":"a","passwordCipher":"p","windowSize":16}}`,
|
|
`{"schemaVersion":"v1","messageType":"ConnectChannel","channelId":"c","connectionId":"x","desiredConnections":1,"channel":{"gatewayHost":"127.0.0.1","gatewayPort":17890,"account":"a","passwordCipher":"p","windowSize":65}}`,
|
|
} {
|
|
response := httptest.NewRecorder()
|
|
handler.ServeHTTP(response, httptest.NewRequest(http.MethodPost, "/connections/connect", strings.NewReader(body)))
|
|
if response.Code != http.StatusBadRequest {
|
|
t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestDisconnectChannelStopsSupplierPool(t *testing.T) {
|
|
var received DisconnectChannelCommand
|
|
handler := handlerWithServer(Server{
|
|
Disconnect: func(_ context.Context, command DisconnectChannelCommand) (ConnectionStateCallback, error) {
|
|
received = command
|
|
return ConnectionStateCallback{
|
|
ChannelID: command.ChannelID,
|
|
ConnectionID: command.ConnectionID,
|
|
Status: "disconnected",
|
|
CurrentConnections: 0,
|
|
}, nil
|
|
},
|
|
})
|
|
resp := httptest.NewRecorder()
|
|
req := httptest.NewRequest(http.MethodPost, "/connections/disconnect", strings.NewReader(`{
|
|
"schemaVersion":"v1",
|
|
"messageType":"DisconnectChannel",
|
|
"traceId":"trace-disconnect",
|
|
"channelId":"channel-1",
|
|
"connectionId":"channel-1:primary",
|
|
"reason":"channel_disabled"
|
|
}`))
|
|
|
|
handler.ServeHTTP(resp, req)
|
|
|
|
if resp.Code != http.StatusOK {
|
|
t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String())
|
|
}
|
|
if received.ChannelID != "channel-1" || received.Reason != "channel_disabled" {
|
|
t.Fatalf("unexpected disconnect command: %+v", received)
|
|
}
|
|
}
|
|
|
|
func TestDisconnectDownstreamAccountEndpointIsAccountScoped(t *testing.T) {
|
|
handler := handlerWithServer(Server{})
|
|
resp := httptest.NewRecorder()
|
|
req := httptest.NewRequest(
|
|
http.MethodPost,
|
|
"/downstream/connections/disconnect",
|
|
strings.NewReader(`{"account":"100001","reason":"application_disabled"}`),
|
|
)
|
|
|
|
handler.ServeHTTP(resp, req)
|
|
|
|
if resp.Code != http.StatusOK {
|
|
t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String())
|
|
}
|
|
var payload struct {
|
|
Account string `json:"account"`
|
|
Disconnected int `json:"disconnected"`
|
|
}
|
|
if err := json.Unmarshal(resp.Body.Bytes(), &payload); err != nil {
|
|
t.Fatalf("decode response: %v", err)
|
|
}
|
|
if payload.Account != "100001" || payload.Disconnected != 0 {
|
|
t.Fatalf("unexpected downstream disconnect response: %+v", payload)
|
|
}
|
|
}
|
|
|
|
func TestRecoveryCandidatesEndpointReturnsView(t *testing.T) {
|
|
handler := handlerWithServer(Server{
|
|
RecoveryCandidates: func(context.Context) ([]inbound.DownstreamPresence, error) {
|
|
return []inbound.DownstreamPresence{{
|
|
Account: "100001",
|
|
GatewayInstanceID: "gateway-a",
|
|
State: "connected",
|
|
UpdatedAt: time.Now().UTC(),
|
|
}}, nil
|
|
},
|
|
})
|
|
|
|
resp := httptest.NewRecorder()
|
|
req := httptest.NewRequest(http.MethodGet, "/downstream/recovery-candidates", nil)
|
|
handler.ServeHTTP(resp, req)
|
|
|
|
if resp.Code != http.StatusOK {
|
|
t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String())
|
|
}
|
|
var payload []inbound.DownstreamPresence
|
|
if err := json.Unmarshal(resp.Body.Bytes(), &payload); err != nil {
|
|
t.Fatalf("decode response: %v", err)
|
|
}
|
|
if len(payload) != 1 || payload[0].Account != "100001" {
|
|
t.Fatalf("unexpected payload: %+v", payload)
|
|
}
|
|
}
|
|
|
|
func TestRecoveryStatusesEndpointReturnsView(t *testing.T) {
|
|
handler := handlerWithServer(Server{
|
|
RecoveryStatuses: func(context.Context) ([]inbound.DownstreamRecoveryStatus, error) {
|
|
return []inbound.DownstreamRecoveryStatus{{
|
|
Account: "100001",
|
|
GatewayInstanceID: "gateway-a",
|
|
State: "waiting_connection",
|
|
AttemptCount: 2,
|
|
}}, nil
|
|
},
|
|
})
|
|
|
|
resp := httptest.NewRecorder()
|
|
req := httptest.NewRequest(http.MethodGet, "/downstream/recovery-statuses", nil)
|
|
handler.ServeHTTP(resp, req)
|
|
|
|
if resp.Code != http.StatusOK {
|
|
t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String())
|
|
}
|
|
var payload []inbound.DownstreamRecoveryStatus
|
|
if err := json.Unmarshal(resp.Body.Bytes(), &payload); err != nil {
|
|
t.Fatalf("decode response: %v", err)
|
|
}
|
|
if len(payload) != 1 || payload[0].State != "waiting_connection" || payload[0].AttemptCount != 2 {
|
|
t.Fatalf("unexpected payload: %+v", payload)
|
|
}
|
|
}
|
|
|
|
func TestRecoveryOverviewEndpointReturnsCombinedView(t *testing.T) {
|
|
handler := handlerWithServer(Server{
|
|
RecoveryCandidates: func(context.Context) ([]inbound.DownstreamPresence, error) {
|
|
return []inbound.DownstreamPresence{{
|
|
Account: "100001",
|
|
GatewayInstanceID: "gateway-a",
|
|
State: "connected",
|
|
}}, nil
|
|
},
|
|
RecoveryStatuses: func(context.Context) ([]inbound.DownstreamRecoveryStatus, error) {
|
|
return []inbound.DownstreamRecoveryStatus{{
|
|
Account: "100001",
|
|
GatewayInstanceID: "gateway-a",
|
|
State: "success",
|
|
}}, nil
|
|
},
|
|
})
|
|
|
|
resp := httptest.NewRecorder()
|
|
req := httptest.NewRequest(http.MethodGet, "/downstream/recovery-overview", nil)
|
|
handler.ServeHTTP(resp, req)
|
|
|
|
if resp.Code != http.StatusOK {
|
|
t.Fatalf("unexpected response status: %d body=%s", resp.Code, resp.Body.String())
|
|
}
|
|
var payload DownstreamRecoveryOverview
|
|
if err := json.Unmarshal(resp.Body.Bytes(), &payload); err != nil {
|
|
t.Fatalf("decode response: %v", err)
|
|
}
|
|
if len(payload.Candidates) != 1 || payload.Candidates[0].Account != "100001" {
|
|
t.Fatalf("unexpected candidate payload: %+v", payload.Candidates)
|
|
}
|
|
if len(payload.Statuses) != 1 || payload.Statuses[0].State != "success" {
|
|
t.Fatalf("unexpected status payload: %+v", payload.Statuses)
|
|
}
|
|
}
|
|
|
|
type testConnectError struct{}
|
|
|
|
func (testConnectError) Error() string {
|
|
return "connect failed"
|
|
}
|
|
|
|
var errTestConnect testConnectError
|
|
|
|
func handlerWithConnect(connect ConnectFunc) http.Handler {
|
|
return handlerWithServer(Server{Connect: connect})
|
|
}
|
|
|
|
func handlerWithServer(server Server) http.Handler {
|
|
mux := http.NewServeMux()
|
|
Register(mux, server)
|
|
return mux
|
|
}
|
|
|
|
func validConnectCommand() string {
|
|
return `{
|
|
"schemaVersion": "v1",
|
|
"messageType": "ConnectChannel",
|
|
"traceId": "trace-1",
|
|
"channelId": "channel-1",
|
|
"connectionId": "channel-1:primary",
|
|
"reason": "channel_created",
|
|
"desiredConnections": 2,
|
|
"channel": {
|
|
"code": "CMPP-A",
|
|
"name": "主通道",
|
|
"gatewayHost": "127.0.0.1",
|
|
"gatewayPort": 17890,
|
|
"account": "sp",
|
|
"passwordCipher": "secret",
|
|
"srcId": "10690000",
|
|
"cmppVersion": "3.0",
|
|
"rateLimitPerSecond": 100
|
|
}
|
|
}`
|
|
}
|