[MM-62408] Server Code Coverage with Fully Parallel Tests (#30078)

* TestPool

* Store infra

* Store tests updates

* Bump maximum concurrent postgres connections

* More infra

* channels/jobs

* channels/app

* channels/api4

* Protect i18n from concurrent access

* Replace some use of os.Setenv

* Remove debug

* Lint fixes

* Fix more linting

* Fix test

* Remove use of Setenv in drafts tests

* Fix flaky TestWebHubCloseConnOnDBFail

* Fix merge

* [MM-62408] Add CI job to generate test coverage (#30284)

* Add CI job to generate test coverage

* Remove use of Setenv in drafts tests

* Fix flaky TestWebHubCloseConnOnDBFail

* Fix more Setenv usage

* Fix more potential flakyness

* Remove parallelism from flaky test

* Remove conflicting env var

* Fix

* Disable parallelism

* Test atomic covermode

* Disable parallelism

* Enable parallelism

* Add upload coverage step

* Fix codecov.yml

* Add codecov.yml

* Remove redundant workspace field

* Add Parallel() util methods and refactor

* Fix formatting

* More formatting fixes

* Fix reporting

server/channels/jobs/base_workers_test.go

@@ -5,6 +5,7 @@ package jobs
 
 import (
 	"errors"
+	"os"
 	"testing"
 
 	"github.com/mattermost/mattermost/server/public/model"
@@ -14,6 +15,10 @@ import (
 )
 
 func TestSimpleWorkerPanic(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	jobServer, mockStore, mockMetrics := makeJobServer(t)
 
 	job := &model.Job{

server/channels/jobs/batch_migration_worker_test.go

@@ -45,6 +45,8 @@ func (ma *MockApp) SetOutOfSync() {
 }
 
 func TestBatchMigrationWorker(t *testing.T) {
+	mainHelper.Parallel(t)
+
 	setupBatchWorker := func(t *testing.T, th *TestHelper, mockApp *MockApp, doMigrationBatch func(model.StringMap, store.Store) (model.StringMap, bool, error)) (model.Worker, *model.Job) {
 		t.Helper()
 

server/channels/jobs/batch_report_worker_test.go

@@ -20,17 +20,22 @@ type ReportMockApp struct{}
 func (rma *ReportMockApp) SaveReportChunk(format string, prefix string, count int, reportData []model.ReportableObject) *model.AppError {
 	return nil
 }
+
 func (rma *ReportMockApp) CompileReportChunks(format string, prefix string, numberOfChunks int, headers []string) *model.AppError {
 	return nil
 }
+
 func (rma *ReportMockApp) SendReportToUser(rctx request.CTX, job *model.Job, format string) *model.AppError {
 	return nil
 }
+
 func (rma *ReportMockApp) CleanupReportChunks(format string, prefix string, numberOfChunks int) *model.AppError {
 	return nil
 }
 
 func TestBatchReportWorker(t *testing.T) {
+	mainHelper.Parallel(t)
+
 	setupBatchWorker := func(
 		t *testing.T,
 		th *TestHelper,
@@ -83,6 +88,8 @@ func TestBatchReportWorker(t *testing.T) {
 	}
 
 	t.Run("should finish when the report is done, incrementing file count along the way", func(t *testing.T) {
+		mainHelper.Parallel(t)
+
 		th := Setup(t).InitBasic(t)
 
 		var worker model.Worker
@@ -113,6 +120,7 @@ func TestBatchReportWorker(t *testing.T) {
 	})
 
 	t.Run("should fail job when get data throws an error", func(t *testing.T) {
+		mainHelper.Parallel(t)
 		th := Setup(t).InitBasic(t)
 
 		var worker model.Worker

server/channels/jobs/batch_worker_test.go

@@ -17,6 +17,8 @@ import (
 // TestBatchWorkerRace tests race conditions during the start/stop
 // cases of the batch worker. Use the -race flag while testing this.
 func TestBatchWorkerRace(t *testing.T) {
+	mainHelper.Parallel(t)
+
 	th := Setup(t)
 
 	worker := jobs.MakeBatchWorker(th.Server.Jobs, th.Server.Store(), 1*time.Second, func(rctx *request.Context, job *model.Job) bool {
@@ -28,6 +30,8 @@ func TestBatchWorkerRace(t *testing.T) {
 }
 
 func TestBatchWorker(t *testing.T) {
+	mainHelper.Parallel(t)
+
 	createBatchWorker := func(t *testing.T, th *TestHelper, doBatch func(rctx *request.Context, job *model.Job) bool) (*jobs.BatchWorker, *model.Job) {
 		t.Helper()
 
@@ -58,6 +62,8 @@ func TestBatchWorker(t *testing.T) {
 	}
 
 	t.Run("stop after first batch", func(t *testing.T) {
+		mainHelper.Parallel(t)
+
 		th := Setup(t).InitBasic(t)
 
 		var worker *jobs.BatchWorker
@@ -84,6 +90,8 @@ func TestBatchWorker(t *testing.T) {
 	})
 
 	t.Run("stop after second batch", func(t *testing.T) {
+		mainHelper.Parallel(t)
+
 		th := Setup(t).InitBasic(t)
 
 		var worker *jobs.BatchWorker
@@ -110,6 +118,8 @@ func TestBatchWorker(t *testing.T) {
 	})
 
 	t.Run("done after first batch", func(t *testing.T) {
+		mainHelper.Parallel(t)
+
 		th := Setup(t).InitBasic(t)
 
 		var worker *jobs.BatchWorker
@@ -133,6 +143,8 @@ func TestBatchWorker(t *testing.T) {
 	})
 
 	t.Run("done after three batches", func(t *testing.T) {
+		mainHelper.Parallel(t)
+
 		th := Setup(t).InitBasic(t)
 
 		var worker *jobs.BatchWorker

server/channels/jobs/export_delete_test.go

@@ -15,6 +15,7 @@ import (
 )
 
 func TestExportDelete(t *testing.T) {
+	mainHelper.Parallel(t)
 	// Create a temporary export directory
 	fileSettingsDir, err := os.MkdirTemp("", "")
 	require.NoError(t, err)

server/channels/jobs/helper_test.go

@@ -29,6 +29,7 @@ type TestHelper struct {
 	BasicTeam  *model.Team
 	BasicUser  *model.User
 	BasicUser2 *model.User
+	Store      store.Store
 
 	SystemAdminUser   *model.User
 	LogBuffer         *mlog.Buffer
@@ -36,18 +37,18 @@ type TestHelper struct {
 	IncludeCacheLayer bool
 	ConfigStore       *config.Store
 
-	tempWorkspace             string
-	oldWatcherPollingInterval int
+	tempWorkspace string
 }
 
-func setupTestHelper(tb testing.TB, dbStore store.Store, enterprise bool, includeCacheLayer bool,
-	updateCfg func(cfg *model.Config), options []app.Option) *TestHelper {
+func setupTestHelper(tb testing.TB, dbStore store.Store, sqlSettings *model.SqlSettings, enterprise bool, includeCacheLayer bool,
+	updateCfg func(cfg *model.Config), options []app.Option,
+) *TestHelper {
 	tempWorkspace, err := os.MkdirTemp("", "jobstest")
 	require.NoError(tb, err)
 
 	configStore := config.NewTestMemoryStore()
 	memoryConfig := configStore.Get()
-	memoryConfig.SqlSettings = *mainHelper.GetSQLSettings()
+	memoryConfig.SqlSettings = *sqlSettings
 	*memoryConfig.PluginSettings.Directory = filepath.Join(tempWorkspace, "plugins")
 	*memoryConfig.PluginSettings.ClientDirectory = filepath.Join(tempWorkspace, "webapp")
 	*memoryConfig.PluginSettings.AutomaticPrepackagedPlugins = false
@@ -96,6 +97,7 @@ func setupTestHelper(tb testing.TB, dbStore store.Store, enterprise bool, includ
 		TestLogger:        testLogger,
 		IncludeCacheLayer: includeCacheLayer,
 		ConfigStore:       configStore,
+		Store:             dbStore,
 		tempWorkspace:     tempWorkspace,
 	}
 
@@ -131,10 +133,6 @@ func setupTestHelper(tb testing.TB, dbStore store.Store, enterprise bool, includ
 		if th.tempWorkspace != "" {
 			os.RemoveAll(th.tempWorkspace)
 		}
-
-		if th.oldWatcherPollingInterval != 0 {
-			jobs.DefaultWatcherPollingInterval = th.oldWatcherPollingInterval
-		}
 	})
 
 	return th
@@ -150,16 +148,22 @@ func SetupWithUpdateCfg(tb testing.TB, updateCfg func(cfg *model.Config), option
 		tb.SkipNow()
 	}
 
-	oldWatcherPollingInterval := jobs.DefaultWatcherPollingInterval
-	jobs.DefaultWatcherPollingInterval = 100
+	var dbStore store.Store
+	var dbSettings *model.SqlSettings
+	if mainHelper.Options.RunParallel {
+		dbStore, _, dbSettings, _ = mainHelper.GetNewStores(tb)
+		tb.Cleanup(func() {
+			dbStore.Close()
+		})
+	} else {
+		dbStore = mainHelper.GetStore()
+		dbSettings = mainHelper.GetSQLSettings()
+		dbStore.DropAllTables()
+		dbStore.MarkSystemRanUnitTests()
+		mainHelper.PreloadMigrations()
+	}
 
-	dbStore := mainHelper.GetStore()
-	dbStore.DropAllTables()
-	dbStore.MarkSystemRanUnitTests()
-	mainHelper.PreloadMigrations()
-
-	th := setupTestHelper(tb, dbStore, false, true, updateCfg, options)
-	th.oldWatcherPollingInterval = oldWatcherPollingInterval
+	th := setupTestHelper(tb, dbStore, dbSettings, false, true, updateCfg, options)
 	return th
 }
 
@@ -228,6 +232,22 @@ func (th *TestHelper) CreateUserOrGuest(tb testing.TB, guest bool) *model.User {
 	return user
 }
 
+func (th *TestHelper) ShutdownApp() {
+	done := make(chan bool)
+	go func() {
+		th.Server.Shutdown()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		// panic instead of fatal to terminate all tests in this package, otherwise the
+		// still running App could spuriously fail subsequent tests.
+		panic("failed to shutdown App within 30 seconds")
+	}
+}
+
 func (th *TestHelper) SetupBatchWorker(tb testing.TB, worker *jobs.BatchWorker) *model.Job {
 	tb.Helper()
 

server/channels/jobs/jobs_test.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"os"
 	"testing"
 
 	"github.com/stretchr/testify/mock"
@@ -64,6 +65,10 @@ func makeTeamEditionJobServer(t *testing.T) (*JobServer, *storetest.Store) {
 }
 
 func TestClaimJob(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("error claiming job", func(t *testing.T) {
 		jobServer, mockStore, _ := makeJobServer(t)
 
@@ -141,6 +146,10 @@ func TestClaimJob(t *testing.T) {
 }
 
 func TestSetJobProgress(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("error setting progress", func(t *testing.T) {
 		jobServer, mockStore, _ := makeJobServer(t)
 
@@ -180,6 +189,10 @@ func TestSetJobProgress(t *testing.T) {
 }
 
 func TestSetJobWarning(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("error setting status", func(t *testing.T) {
 		jobServer, mockStore, _ := makeJobServer(t)
 
@@ -216,6 +229,10 @@ func TestSetJobWarning(t *testing.T) {
 }
 
 func TestSetJobSuccess(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("error setting status", func(t *testing.T) {
 		jobServer, mockStore, _ := makeJobServer(t)
 
@@ -261,6 +278,10 @@ func TestSetJobSuccess(t *testing.T) {
 }
 
 func TestSetJobError(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("nil provided job error", func(t *testing.T) {
 		t.Run("error setting status", func(t *testing.T) {
 			jobServer, mockStore, _ := makeJobServer(t)
@@ -448,6 +469,10 @@ func TestSetJobError(t *testing.T) {
 }
 
 func TestSetJobCanceled(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("error setting status", func(t *testing.T) {
 		jobServer, mockStore, _ := makeJobServer(t)
 
@@ -493,6 +518,10 @@ func TestSetJobCanceled(t *testing.T) {
 }
 
 func TestUpdateInProgressJobData(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("error updating", func(t *testing.T) {
 		jobServer, mockStore, _ := makeJobServer(t)
 
@@ -527,6 +556,10 @@ func TestUpdateInProgressJobData(t *testing.T) {
 }
 
 func TestHandleJobPanic(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("no panic", func(t *testing.T) {
 		logger := mlog.CreateConsoleTestLogger(t)
 		jobServer, _, _ := makeJobServer(t)
@@ -589,6 +622,10 @@ func TestHandleJobPanic(t *testing.T) {
 }
 
 func TestRequestCancellation(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	ctx := request.TestContext(t)
 	t.Run("error cancelling", func(t *testing.T) {
 		jobServer, mockStore, _ := makeJobServer(t)

server/channels/jobs/main_test.go

@@ -4,17 +4,39 @@
 package jobs_test
 
 import (
+	"flag"
+	"os"
+	"strconv"
 	"testing"
 
+	"github.com/mattermost/mattermost/server/public/shared/mlog"
+	"github.com/mattermost/mattermost/server/v8/channels/jobs"
 	"github.com/mattermost/mattermost/server/v8/channels/testlib"
 )
 
 var mainHelper *testlib.MainHelper
 
 func TestMain(m *testing.M) {
-	var options = testlib.HelperOptions{
+	var parallelism int
+	if f := flag.Lookup("test.parallel"); f != nil {
+		parallelism, _ = strconv.Atoi(f.Value.String())
+	}
+	runParallel := os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" && parallelism > 1
+	if runParallel {
+		mlog.Info("Fully parallel tests enabled", mlog.Int("parallelism", parallelism))
+	}
+
+	oldWatcherPollingInterval := jobs.DefaultWatcherPollingInterval
+	jobs.DefaultWatcherPollingInterval = 100
+	defer func() {
+		jobs.DefaultWatcherPollingInterval = oldWatcherPollingInterval
+	}()
+
+	options := testlib.HelperOptions{
 		EnableStore:     true,
 		EnableResources: true,
+		RunParallel:     runParallel,
+		Parallelism:     parallelism,
 	}
 
 	mainHelper = testlib.NewMainHelperWithOptions(&options)

server/channels/jobs/schedulers_test.go

@@ -3,6 +3,7 @@
 package jobs
 
 import (
+	"os"
 	"sync"
 	"testing"
 	"time"
@@ -35,6 +36,10 @@ func (scheduler *MockScheduler) ScheduleJob(c request.CTX, cfg *model.Config, pe
 }
 
 func TestScheduler(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	mockStore := &storetest.Store{}
 	defer mockStore.AssertExpectations(t)
 
@@ -171,6 +176,10 @@ func TestScheduler(t *testing.T) {
 }
 
 func TestRandomDelay(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	cases := []int64{5, 10, 100}
 	for _, c := range cases {
 		out := getRandomDelay(c)

server/channels/jobs/server_test.go

@@ -4,6 +4,7 @@
 package jobs
 
 import (
+	"os"
 	"testing"
 	"time"
 
@@ -11,6 +12,10 @@ import (
 )
 
 func TestStartWorkers(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("uninitialized", func(t *testing.T) {
 		jobServer, _, _ := makeJobServer(t)
 		err := jobServer.StartWorkers()
@@ -43,6 +48,10 @@ func TestStartWorkers(t *testing.T) {
 }
 
 func TestStopWorkers(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("uninitialized", func(t *testing.T) {
 		jobServer, _, _ := makeJobServer(t)
 		err := jobServer.StopWorkers()
@@ -69,6 +78,10 @@ func TestStopWorkers(t *testing.T) {
 }
 
 func TestStartSchedulers(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("uninitialized", func(t *testing.T) {
 		jobServer, _, _ := makeJobServer(t)
 		err := jobServer.StartSchedulers()
@@ -99,6 +112,10 @@ func TestStartSchedulers(t *testing.T) {
 }
 
 func TestStopSchedulers(t *testing.T) {
+	if os.Getenv("ENABLE_FULLY_PARALLEL_TESTS") == "true" {
+		t.Parallel()
+	}
+
 	t.Run("uninitialized", func(t *testing.T) {
 		jobServer, _, _ := makeJobServer(t)
 		err := jobServer.StopSchedulers()