forked from raystack/transformers
-
Notifications
You must be signed in to change notification settings - Fork 1
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
1 parent
5f3867f
commit 13df94b
Showing
6 changed files
with
184 additions
and
36 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,112 @@ | ||
| package client_test | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "log/slog" | ||
| "os" | ||
| "testing" | ||
|
|
||
| "github.com/goto/maxcompute-transformation/internal/client" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestExecute(t *testing.T) { | ||
| t.Run("should return error when reading query file fails", func(t *testing.T) { | ||
| // arrange | ||
| client := client.NewClient(slog.Default(), nil) | ||
| client.OdpsClient = &mockOdpsClient{} | ||
| // act | ||
| err := client.Execute(nil, "", "./nonexistentfile") | ||
| // assert | ||
| assert.Error(t, err) | ||
| }) | ||
| t.Run("should return error when getting partition name fails", func(t *testing.T) { | ||
| // arrange | ||
| client := client.NewClient(slog.Default(), nil) | ||
| client.OdpsClient = &mockOdpsClient{ | ||
| partitionResult: func() ([]string, error) { | ||
| return nil, fmt.Errorf("error get partition name") | ||
| }, | ||
| } | ||
| assert.NoError(t, os.WriteFile("/tmp/query.sql", []byte("SELECT * FROM table;"), 0644)) | ||
| // act | ||
| err := client.Execute(nil, "project_test.table_test", "/tmp/query.sql") | ||
| // assert | ||
| assert.Error(t, err) | ||
| assert.ErrorContains(t, err, "error get partition name") | ||
| }) | ||
| t.Run("should return error when executing query fails", func(t *testing.T) { | ||
| // arrange | ||
| client := client.NewClient(slog.Default(), nil) | ||
| client.OdpsClient = &mockOdpsClient{ | ||
| partitionResult: func() ([]string, error) { | ||
| return nil, nil | ||
| }, | ||
| execSQLResult: func() error { | ||
| return fmt.Errorf("error exec sql") | ||
| }, | ||
| } | ||
| loader := &mockLoader{ | ||
| getQueryResult: func() string { | ||
| return "INSERT INTO table SELECT * FROM table;" | ||
| }, | ||
| } | ||
| assert.NoError(t, os.WriteFile("/tmp/query.sql", []byte("SELECT * FROM table;"), 0644)) | ||
| // act | ||
| err := client.Execute(loader, "project_test.table_test", "/tmp/query.sql") | ||
| // assert | ||
| assert.Error(t, err) | ||
| assert.ErrorContains(t, err, "error exec sql") | ||
| }) | ||
| t.Run("should return nil when everything is successful", func(t *testing.T) { | ||
| // arrange | ||
| client := client.NewClient(slog.Default(), nil) | ||
| client.OdpsClient = &mockOdpsClient{ | ||
| partitionResult: func() ([]string, error) { | ||
| return []string{"event_date"}, nil | ||
| }, | ||
| execSQLResult: func() error { | ||
| return nil | ||
| }, | ||
| } | ||
| loader := &mockLoader{ | ||
| getQueryResult: func() string { | ||
| return "INSERT INTO table SELECT * FROM table;" | ||
| }, | ||
| getPartitionedQueryResult: func() string { | ||
| return "INSERT INTO table PARTITION (event_date) SELECT * FROM table;" | ||
| }, | ||
| } | ||
| assert.NoError(t, os.WriteFile("/tmp/query.sql", []byte("SELECT * FROM table;"), 0644)) | ||
| // act | ||
| err := client.Execute(loader, "project_test.table_test", "/tmp/query.sql") | ||
| // assert | ||
| assert.NoError(t, err) | ||
| }) | ||
| } | ||
|
|
||
| type mockOdpsClient struct { | ||
| partitionResult func() ([]string, error) | ||
| execSQLResult func() error | ||
| } | ||
|
|
||
| func (m *mockOdpsClient) GetPartitionNames(tableID string) ([]string, error) { | ||
| return m.partitionResult() | ||
| } | ||
|
|
||
| func (m *mockOdpsClient) ExecSQL(query string) error { | ||
| return m.execSQLResult() | ||
| } | ||
|
|
||
| type mockLoader struct { | ||
| getQueryResult func() string | ||
| getPartitionedQueryResult func() string | ||
| } | ||
|
|
||
| func (m *mockLoader) GetQuery(tableID, query string) string { | ||
| return m.getQueryResult() | ||
| } | ||
|
|
||
| func (m *mockLoader) GetPartitionedQuery(tableID, query string, partitionName []string) string { | ||
| return m.getPartitionedQueryResult() | ||
| } |
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,43 @@ | ||
| package client | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "log/slog" | ||
|
|
||
| "github.com/aliyun/aliyun-odps-go-sdk/odps" | ||
| ) | ||
|
|
||
| type odpsClient struct { | ||
| logger *slog.Logger | ||
| client *odps.Odps | ||
| } | ||
|
|
||
| func NewODPSClient(client *odps.Odps) *odpsClient { | ||
| return &odpsClient{ | ||
| client: client, | ||
| } | ||
| } | ||
|
|
||
| // ExecSQL executes the given query in syncronous mode (blocking) | ||
| func (c *odpsClient) ExecSQL(query string) error { | ||
| taskIns, err := c.client.ExecSQl(query) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // wait execution success | ||
| c.logger.Info(fmt.Sprintf("taskId: %s", taskIns.Id())) | ||
| return taskIns.WaitForSuccess() | ||
| } | ||
|
|
||
| func (c *odpsClient) GetPartitionNames(tableID string) ([]string, error) { | ||
| table := c.client.Table(tableID) | ||
| if err := table.Load(); err != nil { | ||
| return nil, err | ||
| } | ||
| var partitionNames []string | ||
| for _, partition := range table.Schema().PartitionColumns { | ||
| partitionNames = append(partitionNames, partition.Name) | ||
| } | ||
| return partitionNames, nil | ||
| } |
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