You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: keywatcher_test.go
    path of file: ./repos/gitlabhq/workhorse/internal/redis
    
    the code of the file until where you have to start completion: package redis

import (
	"context"
	"os"
	"sync"
	"testing"
	"time"

	"github.com/redis/go-redis/v9"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"gitlab.com/gitlab-org/gitlab/workhorse/internal/config"
)

var ctx = context.Background()

const (
	runnerKey = "runner:build_queue:10"
)

func initRdb(t *testing.T) *redis.Client {
	buf, err := os.ReadFile("../../config.toml")
	require.NoError(t, err)
	cfg, err := config.LoadConfig(string(buf))
	require.NoError(t, err)
	rdb, err := Configure(cfg.Redis)
	require.NoError(t, err)
	t.Cleanup(func() {
		assert.NoError(t, rdb.Close())
	})
	return rdb
}

func countSubscribers(kw *KeyWatcher, key string) int {
	kw.mu.Lock()
	defer kw.mu.Unlock()
	return len(kw.subscribers[key])
}

// Forces a run of the `Process` loop against a mock PubSubConn.
func proce
    
func TestKeyChangesParallel(t *testing.T) {
	rdb := initRdb(t)

	testCases := []keyChangeTestCase{
		{
			desc:           "massively parallel, sees change with key existing",
			returnValue:    "something",
			watchValue:     "something",
			processedValue: "somethingelse",
			expectedStatus: WatchKeyStatusSeenChange,
		},
		{
			desc:           "massively parallel, sees change with key existing, watching missing keys",
			isKeyMissing:   true,
			watchValue:     "",
			processedValue: "somethingelse",
			expectedStatus: WatchKeyStatusSeenChange,
		},
	}

	for _, tc := range testCases {
		t.Run(tc.desc, func(t *testing.T) {
			runTimes := 100

			if !tc.isKeyMissing {
				rdb.Set(ctx, runnerKey, tc.returnValue, 0)
			}

			defer rdb.FlushDB(ctx)

			wg := &sync.WaitGroup{}
			wg.Add(runTimes)
			ready := make(chan struct{})

			kw := NewKeyWatcher(rdb)
			defer kw.Shutdown()

			for i := 0; i < runTimes; i++ {
				go func() {
					defer wg.Done()
					<-ready
					val, err := kw.WatchKey(ctx, runnerKey, tc.watchValue, time.Second)

					require.NoError(t, err, "Expected no error")
					require.Equal(t, tc.expectedStatus, val, "Expected value")
				}()
			}

			processMessages(t, kw, runTimes, tc.processedValue, ready, wg)
		})
	}
}