story/internal/opsconfig/service_integration_test.go

package opsconfig

import (
	"context"
	"errors"
	"os"
	"testing"

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

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

func TestRoutingConfigurationApprovalPublishAndRollback(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()
	if _, err := NewService(nil, audit.NewRepository()); err == nil {
		t.Fatal("nil database pool was accepted")
	}
	service, err := 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"}
	outsider := identity.Principal{Issuer: admin.Issuer, Subject: "00000000-0000-0000-0000-000000000103"}
	tenantID := "10000000-0000-0000-0000-000000000010"
	configKey := "integration.routing." + uuid.NewString()
	keys := []string{"ops-integration-v1-" + uuid.NewString(), "ops-integration-eval1-" + uuid.NewString(), "ops-integration-v2-" + uuid.NewString(), "ops-integration-eval2-" + uuid.NewString(), "ops-integration-rollback-" + uuid.NewString()}
	defer func() {
		_, _ = pool.Exec(ctx, `DELETE FROM audit_event WHERE request_id LIKE 'ops-integration-%'`)
		_, _ = pool.Exec(ctx, `DELETE FROM idempotency_record WHERE idempotency_key=ANY($1::text[])`, keys)
		_, _ = 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)
	}()
	input := CreateInput{TenantID: tenantID, ConfigKey: configKey, DisplayName: "集成路由配置", ConfigKind: "ROUTING", Payload: map[string]any{"default_engine_mode": "s2s_hybrid", "routes": []any{}}, SecretRefs: map[string]string{}, ChangeSummary: "建立集成路由基线"}
	v1, created, err := service.Create(ctx, admin, keys[0], "ops-integration-create-v1", input)
	if err != nil || !created || v1.Status != "DRAFT" || v1.VersionNo != 1 || v1.RowVersion != 1 {
		t.Fatalf("create v1=%+v created=%v err=%v", v1, created, err)
	}
	replayed, created, err := service.Create(ctx, admin, keys[0], "ops-integration-create-v1-replay", input)
	if err != nil || created || replayed.ID != v1.ID {
		t.Fatalf("replay v1=%+v created=%v err=%v", replayed, created, err)
	}
	if _, _, err := service.Create(ctx, outsider, "ops-integration-outsider", "ops-integration-outsider", input); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider create error=%v", err)
	}
	versions, err := service.List(ctx, approver, tenantID, configKey)
	if err != nil || len(versions) != 1 || versions[0].ID != v1.ID {
		t.Fatalf("versions=%+v err=%v", versions, err)
	}
	v1, err = service.Submit(ctx, admin, tenantID, v1.ID, "ops-integration-submit-v1", 1)
	if err != nil || v1.Status != "PENDING_APPROVAL" || v1.RowVersion != 2 || v1.SubmittedAt == nil {
		t.Fatalf("submitted v1=%+v err=%v", v1, err)
	}
	rollout := RolloutInput{ScopeType: "TENANT", RolloutPercent: 100, MaxConcurrentCalls: 10}
	if _, err := service.Publish(ctx, admin, tenantID, v1.ID, "ops-integration-self-publish", PublishInput{Reason: "禁止自批", Rollout: rollout}, 2); !errors.Is(err, ErrSelfApproval) {
		t.Fatalf("self approval error=%v", err)
	}
	if _, err := service.Publish(ctx, approver, tenantID, v1.ID, "ops-integration-no-eval", PublishInput{Reason: "缺少评测", Rollout: rollout}, 2); !errors.Is(err, ErrPublishGate) {
		t.Fatalf("missing evaluation gate error=%v", err)
	}
	evaluation, created, err := service.CreateEvaluation(ctx, admin, tenantID, v1.ID, keys[1], "ops-integration-eval-v1", EvidenceTypeGeneric, "routing-suite-v1", "PASSED", "integration://reports/routing-v1", map[string]any{"cases": 12, "passed": 12})
	if err != nil || !created || evaluation.Result != "PASSED" {
		t.Fatalf("evaluation=%+v created=%v err=%v", evaluation, created, err)
	}
	evaluations, err := service.ListEvaluations(ctx, approver, tenantID, v1.ID)
	if err != nil || len(evaluations) != 1 || evaluations[0].ID != evaluation.ID {
		t.Fatalf("evaluations=%+v err=%v", evaluations, err)
	}
	v1, err = service.Publish(ctx, approver, tenantID, v1.ID, "ops-integration-publish-v1", PublishInput{Reason: "评测通过后发布", Rollout: rollout}, 2)
	if err != nil || v1.Status != "PUBLISHED" || v1.RowVersion != 3 || v1.PublishedAt == nil {
		t.Fatalf("published v1=%+v err=%v", v1, err)
	}
	rollouts, err := service.ListRollouts(ctx, approver, tenantID, configKey)
	if err != nil || len(rollouts) != 1 || rollouts[0].Status != "ACTIVE" || rollouts[0].ConfigVersionID != v1.ID || rollouts[0].RowVersion != 1 {
		t.Fatalf("v1 rollouts=%+v err=%v", rollouts, err)
	}
	if _, err = service.PauseRollout(ctx, outsider, tenantID, rollouts[0].ID, "ops-integration-pause-outsider", "无权暂停", rollouts[0].RowVersion); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider pause error=%v", err)
	}
	paused, err := service.PauseRollout(ctx, approver, tenantID, rollouts[0].ID, "ops-integration-pause-v1", "观察到异常，暂停新通话命中", rollouts[0].RowVersion)
	if err != nil || paused.Status != "PAUSED" || paused.PausedAt == nil || paused.RowVersion != 2 {
		t.Fatalf("paused rollout=%+v err=%v", paused, err)
	}
	if _, err = service.ResumeRollout(ctx, approver, tenantID, paused.ID, "ops-integration-resume-stale", "使用过期版本尝试恢复", 1); !errors.Is(err, ErrVersionConflict) {
		t.Fatalf("stale rollout resume error=%v", err)
	}
	resumed, err := service.ResumeRollout(ctx, approver, tenantID, paused.ID, "ops-integration-resume-v1", "观察确认正常，恢复投放", paused.RowVersion)
	if err != nil || resumed.Status != "ACTIVE" || resumed.PausedAt != nil || resumed.RowVersion != 3 {
		t.Fatalf("resumed rollout=%+v err=%v", resumed, err)
	}
	paused, err = service.PauseRollout(ctx, approver, tenantID, resumed.ID, "ops-integration-pause-before-v2", "发布第二版前暂停旧投放", resumed.RowVersion)
	if err != nil || paused.Status != "PAUSED" || paused.RowVersion != 4 {
		t.Fatalf("second pause=%+v err=%v", paused, err)
	}

	input.Payload = map[string]any{"default_engine_mode": "cascade", "routes": []any{map[string]any{"goal": "open_conversation", "engine_mode": "cascade"}}}
	input.ChangeSummary = "切换合成级联路由"
	v2, created, err := service.Create(ctx, admin, keys[2], "ops-integration-create-v2", input)
	if err != nil || !created || v2.VersionNo != 2 {
		t.Fatalf("create v2=%+v created=%v err=%v", v2, created, err)
	}
	v2, err = service.Submit(ctx, admin, tenantID, v2.ID, "ops-integration-submit-v2", 1)
	if err != nil {
		t.Fatal(err)
	}
	if _, _, err := service.CreateEvaluation(ctx, admin, tenantID, v2.ID, keys[3], "ops-integration-eval-v2", EvidenceTypeGeneric, "routing-suite-v1", "PASSED", "integration://reports/routing-v2", map[string]any{"cases": 12, "passed": 12}); err != nil {
		t.Fatal(err)
	}
	v2, err = service.Publish(ctx, approver, tenantID, v2.ID, "ops-integration-publish-v2", PublishInput{Reason: "发布第二版", Rollout: rollout}, v2.RowVersion)
	if err != nil || v2.Status != "PUBLISHED" || v2.SupersedesID == nil || *v2.SupersedesID != v1.ID {
		t.Fatalf("published v2=%+v err=%v", v2, err)
	}
	if _, err = service.ResumeRollout(ctx, approver, tenantID, paused.ID, "ops-integration-resume-superseded", "不得恢复已被替代的投放", paused.RowVersion+1); !errors.Is(err, ErrStateConflict) {
		t.Fatalf("superseded rollout resume error=%v", err)
	}
	versions, err = service.List(ctx, approver, tenantID, configKey)
	if err != nil || len(versions) != 2 || versions[1].Status != "RETIRED" {
		t.Fatalf("post-v2 versions=%+v err=%v", versions, err)
	}
	rollback, created, err := service.Rollback(ctx, approver, tenantID, v1.ID, keys[4], "ops-integration-rollback", PublishInput{Reason: "回滚到稳定路由", Rollout: rollout})
	if err != nil || !created || rollback.VersionNo != 3 || rollback.Status != "PUBLISHED" || rollback.RollbackOfID == nil || *rollback.RollbackOfID != v1.ID {
		t.Fatalf("rollback=%+v created=%v err=%v", rollback, created, err)
	}
	replayedRollback, created, err := service.Rollback(ctx, approver, tenantID, v1.ID, keys[4], "ops-integration-rollback-replay", PublishInput{Reason: "回滚到稳定路由", Rollout: rollout})
	if err != nil || created || replayedRollback.ID != rollback.ID {
		t.Fatalf("rollback replay=%+v created=%v err=%v", replayedRollback, created, err)
	}
	rollouts, err = service.ListRollouts(ctx, approver, tenantID, configKey)
	if err != nil || len(rollouts) != 3 || rollouts[0].ConfigVersionID != rollback.ID || rollouts[0].Status != "ACTIVE" {
		t.Fatalf("final rollouts=%+v err=%v", rollouts, err)
	}
}