iotlcm-public/iotlcm
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync: 同步生产版 v1.0.4 并脱敏

Browse Source
This commit is contained in:
MiFaZhan
2026-10-29 15:03:39 +08:00
parent fed9459244
commit 2d045591b7
32 changed files with 695 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cmd/http_test.go
+122
View File
@@ -1,12 +1,134 @@
package cmd
import (
"context"
"crypto/sha256"
"encoding/json"
"fmt"
"net/url"
"os"
"strings"
"testing"
"time"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/net/ghttp"
"github.com/gogf/gf/v2/os/gcfg"
v1 "git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/api/cardquery/v1"
_ "git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/logic/simKeyFill"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/logic/upstreamapi"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/logic/upstreamimpl"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
)
func TestAPNHTTPHarness(t *testing.T) {
if os.Getenv("APN_HTTP_CHAIN_HELPER") != "1" {
t.Skip("未启用本机HTTP模拟链")
}
gateway, err := url.Parse(os.Getenv("APN_HTTP_CHAIN_CHAN"))
if err != nil || gateway.Scheme != "http" || gateway.Hostname() != "127.0.0.1" || gateway.Port() == "" {
t.Fatal("只允许本机模拟chan")
}
adapter, err := gcfg.NewAdapterContent(fmt.Sprintf(`{"gateway":{"address":%q,"timeout":"2s"}}`, gateway.Host))
if err != nil {
t.Fatal(err)
}
g.Cfg().SetAdapter(adapter)
api := upstreamapi.NewUpstreamAPI()
api.RegisterChan("fake-channel", "1", &model.SimCardChannelCache{Instance: upstreamimpl.NewSimYD()})
service.RegisterUpstreamAPI(api)
server := ghttp.GetServer(t.Name())
server.SetAddr("127.0.0.1:0")
bindHTTPRoutes(server)
stop := make(chan struct{})
server.BindHandler("POST:/__test/stop", func(r *ghttp.Request) { close(stop) })
if err := server.Start(); err != nil {
t.Fatal(err)
}
defer server.Shutdown()
fmt.Printf("APN_CHAIN_READY=http://127.0.0.1:%d\n", server.GetListenedPort())
select {
case <-stop:
case <-time.After(3 * time.Minute):
t.Error("HTTP模拟链未正常结束")
}
}
type apnBindingTestFill struct {
service.ISimKeyFill
captured chan model.SimBaseReqJson
}
func (s *apnBindingTestFill) SimKeyFill(_ context.Context, input *model.SimBaseReqJson) (model.SimBaseResJson, error) {
s.captured <- *input
return model.SimBaseResJson{}, nil
}
func TestAPNProductionHTTPBindingPreservesRawPayload(t *testing.T) {
old := service.SimKeyFill()
capture := &apnBindingTestFill{ISimKeyFill: old, captured: make(chan model.SimBaseReqJson, 1)}
service.RegisterSimKeyFill(capture)
t.Cleanup(func() { service.RegisterSimKeyFill(old) })
server := ghttp.GetServer(t.Name())
server.SetAddr("127.0.0.1:0")
bindHTTPRoutes(server)
server.BindHandler("POST:/__test/legacy-binding", func(r *ghttp.Request) {
var req v1.SimBaseReq
if err := r.Parse(&req); err != nil {
t.Error(err)
r.Response.WriteStatus(400)
return
}
capture.captured <- req.SimBaseReqJson
})
if err := server.Start(); err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = server.Shutdown() })
for _, data := range []string{
`{"operationId":"frozen-operation","submitOnly":true,"operType":"1","apnName":"CMIOT"}`,
`{"operationId":"frozen-operation","queryJobId":"saved-job","operType":"1","apnName":"CMIOT"}`,
`{"operationId":"frozen-operation","recoverOnly":true,"operType":"1","apnName":"CMIOT"}`,
} {
for _, path := range []string{"/sim/change/sim-apn", "/sim/query/sim-apn-capability", "/sim/query/sim-session"} {
input := model.SimBaseReqJson{SimKeys: []*model.SimKey{{ChanKey: "fake", ICCID: "fake-card", MSISDN: "frozen-number"}}, ReqData: json.RawMessage(data)}
response, err := g.Client().ContentJson().Post(context.Background(), fmt.Sprintf("http://127.0.0.1:%d%s", server.GetListenedPort(), path), input)
if err != nil {
t.Fatal(err)
}
body := response.ReadAll()
_ = response.Close()
select {
case got := <-capture.captured:
var wantData, gotData map[string]any
_ = json.Unmarshal(input.ReqData, &wantData)
_ = json.Unmarshal(got.ReqData, &gotData)
wantJSON, _ := json.Marshal(wantData)
gotJSON, _ := json.Marshal(gotData)
if path == "/sim/query/sim-session" {
baselineResponse, err := g.Client().ContentJson().Post(context.Background(), fmt.Sprintf("http://127.0.0.1:%d/__test/legacy-binding", server.GetListenedPort()), input)
if err != nil {
t.Fatal(err)
}
_ = baselineResponse.Close()
baseline := <-capture.captured
if string(got.ReqData) != string(baseline.ReqData) {
t.Fatalf("非APN旧绑定行为改变：got=%q baseline=%q", got.ReqData, baseline.ReqData)
}
} else if sha256.Sum256(wantJSON) != sha256.Sum256(gotJSON) {
t.Fatalf("APN冻结请求摘要改变：path=%s want=%s got=%s", path, wantJSON, gotJSON)
}
if len(got.SimKeys) != 1 || *got.SimKeys[0] != *input.SimKeys[0] {
t.Fatalf("代理生产绑定丢字段：%+v", got)
}
default:
t.Fatalf("未经过生产代理controller：%s", body)
}
}
}
}
func TestBindHTTPRoutesRegistersChannelControlRoutes(t *testing.T) {
server := ghttp.GetServer("proxy-channel-refresh-route-test")
server.SetAddr("127.0.0.1:0")
internal/controller/cardquery/cardquery_v1_simcard.go
+11
View File
@@ -2,6 +2,10 @@ package cardquery
import (
"context"
"encoding/json"
"strings"
"github.com/gogf/gf/v2/frame/g"
v1 "git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/api/cardquery/v1"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
)
@@ -46,6 +50,13 @@ import (
// }
func (c *ControllerV1) SimBase(ctx context.Context, req *v1.SimBaseReq) (res *v1.SimBaseRes, err error) {
request := g.RequestFromCtx(ctx)
if request != nil && (request.URL.Path == "/sim/change/sim-apn" || request.URL.Path == "/sim/query/sim-apn-capability") && strings.HasPrefix(strings.ToLower(request.Header.Get("Content-Type")), "application/json") {
// GoFrame通用字段转换不能保留json.RawMessage，APN协议直接按原始JSON绑定。
if err := json.Unmarshal(request.GetBody(), &req.SimBaseReqJson); err != nil {
return nil, err
}
}
out, err := service.SimKeyFill().SimKeyFill(ctx, &req.SimBaseReqJson)
if err != nil {
return nil, err
internal/dao/internal/t_sim_card_info.go
+2 -2
View File
@@ -48,7 +48,7 @@ type TSimCardInfoColumns struct {
DataServiceStatus string // 数据服务状态：1启用、0禁用
PackageTotalFlowMb string // 套餐总流量MB
UsedFlowMb string // 已用流量MB
TrafficCutoffStatus string // 达量断网配置状态：0未配置，1已开启，2已关闭
TrafficCutoffStatus string // 达量断网配置状态：0未配置，1已开启，2已关闭
TrafficCutoffThreshold string // 达量断网阈值
SmsServiceStatus string // 短信服务状态：1启用、0禁用
PackageTotalSms string // 套餐短信总条数
@@ -107,7 +107,7 @@ var tSimCardInfoColumns = TSimCardInfoColumns{
DataServiceStatus: "data_service_status",
PackageTotalFlowMb: "package_total_flow_mb",
UsedFlowMb: "used_flow_mb",
TrafficCutoffStatus: "traffic_cutoff_status",
TrafficCutoffStatus: "traffic_cutoff_status",
TrafficCutoffThreshold: "traffic_cutoff_threshold",
SmsServiceStatus: "sms_service_status",
PackageTotalSms: "package_total_sms",
internal/logic/channelconfig/channel_config.go
+1 -1
View File
@@ -9,9 +9,9 @@ import (
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model/entity"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/dao"
"sync"
"time"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/dao"
"github.com/gogf/gf/v2/crypto/gmd5"
"github.com/gogf/gf/v2/encoding/gjson"
internal/logic/channelconfig/channel_config_test.go
+1 -1
View File
@@ -2,8 +2,8 @@ package channelconfig
import (
"context"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model/entity"
"testing"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model/entity"
)
func TestChannelVersionDetectsCredentialChanges(t *testing.T) {
internal/logic/simKeyFill/simKeyFill.go
+17 -1
View File
@@ -2,11 +2,12 @@ package simKeyFill
import (
"context"
"encoding/json"
"strings"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/dao"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model/entity"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
"strings"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g"
@@ -24,6 +25,21 @@ func init() {
}
func (s *sSimKeyFill) SimKeyFill(ctx context.Context, input *model.SimBaseReqJson) (output model.SimBaseResJson, err error) {
var operation struct {
OperationID string `json:"operationId"`
}
request := g.RequestFromCtx(ctx)
if request != nil && (request.URL.Path == "/sim/change/sim-apn" || request.URL.Path == "/sim/query/sim-apn-capability") && json.Unmarshal(input.ReqData, &operation) == nil && operation.OperationID != "" {
if len(input.SimKeys) == 0 || len(input.SimKeys) > 100 {
return nil, gerror.New("异步子任务卡数必须在1至100之间")
}
for _, key := range input.SimKeys {
if key == nil || key.ChanKey == "" || key.ICCID == "" {
return nil, gerror.New("异步子任务必须保留chanKey和ICCID快照")
}
}
return service.UpstreamAPI().SimBase(ctx, input)
}
simKeys, err := s.batchGetSimKey(ctx, input.SimKeys)
if err != nil {
return nil, err
internal/logic/simKeyFill/simKeyFill_test.go
+38
View File
@@ -0,0 +1,38 @@
package simKeyFill
import (
"context"
"encoding/json"
"net/http/httptest"
"testing"
"github.com/gogf/gf/v2/net/ghttp"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
)
type snapshotTestAPI struct {
service.IUpstreamAPI
input *model.SimBaseReqJson
}
func (a *snapshotTestAPI) SimBase(_ context.Context, in *model.SimBaseReqJson) (model.SimBaseResJson, error) {
a.input = in
return nil, nil
}
func TestAsyncOperationPreservesCardSnapshot(t *testing.T) {
api := &snapshotTestAPI{}
service.RegisterUpstreamAPI(api)
defer service.RegisterUpstreamAPI(nil)
in := &model.SimBaseReqJson{SimKeys: []*model.SimKey{{ChanKey: "test", ICCID: "test-card", MSISDN: "snapshot-number"}}, ReqData: json.RawMessage(`{"operationId":"stable-id","submitOnly":true}`)}
_, err := NewSimKeyFill().SimKeyFill((&ghttp.Request{Request: httptest.NewRequest("POST", "/sim/change/sim-apn", nil)}).Context(), in)
if err != nil || api.input != in || api.input.SimKeys[0].MSISDN != "snapshot-number" {
t.Fatalf("异步请求不应重新填充或改写卡快照：%v", err)
}
}
func TestAsyncOperationRejectsIncompleteSnapshot(t *testing.T) {
_, err := NewSimKeyFill().SimKeyFill((&ghttp.Request{Request: httptest.NewRequest("POST", "/sim/change/sim-apn", nil)}).Context(), &model.SimBaseReqJson{SimKeys: []*model.SimKey{{ICCID: "test-card"}}, ReqData: json.RawMessage(`{"operationId":"stable-id"}`)})
if err == nil {
t.Fatal("缺失通道快照应在调用前拒绝")
}
}
internal/logic/upstreamapi/apn_passthrough_test.go
+55
View File
@@ -0,0 +1,55 @@
package upstreamapi
import (
"context"
"encoding/json"
"net/http/httptest"
"strings"
"testing"
"github.com/gogf/gf/v2/net/ghttp"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
)
type apnPassthroughTestChannel struct{ reqData json.RawMessage }
func (c *apnPassthroughTestChannel) SimBase(_ context.Context, input *model.SimBaseReqJson) (model.SimBaseResJson, error) {
c.reqData = append(json.RawMessage(nil), input.ReqData...)
return model.SimBaseResJson{&model.SimBaseRes[json.RawMessage]{SimKey: input.SimKeys[0], Code: 0, Message: "已受理", ResData: json.RawMessage(`{"orderNum":"job-persisted","pending":true,"uncertain":false}`)}}, nil
}
func TestAPNProxyUnknownChannelPreservesUncertain(t *testing.T) {
api := NewUpstreamAPI()
out, err := api.SimBase((&ghttp.Request{Request: httptest.NewRequest("POST", "/sim/change/sim-apn", nil)}).Context(), &model.SimBaseReqJson{
SimKeys: []*model.SimKey{{ChanKey: "missing_chan", ICCID: "test"}},
ReqData: json.RawMessage(`{"operationId":"stable-id","submitOnly":true}`),
})
if err != nil || len(out) != 1 || !strings.Contains(string(out[0].ResData), `"uncertain":true`) {
t.Fatalf("未知通道不能误报业务失败: result=%+v err=%v", out, err)
}
}
func TestNonAPNProxySameNamedFieldsKeepLegacyFailure(t *testing.T) {
api := NewUpstreamAPI()
ctx := (&ghttp.Request{Request: httptest.NewRequest("POST", "/sim/change/sim-status", nil)}).Context()
out, err := api.SimBase(ctx, &model.SimBaseReqJson{SimKeys: []*model.SimKey{{ChanKey: "missing_chan", ICCID: "test"}}, ReqData: json.RawMessage(`{"operationId":"other-domain","submitOnly":true,"recoverOnly":true,"queryJobId":"unrelated"}`)})
if err != nil || len(out) != 1 || string(out[0].ResData) != `{"status":"-2"}` {
t.Fatalf("非APN同名字段改变既有失败语义：%+v %v", out, err)
}
}
func TestAPNProxyPreservesOperationAndAsyncFields(t *testing.T) {
api := NewUpstreamAPI()
instance := &apnPassthroughTestChannel{}
api.RegisterChan("test_chan", "test", &model.SimCardChannelCache{Instance: instance})
req := json.RawMessage(`{"operationId":"stable-id","submitOnly":true,"recoverOnly":false,"operType":"1"}`)
out, err := api.SimBase((&ghttp.Request{Request: httptest.NewRequest("POST", "/sim/change/sim-apn", nil)}).Context(), &model.SimBaseReqJson{SimKeys: []*model.SimKey{{ChanKey: "test_chan", ICCID: "test"}}, ReqData: req})
if err != nil || len(out) != 1 {
t.Fatalf("proxy请求错误：%v %+v", err, out)
}
if !strings.Contains(string(instance.reqData), `"operationId":"stable-id"`) {
t.Fatalf("operationId丢失：%s", instance.reqData)
}
if !strings.Contains(string(out[0].ResData), `"pending":true`) || !strings.Contains(string(out[0].ResData), `"orderNum":"job-persisted"`) {
t.Fatalf("受理标识丢失：%s", out[0].ResData)
}
}
internal/logic/upstreamapi/errcode_test.go
+2 -2
View File
@@ -18,7 +18,7 @@ func (f *errCodeTestChannel) SimBase(context.Context, *model.SimBaseReqJson) (mo
}
func TestSimBasePreservesCarrierErrCode(t *testing.T) {
const payload = `{"code":0,"message":"success","data":[{"code":9999,"message":"http status 797","errCode":"S00004","simKey":{"chanKey":"channel_003","iccid":"8986112520409968013"},"resData":{}}]}`
const payload = `{"code":0,"message":"success","data":[{"code":9999,"message":"http status 797","errCode":"S00004","simKey":{"chanKey":"channel_003","iccid":"8986000000000000002"},"resData":{}}]}`
var upstream model.SimResponseData[model.SimBaseResJson]
if err := json.Unmarshal([]byte(payload), &upstream); err != nil {
@@ -38,7 +38,7 @@ func TestSimBasePreservesCarrierErrCode(t *testing.T) {
output, err := api.SimBase(context.Background(), &model.SimBaseReqJson{
SimKeys: []*model.SimKey{{
ChanKey: "channel_003",
ICCID: "8986112520409968013",
ICCID: "8986000000000000002",
}},
})
if err != nil {
internal/logic/upstreamapi/upstreamapi.go
+17 -6
View File
@@ -3,14 +3,14 @@ package upstreamapi
import (
"context"
"encoding/json"
"strings"
"sync"
"time"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/consts"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/dao"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model/entity"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
"strings"
"sync"
"time"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g"
@@ -317,6 +317,17 @@ func (u *sUpstreamAPI) packChanKey(ctx context.Context, simKeys []*model.SimKey)
func (u *sUpstreamAPI) SimBase(ctx context.Context, input *model.SimBaseReqJson) (model.SimBaseResJson, error) {
output := make([]*model.SimBaseRes[json.RawMessage], 0)
inputMap := make(map[string]*model.SimBaseReqJson)
// 提交/恢复/查询阶段的代理异常不是运营商办理失败。
failureResData := json.RawMessage(`{"status":"-2"}`)
var apnRequest struct {
SubmitOnly bool `json:"submitOnly"`
RecoverOnly bool `json:"recoverOnly"`
QueryJobID string `json:"queryJobId"`
}
request := g.RequestFromCtx(ctx)
if request != nil && request.URL.Path == "/sim/change/sim-apn" && json.Unmarshal(input.ReqData, &apnRequest) == nil && (apnRequest.SubmitOnly || apnRequest.RecoverOnly || apnRequest.QueryJobID != "") {
failureResData = json.RawMessage(`{"uncertain":true}`)
}
chanKeyMap, err := u.packChanKey(ctx, input.SimKeys)
if err != nil {
return nil, err
@@ -334,7 +345,7 @@ func (u *sUpstreamAPI) SimBase(ctx context.Context, input *model.SimBaseReqJson)
SimKey: simKey,
Code: consts.ApiStatusError,
Message: "通道未配置",
ResData: json.RawMessage(`{"status": "-2"}`),
ResData: failureResData,
})
}
continue
@@ -346,7 +357,7 @@ func (u *sUpstreamAPI) SimBase(ctx context.Context, input *model.SimBaseReqJson)
SimKey: simKey,
Code: consts.ApiStatusError,
Message: "通道未实例化",
ResData: json.RawMessage(`{"status": "-2"}`),
ResData: failureResData,
})
}
continue
@@ -358,7 +369,7 @@ func (u *sUpstreamAPI) SimBase(ctx context.Context, input *model.SimBaseReqJson)
SimKey: simKey,
Code: consts.ApiStatusError,
Message: err.Error(),
ResData: json.RawMessage(`{"status": "-2"}`),
ResData: failureResData,
})
}
} else {
internal/logic/upstreamimpl/apn_http_test.go
+117
View File
@@ -0,0 +1,117 @@
package upstreamimpl
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/net/ghttp"
"github.com/gogf/gf/v2/os/gcfg"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/logic/upstreamapi"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
)
func TestAPNProxyHTTPFailureIsUncertainOnlyOnAPNRoute(t *testing.T) {
oldAdapter := g.Cfg().GetAdapter()
defer g.Cfg().SetAdapter(oldAdapter)
for _, tc := range []struct {
name, path, body string
disconnect, uncertain bool
}{
{"HTTP200顶层9999", "/sim/change/sim-apn", `{"code":9999,"message":"模拟网关未确认"}`, false, true},
{"通道连接断开", "/sim/change/sim-apn", "", true, true},
{"非APN同名字段", "/sim/change/sim-status", `{"code":9999,"message":"模拟非APN网关错误"}`, false, false},
} {
t.Run(tc.name, func(t *testing.T) {
chanServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if tc.disconnect {
conn, _, err := w.(http.Hijacker).Hijack()
if err != nil {
t.Error(err)
return
}
_ = conn.Close()
return
}
_, _ = w.Write([]byte(tc.body))
}))
defer chanServer.Close()
adapter, err := gcfg.NewAdapterContent(`{"gateway":{"address":"` + strings.TrimPrefix(chanServer.URL, "http://") + `","timeout":"1s"}}`)
if err != nil {
t.Fatal(err)
}
g.Cfg().SetAdapter(adapter)
api := upstreamapi.NewUpstreamAPI()
api.RegisterChan("fake-channel", "test", &model.SimCardChannelCache{Instance: &sSimYD{}})
proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var input model.SimBaseReqJson
if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
t.Error(err)
return
}
ctx := (&ghttp.Request{Request: r}).Context()
items, err := api.SimBase(ctx, &input)
if err != nil {
t.Error(err)
return
}
_ = json.NewEncoder(w).Encode(map[string]any{"code": 0, "data": items})
}))
defer proxy.Close()
body := []byte(`{"simKeys":[{"chanKey":"fake-channel","iccid":"fake-card"}],"reqData":{"operationId":"fake-operation","submitOnly":true}}`)
response, err := http.Post(proxy.URL+tc.path, "application/json", bytes.NewReader(body))
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
var output struct {
Code int `json:"code"`
Data model.SimBaseResJson `json:"data"`
}
if err := json.NewDecoder(response.Body).Decode(&output); err != nil {
t.Fatal(err)
}
if response.StatusCode != 200 || output.Code != 0 || len(output.Data) != 1 || output.Data[0].Code != 9999 || output.Data[0].Message == "" {
t.Fatalf("代理异常响应无效：HTTP=%d result=%+v", response.StatusCode, output)
}
isUncertain := strings.Contains(string(output.Data[0].ResData), `"uncertain":true`)
if isUncertain != tc.uncertain {
t.Fatalf("APN路由隔离失效：%s", output.Data[0].ResData)
}
})
}
}
func TestAPNProxyHTTPKeepsDurableJobAndFrozenRequest(t *testing.T) {
oldAdapter := g.Cfg().GetAdapter()
defer g.Cfg().SetAdapter(oldAdapter)
var captured model.SimBaseReqJson
chanServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if err := json.NewDecoder(r.Body).Decode(&captured); err != nil {
t.Error(err)
return
}
_, _ = w.Write([]byte(`{"code":0,"data":[{"code":0,"simKey":{"chanKey":"fake-channel","iccid":"fake-card","msisdn":"frozen-number"},"resData":{"orderNum":"saved-job","pending":true}}]}`))
}))
defer chanServer.Close()
adapter, err := gcfg.NewAdapterContent(`{"gateway":{"address":"` + strings.TrimPrefix(chanServer.URL, "http://") + `","timeout":"1s"}}`)
if err != nil {
t.Fatal(err)
}
g.Cfg().SetAdapter(adapter)
api := upstreamapi.NewUpstreamAPI()
api.RegisterChan("fake-channel", "test", &model.SimCardChannelCache{Instance: &sSimYD{}})
input := &model.SimBaseReqJson{SimKeys: []*model.SimKey{{ChanKey: "fake-channel", ICCID: "fake-card", MSISDN: "frozen-number"}}, ReqData: json.RawMessage(`{"operationId":"stable-operation","recoverOnly":true}`)}
ctx := (&ghttp.Request{Request: httptest.NewRequest("POST", "/sim/change/sim-apn", nil)}).Context()
items, err := api.SimBase(ctx, input)
if err != nil || len(items) != 1 || !strings.Contains(string(items[0].ResData), `"orderNum":"saved-job"`) || !strings.Contains(string(items[0].ResData), `"pending":true`) {
t.Fatalf("HTTP转发丢失jobId：%+v %v", items, err)
}
if string(captured.ReqData) != string(input.ReqData) || captured.SimKeys[0].MSISDN != "frozen-number" {
t.Fatalf("恢复请求被改写：%+v", captured)
}
}
internal/logic/upstreamimpl/gateway_response.go
+66 -1
View File
@@ -1,15 +1,80 @@
package upstreamimpl
import (
"context"
"encoding/json"
"net/http"
"time"
"fmt"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
"github.com/gogf/gf/v2/frame/g"
"strings"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/net/gclient"
)
const (
// gatewayBatchTimeoutConfigKey 限制单次批量转发的最长等待，防止上游异常时无限占用连接。
gatewayBatchTimeoutConfigKey = "gateway.batchTimeout"
// defaultGatewayBatchTimeout 覆盖 100 张卡按单卡 10 秒预算的最坏耗时，并留有余量。
defaultGatewayBatchTimeout = 20 * time.Minute
// defaultDataServiceTimeout 是数据开停转发的独立预算。
defaultDataServiceTimeout = 210 * time.Second
)
// configureGatewayTimeout 让响应头等待与整次请求使用同一预算，避免默认30秒提前中断。
// upstream-chan 对批量操作按卡串行调用运营商接口，耗时随卡数线性增长；
// 因此批量请求不能沿用 gclient 默认的 30 秒响应头超时，否则请求仍在办理就会被判为失败。
func configureGatewayTimeout(ctx context.Context, client *gclient.Client, path string, count int, perCard time.Duration) {
if path == "/sim/change/sim-apn" || path == "/sim/query/sim-apn-capability" {
client.SetContentType("application/json")
}
applyGatewayTimeout(client, path, count, perCard,
configDuration(ctx, "gateway.dataServiceTimeout", defaultDataServiceTimeout),
configDuration(ctx, gatewayBatchTimeoutConfigKey, defaultGatewayBatchTimeout))
}
// applyGatewayTimeout 计算并应用单次转发预算：整次请求与响应头等待必须一致，
// 否则响应头超时会在业务仍在办理时提前中断请求。
func applyGatewayTimeout(client *gclient.Client, path string, count int, perCard, dataServiceTimeout, batchTimeout time.Duration) {
if count < 1 {
count = 1
}
if perCard <= 0 {
perCard = 10 * time.Second
}
timeout := perCard * time.Duration(count)
if path == "/sim/change/sim-apn" {
timeout = dataServiceTimeout
if timeout <= 0 {
timeout = defaultDataServiceTimeout
}
} else if batchTimeout > 0 && timeout > batchTimeout {
timeout = batchTimeout
}
client.SetTimeout(timeout)
if transport, ok := client.Transport.(*http.Transport); ok {
transport.ResponseHeaderTimeout = timeout
}
}
// configDuration 读取时长配置，兼容写成 "20m" 字符串或秒数整数两种形式。
func configDuration(ctx context.Context, key string, fallback time.Duration) time.Duration {
value := g.Cfg().MustGet(ctx, key)
if value == nil || value.IsNil() {
return fallback
}
if duration := value.Duration(); duration >= time.Second {
return duration
}
if seconds := value.Int(); seconds > 0 {
return time.Duration(seconds) * time.Second
}
return fallback
}
func readGatewayResponse(response *gclient.Response) (model.SimBaseResJson, error) {
defer response.Close()
body := response.ReadAll()
internal/logic/upstreamimpl/gateway_timeout_test.go
+97
View File
@@ -0,0 +1,97 @@
package upstreamimpl
import (
"context"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/gogf/gf/v2/frame/g"
)
func TestGatewayTimeoutAllowsDelayedHeaders(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
time.Sleep(40 * time.Millisecond)
_, _ = w.Write([]byte(`{"code":0,"data":[]}`))
}))
defer server.Close()
client := g.Client()
other := g.Client()
original := other.Transport.(*http.Transport)
original.ResponseHeaderTimeout = time.Millisecond
client.Transport.(*http.Transport).ResponseHeaderTimeout = time.Millisecond
configureGatewayTimeout(context.Background(), client, "/sim/change/sim-apn", 1, time.Second)
response, err := client.Post(context.Background(), server.URL, nil)
if err != nil {
t.Fatal(err)
}
defer response.Close()
if original.ResponseHeaderTimeout != time.Millisecond {
t.Fatal("修改了共享Transport")
}
}
func TestNonAPNGatewayTimeoutPreservesHeaderBudget(t *testing.T) {
client := g.Client()
transport := client.Transport.(*http.Transport)
transport.ResponseHeaderTimeout = 17 * time.Second
configureGatewayTimeout(context.Background(), client, "/sim/query/test", 2, 3*time.Second)
if client.Client.Timeout != 6*time.Second || transport.ResponseHeaderTimeout != 6*time.Second {
t.Fatalf("非APN响应头预算必须与请求预算一致：request=%s header=%s", client.Client.Timeout, transport.ResponseHeaderTimeout)
}
}
// TestNonAPNBatchExpandsHeaderBudget 覆盖批量停复机等非 APN 路径：
// 50 张卡按单卡预算放大后，响应头等待不能继续停留在 gclient 的 30 秒默认值。
func TestNonAPNBatchExpandsHeaderBudget(t *testing.T) {
client := g.Client()
transport := client.Transport.(*http.Transport)
transport.ResponseHeaderTimeout = 30 * time.Second
applyGatewayTimeout(client, "/sim/change/sim-status", 50, 10*time.Second, defaultDataServiceTimeout, defaultGatewayBatchTimeout)
if client.Client.Timeout != 500*time.Second {
t.Fatalf("请求预算=%s，期望500s", client.Client.Timeout)
}
if transport.ResponseHeaderTimeout != client.Client.Timeout {
t.Fatalf("响应头预算=%s与请求预算不一致", transport.ResponseHeaderTimeout)
}
}
// TestNonAPNBatchRespectsBatchTimeoutCap 保证异常上游不会无限占用连接：
// 超出批量上限时按上限截断，且响应头预算同步截断。
func TestNonAPNBatchRespectsBatchTimeoutCap(t *testing.T) {
client := g.Client()
transport := client.Transport.(*http.Transport)
applyGatewayTimeout(client, "/sim/query/usage", 1000, 10*time.Second, defaultDataServiceTimeout, defaultGatewayBatchTimeout)
if client.Client.Timeout != defaultGatewayBatchTimeout {
t.Fatalf("请求预算=%s，期望批量上限%s", client.Client.Timeout, defaultGatewayBatchTimeout)
}
if transport.ResponseHeaderTimeout != defaultGatewayBatchTimeout {
t.Fatalf("响应头预算=%s，期望批量上限%s", transport.ResponseHeaderTimeout, defaultGatewayBatchTimeout)
}
}
// TestNonAPNSingleCardKeepsPerCardBudget 保证单卡请求不被放大成批量预算。
func TestNonAPNSingleCardKeepsPerCardBudget(t *testing.T) {
for _, count := range []int{0, 1} {
client := g.Client()
transport := client.Transport.(*http.Transport)
applyGatewayTimeout(client, "/sim/query/test", count, 10*time.Second, defaultDataServiceTimeout, defaultGatewayBatchTimeout)
if client.Client.Timeout != 10*time.Second || transport.ResponseHeaderTimeout != 10*time.Second {
t.Fatalf("count=%d request=%s header=%s", count, client.Client.Timeout, transport.ResponseHeaderTimeout)
}
}
}
func TestDataServiceTimeoutOverridesCardBudget(t *testing.T) {
for _, count := range []int{0, 1, 100} {
client := g.Client()
configureGatewayTimeout(context.Background(), client, "/sim/change/sim-apn", count, time.Millisecond)
if client.Client.Timeout != 210*time.Second {
t.Fatalf("count=%d timeout=%s", count, client.Client.Timeout)
}
if client.Transport.(*http.Transport).ResponseHeaderTimeout != client.Client.Timeout {
t.Fatal("响应头预算不一致")
}
}
}
internal/logic/upstreamimpl/mingqian_test.go
+3 -3
View File
@@ -10,13 +10,13 @@ import (
"strings"
"testing"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/net/ghttp"
"github.com/gogf/gf/v2/os/gcfg"
_ "git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/logic"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model/entity"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/net/ghttp"
"github.com/gogf/gf/v2/os/gcfg"
)
func TestMingqianProxyTemplate(t *testing.T) {
internal/logic/upstreamimpl/sim_dx.go
+3 -3
View File
@@ -3,11 +3,11 @@ package upstreamimpl
import (
"context"
"fmt"
"regexp"
"time"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/logic/upstreamimpl/dx"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
"regexp"
"time"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g"
@@ -85,7 +85,7 @@ func (s *sSimDX) SimBase(ctx context.Context, input *model.SimBaseReqJson) (mode
}
client := g.Client()
client.SetDiscovery(gsvc.GetRegistry())
client.SetTimeout(s.getUpstreamTimeout(ctx) * time.Duration(len(input.SimKeys)))
configureGatewayTimeout(ctx, client, path, len(input.SimKeys), s.getUpstreamTimeout(ctx))
resp, err := client.Post(ctx, url, input)
if err != nil {
g.Log().Errorf(ctx, "SimBase post error %+v,url %s, params %+v, ", err, url, input)
internal/logic/upstreamimpl/sim_yd.go
+4 -4
View File
@@ -3,10 +3,10 @@ package upstreamimpl
import (
"context"
"fmt"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
"sync/atomic"
"time"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g"
@@ -106,11 +106,11 @@ func (s *sSimYD) SimBase(ctx context.Context, input *model.SimBaseReqJson) (mode
g.Log().Debugf(ctx, "SimYD.SimBase, url %s", url)
client := g.Client()
client.SetDiscovery(gsvc.GetRegistry())
client.SetTimeout(s.getUpstreamTimeout(ctx) * time.Duration(len(input.SimKeys)))
configureGatewayTimeout(ctx, client, path, len(input.SimKeys), s.getUpstreamTimeout(ctx))
resp, err := client.Post(ctx, url, input)
if err != nil {
g.Log().Errorf(ctx, "SimYD.SimBase, post error %+v,url %s, params %+v, ", err, url, input)
return nil, gerror.New("请求失败")
return nil, gerror.Wrap(err, "请求通道失败")
}
return readGatewayResponse(resp)
}
internal/pub/pkg/version/version.go
+1 -1
View File
@@ -1,7 +1,7 @@
// package version
// @Link https://github.com/xiujiecn/xiu-admin
// @Copyright Copyright (c) 2025 LiXiujie
// @Author Lxj <li@xiujie.cn>
// @Author Lxj <xxx@example.com>
// @License https://github.com/xiujiecn/xiu-admin/blob/master/LICENSE
package version
internal/pub/xjpubconsts/http.go
+1 -1
View File
@@ -1,7 +1,7 @@
// Package xjpubconsts
// @Link https://github.com/xiujiecn/xiu-admin
// @Copyright Copyright (c) 2025 LiXiujie
// @Author Lxj <li@xiujie.cn>
// @Author Lxj <xxx@example.com>
// @License https://github.com/xiujiecn/xiu-admin/blob/master/LICENSE
package xjpubconsts
internal/pub/xjpublogic/middleware/middleware.go
+1 -1
View File
@@ -1,7 +1,7 @@
// package middleware
// @Link https://github.com/xiujiecn/xiu-admin
// @Copyright Copyright (c) 2025 LiXiujie
// @Author Lxj <li@xiujie.cn>
// @Author Lxj <xxx@example.com>
// @License https://github.com/xiujiecn/xiu-admin/blob/master/LICENSE
package middleware
internal/pub/xjpubmodel/response/common.go
+1 -1
View File
@@ -1,7 +1,7 @@
// package response
// @Link https://github.com/xiujiecn/xiu-admin
// @Copyright Copyright (c) 2025 LiXiujie
// @Author Lxj <li@xiujie.cn>
// @Author Lxj <xxx@example.com>
// @License https://github.com/xiujiecn/xiu-admin/blob/master/LICENSE
package response
internal/pub/xjpubmodel/response/response.go
+1 -1
View File
@@ -1,7 +1,7 @@
// package response
// @Link https://github.com/xiujiecn/xiu-admin
// @Copyright Copyright (c) 2025 LiXiujie
// @Author Lxj <li@xiujie.cn>
// @Author Lxj <xxx@example.com>
// @License https://github.com/xiujiecn/xiu-admin/blob/master/LICENSE
package response