story/internal/httpapi/ops_config_handlers_integration_test.go

package httpapi

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"testing"

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/concurrency"
	"data-terminal-cloud/internal/identity"
	"data-terminal-cloud/internal/opsconfig"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
)

func TestOpsConfigHandlersPublishAndRollback(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	service, err := opsconfig.NewService(pool, audit.NewRepository())
	if err != nil {
		t.Fatal(err)
	}
	admin := identity.Principal{Issuer: "http://localhost:58180/realms/voice-local", Subject: "00000000-0000-0000-0000-000000000104"}
	approver := identity.Principal{Issuer: admin.Issuer, Subject: "00000000-0000-0000-0000-000000000105"}
	tenantID := "10000000-0000-0000-0000-000000000010"
	configKey := "http.integration.routing." + uuid.NewString()
	createKey, evaluationKey, rollbackKey := "http-ops-create-"+uuid.NewString(), "http-ops-eval-"+uuid.NewString(), "http-ops-rollback-"+uuid.NewString()
	defer func() {
		_, _ = pool.Exec(ctx, `DELETE FROM audit_event WHERE request_id LIKE 'http-ops-%'`)
		_, _ = pool.Exec(ctx, `DELETE FROM idempotency_record WHERE idempotency_key=ANY($1::text[])`, []string{createKey, evaluationKey, rollbackKey})
		_, _ = pool.Exec(ctx, `DELETE FROM runtime_config_rollout WHERE config_key=$1`, configKey)
		_, _ = pool.Exec(ctx, `DELETE FROM ops_config_evaluation WHERE config_version_id IN (SELECT id FROM runtime_config_version WHERE config_key=$1)`, configKey)
		_, _ = pool.Exec(ctx, `DELETE FROM runtime_config_version WHERE config_key=$1`, configKey)
	}()
	withRequestID := func(request *http.Request, id string) *http.Request {
		return request.WithContext(context.WithValue(request.Context(), requestIDKey, id))
	}
	setOpsHeaders := func(request *http.Request, key string) {
		request.Header.Set("X-Tenant-Id", tenantID)
		if key != "" {
			request.Header.Set("Idempotency-Key", key)
		}
	}

	configBody, _ := json.Marshal(map[string]any{"config_key": configKey, "display_name": "HTTP路由配置", "config_kind": "ROUTING", "payload": map[string]any{"default_engine_mode": "s2s_hybrid", "routes": []any{}}, "secret_refs": map[string]string{}, "change_summary": "HTTP配置状态机"})
	request := withRequestID(authenticatedRequest(http.MethodPost, "/api/v1/ops/config-versions", string(configBody), admin), "http-ops-create")
	setOpsHeaders(request, createKey)
	recorder := httptest.NewRecorder()
	CreateOpsConfigVersionHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusCreated || recorder.Header().Get("ETag") != concurrency.FormatETag(1) || recorder.Header().Get("Cache-Control") != "no-store" {
		t.Fatalf("create config status=%d headers=%v body=%s", recorder.Code, recorder.Header(), recorder.Body.String())
	}
	var configResponse struct {
		Data opsconfig.Version `json:"data"`
	}
	if err := json.Unmarshal(recorder.Body.Bytes(), &configResponse); err != nil {
		t.Fatal(err)
	}
	version := configResponse.Data
	request = withRequestID(authenticatedRequest(http.MethodPost, "/api/v1/ops/config-versions", string(configBody), admin), "http-ops-create-replay")
	setOpsHeaders(request, createKey)
	recorder = httptest.NewRecorder()
	CreateOpsConfigVersionHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusOK || recorder.Header().Get("Idempotency-Replayed") != "true" {
		t.Fatalf("replay config status=%d body=%s", recorder.Code, recorder.Body.String())
	}
	request = authenticatedRequest(http.MethodGet, "/api/v1/ops/config-versions?config_key="+configKey, "", approver)
	setOpsHeaders(request, "")
	recorder = httptest.NewRecorder()
	ListOpsConfigVersionsHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), version.ID) {
		t.Fatalf("list config status=%d body=%s", recorder.Code, recorder.Body.String())
	}

	request = withRequestID(authenticatedRequest(http.MethodPost, "/api/v1/ops/config-versions/"+version.ID+"/submit", "", admin), "http-ops-submit")
	request.SetPathValue("id", version.ID)
	setOpsHeaders(request, "")
	recorder = httptest.NewRecorder()
	SubmitOpsConfigVersionHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusPreconditionRequired {
		t.Fatalf("missing submit ETag status=%d body=%s", recorder.Code, recorder.Body.String())
	}
	request.Header.Set("If-Match", concurrency.FormatETag(1))
	recorder = httptest.NewRecorder()
	SubmitOpsConfigVersionHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusOK || recorder.Header().Get("ETag") != concurrency.FormatETag(2) {
		t.Fatalf("submit config status=%d body=%s", recorder.Code, recorder.Body.String())
	}
	publishBody := `{"reason":"HTTP评测通过发布","rollout":{"scope_type":"TENANT","rollout_percent":100,"max_concurrent_calls":10}}`
	request = withRequestID(authenticatedRequest(http.MethodPost, "/api/v1/ops/config-versions/"+version.ID+"/publish", publishBody, approver), "http-ops-publish-no-eval")
	request.SetPathValue("id", version.ID)
	setOpsHeaders(request, "")
	request.Header.Set("If-Match", concurrency.FormatETag(2))
	recorder = httptest.NewRecorder()
	PublishOpsConfigVersionHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusConflict || !strings.Contains(recorder.Body.String(), "CONFIG_PUBLISH_GATE_FAILED") {
		t.Fatalf("publish gate status=%d body=%s", recorder.Code, recorder.Body.String())
	}

	evaluationBody := `{"suite_version":"http-suite-v1","result":"PASSED","metrics":{"cases":8,"passed":8},"report_ref":"integration://http-ops/report"}`
	request = withRequestID(authenticatedRequest(http.MethodPost, "/api/v1/ops/config-versions/"+version.ID+"/evaluations", evaluationBody, admin), "http-ops-evaluation")
	request.SetPathValue("id", version.ID)
	setOpsHeaders(request, evaluationKey)
	recorder = httptest.NewRecorder()
	CreateOpsConfigEvaluationHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusCreated {
		t.Fatalf("create evaluation status=%d body=%s", recorder.Code, recorder.Body.String())
	}
	request = authenticatedRequest(http.MethodGet, "/api/v1/ops/config-versions/"+version.ID+"/evaluations", "", approver)
	request.SetPathValue("id", version.ID)
	setOpsHeaders(request, "")
	recorder = httptest.NewRecorder()
	ListOpsConfigEvaluationsHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), "http-suite-v1") {
		t.Fatalf("list evaluations status=%d body=%s", recorder.Code, recorder.Body.String())
	}

	request = withRequestID(authenticatedRequest(http.MethodPost, "/api/v1/ops/config-versions/"+version.ID+"/publish", publishBody, approver), "http-ops-publish")
	request.SetPathValue("id", version.ID)
	setOpsHeaders(request, "")
	request.Header.Set("If-Match", concurrency.FormatETag(2))
	recorder = httptest.NewRecorder()
	PublishOpsConfigVersionHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusOK || recorder.Header().Get("ETag") != concurrency.FormatETag(3) {
		t.Fatalf("publish config status=%d body=%s", recorder.Code, recorder.Body.String())
	}
	request = authenticatedRequest(http.MethodGet, "/api/v1/ops/config-rollouts?config_key="+configKey, "", approver)
	setOpsHeaders(request, "")
	recorder = httptest.NewRecorder()
	ListOpsConfigRolloutsHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), version.ID) {
		t.Fatalf("list rollouts status=%d body=%s", recorder.Code, recorder.Body.String())
	}
	var rolloutResponse struct {
		Data struct {
			Items []opsconfig.Rollout `json:"items"`
		} `json:"data"`
	}
	if err = json.Unmarshal(recorder.Body.Bytes(), &rolloutResponse); err != nil || len(rolloutResponse.Data.Items) != 1 {
		t.Fatalf("decode rollouts=%+v err=%v", rolloutResponse, err)
	}
	activeRollout := rolloutResponse.Data.Items[0]
	pauseBody := `{"reason":"HTTP纵向冒烟暂停"}`
	request = withRequestID(authenticatedRequest(http.MethodPatch, "/api/v1/ops/config-rollouts/"+activeRollout.ID+"/pause", pauseBody, approver), "http-ops-rollout-pause")
	request.SetPathValue("id", activeRollout.ID)
	setOpsHeaders(request, "")
	recorder = httptest.NewRecorder()
	PauseOpsConfigRolloutHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusPreconditionRequired {
		t.Fatalf("missing rollout ETag status=%d body=%s", recorder.Code, recorder.Body.String())
	}
	request.Header.Set("If-Match", concurrency.FormatETag(activeRollout.RowVersion))
	recorder = httptest.NewRecorder()
	PauseOpsConfigRolloutHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusOK || recorder.Header().Get("ETag") != concurrency.FormatETag(2) || !strings.Contains(recorder.Body.String(), `"status":"PAUSED"`) {
		t.Fatalf("pause rollout status=%d headers=%v body=%s", recorder.Code, recorder.Header(), recorder.Body.String())
	}
	resumeBody := `{"reason":"HTTP纵向冒烟恢复"}`
	request = withRequestID(authenticatedRequest(http.MethodPatch, "/api/v1/ops/config-rollouts/"+activeRollout.ID+"/resume", resumeBody, approver), "http-ops-rollout-resume-stale")
	request.SetPathValue("id", activeRollout.ID)
	setOpsHeaders(request, "")
	request.Header.Set("If-Match", concurrency.FormatETag(1))
	recorder = httptest.NewRecorder()
	ResumeOpsConfigRolloutHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusPreconditionFailed || !strings.Contains(recorder.Body.String(), "VERSION_CONFLICT") {
		t.Fatalf("stale rollout resume status=%d body=%s", recorder.Code, recorder.Body.String())
	}
	request = withRequestID(authenticatedRequest(http.MethodPatch, "/api/v1/ops/config-rollouts/"+activeRollout.ID+"/resume", resumeBody, approver), "http-ops-rollout-resume")
	request.SetPathValue("id", activeRollout.ID)
	setOpsHeaders(request, "")
	request.Header.Set("If-Match", concurrency.FormatETag(2))
	recorder = httptest.NewRecorder()
	ResumeOpsConfigRolloutHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusOK || recorder.Header().Get("ETag") != concurrency.FormatETag(3) || !strings.Contains(recorder.Body.String(), `"status":"ACTIVE"`) {
		t.Fatalf("resume rollout status=%d headers=%v body=%s", recorder.Code, recorder.Header(), recorder.Body.String())
	}

	rollbackBody := `{"reason":"HTTP回滚验证","rollout":{"scope_type":"TENANT","rollout_percent":100,"max_concurrent_calls":10}}`
	request = withRequestID(authenticatedRequest(http.MethodPost, "/api/v1/ops/config-versions/"+version.ID+"/rollback", rollbackBody, approver), "http-ops-rollback")
	request.SetPathValue("id", version.ID)
	setOpsHeaders(request, rollbackKey)
	recorder = httptest.NewRecorder()
	RollbackOpsConfigVersionHandler(service).ServeHTTP(recorder, request)
	if recorder.Code != http.StatusCreated || !strings.Contains(recorder.Body.String(), `"version_no":2`) {
		t.Fatalf("rollback status=%d body=%s", recorder.Code, recorder.Body.String())
	}
}