Function TestStateReplication

pkg/alertmanager/state_replication_test.go:98–202

(t *testing.T)

}

func TestStateReplication(t *testing.T) {
	tc := []struct {
		name              string
		replicationFactor int
		message           *clusterpb.Part
		results           map[string]*clusterpb.Part
	}{
		{
			name:              "with a replication factor of <= 1, state is not replicated.",
			replicationFactor: 1,
			message:           &clusterpb.Part{Key: "nflog", Data: []byte("OK")},
			results:           map[string]*clusterpb.Part{},
		},
		{
			name:              "with a replication factor of > 1, state is broadcasted for replication.",
			replicationFactor: 3,
			message:           &clusterpb.Part{Key: "nflog", Data: []byte("OK")},
			results:           map[string]*clusterpb.Part{"user-1": {Key: "nflog", Data: []byte("OK")}},
		},
	}

	for _, tt := range tc {
		t.Run(tt.name, func(t *testing.T) {
			reg := prometheus.NewPedanticRegistry()
			replicator := newFakeReplicator()
			replicator.read = readStateResult{res: nil, err: nil}
			store := newFakeAlertStore()
			s := newReplicatedStates("user-1", tt.replicationFactor, replicator, store, log.NewNopLogger(), reg)

			require.False(t, s.Ready())
			{
				ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
				defer cancel()
				require.Equal(t, context.DeadlineExceeded, s.WaitReady(ctx))
			}

			require.NoError(t, services.StartAndAwaitRunning(context.Background(), s))
			t.Cleanup(func() {
				require.NoError(t, services.StopAndAwaitTerminated(context.Background(), s))
			})

			require.True(t, s.Ready())
			{
				ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
				defer cancel()
				require.NoError(t, s.WaitReady(ctx))
			}

			ch := s.AddState("nflog:user-1", &fakeState{}, reg)

			part := tt.message
			d, err := part.Marshal()
			require.NoError(t, err)
			ch.Broadcast(d)

			require.Eventually(t, func() bool {
				replicator.mtx.Lock()
				defer replicator.mtx.Unlock()