func (s *FrontendHostSuite) TestFrontendHostCreateConsumerGroupWithBadName() {
	testPath := s.generateKey("/foo/bar")
	frontendHost, ctx := s.utilGetContextAndFrontend()
	req := c.NewCreateConsumerGroupRequest()
	req.DestinationPath = common.StringPtr(testPath)
	req.ConsumerGroupName = common.StringPtr("/CGBadName)(*(*&$#)")
	cgd, err := frontendHost.CreateConsumerGroup(ctx, req)
	s.Error(err)
	s.Nil(cgd)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
TestFrontendHostCreateConsumerGroupWithBadName tests that a consumer group can't be created under an existing, good destination path, but with a bad name
package frontendhost
import (
	"fmt"
	"math"
	"strconv"
	"strings"
	"sync/atomic"
	"testing"
	"time"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/configure"
	dconfig "github.com/uber/cherami-server/common/dconfigclient"
	mockcommon "github.com/uber/cherami-server/test/mocks/common"
	mockctrl "github.com/uber/cherami-server/test/mocks/controllerhost"
	mockmeta "github.com/uber/cherami-server/test/mocks/metadata"
	c "github.com/uber/cherami-thrift/.generated/go/cherami"
	"github.com/uber/cherami-thrift/.generated/go/controller"
	"github.com/uber/cherami-thrift/.generated/go/shared"
	"github.com/pborman/uuid"
	log "github.com/sirupsen/logrus"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	"github.com/uber/tchannel-go"
	"github.com/uber/tchannel-go/thrift"
	"golang.org/x/net/context"
)
var logHasBeenSetup bool
type FrontendHostSuite struct {
	*require.Assertions  
	suite.Suite
	seqNum         int32
	cfg            configure.CommonAppConfig
	mockService    *common.MockService
	mockController *mockctrl.MockControllerHost
	mockMeta       *mockmeta.TChanMetadataService
}
func TestFrontendHostSuite(t *testing.T) {
	s := new(FrontendHostSuite)
	suite.Run(t, s)
}
func destCreateRequestToDesc(createRequest *c.CreateDestinationRequest) *shared.DestinationDescription {
	destDesc := shared.NewDestinationDescription()
	destDesc.Path = common.StringPtr(createRequest.GetPath())
	destDesc.DestinationUUID = common.StringPtr(uuid.New())
	destDesc.Type = common.InternalDestinationTypePtr(shared.DestinationType(createRequest.GetType()))
	destDesc.Status = common.InternalDestinationStatusPtr(shared.DestinationStatus_ENABLED)
	destDesc.ConsumedMessagesRetention = common.Int32Ptr(createRequest.GetConsumedMessagesRetention())
	destDesc.UnconsumedMessagesRetention = common.Int32Ptr(createRequest.GetUnconsumedMessagesRetention())
	destDesc.OwnerEmail = common.StringPtr(createRequest.GetOwnerEmail())
	destDesc.ChecksumOption = common.InternalChecksumOptionPtr(shared.ChecksumOption(createRequest.GetChecksumOption()))
	destDesc.KafkaCluster = common.StringPtr(createRequest.GetKafkaCluster())
	destDesc.KafkaTopics = createRequest.KafkaTopics
	destDesc.IsMultiZone = common.BoolPtr(createRequest.GetIsMultiZone())
	if createRequest.IsSetZoneConfigs() {
		destDesc.ZoneConfigs = make([]*shared.DestinationZoneConfig, 0, len(createRequest.GetZoneConfigs().GetConfigs()))
		for _, destZoneCfg := range createRequest.GetZoneConfigs().GetConfigs() {
			destDesc.ZoneConfigs = append(destDesc.ZoneConfigs, convertDestZoneConfigToInternal(destZoneCfg))
		}
	}
	return destDesc
}
func destUpdateRequestToDesc(updateRequest *c.UpdateDestinationRequest) *shared.DestinationDescription {
	destDesc := shared.NewDestinationDescription()
	destDesc.Path = common.StringPtr(updateRequest.GetPath())
	destDesc.DestinationUUID = common.StringPtr(uuid.New())
	destDesc.Status = common.InternalDestinationStatusPtr(shared.DestinationStatus(updateRequest.GetStatus()))
	destDesc.ConsumedMessagesRetention = common.Int32Ptr(updateRequest.GetConsumedMessagesRetention())
	destDesc.UnconsumedMessagesRetention = common.Int32Ptr(updateRequest.GetUnconsumedMessagesRetention())
	destDesc.OwnerEmail = common.StringPtr(updateRequest.GetOwnerEmail())
	destDesc.ChecksumOption = common.InternalChecksumOptionPtr(shared.ChecksumOption(updateRequest.GetChecksumOption()))
	return destDesc
}
func cgCreateRequestToDesc(createRequest *c.CreateConsumerGroupRequest) *shared.ConsumerGroupDescription {
	cgDesc := shared.NewConsumerGroupDescription()
	cgDesc.ConsumerGroupUUID = common.StringPtr(uuid.New())
	cgDesc.DestinationUUID = common.StringPtr(uuid.New())
	cgDesc.ConsumerGroupName = common.StringPtr(createRequest.GetConsumerGroupName())
	cgDesc.StartFrom = common.Int64Ptr(createRequest.GetStartFrom())
	cgDesc.Status = common.InternalConsumerGroupStatusPtr(shared.ConsumerGroupStatus_ENABLED)
	cgDesc.LockTimeoutSeconds = common.Int32Ptr(createRequest.GetLockTimeoutInSeconds())
	cgDesc.MaxDeliveryCount = common.Int32Ptr(createRequest.GetMaxDeliveryCount())
	cgDesc.SkipOlderMessagesSeconds = common.Int32Ptr(createRequest.GetSkipOlderMessagesInSeconds())
	cgDesc.DelaySeconds = common.Int32Ptr(createRequest.GetDelaySeconds())
	cgDesc.DeadLetterQueueDestinationUUID = common.StringPtr(uuid.New())
	cgDesc.OwnerEmail = common.StringPtr(createRequest.GetOwnerEmail())
	cgDesc.ConsumerGroupType = common.InternalConsumerGroupTypePtr(shared.ConsumerGroupType(createRequest.GetConsumerGroupType()))
	cgDesc.IsMultiZone = common.BoolPtr(createRequest.GetIsMultiZone())
	if createRequest.IsSetZoneConfigs() {
		cgDesc.ZoneConfigs = make([]*shared.ConsumerGroupZoneConfig, 0, len(createRequest.GetZoneConfigs().GetConfigs()))
		for _, cgZoneCfg := range createRequest.GetZoneConfigs().GetConfigs() {
			cgDesc.ZoneConfigs = append(cgDesc.ZoneConfigs, convertCGZoneConfigToInternal(cgZoneCfg))
		}
	}
	cgDesc.Options = createRequest.GetOptions()
	return cgDesc
}
func cgUpdateRequestToDesc(updateRequest *c.UpdateConsumerGroupRequest) *shared.ConsumerGroupDescription {
	cgDesc := shared.NewConsumerGroupDescription()
	cgDesc.ConsumerGroupUUID = common.StringPtr(uuid.New())
	cgDesc.DestinationUUID = common.StringPtr(uuid.New())
	cgDesc.ConsumerGroupName = common.StringPtr(updateRequest.GetConsumerGroupName())
	cgDesc.Status = common.InternalConsumerGroupStatusPtr(shared.ConsumerGroupStatus(updateRequest.GetStatus()))
	cgDesc.LockTimeoutSeconds = common.Int32Ptr(updateRequest.GetLockTimeoutInSeconds())
	cgDesc.MaxDeliveryCount = common.Int32Ptr(updateRequest.GetMaxDeliveryCount())
	cgDesc.SkipOlderMessagesSeconds = common.Int32Ptr(updateRequest.GetSkipOlderMessagesInSeconds())
	cgDesc.DelaySeconds = common.Int32Ptr(updateRequest.GetDelaySeconds())
	cgDesc.OwnerEmail = common.StringPtr(updateRequest.GetOwnerEmail())
	cgDesc.Options = updateRequest.GetOptions()
	return cgDesc
}
func (s *FrontendHostSuite) generateKey(prefix string) string {
	seq := atomic.AddInt32(&s.seqNum, 1)
	return prefix + "." + strconv.Itoa(int(seq))
}
func (s *FrontendHostSuite) SetupCommonMock() {
	rpm := common.NewMockRingpopMonitor()
	rpm.Add(common.OutputServiceName, "99999999-9999-9999-9999-999999999999", "127.0.0.1")
	s.mockService = new(common.MockService)
	s.mockController = new(mockctrl.MockControllerHost)
	s.mockMeta = new(mockmeta.TChanMetadataService)
	mockClientFactory := new(mockcommon.MockClientFactory)
	mockClientFactory.On("GetControllerClient").Return(s.mockController, nil)
	s.mockService.On("GetConfig").Return(s.cfg.GetServiceConfig(common.FrontendServiceName))
	s.mockService.On("GetTChannel").Return(tchannel.NewChannel("test-frontend", nil))
	s.mockService.On("GetHostPort").Return("inputhost:port")
	s.mockService.On("GetHostUUID").Return("99999999-9999-9999-9999-999999999999")
	s.mockService.On("GetRingpopMonitor").Return(rpm)
	s.mockService.On("GetClientFactory").Return(mockClientFactory)
	s.mockService.On("GetMetricsReporter").Return(common.NewMetricReporterWithHostname(configure.NewCommonServiceConfig()))
	s.mockService.On("GetDConfigClient").Return(dconfig.NewDconfigClient(configure.NewCommonServiceConfig(), common.FrontendServiceName))
	s.mockService.On("GetAuthManager").Return(common.NewBypassAuthManager())
}
func (s *FrontendHostSuite) SetupTest() {
	s.Assertions = require.New(s.T())  
	s.cfg = common.SetupServerConfig(configure.NewCommonConfigure())
	s.SetupCommonMock()
	log.Infof("Testing has begun")
}
func (s *FrontendHostSuite) TearDownTest() {
	return
}
func utilGetThriftContext() (thrift.Context, context.CancelFunc) {
	return thrift.NewContext(10 * time.Minute)
}
func (s *FrontendHostSuite) utilGetContextAndFrontend() (frontendHost *Frontend, ctx thrift.Context) {
	appConfig := configure.NewCommonAppConfig()
	sCfgInput := configure.NewCommonServiceConfig()
	sCfgInput.SetWebsocketPort(123)
	sCfgOutput := configure.NewCommonServiceConfig()
	sCfgOutput.SetWebsocketPort(456)
	appConfig.SetServiceConfig(common.InputServiceName, sCfgInput)
	appConfig.SetServiceConfig(common.OutputServiceName, sCfgOutput)
	sName := "cherami-test-frontend"
	frontendHost, _ = NewFrontendHost(sName, s.mockService, s.mockMeta, appConfig)
	ctx, _ = utilGetThriftContext()
	return
}
func (s *FrontendHostSuite) TestFrontendHostCreateDestinationRejectNil() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	dst, err := frontendHost.CreateDestination(ctx, nil)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostCreateDestinationRejectBadPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	req := c.NewCreateDestinationRequest()
	req.Path = common.StringPtr("/fsdf8234*@#($*")
	req.ConsumedMessagesRetention = common.Int32Ptr(15)
	req.UnconsumedMessagesRetention = common.Int32Ptr(30)
	req.OwnerEmail = common.StringPtr("test@uber.com")
	req.Type = c.DestinationTypePtr(c.DestinationType_PLAIN)
	dst, err := frontendHost.CreateDestination(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostCreateDestination() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	path := s.generateKey("/foo/bar")
	req := c.NewCreateDestinationRequest()
	req.Path = common.StringPtr(path)
	req.ConsumedMessagesRetention = common.Int32Ptr(15)
	req.UnconsumedMessagesRetention = common.Int32Ptr(30)
	req.OwnerEmail = common.StringPtr("test@uber.com")
	req.Type = c.DestinationTypePtr(c.DestinationType_TIMER)
	req.IsMultiZone = common.BoolPtr(true)
	req.ZoneConfigs = &c.DestinationZoneConfigs{
		Configs: []*c.DestinationZoneConfig{
			{
				Zone:                   common.StringPtr("zone1"),
				AllowPublish:           common.BoolPtr(true),
				AllowConsume:           common.BoolPtr(true),
				AlwaysReplicateTo:      common.BoolPtr(true),
				RemoteExtentReplicaNum: common.Int32Ptr(3),
			},
		},
	}
	s.mockController.On("CreateDestination", mock.Anything, mock.Anything).Return(destCreateRequestToDesc(req), nil).Once().Run(func(args mock.Arguments) {
		createReq := args.Get(1).(*shared.CreateDestinationRequest)
		s.Equal(createReq.GetPath(), req.GetPath())
		s.Equal(createReq.GetConsumedMessagesRetention(), req.GetConsumedMessagesRetention())
		s.Equal(createReq.GetUnconsumedMessagesRetention(), req.GetUnconsumedMessagesRetention())
		s.Equal(createReq.GetType(), shared.DestinationType(req.GetType()))
		s.Equal(createReq.GetOwnerEmail(), req.GetOwnerEmail())
		s.Equal(createReq.GetIsMultiZone(), req.GetIsMultiZone())
		s.Equal(len(createReq.GetZoneConfigs()), len(req.GetZoneConfigs().GetConfigs()))
		s.Equal(createReq.GetZoneConfigs()[0].GetRemoteExtentReplicaNum(), req.GetZoneConfigs().GetConfigs()[0].GetRemoteExtentReplicaNum())
		s.Equal(createReq.GetKafkaCluster(), req.GetKafkaCluster())
		s.True(common.StringSetEqual(createReq.GetKafkaTopics(), req.GetKafkaTopics()))
	})
	dst, err := frontendHost.CreateDestination(ctx, req)
	s.NoError(err)
	s.NotNil(dst)
	if dst != nil {
		s.Equal(dst.GetPath(), req.GetPath())
		s.Equal(dst.GetConsumedMessagesRetention(), req.GetConsumedMessagesRetention())
		s.Equal(dst.GetUnconsumedMessagesRetention(), req.GetUnconsumedMessagesRetention())
		s.Equal(dst.GetType(), req.GetType())
		s.Equal(dst.GetOwnerEmail(), req.GetOwnerEmail())
		s.Equal(dst.GetIsMultiZone(), req.GetIsMultiZone())
		s.Equal(len(dst.ZoneConfigs.GetConfigs()), len(req.ZoneConfigs.GetConfigs()))
		s.Equal(dst.ZoneConfigs.GetConfigs()[0].GetRemoteExtentReplicaNum(), req.ZoneConfigs.GetConfigs()[0].GetRemoteExtentReplicaNum())
		s.Equal(dst.GetKafkaCluster(), req.GetKafkaCluster())
		s.True(common.StringSetEqual(dst.GetKafkaTopics(), req.GetKafkaTopics()))
	}
}
func (s *FrontendHostSuite) TestFrontendHostCreateKafkaDestination() {
	defer s.resetMocks()
	frontendHost, ctx := s.utilGetContextAndFrontend()
	path := s.generateKey("/foo/bar")
	req := c.NewCreateDestinationRequest()
	req.Path = common.StringPtr(path)
	req.ConsumedMessagesRetention = common.Int32Ptr(15)
	req.UnconsumedMessagesRetention = common.Int32Ptr(30)
	req.OwnerEmail = common.StringPtr("test@uber.com")
	for _, f := range []func() (bool, string){
		func() (bool, string) {
			req.Type = c.DestinationTypePtr(c.DestinationType_PLAIN)
			return true, `base case plain`
		}, func() (bool, string) {
			req.KafkaCluster = common.StringPtr(`foo`)
			req.KafkaTopics = nil
			return false, `Should fail because topics not set`
		}, func() (bool, string) {
			req.KafkaCluster = common.StringPtr(`foo`)
			req.KafkaTopics = make([]string, 0)
			return false, `Should fail because topics len zero`
		}, func() (bool, string) {
			req.KafkaCluster = common.StringPtr(``)
			req.KafkaTopics = []string{`bar`}
			return false, `Should fail because cluster len zero`
		}, func() (bool, string) {
			req.KafkaCluster = common.StringPtr(`foo`)
			req.KafkaTopics = []string{`bar`}
			req.IsMultiZone = common.BoolPtr(true)
			req.ZoneConfigs = &c.DestinationZoneConfigs{
				Configs: []*c.DestinationZoneConfig{
					{
						Zone:                   common.StringPtr("zone1"),
						AllowPublish:           common.BoolPtr(true),
						AllowConsume:           common.BoolPtr(true),
						AlwaysReplicateTo:      common.BoolPtr(true),
						RemoteExtentReplicaNum: common.Int32Ptr(3),
					},
				},
			}
			return false, `Should fail because multi-zone is not allowed for a KAFKA destination`
		}, func() (bool, string) {
			req.KafkaCluster = common.StringPtr(`foo`)
			req.KafkaTopics = []string{`bar`}
			req.Type = c.DestinationTypePtr(c.DestinationType_PLAIN)
			return false, `Should fail because KAFKA config only allowed with type KAFKA`
		}, func() (bool, string) {
			return false, `kafka destination requires kafka config`
		}, func() (bool, string) {
			req.KafkaCluster = common.StringPtr(`foo`)
			req.KafkaTopics = []string{`bar`, ``, `quz`}
			return false, `topics should have no empty strings`
		}, func() (bool, string) {
			req.KafkaCluster = common.StringPtr(`foo`)
			req.KafkaTopics = []string{`bar`, `baz`, `quz`}
			return true, `should work with multiple topics`
		}, func() (bool, string) {
			req.KafkaCluster = common.StringPtr(`foo`)
			req.KafkaTopics = []string{`bar`}
			return true, `should work with one topic`
		},
	} {
		req.Type = c.DestinationTypePtr(c.DestinationType_KAFKA)
		req.IsMultiZone = common.BoolPtr(false)
		req.ZoneConfigs = nil
		req.KafkaCluster = nil
		req.KafkaTopics = nil
		shouldPass, testMsg := f()
		s.resetMocks()
		s.mockController.On("CreateDestination", mock.Anything, mock.Anything).Return(destCreateRequestToDesc(req), nil).Once().Run(func(args mock.Arguments) {
			createReq := args.Get(1).(*shared.CreateDestinationRequest)
			s.Equal(createReq.GetPath(), req.GetPath(), testMsg)
			s.Equal(createReq.GetConsumedMessagesRetention(), req.GetConsumedMessagesRetention(), testMsg)
			s.Equal(createReq.GetUnconsumedMessagesRetention(), req.GetUnconsumedMessagesRetention(), testMsg)
			s.Equal(createReq.GetType(), shared.DestinationType(req.GetType()), testMsg)
			s.Equal(createReq.GetOwnerEmail(), req.GetOwnerEmail(), testMsg)
			s.Equal(createReq.GetIsMultiZone(), req.GetIsMultiZone(), testMsg)
			s.Equal(createReq.GetKafkaCluster(), req.GetKafkaCluster(), testMsg)
			s.True(common.StringSetEqual(createReq.GetKafkaTopics(), req.GetKafkaTopics()), testMsg)
		})
		dst, err := frontendHost.CreateDestination(ctx, req)
		if shouldPass {
			s.NoError(err, testMsg)
			s.NotNil(dst, testMsg)
		} else {
			s.Error(err, testMsg)
			s.Nil(dst, testMsg)
		}
		if dst != nil {
			s.Equal(dst.GetPath(), req.GetPath(), testMsg)
			s.Equal(dst.GetConsumedMessagesRetention(), req.GetConsumedMessagesRetention(), testMsg)
			s.Equal(dst.GetUnconsumedMessagesRetention(), req.GetUnconsumedMessagesRetention(), testMsg)
			s.Equal(dst.GetType(), req.GetType(), testMsg)
			s.Equal(dst.GetOwnerEmail(), req.GetOwnerEmail(), testMsg)
			s.Equal(dst.GetIsMultiZone(), req.GetIsMultiZone(), testMsg)
			s.Equal(dst.GetKafkaCluster(), req.GetKafkaCluster(), testMsg)
			s.True(common.StringSetEqual(dst.GetKafkaTopics(), req.GetKafkaTopics()), testMsg)
		}
	}
}
func (s *FrontendHostSuite) TestFrontendHostReadDestinationRejectNil() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	dst, err := frontendHost.ReadDestination(ctx, nil)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadDestinationRejectBadPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	req := c.NewReadDestinationRequest()
	req.Path = common.StringPtr("/fsdf8234*@#($*")
	dst, err := frontendHost.ReadDestination(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadDestinationNoExistPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(nil, &shared.EntityNotExistsError{})
	req := c.NewReadDestinationRequest()
	req.Path = common.StringPtr(s.generateKey("/foo/bad"))
	dst, err := frontendHost.ReadDestination(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.EntityNotExistsError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadDestination() {
	testPath := s.generateKey("/foo/bar")
	frontendHost, ctx := s.utilGetContextAndFrontend()
	createReq := c.NewCreateDestinationRequest()
	createReq.Path = common.StringPtr(testPath)
	createReq.ConsumedMessagesRetention = common.Int32Ptr(15)
	createReq.UnconsumedMessagesRetention = common.Int32Ptr(30)
	createReq.OwnerEmail = common.StringPtr("test@uber.com")
	createReq.Type = c.DestinationTypePtr(c.DestinationType_TIMER)
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(destCreateRequestToDesc(createReq), nil)
	req := c.NewReadDestinationRequest()
	req.Path = common.StringPtr(testPath)
	dst, err := frontendHost.ReadDestination(ctx, req)
	s.NoError(err)
	s.NotNil(dst)
	if dst != nil {
		s.Equal(dst.GetPath(), testPath)
		s.Equal(dst.GetConsumedMessagesRetention(), int32(15))
		s.Equal(dst.GetUnconsumedMessagesRetention(), int32(30))
		s.Equal(dst.GetOwnerEmail(), "test@uber.com")
		s.Equal(dst.GetType(), c.DestinationType_TIMER)
	}
}
func (s *FrontendHostSuite) TestFrontendHostUpdateDestinationRejectBadPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	req := c.NewUpdateDestinationRequest()
	req.Path = common.StringPtr("/fsdf8234*@#($*")
	req.Status = c.DestinationStatusPtr(c.DestinationStatus_SENDONLY)
	req.ConsumedMessagesRetention = common.Int32Ptr(1800)
	req.UnconsumedMessagesRetention = common.Int32Ptr(3600)
	req.OwnerEmail = common.StringPtr("test_up@uber.com")
	dst, err := frontendHost.UpdateDestination(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostUpdateDestinationNoExistPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	req := c.NewUpdateDestinationRequest()
	req.Path = common.StringPtr(s.generateKey("/foo/bad"))
	req.Status = c.DestinationStatusPtr(c.DestinationStatus_SENDONLY)
	req.ConsumedMessagesRetention = common.Int32Ptr(1800)
	req.UnconsumedMessagesRetention = common.Int32Ptr(3600)
	req.OwnerEmail = common.StringPtr("test_up@uber.com")
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(nil, &shared.EntityNotExistsError{})
	dst, err := frontendHost.UpdateDestination(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.EntityNotExistsError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostUpdateDestination() {
	testPath := s.generateKey("/foo/bar")
	frontendHost, ctx := s.utilGetContextAndFrontend()
	req := c.NewUpdateDestinationRequest()
	req.Path = common.StringPtr(testPath)
	req.Status = c.DestinationStatusPtr(c.DestinationStatus_SENDONLY)
	req.ConsumedMessagesRetention = common.Int32Ptr(1800)
	req.UnconsumedMessagesRetention = common.Int32Ptr(3600)
	req.OwnerEmail = common.StringPtr("test_up@uber.com")
	destDesc := destUpdateRequestToDesc(req)
	s.mockController.On("UpdateDestination", mock.Anything, mock.Anything).Return(destDesc, nil)
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(destDesc, nil)
	dst, err := frontendHost.UpdateDestination(ctx, req)
	s.NoError(err)
	s.NotNil(dst)
	if dst != nil {
		s.Equal(dst.GetPath(), testPath)
		s.Equal(dst.GetConsumedMessagesRetention(), int32(1800))
		s.Equal(dst.GetUnconsumedMessagesRetention(), int32(3600))
		s.Equal(dst.GetStatus(), c.DestinationStatus_SENDONLY)
		s.Equal(dst.GetOwnerEmail(), "test_up@uber.com")
	}
}
func (s *FrontendHostSuite) TestFrontendHostDeleteDestinationRejectBadPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	req := c.NewDeleteDestinationRequest()
	req.Path = common.StringPtr("/fsdf8234*@#($*")
	err := frontendHost.DeleteDestination(ctx, req)
	s.Error(err)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostDeleteDestinationNoExistPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	s.mockController.On("DeleteDestination", mock.Anything, mock.Anything).Return(c.NewEntityNotExistsError())
	req := c.NewDeleteDestinationRequest()
	req.Path = common.StringPtr(s.generateKey("/foo/bad"))
	err := frontendHost.DeleteDestination(ctx, req)
	s.Error(err)
}
func (s *FrontendHostSuite) TestFrontendHostDeleteDestination() {
	testPath := s.generateKey("/foo/bar")
	frontendHost, ctx := s.utilGetContextAndFrontend()
	s.mockController.On("DeleteDestination", mock.Anything, mock.Anything).Return(nil)
	req := c.NewDeleteDestinationRequest()
	req.Path = common.StringPtr(testPath)
	err := frontendHost.DeleteDestination(ctx, req)
	s.NoError(err)
}
func (s *FrontendHostSuite) TestFrontendHostReadPublisherOptionsRejectBadPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	req := c.NewReadPublisherOptionsRequest()
	req.Path = common.StringPtr("/fsdf8234*@#($*")
	dst, err := frontendHost.ReadPublisherOptions(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadPublisherOptionsNoExistPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(nil, &shared.EntityNotExistsError{})
	req := c.NewReadPublisherOptionsRequest()
	req.Path = common.StringPtr(s.generateKey("/foo/bad"))
	dst, err := frontendHost.ReadPublisherOptions(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.EntityNotExistsError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadPublisherOptionsKafka() {
	testPath := s.generateKey("/foo/bar")
	testUUID := uuid.New()
	frontendHost, ctx := s.utilGetContextAndFrontend()
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(&shared.DestinationDescription{
		Path:            common.StringPtr(testPath),
		DestinationUUID: common.StringPtr(testUUID),
		Type:            common.InternalDestinationTypePtr(shared.DestinationType_KAFKA),
	}, nil)
	req := c.NewReadPublisherOptionsRequest()
	req.Path = common.StringPtr(testPath)
	dst, err := frontendHost.ReadPublisherOptions(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadPublisherOptionsReceiveOnly() {
	testPath := s.generateKey("/foo/bar")
	testUUID := uuid.New()
	frontendHost, ctx := s.utilGetContextAndFrontend()
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(&shared.DestinationDescription{
		Path:            common.StringPtr(testPath),
		DestinationUUID: common.StringPtr(testUUID),
		Type:            common.InternalDestinationTypePtr(shared.DestinationType_PLAIN),
		Status:          common.InternalDestinationStatusPtr(shared.DestinationStatus_RECEIVEONLY),
	}, nil)
	req := c.NewReadPublisherOptionsRequest()
	req.Path = common.StringPtr(testPath)
	dst, err := frontendHost.ReadPublisherOptions(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadPublisherOptions() {
	testPath := s.generateKey("/foo/bar")
	frontendHost, ctx := s.utilGetContextAndFrontend()
	createReq := c.NewCreateDestinationRequest()
	createReq.Path = common.StringPtr(testPath)
	createReq.ConsumedMessagesRetention = common.Int32Ptr(15)
	createReq.UnconsumedMessagesRetention = common.Int32Ptr(30)
	createReq.OwnerEmail = common.StringPtr("test@uber.com")
	createReq.ChecksumOption = c.ChecksumOption_CRC32IEEE
	createReq.Type = c.DestinationTypePtr(c.DestinationType_TIMER)
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(destCreateRequestToDesc(createReq), nil)
	s.mockController.On("GetInputHosts", mock.Anything, mock.Anything).Return(&controller.GetInputHostsResult_{InputHostIds: []string{"127.0.0.1:1234"}}, nil)
	req := c.NewReadPublisherOptionsRequest()
	req.Path = common.StringPtr(testPath)
	pubOptions, err := frontendHost.ReadPublisherOptions(ctx, req)
	s.NoError(err)
	s.NotNil(pubOptions)
	s.Equal(2, len(pubOptions.GetHostProtocols()))
}
func (s *FrontendHostSuite) TestFrontendHostReadDestinationHostsRejectBadPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	req := c.NewReadDestinationHostsRequest()
	req.Path = common.StringPtr("/fsdf8234*@#($*")
	dst, err := frontendHost.ReadDestinationHosts(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadDestinationHostsNoExistPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(nil, &shared.EntityNotExistsError{})
	req := c.NewReadDestinationHostsRequest()
	req.Path = common.StringPtr(s.generateKey("/foo/bad"))
	dst, err := frontendHost.ReadDestinationHosts(ctx, req)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.EntityNotExistsError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadDestinationHostsKafka() {
	testPath := s.generateKey("/foo/bar")
	testUUID := uuid.New()
	frontendHost, ctx := s.utilGetContextAndFrontend()
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(&shared.DestinationDescription{
		Path:            common.StringPtr(testPath),
		DestinationUUID: common.StringPtr(testUUID),
		Type:            common.InternalDestinationTypePtr(shared.DestinationType_KAFKA),
	}, nil)
	req := c.NewReadDestinationHostsRequest()
	req.Path = common.StringPtr(testPath)
	hosts, err := frontendHost.ReadDestinationHosts(ctx, req)
	s.Error(err)
	s.Nil(hosts)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadDestinationHostsReceiveOnly() {
	testPath := s.generateKey("/foo/bar")
	testUUID := uuid.New()
	frontendHost, ctx := s.utilGetContextAndFrontend()
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(&shared.DestinationDescription{
		Path:            common.StringPtr(testPath),
		DestinationUUID: common.StringPtr(testUUID),
		Type:            common.InternalDestinationTypePtr(shared.DestinationType_PLAIN),
		Status:          common.InternalDestinationStatusPtr(shared.DestinationStatus_RECEIVEONLY),
	}, nil)
	req := c.NewReadDestinationHostsRequest()
	req.Path = common.StringPtr(testPath)
	hosts, err := frontendHost.ReadDestinationHosts(ctx, req)
	s.Error(err)
	s.Nil(hosts)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostReadDestinationHosts() {
	testPath := s.generateKey("/foo/bar")
	frontendHost, ctx := s.utilGetContextAndFrontend()
	createReq := c.NewCreateDestinationRequest()
	createReq.Path = common.StringPtr(testPath)
	createReq.ConsumedMessagesRetention = common.Int32Ptr(15)
	createReq.UnconsumedMessagesRetention = common.Int32Ptr(30)
	createReq.OwnerEmail = common.StringPtr("test@uber.com")
	createReq.Type = c.DestinationTypePtr(c.DestinationType_TIMER)
	s.mockMeta.On("ReadDestination", mock.Anything, mock.Anything).Return(destCreateRequestToDesc(createReq), nil)
	s.mockController.On("GetInputHosts", mock.Anything, mock.Anything).Return(&controller.GetInputHostsResult_{InputHostIds: []string{"127.0.0.1:1234"}}, nil)
	req := c.NewReadDestinationHostsRequest()
	req.Path = common.StringPtr(testPath)
	hosts, err := frontendHost.ReadDestinationHosts(ctx, req)
	s.NoError(err)
	s.NotNil(hosts)
	s.Equal(2, len(hosts.GetHostProtocols()))
}
func (s *FrontendHostSuite) TestFrontendHostCreateConsumerGroupRejectNil() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	dst, err := frontendHost.CreateConsumerGroup(ctx, nil)
	s.Error(err)
	s.Nil(dst)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}
func (s *FrontendHostSuite) TestFrontendHostCreateConsumerGroupRejectBadPath() {
	frontendHost, ctx := s.utilGetContextAndFrontend()
	req := c.NewCreateConsumerGroupRequest()
	req.DestinationPath = common.StringPtr("/fsdf8234*@#($*")
	req.ConsumerGroupName = common.StringPtr(s.generateKey("/GoodName"))
	cgd, err := frontendHost.CreateConsumerGroup(ctx, req)
	s.Error(err)
	s.Nil(cgd)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
	req.ConsumerGroupName = common.StringPtr("/fsdf8234*@#($*")
	req.DestinationPath = common.StringPtr(s.generateKey("/GoodName"))
	cgd, err = frontendHost.CreateConsumerGroup(ctx, req)
	s.Error(err)
	s.Nil(cgd)
	assert.IsType(s.T(), &c.BadRequestError{}, err)
}