package upstreamimpl import ( "context" "errors" "fmt" "net/http" "net/http/httptest" "net/url" "strings" "testing" "time" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model/entity" ) func TestYdLegacyPollingHardDeadlineCancelsInflightRequest(t *testing.T) { defer shortenYdBatchPoll(50 * time.Millisecond)() server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { select { case <-r.Context().Done(): case <-time.After(time.Second): _, _ = w.Write([]byte(`{"status":"0","result":[{"jobStatus":"1"}]}`)) } })) defer server.Close() started := time.Now() result := queryBatchResultYD(context.Background(), newYdApnChannelWithServer(server), "deadline-job") if result.finished || time.Since(started) > 500*time.Millisecond || !strings.Contains(result.reason, "deadline") { t.Fatalf("轮询没有硬截止:elapsed=%s result=%+v", time.Since(started), result) } } func TestYdAsyncCheckpointFailureStopsFurtherCarrierWrites(t *testing.T) { for _, tc := range []struct { name string keys []*model.SimKey firstPath string firstLen int }{ {name: "single_then_single", keys: ydSimKeysWithMsisdn(2), firstPath: ydCommunicationServicePath, firstLen: 1}, {name: "batch_then_single", keys: func() []*model.SimKey { keys := ydSimKeysWithMsisdn(11) keys[10].MSISDN = "" return keys }(), firstPath: ydCommunicationFunctionBatchPath, firstLen: 10}, } { t.Run(tc.name, func(t *testing.T) { capture := &ydApnCapture{} channel, closeServer := newYdApnTestChannel(capture) defer closeServer() sentinel := errors.New("checkpoint unavailable") checkpoints := 0 ctx := context.WithValue(context.Background(), model.ApnOperationCheckpointKey{}, func(output model.SimCardOperateChangeApnOutput) error { checkpoints++ if len(output) != tc.firstLen { t.Errorf("checkpoint contains %d results, want %d", len(output), tc.firstLen) } return sentinel }) items, err := changeSimApnFunctionYD(ctx, channel, &model.SimCardOperateChangeApnInput{ SimKeys: tc.keys, ReqData: model.SimCardOperateChangeApnInputItem{OperType: "0", SubmitOnly: true, OperationID: "test-operation"}, }) if !errors.Is(err, sentinel) || checkpoints != 1 || len(items) != tc.firstLen { t.Fatalf("checkpoint failure not propagated: err=%v checkpoints=%d items=%d", err, checkpoints, len(items)) } if len(capture.paths) != 1 || !strings.HasSuffix(capture.paths[0], tc.firstPath) { t.Fatalf("wrote another carrier request after checkpoint failure: %v", capture.paths) } }) } } func TestYdAsyncSubmitReturnsJobWithoutQuery(t *testing.T) { capture := &ydApnCapture{} channel, closeServer := newYdApnTestChannel(capture) defer closeServer() checkpoints := 0 ctx := context.WithValue(context.Background(), model.ApnOperationCheckpointKey{}, func(output model.SimCardOperateChangeApnOutput) error { checkpoints++ if output[0].ResData.OrderNum != "JOB1" { t.Fatal("checkpoint缺少jobId") } return nil }) items, err := changeSimApnFunctionYD(ctx, channel, &model.SimCardOperateChangeApnInput{ SimKeys: ydSimKeysWithMsisdn(100), ReqData: model.SimCardOperateChangeApnInputItem{OperType: "1", SubmitOnly: true}, }) if checkpoints != 1 { t.Fatalf("拿jobId后未立即持久化:%d", checkpoints) } if err != nil || len(items) != 100 || len(capture.paths) != 1 { t.Fatalf("提交应仅调用一次,items=%d paths=%v err=%v", len(items), capture.paths, err) } for _, item := range items { if !item.ResData.Pending || item.ResData.OrderNum != "JOB1" || item.ResData.Uncertain { t.Fatalf("受理状态错误:%+v", item) } } } func TestYdAsyncQueryOnlyOnceAndPreservesReason(t *testing.T) { for _, body := range []string{ `{"status":"13012","message":"查询限流"}`, `{"status":"0","result":[{"jobStatus":"1"}]}`, `invalid-json`, `{"status":"0","result":[]}`, } { t.Run(body, func(t *testing.T) { calls := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { calls++ if !strings.Contains(r.URL.Path, "sim-batch-result") || r.URL.Query().Get("jobId") != "persisted-job" { t.Errorf("不应重新提交:%s", r.URL.String()) } _, _ = w.Write([]byte(body)) })) defer server.Close() items, err := changeSimApnFunctionYD(context.Background(), newYdApnChannelWithServer(server), &model.SimCardOperateChangeApnInput{ SimKeys: ydSimKeysWithMsisdn(10), ReqData: model.SimCardOperateChangeApnInputItem{QueryJobID: "persisted-job"}, }) if err != nil || calls != 1 || len(items) != 10 { t.Fatalf("查询次数错误:calls=%d err=%v", calls, err) } for _, item := range items { if !item.ResData.Pending || item.Message == "" || item.ResData.OrderNum != "persisted-job" { t.Fatalf("应保留待查状态和具体原因:%+v", item) } } }) } } func TestYdAsyncSubmitUnknownIsNotConfirmedFailure(t *testing.T) { for _, body := range []string{`{"status":"0","result":[]}`, `not-json`, `{}`, `{"status":"bad"}`} { t.Run(body, func(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte(body)) })) defer server.Close() items, err := changeSimApnFunctionYD(context.Background(), newYdApnChannelWithServer(server), &model.SimCardOperateChangeApnInput{ SimKeys: ydSimKeysWithMsisdn(10), ReqData: model.SimCardOperateChangeApnInputItem{OperType: "1", SubmitOnly: true}, }) if err != nil || len(items) != 10 { t.Fatalf("未知提交结果应逐卡返回:%v", err) } for _, item := range items { if !item.ResData.Uncertain || item.Message == "" { t.Fatalf("提交结果未知不得确认办理失败:%+v", item) } } }) } } func TestYdAsyncQueryTerminalResults(t *testing.T) { keys := ydSimKeysWithMsisdn(10) for _, tc := range []struct{ status, want string }{{"2", "0"}, {"3", "4001"}, {"4", "failed"}} { t.Run(tc.status, func(t *testing.T) { calls := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { calls++ if !strings.Contains(r.URL.Path, "sim-batch-result") { t.Error("查询不得提交") } _, _ = fmt.Fprintf(w, `{"status":"0","message":"任务完成","result":[{"jobStatus":%q,"resultList":[{"resultId":%q,"status":%q,"message":"逐卡结果"}]}]}`, tc.status, keys[0].MSISDN, tc.want) })) defer server.Close() items, err := newYdApnChannelWithServer(server).QuerySimApnFunctionResult(context.Background(), &model.SimCardOperateChangeApnInput{SimKeys: keys, ReqData: model.SimCardOperateChangeApnInputItem{QueryJobID: "job"}}) if err != nil || calls != 1 || len(items) != 10 { t.Fatalf("终态查询错误:%v", err) } if items[0].ResData.Pending || items[0].ResData.Uncertain { t.Fatalf("已确认结果仍标未知:%+v", items[0]) } if tc.status == "2" && items[0].Code != 0 { t.Fatal("成功结果未映射") } if tc.status == "3" && items[0].Code != 4001 { t.Fatal("逐卡失败原因未映射") } if tc.status == "4" && items[0].Code == 0 { t.Fatal("任务失败误报成功") } }) } } func TestYdAsyncMalformedCardStatusIsUnconfirmed(t *testing.T) { keys := ydSimKeysWithMsisdn(1) cards := toYdBatchCardMap([]ydBatchResultItem{{ResultId: keys[0].MSISDN, Status: "invalid", Message: "bad status"}}) items := mapYdQueryResult(keys, "job", ydBatchQueryResult{finished: true, cards: cards}) if !items[0].ResData.Uncertain || items[0].Code == 0 || !strings.Contains(items[0].Message, "invalid") { t.Fatalf("无效状态不能按成功处理:%+v", items[0]) } } func TestYdAsyncMissingCardRemainsUnconfirmed(t *testing.T) { keys := ydSimKeysWithMsisdn(2) items := mapYdQueryResult(keys, "job", ydBatchQueryResult{finished: true, cards: map[string]ydBatchCardResult{keys[0].MSISDN: {code: 0}}}) if items[0].Code != 0 || !items[1].ResData.Uncertain || items[1].ResData.OrderNum != "job" { t.Fatalf("缺失结果应待核对:%+v", items) } } func TestYdCommunicationBusinessType(t *testing.T) { for _, tc := range []struct { in string want string valid bool }{ {"0", "0", true}, // 平台 开 -> 7.2.1 businessType 开(同向) {"1", "1", true}, // 平台 停 -> 7.2.1 businessType 停 {"", "", false}, {"2", "", false}, } { got, ok := convertYdCommunicationBusinessType(context.Background(), tc.in) if got != tc.want || ok != tc.valid { t.Errorf("businessType(%q) = (%q,%v), want (%q,%v)", tc.in, got, ok, tc.want, tc.valid) } } } func TestYdCommunicationOperType(t *testing.T) { for _, tc := range []struct { in string want string valid bool }{ {"0", "1", true}, // 平台 开 -> 7.2.2 恢复(翻转) {"1", "0", true}, // 平台 停 -> 7.2.2 暂停 {"", "", false}, {"9", "", false}, } { got, ok := convertYdCommunicationOperType(context.Background(), tc.in) if got != tc.want || ok != tc.valid { t.Errorf("operType(%q) = (%q,%v), want (%q,%v)", tc.in, got, ok, tc.want, tc.valid) } } } type ydApnCapture struct { paths []string query []url.Values msisdns string } // shortenYdBatchPoll 把批量结果轮询周期压缩到毫秒级,返回恢复函数。 func shortenYdBatchPoll(timeout time.Duration) func() { oldInterval, oldTimeout := ydBatchResultPollInterval, ydBatchResultPollTimeout ydBatchResultPollInterval, ydBatchResultPollTimeout = time.Millisecond, timeout return func() { ydBatchResultPollInterval, ydBatchResultPollTimeout = oldInterval, oldTimeout } } func newYdApnChannelWithServer(server *httptest.Server) *sSimYD { return &sSimYD{ channelConfig: &model.SimChannelConfigItemCache{Channel: &entity.TSimChannels{ChanKey: "test", AppKey: "test"}}, tokenProvider: diagnosisTestToken{}, apiConfig: &oneLinkAPIConfig{BaseURL: server.URL + "/v5"}, } } func newYdApnTestChannel(capture *ydApnCapture) (*sSimYD, func()) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { capture.paths = append(capture.paths, r.URL.Path) capture.query = append(capture.query, r.URL.Query()) switch { case strings.Contains(r.URL.Path, "sim-communication-function/batch"): capture.msisdns = r.URL.Query().Get("msisdns") _, _ = w.Write([]byte(`{"status":"0","message":"正确","result":[{"jobId":"JOB1"}]}`)) case strings.Contains(r.URL.Path, "sim-batch-result"): _, _ = w.Write([]byte(ydBatchResultAllSuccessBody(capture.msisdns))) default: _, _ = w.Write([]byte(`{"status":"0","message":"正确","result":[{"orderNum":"ORD1"}]}`)) } })) return newYdApnChannelWithServer(server), server.Close } // ydBatchResultAllSuccessBody 依据批量请求的 msisdns 拼一个"全部成功"的结果查询响应。 func ydBatchResultAllSuccessBody(msisdns string) string { items := make([]string, 0) for _, id := range strings.Split(msisdns, "_") { id = strings.TrimSpace(id) if id == "" { continue } items = append(items, fmt.Sprintf(`{"status":"0","message":"正确","resultType":"1","resultId":%q}`, id)) } return fmt.Sprintf(`{"status":"0","message":"正确","result":[{"jobStatus":"2","resultList":[%s]}]}`, strings.Join(items, ",")) } func ydSimKeysWithMsisdn(n int) []*model.SimKey { keys := make([]*model.SimKey, 0, n) for i := 0; i < n; i++ { keys = append(keys, &model.SimKey{ ICCID: fmt.Sprintf("898604F41523911%05d", i), MSISDN: fmt.Sprintf("14415436%05d", i), IMSI: fmt.Sprintf("46008743601%05d", i), }) } return keys } func ydApnCounts(paths []string) (batch, single, query int) { for _, p := range paths { switch { case strings.Contains(p, "sim-communication-function/batch"): batch++ case strings.Contains(p, "sim-batch-result"): query++ default: single++ } } return } // 少于 10 张:逐张走 7.2.1 单卡。 func TestYdApnChangeSmallBatchUsesSingle(t *testing.T) { defer shortenYdBatchPoll(2 * time.Second)() capture := &ydApnCapture{} sim, closeFn := newYdApnTestChannel(capture) defer closeFn() out, err := sim.ChangeSimApnFunction(context.Background(), &model.SimCardOperateChangeApnInput{ SimKeys: ydSimKeysWithMsisdn(5), ReqData: model.SimCardOperateChangeApnInputItem{OperType: "0", ApnName: "CMIOT"}, }) if err != nil { t.Fatal(err) } if len(capture.paths) != 5 { t.Fatalf("expected 5 single calls, got %d (%v)", len(capture.paths), capture.paths) } for i, p := range capture.paths { if p != "/v5/ec/manage/sim-communication-service" { t.Fatalf("call %d unexpected path %s", i, p) } q := capture.query[i] if q.Get("functionType") != "01" || q.Get("businessType") != "0" || q.Get("apn") != "CMIOT" { t.Fatalf("call %d unexpected query %v", i, q) } if q.Get("iccid") == "" || q.Get("transid") == "" { t.Fatalf("call %d missing iccid/transid: %v", i, q) } } if len(out) != 5 || out[0].Code != 0 || out[0].ResData.OrderNum != "ORD1" { t.Fatalf("unexpected output %+v", out) } } // 10~100 张:一次走 7.2.2 批量,open 翻成 operType=1;受理后必须回查结果。 func TestYdApnChangeTenToHundredUsesBatch(t *testing.T) { defer shortenYdBatchPoll(2 * time.Second)() for _, n := range []int{10, 100} { capture := &ydApnCapture{} sim, closeFn := newYdApnTestChannel(capture) out, err := sim.ChangeSimApnFunction(context.Background(), &model.SimCardOperateChangeApnInput{ SimKeys: ydSimKeysWithMsisdn(n), ReqData: model.SimCardOperateChangeApnInputItem{OperType: "0", ApnName: "CMIOT"}, }) if err != nil { t.Fatal(err) } batch, single, query := ydApnCounts(capture.paths) if batch != 1 || single != 0 || query == 0 { t.Fatalf("n=%d expected 1 batch + result query, got batch=%d single=%d query=%d (%v)", n, batch, single, query, capture.paths) } q := capture.query[0] if q.Get("serviceType") != "11" || q.Get("operType") != "1" || q.Get("apnName") != "CMIOT" { t.Fatalf("n=%d unexpected batch query %v", n, q) } if got := len(strings.Split(q.Get("msisdns"), "_")); got != n { t.Fatalf("n=%d msisdns count = %d", n, got) } if len(out) != n || out[0].Code != 0 || out[0].ResData.OrderNum != "JOB1" { t.Fatalf("n=%d unexpected output %+v", n, out) } closeFn() } } // 批量任务部分失败:结果查询返回的失败项必须如实回填,不能整批记成功。 func TestYdApnChangeBatchReflectsPartialFailure(t *testing.T) { defer shortenYdBatchPoll(2 * time.Second)() keys := ydSimKeysWithMsisdn(10) failId := keys[3].MSISDN server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { case strings.Contains(r.URL.Path, "sim-communication-function/batch"): _, _ = w.Write([]byte(`{"status":"0","message":"正确","result":[{"jobId":"JOB9"}]}`)) case strings.Contains(r.URL.Path, "sim-batch-result"): items := make([]string, 0, len(keys)) for _, k := range keys { if k.MSISDN == failId { items = append(items, fmt.Sprintf(`{"status":"12047","message":"用户该服务已经是暂停状态,不能重复暂停","resultType":"1","resultId":%q}`, k.MSISDN)) continue } items = append(items, fmt.Sprintf(`{"status":"0","message":"正确","resultType":"1","resultId":%q}`, k.MSISDN)) } _, _ = w.Write([]byte(fmt.Sprintf(`{"status":"0","message":"正确","result":[{"jobStatus":"3","resultList":[%s]}]}`, strings.Join(items, ",")))) default: t.Fatalf("unexpected path %s", r.URL.Path) } })) defer server.Close() out, err := newYdApnChannelWithServer(server).ChangeSimApnFunction(context.Background(), &model.SimCardOperateChangeApnInput{ SimKeys: keys, ReqData: model.SimCardOperateChangeApnInputItem{OperType: "0", ApnName: "CMIOT"}, }) if err != nil { t.Fatal(err) } if len(out) != 10 { t.Fatalf("output len = %d, want 10", len(out)) } failed := 0 for _, item := range out { if item.Code != 0 { failed++ if !strings.Contains(item.Message, "不能重复暂停") { t.Fatalf("unexpected failure message %q", item.Message) } } } if failed != 1 { t.Fatalf("failed count = %d, want 1 (%+v)", failed, out) } } // 轮询超时:未取得终态时按失败记录,并注明"未确认",避免误报成功。 func TestYdApnChangeBatchResultTimeout(t *testing.T) { defer shortenYdBatchPoll(20 * time.Millisecond)() server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { case strings.Contains(r.URL.Path, "sim-communication-function/batch"): _, _ = w.Write([]byte(`{"status":"0","message":"正确","result":[{"jobId":"JOB5"}]}`)) case strings.Contains(r.URL.Path, "sim-batch-result"): _, _ = w.Write([]byte(`{"status":"0","message":"正确","result":[{"jobStatus":"1","resultList":[]}]}`)) default: t.Fatalf("unexpected path %s", r.URL.Path) } })) defer server.Close() out, err := newYdApnChannelWithServer(server).ChangeSimApnFunction(context.Background(), &model.SimCardOperateChangeApnInput{ SimKeys: ydSimKeysWithMsisdn(10), ReqData: model.SimCardOperateChangeApnInputItem{OperType: "0", ApnName: "CMIOT"}, }) if err != nil { t.Fatal(err) } if len(out) != 10 { t.Fatalf("output len = %d, want 10", len(out)) } for _, item := range out { if item.Code == 0 { t.Fatalf("timeout must not be reported as success: %+v", item) } if !strings.Contains(item.Message, "查询上游通信结果失败") { t.Fatalf("unexpected timeout message %q", item.Message) } } } // 停用数据:批量 operType 翻成 0,单卡 businessType=1。 func TestYdApnChangeStopDirection(t *testing.T) { defer shortenYdBatchPoll(2 * time.Second)() capture := &ydApnCapture{} sim, closeFn := newYdApnTestChannel(capture) defer closeFn() if _, err := sim.ChangeSimApnFunction(context.Background(), &model.SimCardOperateChangeApnInput{ SimKeys: ydSimKeysWithMsisdn(10), ReqData: model.SimCardOperateChangeApnInputItem{OperType: "1", ApnName: "CMIOT"}, }); err != nil { t.Fatal(err) } capture2 := &ydApnCapture{} sim2, closeFn2 := newYdApnTestChannel(capture2) defer closeFn2() if _, err := sim2.ChangeSimApnFunction(context.Background(), &model.SimCardOperateChangeApnInput{ SimKeys: ydSimKeysWithMsisdn(3), ReqData: model.SimCardOperateChangeApnInputItem{OperType: "1", ApnName: "CMIOT"}, }); err != nil { t.Fatal(err) } if capture.query[0].Get("operType") != "0" { t.Fatalf("batch stop operType = %q, want 0", capture.query[0].Get("operType")) } if capture2.query[0].Get("businessType") != "1" { t.Fatalf("single stop businessType = %q, want 1", capture2.query[0].Get("businessType")) } } // 超过 100 张:按 100 拆块,不足 10 的尾块退回单卡。 func TestYdApnChangeOverHundredSplits(t *testing.T) { defer shortenYdBatchPoll(2 * time.Second)() capture := &ydApnCapture{} sim, closeFn := newYdApnTestChannel(capture) defer closeFn() out, err := sim.ChangeSimApnFunction(context.Background(), &model.SimCardOperateChangeApnInput{ SimKeys: ydSimKeysWithMsisdn(105), ReqData: model.SimCardOperateChangeApnInputItem{OperType: "0", ApnName: "CMIOT"}, }) if err != nil { t.Fatal(err) } batch, single, _ := ydApnCounts(capture.paths) if batch != 1 || single != 5 { t.Fatalf("expected 1 batch + 5 singles, got batch=%d single=%d (%v)", batch, single, capture.paths) } if len(out) != 105 { t.Fatalf("output len = %d, want 105", len(out)) } } // 批量里有卡缺 msisdn:缺号降级走 7.2.1,其余凑够 10 张仍走 7.2.2。 func TestYdApnChangeMissingMsisdnFallsBackToSingle(t *testing.T) { defer shortenYdBatchPoll(2 * time.Second)() capture := &ydApnCapture{} sim, closeFn := newYdApnTestChannel(capture) defer closeFn() keys := ydSimKeysWithMsisdn(15) keys[0].MSISDN = "" keys[5].MSISDN = "" keys[14].MSISDN = "" out, err := sim.ChangeSimApnFunction(context.Background(), &model.SimCardOperateChangeApnInput{ SimKeys: keys, ReqData: model.SimCardOperateChangeApnInputItem{OperType: "0", ApnName: "CMIOT"}, }) if err != nil { t.Fatal(err) } batch, single, _ := ydApnCounts(capture.paths) if batch != 1 || single != 3 { t.Fatalf("expected 1 batch + 3 singles, got batch=%d single=%d (%v)", batch, single, capture.paths) } if len(out) != 15 { t.Fatalf("output len = %d, want 15", len(out)) } for _, item := range out { if item.Code != 0 { t.Fatalf("unexpected failure item %+v", item) } } }