core-old/internal/msg/consumer_server_test.go

package msg_test

import (
	"context"
	"reflect"
	"testing"
	"time"

	"github.com/ThreeDotsLabs/watermill"

	"gitea.dwysokinski.me/twhelp/core/internal/msg/internal/model"

	"github.com/ThreeDotsLabs/watermill/message/subscriber"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"gitea.dwysokinski.me/twhelp/core/internal/msg/internal/mock"

	"gitea.dwysokinski.me/twhelp/core/internal/domain"
	"gitea.dwysokinski.me/twhelp/core/internal/msg"
)

func TestServerConsumer_refresh(t *testing.T) {
	t.Parallel()

	marshaler := msg.GobMarshaler{}
	pubSub := newPubSub(t)

	servers := []domain.Server{
		{
			Key:         "pl1",
			URL:         "https://pl1.plemiona.pl",
			Open:        false,
			VersionCode: "pl",
		},
		{
			Key:         "pl2",
			URL:         "https://pl2.plemiona.pl",
			Open:        true,
			VersionCode: "pl",
		},
		{
			Key:         "pl3",
			URL:         "https://pl3.plemiona.pl",
			Open:        true,
			VersionCode: "pl",
		},
	}
	serverSvc := &mock.FakeServerService{}
	serverSvc.RefreshReturns(servers, nil)

	runRouter(t, msg.NewServerConsumer(marshaler, pubSub, pubSub, watermill.NopLogger{}, serverSvc))

	msgs, err := pubSub.Subscribe(context.Background(), "servers.event.refreshed")
	require.NoError(t, err)

	require.NoError(
		t,
		msg.NewServerPublisher(pubSub, marshaler).
			CmdRefresh(context.Background(), domain.RefreshServersCmdPayload{
				Host:        "plemiona.pl",
				VersionCode: "pl",
			}),
	)

	receivedMsgs, _ := subscriber.BulkRead(msgs, len(servers), 10*time.Second)
	assert.Len(t, receivedMsgs, len(servers))
	for _, m := range receivedMsgs {
		var received model.ServerRefreshedEvPayload
		assert.NoError(t, marshaler.Unmarshal(m, &received))

		found := false
		for _, srv := range servers {
			expected := model.ServerRefreshedEvPayload{
				Key:         srv.Key,
				URL:         srv.URL,
				Open:        srv.Open,
				VersionCode: srv.VersionCode,
			}
			if reflect.DeepEqual(expected, received) {
				found = true
				break
			}
		}
		assert.True(t, found)
	}
}

func TestServerConsumer_updateInfoAndConfig(t *testing.T) {
	t.Parallel()

	t.Run("OK", func(t *testing.T) {
		t.Parallel()

		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()

		marshaler := msg.GobMarshaler{}
		pubSub := newPubSub(t)

		serverSvc := &mock.FakeServerService{}
		serverSvc.UpdateInfoAndConfigReturns(nil)

		waitMiddleware, wait := newWaitForHandlerMiddleware("ServerConsumer.updateInfoAndConfig")
		runRouter(
			t,
			waitMiddleware,
			msg.NewServerConsumer(marshaler, pubSub, pubSub, watermill.NopLogger{}, serverSvc),
		)

		payload := model.ServerRefreshedEvPayload{Open: true, Key: "xxx1"}
		ev, err := marshaler.Marshal(payload)
		require.NoError(t, err)
		require.NoError(t, pubSub.Publish("servers.event.refreshed", ev))
		ev.SetContext(ctx)

		select {
		case <-wait:
		case <-ctx.Done():
			t.Fatal("timeout")
		}

		require.Equal(t, 1, serverSvc.UpdateInfoAndConfigCallCount())
		_, key, url := serverSvc.UpdateInfoAndConfigArgsForCall(0)
		assert.Equal(t, payload.Key, key)
		assert.Equal(t, payload.URL, url)
	})

	t.Run("OK: only open servers should be updated", func(t *testing.T) {
		t.Parallel()

		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()

		marshaler := msg.GobMarshaler{}
		pubSub := newPubSub(t)
		serverSvc := &mock.FakeServerService{}

		waitMiddleware, wait := newWaitForHandlerMiddleware("ServerConsumer.updateInfoAndConfig")
		runRouter(
			t,
			waitMiddleware,
			msg.NewServerConsumer(marshaler, pubSub, pubSub, watermill.NopLogger{}, serverSvc),
		)

		ev, err := marshaler.Marshal(model.ServerRefreshedEvPayload{Open: false})
		require.NoError(t, err)
		require.NoError(t, pubSub.Publish("servers.event.refreshed", ev))
		ev.SetContext(ctx)

		select {
		case <-wait:
		case <-ctx.Done():
			t.Fatal("timeout")
		}

		assert.Equal(t, 0, serverSvc.UpdateInfoAndConfigCallCount())
	})
}

func TestServerConsumer_updateNumPlayers(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	marshaler := msg.GobMarshaler{}
	pubSub := newPubSub(t)

	serverSvc := &mock.FakeServerService{}
	serverSvc.UpdateNumPlayersReturns(nil)

	waitMiddleware, wait := newWaitForHandlerMiddleware("ServerConsumer.updateNumPlayers")
	runRouter(
		t,
		waitMiddleware,
		msg.NewServerConsumer(marshaler, pubSub, pubSub, watermill.NopLogger{}, serverSvc),
	)

	payload := model.PlayersRefreshedEvPayload{
		Key:         "pl151",
		URL:         "https://pl151.plemiona.pl",
		VersionCode: "pl",
		NumPlayers:  12345,
	}
	ev, err := marshaler.Marshal(payload)
	require.NoError(t, err)
	require.NoError(t, pubSub.Publish("players.event.refreshed", ev))
	ev.SetContext(ctx)

	select {
	case <-wait:
	case <-ctx.Done():
		t.Fatal("timeout")
	}

	require.Equal(t, 1, serverSvc.UpdateNumPlayersCallCount())
	_, key, numPlayers := serverSvc.UpdateNumPlayersArgsForCall(0)
	assert.Equal(t, payload.Key, key)
	assert.Equal(t, payload.NumPlayers, numPlayers)
}

func TestServerConsumer_updateNumTribes(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	marshaler := msg.GobMarshaler{}
	pubSub := newPubSub(t)

	serverSvc := &mock.FakeServerService{}
	serverSvc.UpdateNumTribesReturns(nil)

	waitMiddleware, wait := newWaitForHandlerMiddleware("ServerConsumer.updateNumTribes")
	runRouter(
		t,
		waitMiddleware,
		msg.NewServerConsumer(marshaler, pubSub, pubSub, watermill.NopLogger{}, serverSvc),
	)

	payload := model.TribesRefreshedEvPayload{
		Key:         "pl151",
		URL:         "https://pl151.plemiona.pl",
		VersionCode: "pl",
		NumTribes:   12345,
	}
	ev, err := marshaler.Marshal(payload)
	require.NoError(t, err)
	require.NoError(t, pubSub.Publish("tribes.event.refreshed", ev))
	ev.SetContext(ctx)

	select {
	case <-wait:
	case <-ctx.Done():
		t.Fatal("timeout")
	}

	require.Equal(t, 1, serverSvc.UpdateNumTribesCallCount())
	_, key, numTribes := serverSvc.UpdateNumTribesArgsForCall(0)
	assert.Equal(t, payload.Key, key)
	assert.Equal(t, payload.NumTribes, numTribes)
}

func TestServerConsumer_updateNumVillages(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	marshaler := msg.GobMarshaler{}
	pubSub := newPubSub(t)

	serverSvc := &mock.FakeServerService{}
	serverSvc.UpdateNumVillagesReturns(nil)

	waitMiddleware, wait := newWaitForHandlerMiddleware("ServerConsumer.updateNumVillages")
	runRouter(
		t,
		waitMiddleware,
		msg.NewServerConsumer(marshaler, pubSub, pubSub, watermill.NopLogger{}, serverSvc),
	)

	payload := model.VillagesRefreshedEvPayload{
		Key:                  "pl151",
		URL:                  "https://pl151.plemiona.pl",
		VersionCode:          "pl",
		NumVillages:          12345,
		NumPlayerVillages:    12000,
		NumBarbarianVillages: 345,
		NumBonusVillages:     521,
	}
	ev, err := marshaler.Marshal(payload)
	require.NoError(t, err)
	require.NoError(t, pubSub.Publish("villages.event.refreshed", ev))
	ev.SetContext(ctx)

	select {
	case <-wait:
	case <-ctx.Done():
		t.Fatal("timeout")
	}

	require.Equal(t, 1, serverSvc.UpdateNumVillagesCallCount())
	_, key, refreshVillagesRes := serverSvc.UpdateNumVillagesArgsForCall(0)
	assert.Equal(t, payload.Key, key)
	assert.Equal(t, domain.RefreshVillagesResult{
		NumVillages:          payload.NumVillages,
		NumPlayerVillages:    payload.NumPlayerVillages,
		NumBarbarianVillages: payload.NumBarbarianVillages,
		NumBonusVillages:     payload.NumBonusVillages,
	}, refreshVillagesRes)
}

func TestServerConsumer_updateEnnoblementDataUpdatedAt(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	marshaler := msg.GobMarshaler{}
	pubSub := newPubSub(t)

	serverSvc := &mock.FakeServerService{}
	serverSvc.UpdateNumVillagesReturns(nil)

	waitMiddleware, wait := newWaitForHandlerMiddleware("ServerConsumer.updateEnnoblementDataUpdatedAt")
	runRouter(
		t,
		waitMiddleware,
		msg.NewServerConsumer(marshaler, pubSub, pubSub, watermill.NopLogger{}, serverSvc),
	)

	payload := model.EnnoblementsRefreshedEvPayload{
		Key:         "pl151",
		URL:         "https://pl151.plemiona.pl",
		VersionCode: "pl",
	}
	ev, err := marshaler.Marshal(payload)
	require.NoError(t, err)
	require.NoError(t, pubSub.Publish("ennoblements.event.refreshed", ev))
	ev.SetContext(ctx)

	select {
	case <-wait:
	case <-ctx.Done():
		t.Fatal("timeout")
	}

	require.Equal(t, 1, serverSvc.UpdateEnnoblementDataUpdatedAtCallCount())
	_, key := serverSvc.UpdateEnnoblementDataUpdatedAtArgsForCall(0)
	assert.Equal(t, payload.Key, key)
}

func TestServerConsumer_updatePlayerSnapshotsCreatedAt(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	marshaler := msg.GobMarshaler{}
	pubSub := newPubSub(t)

	serverSvc := &mock.FakeServerService{}
	serverSvc.UpdateNumVillagesReturns(nil)

	waitMiddleware, wait := newWaitForHandlerMiddleware("ServerConsumer.updatePlayerSnapshotsCreatedAt")
	runRouter(
		t,
		waitMiddleware,
		msg.NewServerConsumer(marshaler, pubSub, pubSub, watermill.NopLogger{}, serverSvc),
	)

	payload := model.SnapshotsCreatedEvPayload{
		Key:         "pl151",
		VersionCode: "pl",
	}
	ev, err := marshaler.Marshal(payload)
	require.NoError(t, err)
	require.NoError(t, pubSub.Publish("players.event.snapshots_created", ev))
	ev.SetContext(ctx)

	select {
	case <-wait:
	case <-ctx.Done():
		t.Fatal("timeout")
	}

	require.Equal(t, 1, serverSvc.UpdatePlayerSnapshotsCreatedAtCallCount())
	_, key := serverSvc.UpdatePlayerSnapshotsCreatedAtArgsForCall(0)
	assert.Equal(t, payload.Key, key)
}

func TestServerConsumer_updateTribeSnapshotsCreatedAt(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	marshaler := msg.GobMarshaler{}
	pubSub := newPubSub(t)

	serverSvc := &mock.FakeServerService{}
	serverSvc.UpdateNumVillagesReturns(nil)

	waitMiddleware, wait := newWaitForHandlerMiddleware("ServerConsumer.updateTribeSnapshotsCreatedAt")
	runRouter(
		t,
		waitMiddleware,
		msg.NewServerConsumer(marshaler, pubSub, pubSub, watermill.NopLogger{}, serverSvc),
	)

	payload := model.SnapshotsCreatedEvPayload{
		Key:         "pl151",
		VersionCode: "pl",
	}
	ev, err := marshaler.Marshal(payload)
	require.NoError(t, err)
	require.NoError(t, pubSub.Publish("tribes.event.snapshots_created", ev))
	ev.SetContext(ctx)

	select {
	case <-wait:
	case <-ctx.Done():
		t.Fatal("timeout")
	}

	require.Equal(t, 1, serverSvc.UpdateTribeSnapshotsCreatedAtCallCount())
	_, key := serverSvc.UpdateTribeSnapshotsCreatedAtArgsForCall(0)
	assert.Equal(t, payload.Key, key)
}