MiFaZhan
438 lines
18 KiB
Go
438 lines
18 KiB
Go
package httpapi
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"net/url"
|
|
"reflect"
|
|
"strings"
|
|
"testing"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/contract"
|
|
)
|
|
|
|
const (
|
|
testToken = "test-secret"
|
|
runPath = "/internal/v1/full-sync-runs/run-1"
|
|
failurePath = runPath + "/failures"
|
|
replayPath = runPath + "/replay"
|
|
actionPath = "/internal/v1/full-sync-actions/action-1"
|
|
)
|
|
|
|
type runFunc func(context.Context, string) (contract.RunView, error)
|
|
|
|
func (f runFunc) GetRun(ctx context.Context, id string) (contract.RunView, error) {
|
|
return f(ctx, id)
|
|
}
|
|
|
|
type runListFunc func(context.Context, contract.RunListFilter) (contract.RunList, error)
|
|
|
|
func (f runListFunc) ListRuns(ctx context.Context, filter contract.RunListFilter) (contract.RunList, error) {
|
|
return f(ctx, filter)
|
|
}
|
|
|
|
type failureFunc func(context.Context, string, contract.PageRequest) (contract.FailurePage, error)
|
|
|
|
func (f failureFunc) ListFailures(ctx context.Context, id string, page contract.PageRequest) (contract.FailurePage, error) {
|
|
return f(ctx, id, page)
|
|
}
|
|
|
|
type replayFunc func(context.Context, contract.ReplayRequest) (contract.ReplayAccepted, error)
|
|
|
|
func (f replayFunc) RequestReplay(ctx context.Context, req contract.ReplayRequest) (contract.ReplayAccepted, error) {
|
|
return f(ctx, req)
|
|
}
|
|
|
|
type actionFunc func(context.Context, string) (contract.ReplayAction, error)
|
|
|
|
func (f actionFunc) GetReplayAction(ctx context.Context, id string) (contract.ReplayAction, error) {
|
|
return f(ctx, id)
|
|
}
|
|
|
|
func testDependencies() Dependencies {
|
|
return Dependencies{
|
|
Runs: runFunc(func(context.Context, string) (contract.RunView, error) {
|
|
return contract.RunView{}, contract.ErrNotFound
|
|
}),
|
|
List: runListFunc(func(context.Context, contract.RunListFilter) (contract.RunList, error) {
|
|
return contract.RunList{}, contract.ErrNotFound
|
|
}),
|
|
Failures: failureFunc(func(context.Context, string, contract.PageRequest) (contract.FailurePage, error) {
|
|
return contract.FailurePage{}, contract.ErrNotFound
|
|
}),
|
|
Replays: replayFunc(func(context.Context, contract.ReplayRequest) (contract.ReplayAccepted, error) {
|
|
return contract.ReplayAccepted{}, contract.ErrNotFound
|
|
}),
|
|
Actions: actionFunc(func(context.Context, string) (contract.ReplayAction, error) {
|
|
return contract.ReplayAction{}, contract.ErrNotFound
|
|
}),
|
|
}
|
|
}
|
|
|
|
func testHandler(t *testing.T, deps Dependencies) http.Handler {
|
|
t.Helper()
|
|
handler, err := NewHandler(testToken, deps)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return handler
|
|
}
|
|
|
|
func request(handler http.Handler, method, path, body string, headers ...string) *httptest.ResponseRecorder {
|
|
req := httptest.NewRequest(method, path, strings.NewReader(body))
|
|
req.Header.Set("Authorization", "Bearer "+testToken)
|
|
for i := 0; i < len(headers); i += 2 {
|
|
req.Header.Set(headers[i], headers[i+1])
|
|
}
|
|
recorder := httptest.NewRecorder()
|
|
handler.ServeHTTP(recorder, req)
|
|
return recorder
|
|
}
|
|
|
|
func assertError(t *testing.T, response *httptest.ResponseRecorder, status int, code string) {
|
|
t.Helper()
|
|
if response.Code != status {
|
|
t.Fatalf("status = %d, want %d: %s", response.Code, status, response.Body)
|
|
}
|
|
if !strings.HasPrefix(response.Header().Get("Content-Type"), "application/json") {
|
|
t.Fatalf("missing JSON content type: %v", response.Header())
|
|
}
|
|
var body ErrorResponse
|
|
if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if body.Error.Code != code || body.Error.Message == "" {
|
|
t.Fatalf("error = %+v, want code %q and message", body, code)
|
|
}
|
|
}
|
|
|
|
func TestAuthenticationProtectsEveryRoute(t *testing.T) {
|
|
deps := testDependencies()
|
|
deps.Runs = runFunc(func(context.Context, string) (contract.RunView, error) {
|
|
t.Fatal("unauthenticated request reached service")
|
|
return contract.RunView{}, nil
|
|
})
|
|
handler := testHandler(t, deps)
|
|
for _, path := range []string{runPath, failurePath, replayPath, actionPath, "/unknown"} {
|
|
for _, auth := range []string{"", "Bearer", "Bearer wrong", "Basic " + testToken, "Bearer " + testToken + " extra"} {
|
|
t.Run(path+"/"+auth, func(t *testing.T) {
|
|
response := request(handler, http.MethodGet, path, "", "Authorization", auth)
|
|
assertError(t, response, 401, "unauthorized")
|
|
if response.Header().Get("WWW-Authenticate") == "" {
|
|
t.Fatal("missing authentication challenge")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
req := httptest.NewRequest(http.MethodGet, runPath, nil)
|
|
req.Header.Add("Authorization", "Bearer "+testToken)
|
|
req.Header.Add("Authorization", "Bearer "+testToken)
|
|
response := httptest.NewRecorder()
|
|
handler.ServeHTTP(response, req)
|
|
assertError(t, response, 401, "unauthorized")
|
|
}
|
|
|
|
func TestRunListQuery(t *testing.T) {
|
|
deps := testDependencies()
|
|
deps.List = runListFunc(func(_ context.Context, filter contract.RunListFilter) (contract.RunList, error) {
|
|
if filter.SyncKind != contract.SyncKindTrafficUsage || filter.Limit != 7 {
|
|
t.Fatalf("filter = %+v", filter)
|
|
}
|
|
return contract.RunList{Items: []contract.RunView{{Source: contract.SourceLive, Run: contract.RunMeta{RunID: "run-1", Operation: "current_month_usage_full_sync"}}}, Total: 1}, nil
|
|
})
|
|
response := request(testHandler(t, deps), http.MethodGet, "/internal/v1/full-sync-runs?syncKind=traffic_usage&limit=7", "")
|
|
if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), "run-1") {
|
|
t.Fatalf("response = %d %s", response.Code, response.Body)
|
|
}
|
|
for _, query := range []string{"syncKind=invalid", "limit=101", "limit=abc", "limit=1&limit=2"} {
|
|
assertError(t, request(testHandler(t, deps), http.MethodGet, "/internal/v1/full-sync-runs?"+query, ""), 400, "invalid_argument")
|
|
}
|
|
}
|
|
|
|
func TestRoutesAndValidation(t *testing.T) {
|
|
handler := testHandler(t, testDependencies())
|
|
for _, route := range []struct{ method, path string }{
|
|
{http.MethodGet, runPath}, {http.MethodGet, failurePath},
|
|
{http.MethodPost, replayPath}, {http.MethodGet, actionPath},
|
|
} {
|
|
t.Run(route.path, func(t *testing.T) {
|
|
response := request(handler, route.method, route.path, `{"requestedBy":"operator"}`, contract.IdempotencyHeader, "key")
|
|
assertError(t, response, 404, "not_found")
|
|
wrongMethod := http.MethodPost
|
|
if route.method == http.MethodPost {
|
|
wrongMethod = http.MethodGet
|
|
}
|
|
response = request(handler, wrongMethod, route.path, "")
|
|
assertError(t, response, 405, "method_not_allowed")
|
|
if response.Header().Get("Allow") != route.method {
|
|
t.Fatalf("Allow = %q", response.Header().Get("Allow"))
|
|
}
|
|
})
|
|
}
|
|
for _, path := range []string{"/unknown", "/internal/v1/full-sync-runs", runPath + "/other", runPath + "/"} {
|
|
assertError(t, request(handler, http.MethodGet, path, ""), 404, "not_found")
|
|
}
|
|
for _, path := range []string{
|
|
"/internal/v1/full-sync-runs/%20run",
|
|
"/internal/v1/full-sync-runs/a%2Fb/failures",
|
|
"/internal/v1/full-sync-actions/%7Baction%7D",
|
|
} {
|
|
assertError(t, request(handler, http.MethodGet, path, ""), 400, "invalid_argument")
|
|
}
|
|
assertError(t, request(handler, http.MethodPost, "/internal/v1/full-sync-runs/a%5Cb/replay", `{"requestedBy":"operator"}`, contract.IdempotencyHeader, "key"), 400, "invalid_argument")
|
|
}
|
|
|
|
func TestRunAndActionResponsesPropagateContext(t *testing.T) {
|
|
type contextKey struct{}
|
|
deps := testDependencies()
|
|
deps.Runs = runFunc(func(ctx context.Context, id string) (contract.RunView, error) {
|
|
if id != "run-1" || ctx.Value(contextKey{}) != "request" {
|
|
t.Fatal("run identity or request context lost")
|
|
}
|
|
return contract.RunView{Source: contract.SourceLive, Run: contract.RunMeta{RunID: id}}, nil
|
|
})
|
|
deps.Actions = actionFunc(func(ctx context.Context, id string) (contract.ReplayAction, error) {
|
|
if id != "action-1" || ctx.Value(contextKey{}) != "request" {
|
|
t.Fatal("action identity or request context lost")
|
|
}
|
|
return contract.ReplayAction{ActionID: id, Status: contract.ActionProcessing, IdempotencyKey: "private-key", Fingerprint: "private-fingerprint", LeaseToken: "private-lease", Revision: 99}, nil
|
|
})
|
|
handler := testHandler(t, deps)
|
|
for _, path := range []string{runPath, actionPath} {
|
|
req := httptest.NewRequest(http.MethodGet, path, nil).WithContext(context.WithValue(context.Background(), contextKey{}, "request"))
|
|
req.Header.Set("Authorization", "bearer "+testToken)
|
|
response := httptest.NewRecorder()
|
|
handler.ServeHTTP(response, req)
|
|
if response.Code != 200 || response.Header().Get("Cache-Control") != "no-store" {
|
|
t.Fatalf("response = %d %s", response.Code, response.Body)
|
|
}
|
|
if strings.Contains(response.Body.String(), "private-") || strings.Contains(response.Body.String(), "revision") {
|
|
t.Fatalf("internal fields leaked: %s", response.Body)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestFailurePagination(t *testing.T) {
|
|
for _, test := range []struct {
|
|
query string
|
|
limit int
|
|
cursor contract.Cursor
|
|
}{
|
|
{"", 100, ""}, {"?limit=", 100, ""}, {"?limit=0", 100, ""},
|
|
{"?limit=1", 1, ""}, {"?limit=500", 500, ""},
|
|
{"?limit=17&cursor=" + url.QueryEscape("opaque/+ token==:雪"), 17, "opaque/+ token==:雪"},
|
|
} {
|
|
t.Run(test.query, func(t *testing.T) {
|
|
deps := testDependencies()
|
|
calls := 0
|
|
deps.Failures = failureFunc(func(_ context.Context, id string, page contract.PageRequest) (contract.FailurePage, error) {
|
|
calls++
|
|
if id != "run-1" || page.Limit != test.limit || page.Cursor != test.cursor {
|
|
t.Fatalf("query = %q %+v", id, page)
|
|
}
|
|
return contract.FailurePage{NextCursor: "next:opaque/+="}, nil
|
|
})
|
|
response := request(testHandler(t, deps), http.MethodGet, failurePath+test.query, "")
|
|
if response.Code != 200 || calls != 1 {
|
|
t.Fatalf("response = %d, calls = %d", response.Code, calls)
|
|
}
|
|
var page contract.FailurePage
|
|
if err := json.Unmarshal(response.Body.Bytes(), &page); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if page.Items == nil || len(page.Items) != 0 || page.NextCursor != "next:opaque/+=" {
|
|
t.Fatalf("empty scan page lost array or continuation: %s", response.Body)
|
|
}
|
|
})
|
|
}
|
|
deps := testDependencies()
|
|
deps.Failures = failureFunc(func(context.Context, string, contract.PageRequest) (contract.FailurePage, error) {
|
|
t.Fatal("invalid pagination reached service")
|
|
return contract.FailurePage{}, nil
|
|
})
|
|
handler := testHandler(t, deps)
|
|
for _, query := range []string{"limit=-1", "limit=501", "limit=abc", "limit=1.5", "limit=999999999999999999999999", "limit=1&limit=2", "cursor=a&cursor=b", "cursor=%ZZ", "cursor=a;b"} {
|
|
assertError(t, request(handler, http.MethodGet, failurePath+"?"+query, ""), 400, "invalid_argument")
|
|
}
|
|
}
|
|
|
|
func TestFailureCursorValidationAndLiveOnly(t *testing.T) {
|
|
deps := testDependencies()
|
|
deps.Runs = runFunc(func(context.Context, string) (contract.RunView, error) {
|
|
t.Fatal("failure route must not query run summaries or fall back to archives")
|
|
return contract.RunView{}, nil
|
|
})
|
|
deps.Failures = failureFunc(func(_ context.Context, _ string, page contract.PageRequest) (contract.FailurePage, error) {
|
|
if page.Cursor == "another-run-token" {
|
|
return contract.FailurePage{}, fmt.Errorf("wrong query binding: %w", contract.ErrInvalidArgument)
|
|
}
|
|
return contract.FailurePage{}, fmt.Errorf("expired details: %w", contract.ErrFailureDetailsUnavailable)
|
|
})
|
|
handler := testHandler(t, deps)
|
|
assertError(t, request(handler, http.MethodGet, failurePath+"?cursor=another-run-token", ""), 400, "invalid_argument")
|
|
assertError(t, request(handler, http.MethodGet, failurePath, ""), 410, "failure_details_unavailable")
|
|
}
|
|
|
|
func TestReplayValidation(t *testing.T) {
|
|
deps := testDependencies()
|
|
deps.Replays = replayFunc(func(context.Context, contract.ReplayRequest) (contract.ReplayAccepted, error) {
|
|
t.Fatal("invalid replay reached service")
|
|
return contract.ReplayAccepted{}, nil
|
|
})
|
|
handler := testHandler(t, deps)
|
|
for _, body := range []string{"", "{", "null", "[]", `{}`, `{"requestedBy":""}`, `{"requestedBy":" "}`, `{"requestedBy":12}`, `{"requestedBy":"a","targetRunId":"injected"}`, `{"requestedBy":"a"} {}`, `{"requestedBy":"a"} trailing`, "{\"requestedBy\":\"\xff\"}"} {
|
|
t.Run(body, func(t *testing.T) {
|
|
assertError(t, request(handler, http.MethodPost, replayPath, body, contract.IdempotencyHeader, "key"), 400, "invalid_argument")
|
|
})
|
|
}
|
|
for _, key := range []string{"", " ", strings.Repeat("x", contract.MaxIdempotencyKeyBytes+1), "bad\tkey"} {
|
|
assertError(t, request(handler, http.MethodPost, replayPath, `{"requestedBy":"operator"}`, contract.IdempotencyHeader, key), 400, "invalid_argument")
|
|
}
|
|
assertError(t, request(handler, http.MethodPost, replayPath, `{"requestedBy":"operator"}`), 400, "invalid_argument")
|
|
assertError(t, request(handler, http.MethodPost, replayPath, `{"requestedBy":"operator"}`, contract.IdempotencyHeader, "key", "Content-Type", "text/plain"), 400, "invalid_argument")
|
|
assertError(t, request(handler, http.MethodPost, replayPath, `{"requestedBy":"`+strings.Repeat("x", maxReplayBodyBytes)+`"}`, contract.IdempotencyHeader, "key"), 413, "body_too_large")
|
|
req := httptest.NewRequest(http.MethodPost, replayPath, strings.NewReader(`{"requestedBy":"operator"}`))
|
|
req.Header.Set("Authorization", "Bearer "+testToken)
|
|
req.Header.Add(contract.IdempotencyHeader, "key")
|
|
req.Header.Add(contract.IdempotencyHeader, "key")
|
|
response := httptest.NewRecorder()
|
|
handler.ServeHTTP(response, req)
|
|
assertError(t, response, 400, "invalid_argument")
|
|
}
|
|
|
|
// The mock owns idempotency; the HTTP adapter must pass exact request identity
|
|
// and return the service's existing action/target even across handler instances.
|
|
func TestReplayAcceptedIdempotencyAndConflict(t *testing.T) {
|
|
deps := testDependencies()
|
|
var (
|
|
stored contract.ReplayRequest
|
|
calls int
|
|
)
|
|
deps.Replays = replayFunc(func(_ context.Context, req contract.ReplayRequest) (contract.ReplayAccepted, error) {
|
|
calls++
|
|
if calls == 1 {
|
|
stored = req
|
|
}
|
|
if req != stored {
|
|
return contract.ReplayAccepted{}, fmt.Errorf("changed fingerprint: %w", contract.ErrConflict)
|
|
}
|
|
return contract.ReplayAccepted{ActionID: "action-1", TargetRunID: "replay-1", Status: contract.ActionAccepted, StatusURL: "https://untrusted.invalid"}, nil
|
|
})
|
|
var first contract.ReplayAccepted
|
|
for i := 0; i < 2; i++ {
|
|
response := request(testHandler(t, deps), http.MethodPost, replayPath, "{\"requestedBy\":\" operator \"}\n", contract.IdempotencyHeader, "key", "Content-Type", "application/json; charset=utf-8")
|
|
if response.Code != 202 {
|
|
t.Fatalf("response = %d %s", response.Code, response.Body)
|
|
}
|
|
var accepted contract.ReplayAccepted
|
|
if err := json.Unmarshal(response.Body.Bytes(), &accepted); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if accepted.ActionID != "action-1" || accepted.TargetRunID != "replay-1" || accepted.StatusURL != actionPath || response.Header().Get("Location") != actionPath {
|
|
t.Fatalf("acceptance = %+v, headers = %v", accepted, response.Header())
|
|
}
|
|
if i == 0 {
|
|
first = accepted
|
|
} else if !reflect.DeepEqual(first, accepted) {
|
|
t.Fatalf("repeat changed identity: %+v vs %+v", first, accepted)
|
|
}
|
|
}
|
|
if stored.SourceRunID != "run-1" || stored.RequestedBy != " operator " || stored.IdempotencyKey != "key" {
|
|
t.Fatalf("request identity changed: %+v", stored)
|
|
}
|
|
assertError(t, request(testHandler(t, deps), http.MethodPost, replayPath, `{"requestedBy":"operator"}`, contract.IdempotencyHeader, "key"), 409, "conflict")
|
|
if calls != 3 {
|
|
t.Fatalf("replay calls = %d", calls)
|
|
}
|
|
}
|
|
|
|
func TestWrappedServiceErrorsUseConsistentEnvelope(t *testing.T) {
|
|
for _, test := range []struct {
|
|
err error
|
|
status int
|
|
code string
|
|
}{
|
|
{contract.ErrInvalidArgument, 400, "invalid_argument"},
|
|
{contract.ErrNotFound, 404, "not_found"},
|
|
{contract.ErrConflict, 409, "conflict"},
|
|
{contract.ErrLeaseLost, 409, "conflict"},
|
|
{contract.ErrFailureDetailsUnavailable, 410, "failure_details_unavailable"},
|
|
{contract.ErrBusy, 503, "busy"},
|
|
{fmt.Errorf("private database password"), 500, "internal_error"},
|
|
} {
|
|
t.Run(test.code, func(t *testing.T) {
|
|
err := fmt.Errorf("private database detail: %w", test.err)
|
|
deps := Dependencies{
|
|
Runs: runFunc(func(context.Context, string) (contract.RunView, error) { return contract.RunView{}, err }),
|
|
List: runListFunc(func(context.Context, contract.RunListFilter) (contract.RunList, error) {
|
|
return contract.RunList{}, err
|
|
}),
|
|
Failures: failureFunc(func(context.Context, string, contract.PageRequest) (contract.FailurePage, error) {
|
|
return contract.FailurePage{}, err
|
|
}),
|
|
Replays: replayFunc(func(context.Context, contract.ReplayRequest) (contract.ReplayAccepted, error) {
|
|
return contract.ReplayAccepted{}, err
|
|
}),
|
|
Actions: actionFunc(func(context.Context, string) (contract.ReplayAction, error) { return contract.ReplayAction{}, err }),
|
|
}
|
|
handler := testHandler(t, deps)
|
|
for _, path := range []string{runPath, failurePath, actionPath, replayPath} {
|
|
method := http.MethodGet
|
|
if path == replayPath {
|
|
method = http.MethodPost
|
|
}
|
|
response := request(handler, method, path, `{"requestedBy":"operator"}`, contract.IdempotencyHeader, "key")
|
|
assertError(t, response, test.status, test.code)
|
|
if strings.Contains(response.Body.String(), "private") {
|
|
t.Fatal("service details leaked")
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestInvalidServiceOutputReturns500(t *testing.T) {
|
|
deps := testDependencies()
|
|
deps.Replays = replayFunc(func(context.Context, contract.ReplayRequest) (contract.ReplayAccepted, error) {
|
|
return contract.ReplayAccepted{ActionID: "bad/id"}, nil
|
|
})
|
|
deps.Runs = runFunc(func(context.Context, string) (contract.RunView, error) {
|
|
return contract.RunView{Rounds: []contract.RoundStats{{ErrorCounts: contract.ErrorCounts{contract.ErrorOther: -1}}}}, nil
|
|
})
|
|
handler := testHandler(t, deps)
|
|
assertError(t, request(handler, http.MethodPost, replayPath, `{"requestedBy":"operator"}`, contract.IdempotencyHeader, "key"), 500, "internal_error")
|
|
assertError(t, request(handler, http.MethodGet, runPath, ""), 500, "internal_error")
|
|
}
|
|
|
|
func TestNewHandlerRejectsInvalidConfiguration(t *testing.T) {
|
|
if _, err := NewHandler("", testDependencies()); err != nil {
|
|
t.Fatalf("empty token should use Docker-network-only access: %v", err)
|
|
}
|
|
for _, token := range []string{" ", "a b", "token\n"} {
|
|
if _, err := NewHandler(token, testDependencies()); err == nil {
|
|
t.Errorf("accepted token %q", token)
|
|
}
|
|
}
|
|
for i := 0; i < 5; i++ {
|
|
deps := testDependencies()
|
|
switch i {
|
|
case 0:
|
|
deps.Runs = nil
|
|
case 1:
|
|
deps.Failures = nil
|
|
case 2:
|
|
deps.Replays = nil
|
|
case 3:
|
|
deps.Actions = nil
|
|
case 4:
|
|
deps.List = nil
|
|
}
|
|
if _, err := NewHandler(testToken, deps); err == nil {
|
|
t.Errorf("accepted missing dependency %d", i)
|
|
}
|
|
}
|
|
}
|