package console

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/models/domain"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db/mock_db"
	"go.uber.org/mock/gomock"
)

func TestBizChangefeedSpecController_List(t *testing.T) {
	trans := mock_db.NewMockTransProvider(gomock.NewController(t))
	mockBizChangefeedModel := biz_changefeed_mock.NewBizChangefeedModel(t)
	controller := &bizChangefeedSpecControllerImpl{
		BaseController:  NewBaseController(trans),
		changefeedModel: mockBizChangefeedModel,
	}

	mockReq := &dataflow.ChangefeedInstanceTypeListReq{
		ClusterId: 111,
	}
	listResMatcher := mock.MatchedBy(func(req *domain.ListBizChangefeedSpecificationsReq) bool {
		return req.ClusterID == mockReq.ClusterId
	})
	t.Run("list failed", func(t *testing.T) {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := mockBizChangefeedModel.
			On("ListBizChangefeedSpecifications", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := controller.List(context.Background(), mockReq)
		assert.Error(t, err)
		assert.Equal(t, mockErr, err)
	})

	mockListResp := &domain.ListBizChangefeedSpecificationsResp{
		Total: 5,
		Specifications: []*domain.BizChangefeedSpecification{
			{
				SpecificationCode: "a",
				RCUNumber:         2,
				Replicas:          1,
				RPSLimit:          5000,
			},
			{
				SpecificationCode: "b",
				RCUNumber:         2,
				Replicas:          2,
				RPSLimit:          10000,
			},
			{
				SpecificationCode: "c",
				RCUNumber:         4,
				Replicas:          2,
				RPSLimit:          20000,
			},
			{
				SpecificationCode: "d",
				RCUNumber:         4,
				Replicas:          3,
				RPSLimit:          30000,
			},
			{
				SpecificationCode: "e",
				RCUNumber:         4,
				Replicas:          4,
				RPSLimit:          40000,
			},
		},
	}
	t.Run("list failed", func(t *testing.T) {
		call := mockBizChangefeedModel.
			On("ListBizChangefeedSpecifications", mock.Anything, mock.Anything).
			Return(mockListResp, nil)
		defer call.Unset()
		resp, err := controller.List(context.Background(), mockReq)
		assert.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Equal(t, mockListResp.Total, resp.Total)
		for i := range resp.Items {
			assert.Equal(t, mockListResp.Specifications[i].SpecificationCode, resp.Items[i].Name)
			assert.Equal(t, mockListResp.Specifications[i].RCUNumber, resp.Items[i].RcuNum)
			assert.Equal(t, mockListResp.Specifications[i].Replicas, resp.Items[i].Replicas)
			assert.Equal(t, mockListResp.Specifications[i].RPSLimit, resp.Items[i].RpsLimit)
		}
		mockBizChangefeedModel.AssertCalled(t, "ListBizChangefeedSpecifications", mock.Anything, listResMatcher)
	})
}