package create

import (
	"context"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	biz_changefeed_mock "github.com/tidbcloud/dataflow-service/app/models/biz_changefeed/mock"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/app/workflow/action/biz_changefeed"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

func TestActionFinalStepRun(t *testing.T) {
	mockBizChangefeedRepo := repo_mock.NewBizChangefeedRepo(t)
	mockBizChangefeedHooker := biz_changefeed_mock.NewBizChangefeedHooker(t)
	action := &ActionFinalStep{
		bizChangefeedRepo:   mockBizChangefeedRepo,
		bizChangefeedHooker: mockBizChangefeedHooker,
	}
	params := &biz_changefeed.CommonParams{
		BizChangefeedID: 111,
	}
	logger := zap.NewNop()

	t.Run("load entity error", func(t *testing.T) {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := mockBizChangefeedRepo.
			On("Load", mock.Anything, params.BizChangefeedID).
			Return(nil, mockErr)
		defer call.Unset()
		err := action.Run(context.Background(), logger, params)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "load biz_changefeed entity")
	})

	mockEnt := &entity.BizChangefeed{}
	loadCall := mockBizChangefeedRepo.
		On("Load", mock.Anything, params.BizChangefeedID).
		Return(mockEnt, nil)
	defer loadCall.Unset()

	t.Run("update entity error", func(t *testing.T) {
		mockEnt.OperationStatus = ""
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := mockBizChangefeedRepo.On("Update", mock.Anything, mock.Anything).Return(mockErr)
		defer call.Unset()
		err := action.Run(context.Background(), logger, params)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "update biz_changefeed status")
	})

	hookCall := mockBizChangefeedHooker.
		On("HandleBizChangefeedEvent", mock.Anything, entity.CreateChangefeed, uint64(0), mockEnt).
		Return()
	defer hookCall.Unset()
	t.Run("entity status is failed", func(t *testing.T) {
		mockEnt.OperationStatus = entity.BizChangefeedOperationStatusFailed
		err := action.Run(context.Background(), logger, params)
		assert.NoError(t, err)
	})

	call := mockBizChangefeedRepo.On("Update", mock.Anything, mock.Anything).Return(nil)
	defer call.Unset()
	t.Run("update successfully", func(t *testing.T) {
		mockEnt.OperationStatus = ""
		err := action.Run(context.Background(), logger, params)
		assert.NoError(t, err)
		matcher := func(ent *entity.BizChangefeed) bool {
			return ent.OperationStatus == entity.BizChangefeedOperationStatusNormal
		}
		mockBizChangefeedRepo.AssertCalled(t, "Update", mock.Anything, mock.MatchedBy(matcher))
	})
}