package iot import ( "context" "encoding/json" "net/http" "net/http/httptest" "testing" "github.com/gogf/gf/v2/errors/gcode" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" v1 "git.5iot.com/iotlcm-public/iotlcm-admin-svr/api/iot/v1" "git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/service/fullsyncretry" ) func TestFullSyncReplayRequestMaxRoundsThroughController(t *testing.T) { var bodies []map[string]any server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost || r.URL.Path != "/internal/v1/full-sync-runs/source/replay" || r.Header.Get("Content-Type") != "application/json" { t.Errorf("unexpected Worker request: %s %s", r.Method, r.URL) } var body map[string]any if err := json.NewDecoder(r.Body).Decode(&body); err != nil { t.Error(err) } bodies = append(bodies, body) w.WriteHeader(http.StatusAccepted) _, _ = w.Write([]byte(`{"replayRunId":"replay-1","status":"accepted"}`)) })) defer server.Close() controller := &FullSyncReplayController{client: fullsyncretry.Client{BaseURL: server.URL}} tests := []struct { body string rounds int }{ {`{"sourceRunId":"source"}`, 0}, {`{"sourceRunId":"source","maxRounds":0}`, 0}, {`{"sourceRunId":"source","maxRounds":1}`, 1}, {`{"sourceRunId":"source","maxRounds":3}`, 3}, {`{"sourceRunId":"source","maxRounds":7}`, 7}, {`{"sourceRunId":"source","maxRounds":100}`, 100}, } for i, tt := range tests { var req v1.FullSyncReplayReq if err := json.Unmarshal([]byte(tt.body), &req); err != nil { t.Fatal(err) } if err := g.Validator().Data(req).Run(context.Background()); err != nil { t.Fatalf("valid request %s: %v", tt.body, err) } res, err := controller.Replay(fullSyncActorContext(), &req) if err != nil || res.ReplayRunID != "replay-1" || res.Status != "accepted" { t.Fatalf("replay = %+v, err = %v", res, err) } body := bodies[i] if body["requestedBy"] != "42" { t.Fatalf("actor not forwarded: %+v", body) } if tt.rounds == 0 { if len(body) != 1 { t.Fatalf("legacy Worker must receive no extra field: %+v", body) } } else if len(body) != 2 || body["maxRounds"] != float64(tt.rounds) { t.Fatalf("rounds not forwarded: %+v", body) } } for _, rounds := range []int{-1, 101} { req := v1.FullSyncReplayReq{SourceRunID: "source", MaxRounds: rounds} if err := g.Validator().Data(req).Run(context.Background()); err == nil { t.Fatalf("API accepted maxRounds=%d", rounds) } if _, err := controller.Replay(fullSyncActorContext(), &req); gerror.Code(err) != gcode.CodeInvalidParameter { t.Fatalf("controller accepted maxRounds=%d: %v", rounds, err) } } if _, err := controller.Replay(context.Background(), &v1.FullSyncReplayReq{SourceRunID: "source", MaxRounds: 3}); gerror.Code(err) != gcode.CodeNotAuthorized { t.Fatalf("anonymous replay = %v", err) } if len(bodies) != len(tests) { t.Fatalf("invalid/anonymous request reached Worker: %d calls", len(bodies)) } }