Function TestShardedIntegration

modules/backendscheduler/integration_test.go:21–65

TestShardedIntegration verifies that sharded work can be used as a drop-in replacement

(t *testing.T)

// TestShardedIntegration verifies that sharded work can be used as a drop-in replacement
func TestShardedIntegration(t *testing.T) {
	tests := []struct {
		name string
	}{
		{"sharded_work"},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tmpDir := t.TempDir()

			// Create config with or without sharding
			cfg := Config{}
			cfg.RegisterFlagsAndApplyDefaults("", &flag.FlagSet{})
			cfg.LocalWorkPath = tmpDir + "/work"
			// Prevent the compaction provider from picking up test blocks as in-flight
			// compaction jobs (which would filter them from SubmitRedaction's view).
			cfg.ProviderConfig.Compaction.MinInputBlocks = 100

			var (
				ctx, cancel   = context.WithCancel(context.Background())
				store, rr, ww = newStore(ctx, t, tmpDir)
			)
			defer func() {
				cancel()
				store.Shutdown()
			}()

			limits, err := overrides.NewOverrides(overrides.Config{Defaults: overrides.Overrides{}}, nil, prometheus.DefaultRegisterer)
			require.NoError(t, err)

			scheduler, err := New(cfg, store, limits, rr, ww)
			require.NoError(t, err)

			err = scheduler.starting(ctx)
			require.NoError(t, err)

			testJobOperations(ctx, t, scheduler)

			testPersistenceAndRecovery(ctx, t, scheduler, cfg, store, limits, rr, ww)

			testSubmitRedactionPersistence(ctx, t, scheduler, cfg, store, limits, rr, ww)
		})
	}
}

func testJobOperations(ctx context.Context, t *testing.T, scheduler *BackendScheduler) {
	// Add some jobs