-
Notifications
You must be signed in to change notification settings - Fork 322
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
142 additions
and
17 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| package router | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/rudderlabs/rudder-go-kit/logger" | ||
| "github.com/rudderlabs/rudder-server/warehouse/internal/model" | ||
| "github.com/rudderlabs/rudder-server/warehouse/schema" | ||
| ) | ||
|
|
||
| type mockFetchSchemaRepo struct { | ||
| schemaInWarehouse model.Schema | ||
| err error | ||
| } | ||
|
|
||
| func (m *mockFetchSchemaRepo) FetchSchema(_ context.Context) (model.Schema, error) { | ||
| if m.err != nil { | ||
| return model.Schema{}, m.err | ||
| } | ||
| return m.schemaInWarehouse, nil | ||
| } | ||
|
|
||
| type mockSyncSchemaRepo struct { | ||
| getLocalSchemaerr error | ||
| updateLocalSchemaErr error | ||
| fetchSchemaErr error | ||
| hasChanged bool | ||
| schemaMap map[string]model.Schema | ||
| } | ||
|
|
||
| func schemaKey(sourceID, destinationID, namespace string) string { | ||
| return fmt.Sprintf("%s_%s_%s", sourceID, destinationID, namespace) | ||
| } | ||
|
|
||
| func (m *mockSyncSchemaRepo) GetLocalSchema(_ context.Context) (model.Schema, error) { | ||
| if m.getLocalSchemaerr != nil { | ||
| return model.Schema{}, m.getLocalSchemaerr | ||
| } | ||
|
|
||
| return model.Schema{}, nil | ||
| } | ||
|
|
||
| func (m *mockSyncSchemaRepo) UpdateLocalSchemaWithWarehouse(_ context.Context, _ model.Schema) error { | ||
| if m.updateLocalSchemaErr != nil && m.hasChanged { | ||
| return m.updateLocalSchemaErr | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (m *mockSyncSchemaRepo) HasSchemaChanged(_ model.Schema) bool { | ||
| return m.hasChanged | ||
| } | ||
|
|
||
| func (m *mockSyncSchemaRepo) FetchSchemaFromWarehouse(_ context.Context, _ schema.FetchSchemaRepo) (model.Schema, error) { | ||
| if m.fetchSchemaErr != nil { | ||
| return nil, m.fetchSchemaErr | ||
| } | ||
|
|
||
| return m.schemaMap[schemaKey("test-sourceID", "test-destinationID", "test-namespace")], nil | ||
| } | ||
|
|
||
| func TestSync_SyncRemoteSchema(t *testing.T) { | ||
| t.Run("fetching schema from local fails", func(t *testing.T) { | ||
| mock := &mockSyncSchemaRepo{ | ||
| getLocalSchemaerr: fmt.Errorf("error fetching schema from local"), | ||
| } | ||
| mockFetchSchema := &mockFetchSchemaRepo{} | ||
| r := &Router{ | ||
| logger: logger.NOP, | ||
| } | ||
| err := r.syncRemoteSchema(context.Background(), mockFetchSchema, mock) | ||
| require.Error(t, err) | ||
| require.True(t, errors.Is(err, mock.getLocalSchemaerr)) | ||
| }) | ||
| t.Run("fetching schema from warehouse fails", func(t *testing.T) { | ||
| mock := &mockSyncSchemaRepo{ | ||
| fetchSchemaErr: fmt.Errorf("error fetching schema from warehouse"), | ||
| } | ||
| mockFetchSchema := &mockFetchSchemaRepo{} | ||
| r := &Router{ | ||
| logger: logger.NOP, | ||
| } | ||
| err := r.syncRemoteSchema(context.Background(), mockFetchSchema, mock) | ||
| require.Error(t, err) | ||
| require.True(t, errors.Is(err, mock.fetchSchemaErr)) | ||
| }) | ||
| t.Run("schema has changed and updating errors", func(t *testing.T) { | ||
| mock := &mockSyncSchemaRepo{ | ||
| hasChanged: true, | ||
| updateLocalSchemaErr: fmt.Errorf("error updating local schema"), | ||
| } | ||
| mockFetchSchema := &mockFetchSchemaRepo{} | ||
| r := &Router{ | ||
| logger: logger.NOP, | ||
| } | ||
| err := r.syncRemoteSchema(context.Background(), mockFetchSchema, mock) | ||
| require.Error(t, err) | ||
| }) | ||
| t.Run("schema has changed and updating errors", func(t *testing.T) { | ||
| mock := &mockSyncSchemaRepo{ | ||
| hasChanged: false, | ||
| updateLocalSchemaErr: fmt.Errorf("error updating local schema"), | ||
| } | ||
| mockFetchSchema := &mockFetchSchemaRepo{} | ||
| r := &Router{ | ||
| logger: logger.NOP, | ||
| } | ||
| err := r.syncRemoteSchema(context.Background(), mockFetchSchema, mock) | ||
| require.NoError(t, err) | ||
| }) | ||
| t.Run("fetching schema succeeds with no error", func(t *testing.T) { | ||
| mock := &mockSyncSchemaRepo{} | ||
| mockFetchSchema := &mockFetchSchemaRepo{} | ||
| r := &Router{ | ||
| logger: logger.NOP, | ||
| } | ||
| err := r.syncRemoteSchema(context.Background(), mockFetchSchema, mock) | ||
| require.NoError(t, err) | ||
| }) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters