package impl import ( "context" "encoding/json" "errors" "fmt" "math" "strings" "testing" "time" flowentity "github.com/shiningrush/fastflow/pkg/entity" "github.com/shiningrush/fastflow/pkg/entity/run" "github.com/shiningrush/fastflow/pkg/mod" "github.com/stretchr/testify/require" "github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed" data_pipeline "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline" "github.com/tidbcloud/dataflow-service-ng/app/models/domain" data_pipeline_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/data_pipeline" "github.com/tidbcloud/dataflow-service-ng/app/repo/entity" "github.com/tidbcloud/dataflow-service-ng/app/workflow" "github.com/tidbcloud/dataflow-service-ng/pkg/lake" app_log "github.com/tidbcloud/dataflow-service-ng/pkg/log" "github.com/tidbcloud/pkgv2/bizerr" mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1" "go.uber.org/zap" "go.uber.org/zap/zaptest/observer" "gorm.io/driver/mysql" "gorm.io/gorm" gorm_logger "gorm.io/gorm/logger" ) const ( testClusterID = uint64(101) testTenantID = uint64(201) testProjectID = uint64(301) testUserID = uint64(401) testPipelineID = uint64(501) testSourcePassword = "test-source-password" testStagingSecret = "test-staging-secret" testSourceCipher = "encrypted-source-value" testStagingCipher = "encrypted-staging-value" ) var testNow = time.Date(2026, time.August, 2, 8, 9, 10, 987654321, time.FixedZone("test-zone", 8*60*60)) func TestCreateDataPipelinePersistsCompletePlanWithoutMutatingPlaintextInput(t *testing.T) { h := newModelHarness() req := validCreateRequest() original := cloneCreateRequest(req) resp, err := h.model.CreateDataPipeline(context.Background(), req) require.NoError(t, err) require.Equal(t, testPipelineID, resp.PipelineID) require.Equal(t, original, req, "Create must not rewrite caller-owned plaintext or filter slices") require.Len(t, h.repo.creates, 1) require.Empty(t, h.repo.saves) persisted := h.repo.creates[0] require.Equal(t, entity.ID(testPipelineID), persisted.ID) require.Equal(t, entity.ID(testTenantID), persisted.TenantID) require.Equal(t, entity.ID(testClusterID), persisted.ClusterID) require.Equal(t, entity.ID(testUserID), persisted.CreatorID) require.Equal(t, "test-pipeline", persisted.Name) require.Equal(t, entity.DataPipelineStatusCreating, persisted.Status) require.Equal(t, "standard-2", persisted.ChangefeedSpecification) require.Equal(t, uint16(1), persisted.Params.Version) require.Equal(t, entity.DataPipelineSyncModeFullAndIncremental, persisted.Params.SyncMode) require.Equal(t, "test-source-user", persisted.Params.Source.Username) require.Equal(t, testSourceCipher, persisted.Params.Source.PasswordCiphertext) require.Equal(t, []string{"test_app.*", "test_audit.orders"}, persisted.Params.Source.TableFilter) require.False(t, persisted.Params.Source.CaseSensitive) require.Equal(t, entity.DataPipelineStagingTypeCustomerManagedS3, persisted.Params.Staging.Type) require.Equal(t, "test-pipeline-bucket", persisted.Params.Staging.S3.Bucket) require.Equal(t, "customer/base", persisted.Params.Staging.S3.BasePrefix) require.Equal(t, "us-west-2", persisted.Params.Staging.S3.Region) require.Equal(t, "test-access-key", persisted.Params.Staging.S3.AccessKeyID) require.Equal(t, testStagingCipher, persisted.Params.Staging.S3.SecretAccessKeyCiphertext) require.Equal(t, entity.DataPipelineDestinationTypeTiDBLake, persisted.Params.Destination.Type) require.Equal(t, "test-warehouse", persisted.Params.Destination.Warehouse) require.Equal(t, domain.DataPipelineTargetNaming{ DatabasePrefix: "lake_", TablePrefix: "raw_", }, persisted.Params.Destination.TargetNaming) require.Equal(t, entity.DataPipelineExecutionPlan{ Version: 1, DestinationType: entity.DataPipelineDestinationTypeTiDBLake, Snapshot: entity.DataPipelineSnapshotPlan{ Engine: entity.DataPipelineSnapshotEngineDumpling, Format: entity.DataPipelineSnapshotFormatCSV, }, Incremental: &entity.DataPipelineIncrementalPlan{ Engine: entity.DataPipelineIncrementalEngineTiCDC, Format: entity.DataPipelineIncrementalFormatCanalJSON, }, }, persisted.ExecutionPlan) require.Equal(t, entity.DeliveryIntervalPlan{ Version: 1, EndToEndSeconds: 60, ProducerFlushIntervalSeconds: 30, ConsumerPollIntervalSeconds: 30, }, persisted.IntervalPlan) wantSnapshotTime := testNow.UTC().Truncate(time.Millisecond) require.Equal(t, uint64(wantSnapshotTime.UnixMilli())<<18, persisted.SnapshotTSO) require.Equal(t, "customer/base/tidb-data-pipelines/staging/101/501/", persisted.StagingRoot) require.Nil(t, persisted.LastError) // Every child identity is settled by the insert, not discovered later. require.Equal(t, "tidb-dp-export-501", persisted.Children.ExportName) require.Equal(t, "tidb-dp-changefeed-501", persisted.Children.ChangefeedName) require.NotEmpty(t, persisted.Children.LakeDataSourceName) require.NotEmpty(t, persisted.Children.LakeTaskName) require.NotZero(t, persisted.Children.ChangefeedID) // The two Lake IDs are Lake's to assign. require.NotEmpty(t, persisted.Children.ExportID) require.Empty(t, persisted.Observation) require.Equal(t, []string{testSourcePassword, testStagingSecret}, h.encryptor.encryptInputs) require.Len(t, h.validator.sourceInputs, 1) require.Equal(t, domain.DataPipelineSourceInput{ Username: "test-source-user", Password: testSourcePassword, TableFilter: []string{"test_app.*", "test_audit.orders"}, }, h.validator.sourceInputs[0]) require.Equal(t, []string{"standard-2"}, h.validator.specifications) require.Len(t, h.validator.stagingCalls, 1) require.Equal(t, data_pipeline.ValidationScope{ TenantID: testTenantID, ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, LakeOrgID: "7", }, h.validator.stagingCalls[0].scope) require.Len(t, h.workflow.calls, 1) assertWorkflowCall( t, h.workflow.calls[0], workflow.DagCreateDataPipeline, workflow.BusinessActionCreateDataPipeline, testClusterID, testPipelineID, ) require.Equal(t, []string{ "cluster.resolve", "validator.source", "validator.specification", "repo.allocate", "repo.allocate", "repo.allocate", "validator.staging", "encrypt.encrypt", "encrypt.encrypt", "repo.create", "workflow.run", }, h.events.values) serialized, marshalErr := json.Marshal(persisted) require.NoError(t, marshalErr) require.NotContains(t, string(serialized), testSourcePassword) require.NotContains(t, string(serialized), testStagingSecret) require.Contains(t, string(serialized), testSourceCipher) require.Contains(t, string(serialized), testStagingCipher) } func TestCreateDataPipelineRejectsPlanningFailuresBeforeExternalSideEffects(t *testing.T) { tests := map[string]func(*domain.CreateDataPipelineReq){ "infeasible end-to-end interval": func(req *domain.CreateDataPipelineReq) { req.Destination.FlushIntervalSeconds = 6 }, "flush interval exceeds signed planner range": func(req *domain.CreateDataPipelineReq) { req.Destination.FlushIntervalSeconds = math.MaxInt32 + 1 }, } for name, mutate := range tests { t.Run(name, func(t *testing.T) { h := newModelHarness() req := validCreateRequest() mutate(req) resp, err := h.model.CreateDataPipeline(context.Background(), req) require.Nil(t, resp) require.Error(t, err) require.Empty(t, h.validator.sourceInputs) require.Empty(t, h.validator.stagingCalls) require.Empty(t, h.encryptor.encryptInputs) require.Zero(t, h.repo.allocateCalls) require.Empty(t, h.repo.creates) require.Empty(t, h.workflow.calls) }) } } // TestCreateDataPipelineDoesNotPersistWhenStagingValidationFails pins that a // destination rejection happens before anything durable exists. func TestCreateDataPipelineDoesNotPersistWhenStagingValidationFails(t *testing.T) { h := newModelHarness() h.validator.stagingErr = bizerr.New( bizerr.ErrCodeParams, "LAKE_MULTI_DATABASE_NOT_SUPPORTED: test Lake contract supports one database", ) resp, err := h.model.CreateDataPipeline(context.Background(), validCreateRequest()) require.Nil(t, resp) require.Error(t, err) require.Contains(t, err.Error(), "LAKE_MULTI_DATABASE_NOT_SUPPORTED") require.Len(t, h.validator.stagingCalls, 1) require.Empty(t, h.encryptor.encryptInputs) require.Empty(t, h.repo.creates) require.Empty(t, h.workflow.calls) } func TestCreateDataPipelineExternalValidationErrorsNeverPersistCredentials(t *testing.T) { tests := map[string]func(*fakeExternalValidator){ "source validation": func(validator *fakeExternalValidator) { validator.sourceErr = bizerr.New(bizerr.ErrCodeParams, "test source validation failed") }, "changefeed specification validation": func(validator *fakeExternalValidator) { validator.specificationErr = bizerr.New(bizerr.ErrCodeParams, "test specification validation failed") }, "staging validation": func(validator *fakeExternalValidator) { validator.stagingErr = bizerr.New(bizerr.ErrCodeParams, "test staging validation failed") }, } for name, inject := range tests { t.Run(name, func(t *testing.T) { h := newModelHarness() inject(h.validator) req := validCreateRequest() original := cloneCreateRequest(req) resp, err := h.model.CreateDataPipeline(context.Background(), req) require.Nil(t, resp) require.Error(t, err) require.Equal(t, original, req) require.Empty(t, h.repo.creates) require.Empty(t, h.repo.saves) require.Empty(t, h.encryptor.encryptInputs) require.Empty(t, h.workflow.calls) require.NotContains(t, err.Error(), testSourcePassword) require.NotContains(t, err.Error(), testStagingSecret) }) } } func TestCreateDataPipelineWorkflowStartFailureLeavesRecoverableCreating(t *testing.T) { h := newModelHarness() h.workflow.err = errors.New("test workflow unavailable") core, logs := observer.New(zap.ErrorLevel) ctx := app_log.WithLogger(context.Background(), zap.New(core)) resp, err := h.model.CreateDataPipeline(ctx, validCreateRequest()) require.NoError(t, err) require.Equal(t, testPipelineID, resp.PipelineID) require.Len(t, h.repo.creates, 1) require.Equal(t, entity.DataPipelineStatusCreating, h.repo.creates[0].Status) // The row must stay transitional: persisting Failed here would turn a // transient workflow-store error into a permanent verdict on a Pipeline // whose children were never touched. require.Empty(t, h.repo.saves) require.NotEmpty(t, logs.All()) require.Equal(t, []string{ "cluster.resolve", "validator.source", "validator.specification", "repo.allocate", "repo.allocate", "repo.allocate", "validator.staging", "encrypt.encrypt", "encrypt.encrypt", "repo.create", "workflow.run", }, h.events.values) } func TestGetDataPipelineUsesTenantAndClusterScopeAndReturnsCompleteObservation(t *testing.T) { h := newModelHarness() pipeline := observedPipeline(testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusRunning) h.repo.getResult = pipeline resp, err := h.model.GetDataPipeline(context.Background(), &domain.GetDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, }) require.NoError(t, err) require.NotNil(t, resp) require.Equal(t, domain.DataPipelineEntity2Domain(pipeline), resp.DataPipeline.Pipeline) require.Equal(t, domain.DataPipelineChildResourceDetails{ Export: domain.DataPipelineResourceDetails{Status: "Completed", ErrorMessage: "test export warning"}, Changefeed: domain.DataPipelineResourceDetails{Status: "Running", ErrorMessage: "test changefeed warning"}, Lake: domain.DataPipelineLakeResourceDetails{ DataSourceStatus: "Ready", IntegrationStatus: "TaskRunning", ErrorMessage: "test Lake warning", }, }, resp.DataPipeline.ChildResources) require.Equal(t, domain.DataPipelineObservationDetails{ ChangefeedCheckpointTSO: pipeline.Observation.ChangefeedCheckpointTSO, ChangefeedCheckpointTime: pipeline.Observation.ChangefeedCheckpointAt, LakeResolvedTSO: pipeline.Observation.LakeResolvedTSO, LakeResolvedTime: pipeline.Observation.LakeResolvedAt, LatencySeconds: uint64(pipeline.Observation.LatencySeconds), ObservedAt: pipeline.Observation.LastSucceededAt, Stale: pipeline.Observation.Stale, Error: pipeline.Observation.PollError, }, resp.DataPipeline.Observation) require.NotSame(t, pipeline.Observation.ChangefeedCheckpointAt, resp.DataPipeline.Observation.ChangefeedCheckpointTime) require.NotSame(t, pipeline.Observation.LakeResolvedAt, resp.DataPipeline.Observation.LakeResolvedTime) require.NotSame(t, pipeline.Observation.PollError, resp.DataPipeline.Observation.Error) require.Len(t, h.repo.getOps, 1) query := queryPipelineSQL(t, h.repo.getOps[0]) require.Contains(t, query, "id IN (501)") require.Contains(t, query, "tenant_id = 201") require.Contains(t, query, "cluster_id = 101") } func TestListDataPipelinesScopesCountsAndRowsWithStatusAndPagination(t *testing.T) { h := newModelHarness() status := domain.DataPipelineStatusRunning wanted := observedPipeline(502, testTenantID, testClusterID, entity.DataPipelineStatusRunning) h.repo.countResult = 3 h.repo.listResult = []entity.DataPipeline{*wanted} resp, err := h.model.ListDataPipelines(context.Background(), &domain.ListDataPipelinesReq{ ClusterID: testClusterID, UserID: testUserID, Page: 2, PerPage: 2, Status: &status, }) require.NoError(t, err) require.Equal(t, uint32(3), resp.Total) require.Len(t, resp.DataPipelines, 1) require.Equal(t, wanted.ID.Uint64(), resp.DataPipelines[0].Pipeline.ID) require.Equal(t, wanted.Observation.ChangefeedCheckpointTSO, resp.DataPipelines[0].Observation.ChangefeedCheckpointTSO) require.Equal(t, wanted.Observation.LakeResolvedTSO, resp.DataPipelines[0].Observation.LakeResolvedTSO) require.Equal(t, uint64(wanted.Observation.LatencySeconds), resp.DataPipelines[0].Observation.LatencySeconds) require.Len(t, h.repo.countOps, 1) require.Len(t, h.repo.listOps, 1) countQuery := queryPipelineSQL(t, h.repo.countOps[0]) require.Contains(t, countQuery, "tenant_id = 201") require.Contains(t, countQuery, "cluster_id = 101") require.Contains(t, countQuery, "status IN ('running')") require.NotContains(t, countQuery, "LIMIT") require.NotContains(t, countQuery, "ORDER BY") listQuery := queryPipelineSQL(t, h.repo.listOps[0]) require.Contains(t, listQuery, "tenant_id = 201") require.Contains(t, listQuery, "cluster_id = 101") require.Contains(t, listQuery, "status IN ('running')") require.Contains(t, listQuery, "ORDER BY id DESC") require.Contains(t, listQuery, "LIMIT 2 OFFSET 2") } func TestDataPipelineDetailsClampsNegativeLatency(t *testing.T) { pipeline := observedPipeline(testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusRunning) pipeline.Observation.LatencySeconds = -1 require.Zero(t, dataPipelineDetails(pipeline).Observation.LatencySeconds) } func TestPauseDataPipelineStatusMatrixAndWorkflowContract(t *testing.T) { for _, status := range allPipelineStatuses() { t.Run(string(status), func(t *testing.T) { h := newModelHarness() h.repo.getResult = observedPipeline(testPipelineID, testTenantID, testClusterID, status) err := h.model.PauseDataPipeline(context.Background(), &domain.PauseDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, }) if status != entity.DataPipelineStatusRunning { require.Error(t, err) require.Empty(t, h.repo.saves) require.Empty(t, h.workflow.calls) return } require.NoError(t, err) require.Len(t, h.repo.saves, 1) require.Equal(t, entity.DataPipelineStatusPausing, h.repo.saves[0].Status) require.Nil(t, h.repo.saves[0].LastError) require.Len(t, h.workflow.calls, 1) assertWorkflowCall( t, h.workflow.calls[0], workflow.DagPauseDataPipeline, workflow.BusinessActionPauseDataPipeline, testClusterID, testPipelineID, ) require.Equal(t, []string{"cluster.resolve", "repo.get", "repo.save", "workflow.run"}, h.events.values) }) } t.Run("full-only running pipeline cannot pause", func(t *testing.T) { h := newModelHarness() pipeline := observedPipeline(testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusRunning) pipeline.Params.SyncMode = entity.DataPipelineSyncModeFull pipeline.ExecutionPlan.Incremental = nil h.repo.getResult = pipeline err := h.model.PauseDataPipeline(context.Background(), &domain.PauseDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, }) require.Error(t, err) require.Empty(t, h.repo.saves) require.Empty(t, h.workflow.calls) }) } func TestEditDataPipelineStatusMatrix(t *testing.T) { for _, status := range allPipelineStatuses() { // Failed is editable on purpose: replacing a revision the children // rejected is the only way back. if status == entity.DataPipelineStatusPaused || status == entity.DataPipelineStatusFailed { continue } t.Run(string(status), func(t *testing.T) { h := newModelHarness() h.repo.getResult = observedPipeline(testPipelineID, testTenantID, testClusterID, status) err := h.model.EditDataPipeline(context.Background(), validEditRequest()) require.Error(t, err) require.Empty(t, h.repo.saves) require.Empty(t, h.workflow.calls) require.Empty(t, h.encryptor.decryptInputs) }) } } func TestEditDataPipelineUpdatesOnlyEditableConfigurationAndDesiredRevision(t *testing.T) { h := newModelHarness() pipeline := observedPipeline(testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusPaused) pipeline.LastError = &entity.DataPipelineError{Code: "TEST_OLD_ERROR", Message: "test old error"} before := clonePipeline(pipeline) h.repo.getResult = pipeline req := validEditRequest() err := h.model.EditDataPipeline(context.Background(), req) require.NoError(t, err) // Edit is asynchronous: it persists the desired revision and hands the // application of it to the Edit DAG, so a rejected value fails while the // Pipeline is still editable rather than during a later Resume. require.Len(t, h.workflow.calls, 1) require.Equal(t, workflow.DagEditDataPipeline, h.workflow.calls[0].dagID) require.Len(t, h.repo.saves, 1) updated := h.repo.saves[0] require.Equal(t, entity.DataPipelineStatusEditing, updated.Status) require.Equal(t, "test-pipeline-edited", updated.Name) require.Equal(t, []string{"test_app.orders", "test_audit.*"}, updated.Params.Source.TableFilter) require.True(t, updated.Params.Source.CaseSensitive) require.Equal(t, "test-warehouse-edited", updated.Params.Destination.Warehouse) require.Equal(t, domain.DataPipelineTargetNaming{ DatabasePrefix: "archive_", TableSuffix: "_v2", }, updated.Params.Destination.TargetNaming) // Resizing is ScaleDataPipeline; Edit must leave the specification alone. require.Equal(t, before.ChangefeedSpecification, updated.ChangefeedSpecification) require.Equal(t, entity.DeliveryIntervalPlan{ Version: 1, EndToEndSeconds: 90, ProducerFlushIntervalSeconds: 45, ConsumerPollIntervalSeconds: 45, }, updated.IntervalPlan) require.Nil(t, updated.LastError) // Immutable/internal choices and credentials remain byte-for-byte stable. require.Equal(t, before.ID, updated.ID) require.Equal(t, before.TenantID, updated.TenantID) require.Equal(t, before.ClusterID, updated.ClusterID) require.Equal(t, before.CreatorID, updated.CreatorID) require.Equal(t, before.Params.Version, updated.Params.Version) require.Equal(t, before.Params.SyncMode, updated.Params.SyncMode) require.Equal(t, before.Params.Source.Username, updated.Params.Source.Username) require.Equal(t, before.Params.Source.PasswordCiphertext, updated.Params.Source.PasswordCiphertext) require.Equal(t, before.Params.Staging, updated.Params.Staging) require.Equal(t, before.Params.Destination.Type, updated.Params.Destination.Type) require.Equal(t, before.ExecutionPlan, updated.ExecutionPlan) require.Equal(t, before.SnapshotTSO, updated.SnapshotTSO) require.Equal(t, before.StagingRoot, updated.StagingRoot) require.Equal(t, before.Children, updated.Children) require.Equal(t, before.Observation, updated.Observation) require.Equal(t, []string{testSourceCipher, testStagingCipher}, h.encryptor.decryptInputs) require.Len(t, h.validator.sourceInputs, 1) require.Equal(t, domain.DataPipelineSourceInput{ Username: "test-source-user", Password: testSourcePassword, TableFilter: []string{"test_app.orders", "test_audit.*"}, CaseSensitive: true, }, h.validator.sourceInputs[0]) require.Empty(t, h.validator.specifications) require.Len(t, h.validator.stagingCalls, 1) require.Equal(t, data_pipeline.ValidationScope{ TenantID: testTenantID, ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, LakeOrgID: "7", }, h.validator.stagingCalls[0].scope) require.Equal(t, domain.DataPipelineStagingInput{ Type: entity.DataPipelineStagingTypeCustomerManagedS3, S3: &domain.DataPipelineS3StagingInput{ URI: "s3://test-pipeline-bucket/customer/base", Region: "us-west-2", AuthType: domain.DataPipelineS3AuthTypeAccessKey, AccessKeyID: "test-access-key", SecretAccessKey: testStagingSecret, }, }, h.validator.stagingCalls[0].staging) require.Equal(t, []string{ "cluster.resolve", "repo.get", "encrypt.decrypt", "encrypt.decrypt", "validator.source", "validator.staging", "repo.save", "workflow.run", }, h.events.values) } func TestEditDataPipelineExternalValidationErrorsDoNotSave(t *testing.T) { tests := map[string]func(*fakeExternalValidator){ "source": func(v *fakeExternalValidator) { v.sourceErr = bizerr.New(bizerr.ErrCodeParams, "test edited source invalid") }, "staging": func(v *fakeExternalValidator) { v.stagingErr = bizerr.New(bizerr.ErrCodeParams, "test edited staging invalid") }, } for name, inject := range tests { t.Run(name, func(t *testing.T) { h := newModelHarness() h.repo.getResult = observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusPaused, ) inject(h.validator) err := h.model.EditDataPipeline(context.Background(), validEditRequest()) require.Error(t, err) require.Empty(t, h.repo.saves) require.Empty(t, h.workflow.calls) }) } } func TestScaleDataPipelineDelegatesToTheChangefeedScaleWorkflow(t *testing.T) { h := newModelHarness() pipeline := observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusRunning, ) pipeline.ChangefeedSpecification = "standard-2" h.repo.getResult = pipeline err := h.model.ScaleDataPipeline(context.Background(), &domain.ScaleDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, ChangefeedSpecification: " standard-4 ", }) require.NoError(t, err) require.Len(t, h.changefeeds.requests, 1) require.Equal(t, &domain.ScaleBizChangefeedReq{ BizChangefeedID: pipeline.Children.ChangefeedID, SpecificationCode: "standard-4", UserID: testUserID, }, h.changefeeds.requests[0]) require.Equal(t, []string{"standard-4"}, h.validator.specifications) require.Len(t, h.repo.saves, 1) require.Equal(t, "standard-4", h.repo.saves[0].ChangefeedSpecification) // Scaling must not disturb the Pipeline lifecycle: the child Changefeed owns // its own Scaling state and reports progress through the observation model. require.Equal(t, entity.DataPipelineStatusRunning, h.repo.saves[0].Status) require.Empty(t, h.workflow.calls) require.Equal(t, []string{ "cluster.resolve", "repo.get", "validator.specification", "changefeed.scale", "repo.save", }, h.events.values) } func TestScaleDataPipelineRejectsUnscalablePipelines(t *testing.T) { t.Run("only a running pipeline can scale", func(t *testing.T) { for _, status := range allPipelineStatuses() { if status == entity.DataPipelineStatusRunning { continue } h := newModelHarness() h.repo.getResult = observedPipeline( testPipelineID, testTenantID, testClusterID, status, ) err := h.model.ScaleDataPipeline(context.Background(), &domain.ScaleDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, ChangefeedSpecification: "standard-4", }) require.Error(t, err, status) require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode()) require.Empty(t, h.changefeeds.requests) require.Empty(t, h.repo.saves) } }) t.Run("a full-only pipeline has no changefeed capacity", func(t *testing.T) { h := newModelHarness() pipeline := observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusRunning, ) pipeline.ExecutionPlan.Incremental = nil pipeline.Params.SyncMode = entity.DataPipelineSyncModeFull h.repo.getResult = pipeline err := h.model.ScaleDataPipeline(context.Background(), &domain.ScaleDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, ChangefeedSpecification: "standard-4", }) require.Error(t, err) require.Empty(t, h.changefeeds.requests) }) t.Run("an unchanged specification is a no-op", func(t *testing.T) { h := newModelHarness() pipeline := observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusRunning, ) pipeline.ChangefeedSpecification = "standard-2" h.repo.getResult = pipeline require.NoError(t, h.model.ScaleDataPipeline( context.Background(), &domain.ScaleDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, ChangefeedSpecification: "standard-2", })) require.Empty(t, h.changefeeds.requests) require.Empty(t, h.repo.saves) }) t.Run("a child failure leaves the pipeline specification untouched", func(t *testing.T) { h := newModelHarness() pipeline := observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusRunning, ) pipeline.ChangefeedSpecification = "standard-2" h.repo.getResult = pipeline h.changefeeds.err = bizerr.New(bizerr.ErrCodeParams, "test changefeed cannot scale") err := h.model.ScaleDataPipeline(context.Background(), &domain.ScaleDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, ChangefeedSpecification: "standard-4", }) require.Error(t, err) require.Empty(t, h.repo.saves) }) } // TestAllPipelineStatusesIsExhaustive guards the state matrices: every matrix // test iterates this helper, so a status missing from it is a status no public // action is ever exercised against. func TestAllPipelineStatusesIsExhaustive(t *testing.T) { require.ElementsMatch(t, []entity.DataPipelineStatus{ entity.DataPipelineStatusCreating, entity.DataPipelineStatusRunning, entity.DataPipelineStatusPausing, entity.DataPipelineStatusPaused, entity.DataPipelineStatusEditing, entity.DataPipelineStatusResuming, entity.DataPipelineStatusDeleting, entity.DataPipelineStatusFailed, entity.DataPipelineStatusCompleted, }, allPipelineStatuses()) } // TestRetryTargetCoversEveryLifecycleStage drives the table from lifecycleStages // itself, so a stage added to a workflow without a Retry mapping fails here // instead of producing a Pipeline an operator cannot rescue. func TestEditDataPipelineRecoversARejectedRevision(t *testing.T) { t.Run("a failed edit can be corrected", func(t *testing.T) { h := newModelHarness() pipeline := observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusFailed, ) pipeline.LastError = &entity.DataPipelineError{Code: "DATA_PIPELINE_CHILD_MISMATCH"} h.repo.getResult = pipeline require.NoError(t, h.model.EditDataPipeline(context.Background(), validEditRequest())) require.Len(t, h.repo.saves, 1) saved := h.repo.saves[0] require.Equal(t, entity.DataPipelineStatusEditing, saved.Status) require.Nil(t, saved.LastError) require.Len(t, h.workflow.calls, 1) require.Equal(t, workflow.DagEditDataPipeline, h.workflow.calls[0].dagID) }) } func TestValidateDataPipelineDestinationChecksConfigurationBeforeAnyResourceExists(t *testing.T) { valid := func() *domain.ValidateDataPipelineDestinationReq { return &domain.ValidateDataPipelineDestinationReq{ ClusterID: testClusterID, UserID: testUserID, SyncMode: domain.DataPipelineSyncModeFullAndIncremental, Destination: domain.DataPipelineDestinationInput{ Type: domain.DataPipelineDestinationTypeTiDBLake, LakeOrgID: "7", Warehouse: "test-warehouse", TargetNaming: domain.DataPipelineTargetNaming{ DatabasePrefix: "lake_", TablePrefix: "raw_", }, FlushIntervalSeconds: 60, }, } } h := newModelHarness() require.NoError(t, h.model.ValidateDataPipelineDestination(context.Background(), valid())) // Validation must not touch persistence or start any workflow. require.Empty(t, h.repo.saves) require.Empty(t, h.workflow.calls) tests := map[string]func(*domain.ValidateDataPipelineDestinationReq){ "unsupported destination": func(r *domain.ValidateDataPipelineDestinationReq) { r.Destination.Type = domain.DataPipelineDestinationTypeSnowflake }, "missing warehouse": func(r *domain.ValidateDataPipelineDestinationReq) { r.Destination.Warehouse = " " }, "impossible interval": func(r *domain.ValidateDataPipelineDestinationReq) { r.Destination.FlushIntervalSeconds = 1 }, "full mode rejects an interval": func(r *domain.ValidateDataPipelineDestinationReq) { r.SyncMode = domain.DataPipelineSyncModeFull }, } for name, inject := range tests { t.Run(name, func(t *testing.T) { h := newModelHarness() req := valid() inject(req) err := h.model.ValidateDataPipelineDestination(context.Background(), req) require.Error(t, err) require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode()) }) } } func TestResumeDataPipelineStatusMatrixAndWorkflowContract(t *testing.T) { for _, status := range allPipelineStatuses() { t.Run(string(status), func(t *testing.T) { h := newModelHarness() h.repo.getResult = observedPipeline(testPipelineID, testTenantID, testClusterID, status) err := h.model.ResumeDataPipeline(context.Background(), &domain.ResumeDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, }) if status != entity.DataPipelineStatusPaused { require.Error(t, err) require.Empty(t, h.repo.saves) require.Empty(t, h.workflow.calls) return } require.NoError(t, err) require.Len(t, h.repo.saves, 1) require.Equal(t, entity.DataPipelineStatusResuming, h.repo.saves[0].Status) // Resume must start at the first stage of the Resume DAG; the applying // stages belong to Edit and the Resume DAG would skip straight past them. require.Len(t, h.workflow.calls, 1) assertWorkflowCall( t, h.workflow.calls[0], workflow.DagResumeDataPipeline, workflow.BusinessActionResumeDataPipeline, testClusterID, testPipelineID, ) require.Equal(t, []string{"cluster.resolve", "repo.get", "repo.save", "workflow.run"}, h.events.values) }) } t.Run("full-only paused pipeline cannot resume", func(t *testing.T) { h := newModelHarness() pipeline := observedPipeline(testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusPaused) pipeline.Params.SyncMode = entity.DataPipelineSyncModeFull pipeline.ExecutionPlan.Incremental = nil h.repo.getResult = pipeline err := h.model.ResumeDataPipeline(context.Background(), &domain.ResumeDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, }) require.Error(t, err) require.Empty(t, h.repo.saves) require.Empty(t, h.workflow.calls) }) } func TestDeleteDataPipelineStatusMatrixAndWorkflowContract(t *testing.T) { for _, status := range allPipelineStatuses() { t.Run(string(status), func(t *testing.T) { h := newModelHarness() h.repo.getResult = observedPipeline(testPipelineID, testTenantID, testClusterID, status) err := h.model.DeleteDataPipeline(context.Background(), &domain.DeleteDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, }) if status == entity.DataPipelineStatusDeleting { // Already deleting: refused, so nothing is written and no // second workflow races the one already running. require.Error(t, err) require.Empty(t, h.repo.saves) require.Empty(t, h.workflow.calls) require.Equal(t, []string{"cluster.resolve", "repo.get"}, h.events.values) return } require.NoError(t, err) require.Len(t, h.repo.saves, 1) require.Equal(t, entity.DataPipelineStatusDeleting, h.repo.saves[0].Status) require.Len(t, h.workflow.calls, 1) assertWorkflowCall( t, h.workflow.calls[0], workflow.DagDeleteDataPipeline, workflow.BusinessActionDeleteDataPipeline, testClusterID, testPipelineID, ) require.Equal(t, []string{"cluster.resolve", "repo.get", "repo.save", "workflow.run"}, h.events.values) }) } } func TestValidationAPIsKeepSecretsAtTheExternalValidatorBoundary(t *testing.T) { t.Run("source connection", func(t *testing.T) { h := newModelHarness() core, logs := observer.New(zap.DebugLevel) ctx := app_log.WithLogger(context.Background(), zap.New(core)) req := &domain.ValidateDataPipelineSourceConnectionReq{ ClusterID: testClusterID, UserID: testUserID, Username: "test-source-user", Password: testSourcePassword, } original := *req err := h.model.ValidateDataPipelineSourceConnection(ctx, req) require.NoError(t, err) require.Equal(t, original, *req) // No filter: this endpoint checks the account, not the selection. require.Equal(t, []domain.DataPipelineSourceInput{{ Username: "test-source-user", Password: testSourcePassword, }}, h.validator.sourceInputs) assertValidationHasNoPersistenceOrSecretLogs(t, h, logs, testSourcePassword) }) t.Run("source validator error is not expanded with request credentials", func(t *testing.T) { h := newModelHarness() h.validator.sourceErr = bizerr.New(bizerr.ErrCodeParams, "test source unavailable") core, logs := observer.New(zap.DebugLevel) ctx := app_log.WithLogger(context.Background(), zap.New(core)) err := h.model.ValidateDataPipelineSourceConnection(ctx, &domain.ValidateDataPipelineSourceConnectionReq{ ClusterID: testClusterID, UserID: testUserID, Username: "test-source-user", Password: testSourcePassword, }) require.Error(t, err) require.NotContains(t, err.Error(), testSourcePassword) assertValidationHasNoPersistenceOrSecretLogs(t, h, logs, testSourcePassword) }) t.Run("S3 staging", func(t *testing.T) { h := newModelHarness() core, logs := observer.New(zap.DebugLevel) ctx := app_log.WithLogger(context.Background(), zap.New(core)) req := &domain.ValidateDataPipelineStagingReq{ ClusterID: testClusterID, UserID: testUserID, LakeOrgID: "7", Staging: validCreateRequest().Staging, } original := cloneStagingInput(req.Staging) err := h.model.ValidateDataPipelineStaging(ctx, req) require.NoError(t, err) require.Equal(t, original, req.Staging) require.Len(t, h.validator.stagingCalls, 1) require.Equal(t, data_pipeline.ValidationScope{ TenantID: testTenantID, ClusterID: testClusterID, UserID: testUserID, LakeOrgID: "7", }, h.validator.stagingCalls[0].scope) require.Equal(t, original, h.validator.stagingCalls[0].staging) assertValidationHasNoPersistenceOrSecretLogs(t, h, logs, testStagingSecret) }) t.Run("S3 validator error is not expanded with request credentials", func(t *testing.T) { h := newModelHarness() h.validator.stagingErr = bizerr.New(bizerr.ErrCodeParams, "test S3 unavailable") core, logs := observer.New(zap.DebugLevel) ctx := app_log.WithLogger(context.Background(), zap.New(core)) err := h.model.ValidateDataPipelineStaging(ctx, &domain.ValidateDataPipelineStagingReq{ ClusterID: testClusterID, UserID: testUserID, LakeOrgID: "7", Staging: validCreateRequest().Staging, }) require.Error(t, err) require.NotContains(t, err.Error(), testStagingSecret) assertValidationHasNoPersistenceOrSecretLogs(t, h, logs, testStagingSecret) }) } func TestCreateDataPipelineFailureBoundariesDoNotStartWorkflow(t *testing.T) { t.Run("runtime dependency unavailable", func(t *testing.T) { h := newModelHarness() h.model.(*dataPipelineModel).encryptor = nil resp, err := h.model.CreateDataPipeline(context.Background(), validCreateRequest()) require.Nil(t, resp) require.Error(t, err) require.Empty(t, h.validator.sourceInputs) require.Zero(t, h.repo.allocateCalls) }) t.Run("invalid environment cannot allocate a staging path", func(t *testing.T) { h := newModelHarness() h.model.(*dataPipelineModel).runtime.Environment = "../staging" resp, err := h.model.CreateDataPipeline(context.Background(), validCreateRequest()) require.Nil(t, resp) require.Error(t, err) // The Pipeline, Export and Changefeed IDs are all allocated up front. require.Equal(t, 3, h.repo.allocateCalls) require.Empty(t, h.validator.stagingCalls) require.Empty(t, h.repo.creates) }) t.Run("invalid snapshot clock", func(t *testing.T) { h := newModelHarness() h.clock.now = time.UnixMilli(-1) resp, err := h.model.CreateDataPipeline(context.Background(), validCreateRequest()) require.Nil(t, resp) require.Error(t, err) require.Len(t, h.validator.stagingCalls, 0) require.Empty(t, h.encryptor.encryptInputs) require.Empty(t, h.repo.creates) }) t.Run("source credential encryption", func(t *testing.T) { h := newModelHarness() h.encryptor.encryptErrors = map[string]bizerr.BizError{ testSourcePassword: bizerr.New(bizerr.ErrCodeUnknown, "test source encryption failed"), } resp, err := h.model.CreateDataPipeline(context.Background(), validCreateRequest()) require.Nil(t, resp) require.Error(t, err) require.Equal(t, []string{testSourcePassword}, h.encryptor.encryptInputs) require.Empty(t, h.repo.creates) }) t.Run("staging credential encryption", func(t *testing.T) { h := newModelHarness() h.encryptor.encryptErrors = map[string]bizerr.BizError{ testStagingSecret: bizerr.New(bizerr.ErrCodeUnknown, "test staging encryption failed"), } resp, err := h.model.CreateDataPipeline(context.Background(), validCreateRequest()) require.Nil(t, resp) require.Error(t, err) require.Equal(t, []string{testSourcePassword, testStagingSecret}, h.encryptor.encryptInputs) require.Empty(t, h.repo.creates) }) t.Run("repository create", func(t *testing.T) { h := newModelHarness() h.repo.createErr = bizerr.New(bizerr.ErrCodeUnknown, "test create persistence failed") resp, err := h.model.CreateDataPipeline(context.Background(), validCreateRequest()) require.Nil(t, resp) require.Error(t, err) require.Len(t, h.repo.creates, 1) require.Empty(t, h.workflow.calls) }) } func TestReadDataPipelineFailureBoundariesAndTotalClamp(t *testing.T) { t.Run("Get validates request", func(t *testing.T) { h := newModelHarness() resp, err := h.model.GetDataPipeline(context.Background(), nil) require.Nil(t, resp) require.Error(t, err) require.Empty(t, h.clusters.calls) }) t.Run("Get propagates scoped repository error", func(t *testing.T) { h := newModelHarness() h.repo.getErr = bizerr.New(bizerr.ErrCodeRecordNotFound, "test pipeline not found") resp, err := h.model.GetDataPipeline(context.Background(), &domain.GetDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, }) require.Nil(t, resp) require.Error(t, err) require.Len(t, h.repo.getOps, 1) }) t.Run("List validates pagination", func(t *testing.T) { h := newModelHarness() resp, err := h.model.ListDataPipelines(context.Background(), &domain.ListDataPipelinesReq{ ClusterID: testClusterID, Page: 1, PerPage: maxPageSize + 1, }) require.Nil(t, resp) require.Error(t, err) require.Empty(t, h.clusters.calls) }) t.Run("List propagates count error", func(t *testing.T) { h := newModelHarness() h.repo.countErr = bizerr.New(bizerr.ErrCodeUnknown, "test count failed") resp, err := h.model.ListDataPipelines(context.Background(), &domain.ListDataPipelinesReq{ ClusterID: testClusterID, Page: 1, PerPage: 20, }) require.Nil(t, resp) require.Error(t, err) require.Empty(t, h.repo.listOps) }) t.Run("List propagates row query error", func(t *testing.T) { h := newModelHarness() h.repo.listErr = bizerr.New(bizerr.ErrCodeUnknown, "test list failed") resp, err := h.model.ListDataPipelines(context.Background(), &domain.ListDataPipelinesReq{ ClusterID: testClusterID, Page: 1, PerPage: 20, }) require.Nil(t, resp) require.Error(t, err) require.Len(t, h.repo.countOps, 1) }) t.Run("List clamps total to public uint32 range", func(t *testing.T) { h := newModelHarness() h.repo.countResult = int64(math.MaxUint32) + 1 resp, err := h.model.ListDataPipelines(context.Background(), &domain.ListDataPipelinesReq{ ClusterID: testClusterID, Page: 1, PerPage: 20, }) require.NoError(t, err) require.Equal(t, uint32(math.MaxUint32), resp.Total) }) } func TestScopeFeatureAndDependencyGuards(t *testing.T) { t.Run("cluster resolver error", func(t *testing.T) { h := newModelHarness() h.clusters.err = bizerr.New(bizerr.ErrCodeUnknown, "test cluster lookup failed") _, _, err := h.model.(*dataPipelineModel).resolveScope(context.Background(), testClusterID) require.Error(t, err) }) invalidClusters := map[string]*mgmt.Cluster{ "nil cluster": nil, "mismatched cluster ID": {ClusterId: testClusterID + 1, OrganizationId: testTenantID, DeployType: mgmt.DeployType_DEPLOY_TYPE_NEXTGEN}, "missing organization ID": {ClusterId: testClusterID, DeployType: mgmt.DeployType_DEPLOY_TYPE_NEXTGEN}, "non-Premium cluster": {ClusterId: testClusterID, OrganizationId: testTenantID, DeployType: mgmt.DeployType_DEPLOY_TYPE_DEDICATED}, } for name, cluster := range invalidClusters { t.Run(name, func(t *testing.T) { h := newModelHarness() h.clusters.cluster = cluster _, _, err := h.model.(*dataPipelineModel).resolveScope(context.Background(), testClusterID) require.Error(t, err) }) } t.Run("nil resolver and zero scope", func(t *testing.T) { h := newModelHarness() model := h.model.(*dataPipelineModel) model.clusters = nil _, _, err := model.resolveScope(context.Background(), testClusterID) require.Error(t, err) _, err = model.loadScoped(context.Background(), 0, testPipelineID) require.Error(t, err) }) t.Run("nil validator is rejected", func(t *testing.T) { h := newModelHarness() h.model.(*dataPipelineModel).validator = nil err := h.model.(*dataPipelineModel).ensureValidator() require.Error(t, err) }) } func TestLifecycleFailureBoundaries(t *testing.T) { t.Run("nil requests", func(t *testing.T) { h := newModelHarness() require.Error(t, h.model.PauseDataPipeline(context.Background(), nil)) require.Error(t, h.model.ResumeDataPipeline(context.Background(), nil)) require.Error(t, h.model.DeleteDataPipeline(context.Background(), nil)) require.Error(t, h.model.EditDataPipeline(context.Background(), nil)) require.Empty(t, h.clusters.calls) }) t.Run("transition persistence fails before workflow", func(t *testing.T) { h := newModelHarness() h.repo.getResult = observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusRunning, ) h.repo.saveErr = bizerr.New(bizerr.ErrCodeUnknown, "test transition save failed") err := h.model.PauseDataPipeline(context.Background(), &domain.PauseDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, }) require.Error(t, err) require.Len(t, h.repo.saves, 1) require.Empty(t, h.workflow.calls) }) t.Run("workflow failure leaves the transition recoverable", func(t *testing.T) { h := newModelHarness() h.repo.getResult = observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusPaused, ) h.workflow.err = errors.New("test resume workflow unavailable") core, logs := observer.New(zap.ErrorLevel) ctx := app_log.WithLogger(context.Background(), zap.New(core)) err := h.model.ResumeDataPipeline(ctx, &domain.ResumeDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, }) // The durable Resuming row is what a re-issued Resume keys on, so a lost // DAG submission must not be reported or persisted as a terminal failure. require.NoError(t, err) require.Len(t, h.repo.saves, 1) require.Equal(t, entity.DataPipelineStatusResuming, h.repo.saves[0].Status) require.Nil(t, h.repo.saves[0].LastError) require.NotEmpty(t, logs.All()) }) t.Run("scoped load error stops delete", func(t *testing.T) { h := newModelHarness() h.repo.getErr = bizerr.New(bizerr.ErrCodeRecordNotFound, "test pipeline missing") err := h.model.DeleteDataPipeline(context.Background(), &domain.DeleteDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, }) require.Error(t, err) require.Empty(t, h.repo.saves) require.Empty(t, h.workflow.calls) }) } func TestEditDataPipelinePlannerCredentialAndPersistenceFailures(t *testing.T) { tests := map[string]func(*modelHarness, *domain.EditDataPipelineReq){ "invalid interval": func(_ *modelHarness, req *domain.EditDataPipelineReq) { req.Destination.FlushIntervalSeconds = 6 }, "source decrypt": func(h *modelHarness, _ *domain.EditDataPipelineReq) { h.encryptor.decryptErrors = map[string]bizerr.BizError{ testSourceCipher: bizerr.New(bizerr.ErrCodeUnknown, "test source decrypt failed"), } }, "staging decrypt": func(h *modelHarness, _ *domain.EditDataPipelineReq) { h.encryptor.decryptErrors = map[string]bizerr.BizError{ testStagingCipher: bizerr.New(bizerr.ErrCodeUnknown, "test staging decrypt failed"), } }, "invalid persisted staging": func(h *modelHarness, _ *domain.EditDataPipelineReq) { h.repo.getResult.Params.Staging.S3.Bucket = "test bucket with spaces" }, "repository save": func(h *modelHarness, _ *domain.EditDataPipelineReq) { h.repo.saveErr = bizerr.New(bizerr.ErrCodeUnknown, "test edit save failed") }, } for name, inject := range tests { t.Run(name, func(t *testing.T) { h := newModelHarness() h.repo.getResult = observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusPaused, ) req := validEditRequest() inject(h, req) err := h.model.EditDataPipeline(context.Background(), req) require.Error(t, err) require.Empty(t, h.workflow.calls) }) } } func TestValidationAPIGuards(t *testing.T) { t.Run("source request", func(t *testing.T) { h := newModelHarness() require.Error(t, h.model.ValidateDataPipelineSourceConnection(context.Background(), nil)) require.Empty(t, h.clusters.calls) }) t.Run("source validator dependency", func(t *testing.T) { h := newModelHarness() h.model.(*dataPipelineModel).validator = nil err := h.model.ValidateDataPipelineSourceConnection( context.Background(), &domain.ValidateDataPipelineSourceConnectionReq{ ClusterID: testClusterID, Username: "test-source-user", Password: testSourcePassword, }, ) require.Error(t, err) }) t.Run("staging request", func(t *testing.T) { h := newModelHarness() require.Error(t, h.model.ValidateDataPipelineStaging(context.Background(), nil)) require.Empty(t, h.clusters.calls) }) t.Run("staging validator dependency", func(t *testing.T) { h := newModelHarness() h.model.(*dataPipelineModel).validator = nil err := h.model.ValidateDataPipelineStaging(context.Background(), &domain.ValidateDataPipelineStagingReq{ ClusterID: testClusterID, UserID: testUserID, Staging: validCreateRequest().Staging, }) require.Error(t, err) }) } func TestRequestValidationGuards(t *testing.T) { createTests := map[string]func(*domain.CreateDataPipelineReq){ "missing IDs": func(req *domain.CreateDataPipelineReq) { req.ClusterID = 0 }, "missing name": func(req *domain.CreateDataPipelineReq) { req.Name = " " }, "long name": func(req *domain.CreateDataPipelineReq) { req.Name = strings.Repeat("x", maxPipelineNameLength+1) }, "missing source username": func(req *domain.CreateDataPipelineReq) { req.Source.Username = " " }, "missing source password": func(req *domain.CreateDataPipelineReq) { req.Source.Password = "" }, "missing warehouse": func(req *domain.CreateDataPipelineReq) { req.Destination.Warehouse = " " }, "missing incremental specification": func(req *domain.CreateDataPipelineReq) { req.ChangefeedSpecification = " " }, } for name, mutate := range createTests { t.Run("Create/"+name, func(t *testing.T) { req := validCreateRequest() mutate(req) require.Error(t, validateCreateRequest(req)) }) } editTests := map[string]func(*domain.EditDataPipelineReq){ "missing IDs": func(req *domain.EditDataPipelineReq) { req.PipelineID = 0 }, "missing name": func(req *domain.EditDataPipelineReq) { req.Name = " " }, "long name": func(req *domain.EditDataPipelineReq) { req.Name = strings.Repeat("x", maxPipelineNameLength+1) }, "missing warehouse": func(req *domain.EditDataPipelineReq) { req.Destination.Warehouse = " " }, } for name, mutate := range editTests { t.Run("Edit/"+name, func(t *testing.T) { req := validEditRequest() mutate(req) require.Error(t, validateEditRequest(req)) }) } stagingTests := map[string]domain.DataPipelineStagingInput{ "unsupported staging": {Type: domain.DataPipelineStagingTypeTiDBManaged}, "missing S3": {Type: domain.DataPipelineStagingTypeCustomerManagedS3}, "missing credentials": { Type: domain.DataPipelineStagingTypeCustomerManagedS3, S3: &domain.DataPipelineS3StagingInput{URI: "s3://test-pipeline-bucket"}, }, "invalid URI": { Type: domain.DataPipelineStagingTypeCustomerManagedS3, S3: &domain.DataPipelineS3StagingInput{ URI: "https://test-pipeline-bucket", Region: "us-west-2", AuthType: domain.DataPipelineS3AuthTypeAccessKey, AccessKeyID: "test-access-key", SecretAccessKey: testStagingSecret, }, }, } for name, staging := range stagingTests { t.Run("Staging/"+name, func(t *testing.T) { _, _, err := validateStagingInput(staging) require.Error(t, err) }) } h := newModelHarness() require.Equal(t, "dataPipelineModel{[REDACTED]}", h.model.(*dataPipelineModel).String()) require.Nil(t, clonePipelineError(nil)) } // fakeScalableChangefeedModel only implements the one method the Pipeline model // calls. The embedded nil interface makes any other call panic loudly instead of // silently succeeding. type fakeScalableChangefeedModel struct { biz_changefeed.BizChangefeedModel events *eventLog requests []*domain.ScaleBizChangefeedReq err bizerr.BizError adminProfile *domain.DescribeAdminProfileResp adminProfileErr bizerr.BizError } func (f *fakeScalableChangefeedModel) DescribeAdminProfile( _ context.Context, _ *domain.DescribeAdminProfileReq, ) (*domain.DescribeAdminProfileResp, bizerr.BizError) { if f.adminProfileErr != nil { return nil, f.adminProfileErr } if f.adminProfile != nil { return f.adminProfile, nil } return &domain.DescribeAdminProfileResp{ ProviderType: "aws", CloudIAMIdentity: "arn:aws:iam::111122223333:role/test-changefeed", CloudAccountID: "111122223333", AWSExternalID: "test-external-id", }, nil } func (f *fakeScalableChangefeedModel) ScaleBizChangefeed( _ context.Context, req *domain.ScaleBizChangefeedReq, ) (*domain.ScaleBizChangefeedResp, bizerr.BizError) { f.requests = append(f.requests, req) f.events.add("changefeed.scale") if f.err != nil { return nil, f.err } return &domain.ScaleBizChangefeedResp{WorkflowID: "test-scale-workflow"}, nil } type modelHarness struct { model data_pipeline.Model events *eventLog repo *fakeDataPipelineRepo clusters *fakeClusterResolver changefeeds *fakeScalableChangefeedModel validator *fakeExternalValidator encryptor *fakeEncryptor workflow *fakeWorkflow clock *fakeClock } func newModelHarness() *modelHarness { events := new(eventLog) repo := &fakeDataPipelineRepo{nextID: entity.ID(testPipelineID), events: events} clusters := &fakeClusterResolver{ cluster: &mgmt.Cluster{ ClusterId: testClusterID, ProjectId: testProjectID, OrganizationId: testTenantID, CloudProvider: "aws", Region: "us-west-2", DeployType: mgmt.DeployType_DEPLOY_TYPE_NEXTGEN, }, events: events, } validator := &fakeExternalValidator{events: events} encryptor := &fakeEncryptor{ encryptResults: map[string]string{ testSourcePassword: testSourceCipher, testStagingSecret: testStagingCipher, }, decryptResults: map[string]string{ testSourceCipher: testSourcePassword, testStagingCipher: testStagingSecret, }, events: events, } workflowModel := &fakeWorkflow{events: events} clock := &fakeClock{now: testNow} runtime := RuntimeConfig{ Environment: "staging", } changefeeds := &fakeScalableChangefeedModel{events: events} return &modelHarness{ model: NewDataPipelineModel( repo, clusters, changefeeds, lake.NewFakeClient(), validator, encryptor, workflowModel, runtime, clock, ), events: events, repo: repo, clusters: clusters, changefeeds: changefeeds, validator: validator, encryptor: encryptor, workflow: workflowModel, clock: clock, } } func validCreateRequest() *domain.CreateDataPipelineReq { return &domain.CreateDataPipelineReq{ ClusterID: testClusterID, UserID: testUserID, Name: " test-pipeline ", Source: domain.DataPipelineSourceInput{ Username: "test-source-user", Password: testSourcePassword, TableFilter: []string{"test_app.*", "test_audit.orders"}, }, Destination: domain.DataPipelineDestinationInput{ Type: domain.DataPipelineDestinationTypeTiDBLake, LakeOrgID: " 7 ", Warehouse: " test-warehouse ", TargetNaming: domain.DataPipelineTargetNaming{ DatabasePrefix: "lake_", TablePrefix: "raw_", }, FlushIntervalSeconds: 60, }, Staging: domain.DataPipelineStagingInput{ Type: domain.DataPipelineStagingTypeCustomerManagedS3, S3: &domain.DataPipelineS3StagingInput{ URI: " s3://test-pipeline-bucket/customer//base/ ", Region: "us-west-2", AuthType: domain.DataPipelineS3AuthTypeAccessKey, AccessKeyID: "test-access-key", SecretAccessKey: testStagingSecret, }, }, SyncMode: domain.DataPipelineSyncModeFullAndIncremental, ChangefeedSpecification: " standard-2 ", } } func validEditRequest() *domain.EditDataPipelineReq { return &domain.EditDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, Name: " test-pipeline-edited ", Source: domain.DataPipelineEditableSourceInput{ TableFilter: []string{"test_app.orders", "test_audit.*"}, CaseSensitive: true, }, Destination: domain.DataPipelineEditableDestinationInput{ Warehouse: " test-warehouse-edited ", TargetNaming: domain.DataPipelineTargetNaming{ DatabasePrefix: "archive_", TableSuffix: "_v2", }, FlushIntervalSeconds: 90, }, } } func observedPipeline( id, tenantID, clusterID uint64, status entity.DataPipelineStatus, ) *entity.DataPipeline { createdAt := testNow.UTC().Add(-2 * time.Hour) updatedAt := createdAt.Add(time.Hour) lastAttempt := updatedAt.Add(31 * time.Minute) lastSucceeded := updatedAt.Add(30 * time.Minute) checkpointAt := lastSucceeded.Add(-20 * time.Second) lakeResolvedAt := lastSucceeded.Add(-50 * time.Second) errorAt := lastAttempt return &entity.DataPipeline{ Versioned: entity.Versioned{Versioning: 7}, ID: entity.ID(id), TenantID: entity.ID(tenantID), ClusterID: entity.ID(clusterID), CreatorID: entity.ID(testUserID), CreatedAt: createdAt, UpdatedAt: &updatedAt, Name: "test-pipeline-existing", Status: status, ChangefeedSpecification: "standard-2", Params: entity.DataPipelineParams{ Version: 1, SyncMode: entity.DataPipelineSyncModeFullAndIncremental, Source: entity.DataPipelineSourceConfig{ Username: "test-source-user", PasswordCiphertext: testSourceCipher, TableFilter: []string{"test_app.*", "test_audit.orders"}, }, Staging: entity.DataPipelineStagingConfig{ Type: entity.DataPipelineStagingTypeCustomerManagedS3, S3: entity.DataPipelineS3Config{ Bucket: "test-pipeline-bucket", BasePrefix: "customer/base", Region: "us-west-2", AccessKeyID: "test-access-key", SecretAccessKeyCiphertext: testStagingCipher, }, }, Destination: entity.DataPipelineDestinationConfig{ Type: entity.DataPipelineDestinationTypeTiDBLake, Warehouse: "test-warehouse", LakeOrgID: "7", TargetNaming: entity.DataPipelineTargetNaming{ DatabasePrefix: "lake_", TablePrefix: "raw_", }, }, }, ExecutionPlan: entity.DataPipelineExecutionPlan{ Version: 1, DestinationType: entity.DataPipelineDestinationTypeTiDBLake, Snapshot: entity.DataPipelineSnapshotPlan{ Engine: entity.DataPipelineSnapshotEngineDumpling, Format: entity.DataPipelineSnapshotFormatCSV, }, Incremental: &entity.DataPipelineIncrementalPlan{ Engine: entity.DataPipelineIncrementalEngineTiCDC, Format: entity.DataPipelineIncrementalFormatCanalJSON, }, }, IntervalPlan: entity.DeliveryIntervalPlan{ Version: 1, EndToEndSeconds: 60, ProducerFlushIntervalSeconds: 30, ConsumerPollIntervalSeconds: 30, }, SnapshotTSO: uint64(createdAt.UnixMilli()) << 18, StagingRoot: fmt.Sprintf("customer/base/tidb-data-pipelines/staging/%d/%d/", clusterID, id), Children: entity.DataPipelineChildResources{ ExportID: fmt.Sprintf("test-export-%d", id), ChangefeedID: id + 100, }, Observation: entity.DataPipelineObservation{ LastAttemptAt: &lastAttempt, LastSucceededAt: &lastSucceeded, ExportStatus: "Completed", ExportErrorMessage: "test export warning", ChangefeedStatus: "Running", ChangefeedErrorMessage: "test changefeed warning", LakeDataSourceStatus: "Ready", LakeIntegrationStatus: "TaskRunning", LakeErrorMessage: "test Lake warning", ChangefeedCheckpointTSO: uint64(checkpointAt.UnixMilli()) << 18, ChangefeedCheckpointAt: &checkpointAt, LakeResolvedTSO: uint64(lakeResolvedAt.UnixMilli()) << 18, LakeResolvedAt: &lakeResolvedAt, LatencySeconds: 30, Stale: true, PollError: &entity.DataPipelineError{ Code: "TEST_OBSERVE_FAILED", Message: "test observation failed", Retryable: true, OccurredAt: &errorAt, }, }, } } func allPipelineStatuses() []entity.DataPipelineStatus { return []entity.DataPipelineStatus{ entity.DataPipelineStatusCreating, entity.DataPipelineStatusRunning, entity.DataPipelineStatusPausing, entity.DataPipelineStatusPaused, entity.DataPipelineStatusEditing, entity.DataPipelineStatusResuming, entity.DataPipelineStatusDeleting, entity.DataPipelineStatusFailed, entity.DataPipelineStatusCompleted, } } func cloneCreateRequest(req *domain.CreateDataPipelineReq) *domain.CreateDataPipelineReq { if req == nil { return nil } cloned := *req cloned.Source.TableFilter = append([]string(nil), req.Source.TableFilter...) cloned.Staging = cloneStagingInput(req.Staging) return &cloned } func cloneStagingInput(staging domain.DataPipelineStagingInput) domain.DataPipelineStagingInput { cloned := staging if staging.S3 != nil { s3 := *staging.S3 cloned.S3 = &s3 } return cloned } func clonePipeline(pipeline *entity.DataPipeline) *entity.DataPipeline { if pipeline == nil { return nil } cloned := *pipeline cloned.Params.Source.TableFilter = append([]string(nil), pipeline.Params.Source.TableFilter...) if pipeline.ExecutionPlan.Incremental != nil { incremental := *pipeline.ExecutionPlan.Incremental cloned.ExecutionPlan.Incremental = &incremental } cloned.UpdatedAt = cloneTime(pipeline.UpdatedAt) cloned.LastError = cloneEntityPipelineError(pipeline.LastError) cloned.Observation.LastAttemptAt = cloneTime(pipeline.Observation.LastAttemptAt) cloned.Observation.LastSucceededAt = cloneTime(pipeline.Observation.LastSucceededAt) cloned.Observation.ChangefeedCheckpointAt = cloneTime(pipeline.Observation.ChangefeedCheckpointAt) cloned.Observation.LakeResolvedAt = cloneTime(pipeline.Observation.LakeResolvedAt) cloned.Observation.PollError = cloneEntityPipelineError(pipeline.Observation.PollError) return &cloned } func cloneEntityPipelineError(pipelineErr *entity.DataPipelineError) *entity.DataPipelineError { if pipelineErr == nil { return nil } cloned := *pipelineErr cloned.OccurredAt = cloneTime(pipelineErr.OccurredAt) return &cloned } func assertWorkflowCall( t *testing.T, call workflowCall, dagID string, action workflow.BusinessAction, clusterID, pipelineID uint64, ) { t.Helper() require.Equal(t, dagID, call.dagID) require.Equal(t, map[string]string{ workflow.KeyDataPipelineID: fmt.Sprintf("%d", pipelineID), }, call.specVars) require.Equal(t, map[string]string{ workflow.ClusterID: fmt.Sprintf("%d", clusterID), workflow.BusinessTypeKey: string(workflow.BusinessTypeDataPipeline), workflow.BusinessActionKey: string(action), workflow.BusinessID: fmt.Sprintf("%d", pipelineID), }, call.tags) } func assertValidationHasNoPersistenceOrSecretLogs( t *testing.T, h *modelHarness, logs *observer.ObservedLogs, secret string, ) { t.Helper() require.Zero(t, h.repo.allocateCalls) require.Empty(t, h.repo.creates) require.Empty(t, h.repo.saves) require.Empty(t, h.repo.getOps) require.Empty(t, h.encryptor.encryptInputs) require.Empty(t, h.encryptor.decryptInputs) require.Empty(t, h.workflow.calls) for _, entry := range logs.All() { require.NotContains(t, entry.Message, secret) for _, field := range entry.Context { require.NotContains(t, field.String, secret) require.NotContains(t, fmt.Sprint(field.Interface), secret) } } } func queryPipelineSQL(t *testing.T, ops []data_pipeline_repo.FilterOp) string { t.Helper() capture := new(sqlCaptureLogger) db, err := gorm.Open(mysql.New(mysql.Config{ DSN: "test:test@tcp(127.0.0.1:3306)/test?parseTime=true", SkipInitializeWithVersion: true, }), &gorm.Config{ DryRun: true, DisableAutomaticPing: true, Logger: capture, }) require.NoError(t, err) querySet := entity.NewDataPipelineQuerySet(db) for _, op := range ops { querySet = op(querySet) } var selected []entity.DataPipeline require.NoError(t, querySet.All(&selected)) require.NotEmpty(t, capture.sql) return capture.sql } type sqlCaptureLogger struct { sql string } func (l *sqlCaptureLogger) LogMode(gorm_logger.LogLevel) gorm_logger.Interface { return l } func (*sqlCaptureLogger) Info(context.Context, string, ...interface{}) {} func (*sqlCaptureLogger) Warn(context.Context, string, ...interface{}) {} func (*sqlCaptureLogger) Error(context.Context, string, ...interface{}) {} func (l *sqlCaptureLogger) Trace( _ context.Context, _ time.Time, fc func() (string, int64), _ error, ) { l.sql, _ = fc() } type eventLog struct { values []string } func (l *eventLog) add(event string) { if l != nil { l.values = append(l.values, event) } } type fakeDataPipelineRepo struct { nextID entity.ID events *eventLog allocateCalls int creates []*entity.DataPipeline saves []*entity.DataPipeline getOps [][]data_pipeline_repo.FilterOp listOps [][]data_pipeline_repo.FilterOp countOps [][]data_pipeline_repo.FilterOp getResult *entity.DataPipeline listResult []entity.DataPipeline countResult int64 createErr bizerr.BizError saveErr bizerr.BizError getErr bizerr.BizError listErr bizerr.BizError countErr bizerr.BizError } func (r *fakeDataPipelineRepo) AllocateID() entity.ID { r.events.add("repo.allocate") r.allocateCalls++ return r.nextID } func (r *fakeDataPipelineRepo) Create(_ context.Context, pipeline *entity.DataPipeline) bizerr.BizError { r.events.add("repo.create") r.creates = append(r.creates, clonePipeline(pipeline)) return r.createErr } func (r *fakeDataPipelineRepo) Load(_ context.Context, _ uint64) (*entity.DataPipeline, bizerr.BizError) { return clonePipeline(r.getResult), r.getErr } func (r *fakeDataPipelineRepo) Get( _ context.Context, ops ...data_pipeline_repo.FilterOp, ) (*entity.DataPipeline, bizerr.BizError) { r.events.add("repo.get") r.getOps = append(r.getOps, append([]data_pipeline_repo.FilterOp(nil), ops...)) return clonePipeline(r.getResult), r.getErr } func (r *fakeDataPipelineRepo) List( _ context.Context, ops ...data_pipeline_repo.FilterOp, ) ([]entity.DataPipeline, bizerr.BizError) { r.events.add("repo.list") r.listOps = append(r.listOps, append([]data_pipeline_repo.FilterOp(nil), ops...)) rows := make([]entity.DataPipeline, 0, len(r.listResult)) for index := range r.listResult { rows = append(rows, *clonePipeline(&r.listResult[index])) } return rows, r.listErr } func (r *fakeDataPipelineRepo) Count( _ context.Context, ops ...data_pipeline_repo.FilterOp, ) (int64, bizerr.BizError) { r.events.add("repo.count") r.countOps = append(r.countOps, append([]data_pipeline_repo.FilterOp(nil), ops...)) return r.countResult, r.countErr } func (r *fakeDataPipelineRepo) Save(_ context.Context, pipeline *entity.DataPipeline) bizerr.BizError { r.events.add("repo.save") r.saves = append(r.saves, clonePipeline(pipeline)) return r.saveErr } func (r *fakeDataPipelineRepo) Delete(_ context.Context, _ uint64) bizerr.BizError { return nil } type fakeClusterResolver struct { cluster *mgmt.Cluster err bizerr.BizError calls []uint64 events *eventLog } func (r *fakeClusterResolver) GetClusterByID( _ context.Context, clusterID uint64, ) (*mgmt.Cluster, bizerr.BizError) { r.events.add("cluster.resolve") r.calls = append(r.calls, clusterID) return r.cluster, r.err } type stagingValidationCall struct { scope data_pipeline.ValidationScope staging domain.DataPipelineStagingInput } type fakeExternalValidator struct { sourceErr bizerr.BizError sourceMatched int32 stagingErr bizerr.BizError specificationErr bizerr.BizError sourceInputs []domain.DataPipelineSourceInput stagingCalls []stagingValidationCall specifications []string events *eventLog } func (v *fakeExternalValidator) ValidateSource( _ context.Context, _ uint64, source domain.DataPipelineSourceInput, ) (int32, bizerr.BizError) { v.events.add("validator.source") v.sourceInputs = append(v.sourceInputs, source) return v.sourceMatched, v.sourceErr } func (v *fakeExternalValidator) ValidateStaging( _ context.Context, scope data_pipeline.ValidationScope, staging domain.DataPipelineStagingInput, ) bizerr.BizError { v.events.add("validator.staging") v.stagingCalls = append(v.stagingCalls, stagingValidationCall{ scope: scope, staging: staging, }) return v.stagingErr } func (v *fakeExternalValidator) ValidateChangefeedSpecification( _ context.Context, _ uint64, specification string, ) bizerr.BizError { v.events.add("validator.specification") v.specifications = append(v.specifications, specification) return v.specificationErr } type fakeEncryptor struct { encryptResults map[string]string decryptResults map[string]string encryptErr bizerr.BizError decryptErr bizerr.BizError encryptErrors map[string]bizerr.BizError decryptErrors map[string]bizerr.BizError encryptInputs []string decryptInputs []string events *eventLog } func (e *fakeEncryptor) EncryptBase64(text string) (string, bizerr.BizError) { e.events.add("encrypt.encrypt") e.encryptInputs = append(e.encryptInputs, text) if err := e.encryptErrors[text]; err != nil { return "", err } if e.encryptErr != nil { return "", e.encryptErr } return e.encryptResults[text], nil } func (e *fakeEncryptor) DecryptBase64(ciphertext string) (string, bizerr.BizError) { e.events.add("encrypt.decrypt") e.decryptInputs = append(e.decryptInputs, ciphertext) if err := e.decryptErrors[ciphertext]; err != nil { return "", err } if e.decryptErr != nil { return "", e.decryptErr } return e.decryptResults[ciphertext], nil } func (e *fakeEncryptor) EncryptBase64Inplace(text *string) bizerr.BizError { if text == nil || *text == "" { return nil } ciphertext, err := e.EncryptBase64(*text) if err == nil { *text = ciphertext } return err } func (e *fakeEncryptor) DecryptBase64Inplace(ciphertext *string) bizerr.BizError { if ciphertext == nil || *ciphertext == "" { return nil } plaintext, err := e.DecryptBase64(*ciphertext) if err == nil { *ciphertext = plaintext } return err } type workflowCall struct { dagID string specVars map[string]string tags map[string]string } type fakeWorkflow struct { calls []workflowCall err error events *eventLog instances []*flowentity.DagInstance listErr error listInputs []*mod.ListDagInstanceInput } func (*fakeWorkflow) Close() {} func (*fakeWorkflow) RegisterDag(*flowentity.Dag, []run.Action) error { return nil } func (*fakeWorkflow) RunDag(string, map[string]string) (*flowentity.DagInstance, error) { return &flowentity.DagInstance{}, nil } func (w *fakeWorkflow) RunDagWithTags( dagID string, specVars map[string]string, tags map[string]string, ) (*flowentity.DagInstance, error) { if w.events != nil { w.events.add("workflow.run") } w.calls = append(w.calls, workflowCall{ dagID: dagID, specVars: cloneStringMap(specVars), tags: cloneStringMap(tags), }) return &flowentity.DagInstance{}, w.err } func (*fakeWorkflow) RetryDagIns(string, ...mod.CommandOptSetter) error { return nil } func (*fakeWorkflow) CancelDagIns(string, ...mod.CommandOptSetter) error { return nil } func (w *fakeWorkflow) ListDagIns(input *mod.ListDagInstanceInput) ([]*flowentity.DagInstance, error) { w.listInputs = append(w.listInputs, input) return w.instances, w.listErr } func cloneStringMap(input map[string]string) map[string]string { cloned := make(map[string]string, len(input)) for key, value := range input { cloned[key] = value } return cloned } type fakeClock struct { now time.Time calls int } func (c *fakeClock) Now() time.Time { c.calls++ return c.now } func TestValidationHelpersRejectMalformedRequestsWithoutSecretsInErrors(t *testing.T) { tests := []struct { name string err bizerr.BizError }{ {name: "nil create", err: validateCreateRequest(nil)}, {name: "nil edit", err: validateEditRequest(nil)}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { require.Error(t, test.err) require.NotContains(t, test.err.Error(), testSourcePassword) require.NotContains(t, test.err.Error(), testStagingSecret) }) } invalidStaging := validCreateRequest().Staging invalidStaging.S3.URI = "s3://test-user:test-secret@test-pipeline-bucket/path" _, _, err := validateStagingInput(invalidStaging) require.Error(t, err) require.False(t, strings.Contains(err.Error(), testStagingSecret)) } // TestValidateStagingAuthRejectsAmbiguousCredentials pins that exactly one // credential shape is accepted. Taking both would leave it undecided which one // the children were actually configured with. func TestValidateStagingAuthRejectsAmbiguousCredentials(t *testing.T) { t.Parallel() tests := map[string]struct { s3 domain.DataPipelineS3StagingInput wantErr bool }{ "access key": {s3: domain.DataPipelineS3StagingInput{ AuthType: domain.DataPipelineS3AuthTypeAccessKey, AccessKeyID: "AKIA", SecretAccessKey: "secret", }}, "role arn": {s3: domain.DataPipelineS3StagingInput{ AuthType: domain.DataPipelineS3AuthTypeRoleARN, RoleARN: "arn:aws:iam::000000000000:role/r", }}, "no auth type at all": { s3: domain.DataPipelineS3StagingInput{AccessKeyID: "AKIA", SecretAccessKey: "s"}, wantErr: true, }, "access key mode missing the secret": { s3: domain.DataPipelineS3StagingInput{ AuthType: domain.DataPipelineS3AuthTypeAccessKey, AccessKeyID: "AKIA", }, wantErr: true, }, "role arn mode missing the role": { s3: domain.DataPipelineS3StagingInput{AuthType: domain.DataPipelineS3AuthTypeRoleARN}, wantErr: true, }, "both shapes supplied under access key": { s3: domain.DataPipelineS3StagingInput{ AuthType: domain.DataPipelineS3AuthTypeAccessKey, AccessKeyID: "AKIA", SecretAccessKey: "s", RoleARN: "arn:aws:iam::000000000000:role/r", }, wantErr: true, }, "both shapes supplied under role arn": { s3: domain.DataPipelineS3StagingInput{ AuthType: domain.DataPipelineS3AuthTypeRoleARN, RoleARN: "arn:aws:iam::000000000000:role/r", AccessKeyID: "AKIA", }, wantErr: true, }, } for name, test := range tests { t.Run(name, func(t *testing.T) { t.Parallel() err := validateStagingAuth(test.s3) if test.wantErr { require.Error(t, err) require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode()) return } require.NoError(t, err) }) } } // TestGetDataPipelineStagingTrustReportsEveryPrincipal pins that the answer // carries all three identities that assume the customer's staging role. They // come from two systems that report them separately, and a customer who trusts // only what one of them says gets a role that validates and then fails during // delivery - so a partial answer is the failure this call exists to prevent. func TestGetDataPipelineStagingTrustReportsEveryPrincipal(t *testing.T) { h := newModelHarness() resp, err := h.model.GetDataPipelineStagingTrust(context.Background(), &domain.GetDataPipelineStagingTrustReq{ClusterID: testClusterID, UserID: testUserID, LakeOrgID: "7"}) require.NoError(t, err) require.Len(t, resp.Principals, 3) require.Equal(t, "arn:aws:iam::111122223333:role/test-changefeed", resp.Principals[0].RoleARN) require.Equal(t, "test-external-id", resp.Principals[0].ExternalID) // The destination reports two: one validates the connection, the other // loads the data, and they assume the role from different places. require.NotEqual(t, resp.Principals[1].RoleARN, resp.Principals[2].RoleARN) for _, principal := range resp.Principals { require.NotEmpty(t, principal.Name) require.NotEmpty(t, principal.RoleARN) require.NotEmpty(t, principal.ExternalID) } } // TestGetDataPipelineStagingTrustFailsWhenAPrincipalIsMissing covers the // destination not being ready to report its roles yet. Answering with the // producer's identity alone would read as a complete setup instruction and // send the customer away with a role that cannot load data. func TestGetDataPipelineStagingTrustFailsWhenAPrincipalIsMissing(t *testing.T) { h := newModelHarness() lakeClient := lake.NewFakeClient() lakeClient.FailNext(lake.OperationSetupInfo, errors.New("tenant role arn is not ready")) model := NewDataPipelineModel( h.repo, h.clusters, h.changefeeds, lakeClient, h.validator, h.encryptor, h.workflow, RuntimeConfig{Environment: "staging"}, h.clock, ) _, err := model.GetDataPipelineStagingTrust(context.Background(), &domain.GetDataPipelineStagingTrustReq{ClusterID: testClusterID, UserID: testUserID, LakeOrgID: "7"}) require.Error(t, err) } // TestGetDataPipelineStagingTrustFailsWithoutChangefeedIdentity covers the same // gap on the producer side: a cluster whose changefeed identity is not // provisioned yet must not yield an answer listing only the destination roles. func TestGetDataPipelineStagingTrustFailsWithoutChangefeedIdentity(t *testing.T) { h := newModelHarness() h.changefeeds.adminProfile = &domain.DescribeAdminProfileResp{ProviderType: "aws"} _, err := h.model.GetDataPipelineStagingTrust(context.Background(), &domain.GetDataPipelineStagingTrustReq{ClusterID: testClusterID, UserID: testUserID, LakeOrgID: "7"}) require.Error(t, err) require.Contains(t, err.Error(), "changefeed IAM identity") } // TestChangefeedRoleARNIsPastable covers what this endpoint exists to produce: // a string a customer can paste into an IAM trust policy. The regional service // reports the changefeed identity as a bare role name - observed on staging as // "tidbcloud--" - which names nothing in a trust policy. func TestChangefeedRoleARNIsPastable(t *testing.T) { tests := map[string]struct { profile *domain.DescribeAdminProfileResp region string want string }{ "bare role name becomes an ARN": { profile: &domain.DescribeAdminProfileResp{ ProviderType: "aws", CloudAccountID: "790636686964", CloudIAMIdentity: "tidbcloud-b8519f03-ee7c8a09", }, region: "us-west-2", want: "arn:aws:iam::790636686964:role/tidbcloud-b8519f03-ee7c8a09", }, "an ARN is left alone": { profile: &domain.DescribeAdminProfileResp{ ProviderType: "aws", CloudAccountID: "790636686964", CloudIAMIdentity: "arn:aws:iam::790636686964:role/already-an-arn", }, region: "us-west-2", want: "arn:aws:iam::790636686964:role/already-an-arn", }, "china lives in its own partition": { profile: &domain.DescribeAdminProfileResp{ ProviderType: "aws", CloudAccountID: "790636686964", CloudIAMIdentity: "tidbcloud-cn", }, region: "cn-northwest-1", want: "arn:aws-cn:iam::790636686964:role/tidbcloud-cn", }, "a service account is not an ARN": { profile: &domain.DescribeAdminProfileResp{ ProviderType: "gcp", CloudAccountID: "some-project", CloudIAMIdentity: "cdc@some-project.iam.gserviceaccount.com", }, region: "us-central1", want: "cdc@some-project.iam.gserviceaccount.com", }, "an unknown account yields no invented ARN": { profile: &domain.DescribeAdminProfileResp{ ProviderType: "aws", CloudIAMIdentity: "tidbcloud-no-account", }, region: "us-west-2", want: "tidbcloud-no-account", }, } for name, test := range tests { t.Run(name, func(t *testing.T) { require.Equal(t, test.want, changefeedRoleARN(test.profile, test.region)) }) } } // The notification queue is the one part of the staging area an Edit may // change: it does not move where data lands, so attaching or detaching it // leaves everything already staged meaningful. func TestEditDataPipelineAttachesAndDetachesTheNotificationQueue(t *testing.T) { const queue = "https://sqs.us-west-2.amazonaws.com/000000000000/test-queue.fifo" t.Run("attaching a queue moves the whole budget to the producer", func(t *testing.T) { h := newModelHarness() pipeline := observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusPaused) before := clonePipeline(pipeline) h.repo.getResult = pipeline req := validEditRequest() req.Staging = domain.DataPipelineEditableStagingInput{ SQSQueueURL: queue, SQSRegion: "us-west-2", } require.NoError(t, h.model.EditDataPipeline(context.Background(), req)) require.Len(t, h.repo.saves, 1) updated := h.repo.saves[0] require.Equal(t, queue, updated.Params.Staging.S3.SQSQueueURL) require.Equal(t, "us-west-2", updated.Params.Staging.S3.SQSRegion) // Where the data lands, and the credentials that reach it, are not // what this edit touched. require.Equal(t, before.Params.Staging.S3.Bucket, updated.Params.Staging.S3.Bucket) require.Equal(t, before.Params.Staging.S3.BasePrefix, updated.Params.Staging.S3.BasePrefix) require.Equal(t, before.Params.Staging.S3.SecretAccessKeyCiphertext, updated.Params.Staging.S3.SecretAccessKeyCiphertext) // A notified destination does not poll, so nothing of the budget is // spent on discovery. require.Equal(t, entity.DeliveryIntervalPlan{ Version: 1, EndToEndSeconds: 90, ProducerFlushIntervalSeconds: 90, ConsumerPollIntervalSeconds: 0, }, updated.IntervalPlan) // The queue is validated as part of the staging area, so an // unreachable one fails the Edit rather than leaving the destination // waiting for notifications that never arrive. require.Len(t, h.validator.stagingCalls, 1) require.Equal(t, queue, h.validator.stagingCalls[0].staging.S3.SQSQueueURL) }) t.Run("omitting the queue detaches it and restores the split", func(t *testing.T) { h := newModelHarness() pipeline := observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusPaused) pipeline.Params.Staging.S3.SQSQueueURL = queue pipeline.Params.Staging.S3.SQSRegion = "us-west-2" h.repo.getResult = pipeline require.NoError(t, h.model.EditDataPipeline(context.Background(), validEditRequest())) require.Len(t, h.repo.saves, 1) updated := h.repo.saves[0] require.Empty(t, updated.Params.Staging.S3.SQSQueueURL) require.Empty(t, updated.Params.Staging.S3.SQSRegion) require.Equal(t, entity.DeliveryIntervalPlan{ Version: 1, EndToEndSeconds: 90, ProducerFlushIntervalSeconds: 45, ConsumerPollIntervalSeconds: 45, }, updated.IntervalPlan) require.Len(t, h.validator.stagingCalls, 1) require.Empty(t, h.validator.stagingCalls[0].staging.S3.SQSQueueURL) }) } func TestValidateStagingNotificationAcceptsEitherQueueType(t *testing.T) { t.Parallel() tests := map[string]struct { s3 domain.DataPipelineS3StagingInput wantErr bool }{ // Polling is the default, and the overwhelmingly common case. "no queue at all": {}, // A standard queue is what S3 event notifications can target - S3 // rejects a FIFO destination outright - so this is the configuration a // customer can wire directly, and it must not be refused here. "a standard queue": {s3: domain.DataPipelineS3StagingInput{ SQSQueueURL: "https://sqs.us-west-2.amazonaws.com/000000000000/q", SQSRegion: "us-west-2", }}, // A FIFO queue is reachable through EventBridge, so it is accepted // too. Which of the two the destination can consume is its own probe // to answer, not something decidable from the name. "a FIFO queue": {s3: domain.DataPipelineS3StagingInput{ SQSQueueURL: "https://sqs.us-west-2.amazonaws.com/000000000000/q.fifo", }}, "a region naming no queue": { s3: domain.DataPipelineS3StagingInput{SQSRegion: "us-west-2"}, wantErr: true, }, } for name, test := range tests { t.Run(name, func(t *testing.T) { t.Parallel() err := validateStagingNotification(test.s3) if test.wantErr { require.Error(t, err) require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode()) return } require.NoError(t, err) }) } } // The trust answer names the actions as well as the identities. A policy that // trusts the right roles but grants the wrong actions passes validation and // then fails partway through delivery, which is the failure this call exists // to prevent. func TestGetDataPipelineStagingTrustReportsTheRequiredActions(t *testing.T) { h := newModelHarness() resp, err := h.model.GetDataPipelineStagingTrust(context.Background(), &domain.GetDataPipelineStagingTrustReq{ClusterID: testClusterID, UserID: testUserID, LakeOrgID: "7"}) require.NoError(t, err) // The producers write the staged objects and the destination reads them, // so both halves have to be there. require.Contains(t, resp.BucketActions, "s3:PutObject") require.Contains(t, resp.BucketActions, "s3:GetObject") require.Contains(t, resp.BucketActions, "s3:ListBucket") // A consumer that may receive but not delete reprocesses the same // notification until the queue's retention expires. require.Contains(t, resp.QueueActions, "sqs:ReceiveMessage") require.Contains(t, resp.QueueActions, "sqs:DeleteMessage") // The queue actions are reported whether or not a queue is in use: the // customer writes this policy before the Pipeline exists, so there is // nothing yet to condition the answer on. require.NotEmpty(t, resp.QueueActions) } // A warehouse name does not identify a destination on its own: a TiDB // organization may own several Lakes, and two of them can hold a warehouse of // the same name. Every entry point that reaches Lake therefore has to be told // which one, and the ones that run before a Pipeline exists cannot read it off // a row. func TestDataPipelineEntryPointsRequireTheDestinationLake(t *testing.T) { ctx := context.Background() t.Run("create", func(t *testing.T) { h := newModelHarness() req := validCreateRequest() req.Destination.LakeOrgID = " " _, err := h.model.CreateDataPipeline(ctx, req) require.ErrorContains(t, err, "destination Lake is required") require.Empty(t, h.repo.creates, "a rejected request must not reach persistence") }) t.Run("validate destination", func(t *testing.T) { h := newModelHarness() req := &domain.ValidateDataPipelineDestinationReq{ ClusterID: testClusterID, UserID: testUserID, SyncMode: domain.DataPipelineSyncModeFullAndIncremental, Destination: domain.DataPipelineDestinationInput{ Type: domain.DataPipelineDestinationTypeTiDBLake, Warehouse: "test-warehouse", FlushIntervalSeconds: 60, }, } require.ErrorContains(t, h.model.ValidateDataPipelineDestination(ctx, req), "destination Lake is required") }) t.Run("validate staging", func(t *testing.T) { h := newModelHarness() err := h.model.ValidateDataPipelineStaging(ctx, &domain.ValidateDataPipelineStagingReq{ ClusterID: testClusterID, UserID: testUserID, Staging: validCreateRequest().Staging, }) require.ErrorContains(t, err, "destination Lake is required") require.Empty(t, h.validator.stagingCalls, "nothing should be validated against no Lake") }) t.Run("staging trust", func(t *testing.T) { h := newModelHarness() _, err := h.model.GetDataPipelineStagingTrust(ctx, &domain.GetDataPipelineStagingTrustReq{ClusterID: testClusterID, UserID: testUserID}) require.ErrorContains(t, err, "destination Lake is required") }) } // The Lake is fixed at creation. An Edit re-validates the staging area, and it // must do so against the Pipeline's own Lake rather than anything the request // could carry - moving a Pipeline between Lakes would strand its datasource, // its task, and everything already loaded. func TestEditDataPipelineValidatesAgainstTheStoredLake(t *testing.T) { h := newModelHarness() pipeline := observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusPaused) pipeline.Params.Destination.LakeOrgID = "9" h.repo.getResult = pipeline require.NoError(t, h.model.EditDataPipeline(context.Background(), validEditRequest())) require.Len(t, h.validator.stagingCalls, 1) require.Equal(t, "9", h.validator.stagingCalls[0].scope.LakeOrgID) require.Len(t, h.repo.saves, 1) require.Equal(t, "9", h.repo.saves[0].Params.Destination.LakeOrgID, "an Edit must not move the Pipeline to another Lake") } // Repeating a Delete must not start a second workflow. Staging ended up with // two instances retrying the same delete step, 206 times and 127 times, over // the same row and the same children. func TestDeleteDataPipelineRefusesARepeat(t *testing.T) { h := newModelHarness() h.repo.getResult = observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusDeleting) err := h.model.DeleteDataPipeline(context.Background(), &domain.DeleteDataPipelineReq{ ClusterID: testClusterID, PipelineID: testPipelineID, UserID: testUserID, }) require.Error(t, err) require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode()) require.Contains(t, err.Error(), "deleting") // The point of the refusal: no duplicate DAG, and nothing written. require.Empty(t, h.workflow.calls) require.Empty(t, h.repo.saves) } // The validator has always decided "is a queue configured?" with TrimSpace, // so UsesSQS and the persisted value have to agree with it. A value that // differed would plan a notified Pipeline - consumer interval zero - out of a // queue the validator had treated as absent, and send Lake a URL with spaces // in it that the spec matcher would then compare against. func TestSurroundingWhitespaceDoesNotCreateAPhantomQueue(t *testing.T) { t.Parallel() t.Run("whitespace only is not a queue", func(t *testing.T) { t.Parallel() require.False(t, domain.DataPipelineS3StagingInput{SQSQueueURL: " "}.UsesSQS()) require.False(t, domain.DataPipelineEditableStagingInput{SQSQueueURL: " \t "}.UsesSQS()) require.False(t, entity.DataPipelineS3Config{SQSQueueURL: " "}.UsesSQS()) }) t.Run("a padded queue is stored trimmed", func(t *testing.T) { t.Parallel() const queue = "https://sqs.us-west-2.amazonaws.com/000000000000/q.fifo" h := newModelHarness() h.repo.getResult = observedPipeline( testPipelineID, testTenantID, testClusterID, entity.DataPipelineStatusPaused) req := validEditRequest() req.Staging = domain.DataPipelineEditableStagingInput{ SQSQueueURL: " " + queue + " ", SQSRegion: " us-west-2 ", } require.NoError(t, h.model.EditDataPipeline(context.Background(), req)) require.Len(t, h.repo.saves, 1) stored := h.repo.saves[0].Params.Staging.S3 require.Equal(t, queue, stored.SQSQueueURL) require.Equal(t, "us-west-2", stored.SQSRegion) // What Lake is told, what is stored, and what the matcher compares are // then all the same string. require.Equal(t, queue, h.validator.stagingCalls[0].staging.S3.SQSQueueURL) }) }