diff --git a/internal/flink/command_application_test.go b/internal/flink/command_application_test.go new file mode 100644 index 0000000000..9a568f7302 --- /dev/null +++ b/internal/flink/command_application_test.go @@ -0,0 +1,24 @@ +package flink + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + pflink "github.com/confluentinc/cli/v4/pkg/flink" +) + +func TestApplicationCreateRejectsMissingMetadataNameFromResourceFile(t *testing.T) { + resourceFilePath := filepath.Join(t.TempDir(), "application.yaml") + err := os.WriteFile(resourceFilePath, []byte("metadata: {}\n"), 0600) + require.NoError(t, err) + + application, err := readApplicationResourceFile(resourceFilePath) + require.NoError(t, err) + + _, err = (&pflink.CmfRestClient{}).CreateApplication(context.Background(), "environment", application) + require.EqualError(t, err, `application name is required: ensure the resource file contains a non-empty "metadata.name" field`) +} diff --git a/pkg/flink/cmf_rest_client.go b/pkg/flink/cmf_rest_client.go index 83a76a0672..ff58f276eb 100644 --- a/pkg/flink/cmf_rest_client.go +++ b/pkg/flink/cmf_rest_client.go @@ -158,11 +158,22 @@ func (cmfClient *CmfRestClient) CmfApiContext() context.Context { return context.WithValue(context.Background(), cmfsdk.ContextAccessToken, cmfClient.AuthToken) } +func getApplicationName(application cmfsdk.FlinkApplication) (string, error) { + applicationName, ok := application.Metadata["name"].(string) + if !ok || applicationName == "" { + return "", fmt.Errorf(`application name is required: ensure the resource file contains a non-empty "metadata.name" field`) + } + return applicationName, nil +} + // CreateApplication Create a Flink application in the specified environment. // Internally, since the call for Create and Update is the same, we check if the environment doesn't contain said application before creation. func (cmfClient *CmfRestClient) CreateApplication(ctx context.Context, environment string, application cmfsdk.FlinkApplication) (cmfsdk.FlinkApplication, error) { // Get the name of the application - applicationName := application.Metadata["name"].(string) + applicationName, err := getApplicationName(application) + if err != nil { + return cmfsdk.FlinkApplication{}, err + } _, httpResponse, _ := cmfClient.FlinkApplicationsApi.GetApplication(ctx, environment, applicationName).Execute() // check if the application exists by checking the status code if httpResponse != nil && httpResponse.StatusCode == http.StatusOK { @@ -203,7 +214,10 @@ func (cmfClient *CmfRestClient) ListApplications(ctx context.Context, environmen // Internally, since the call for Create and Update is the same, we check if the environment contains said application before updation. func (cmfClient *CmfRestClient) UpdateApplication(ctx context.Context, environment string, application cmfsdk.FlinkApplication) (cmfsdk.FlinkApplication, error) { // Get the name of the application - applicationName := application.Metadata["name"].(string) + applicationName, err := getApplicationName(application) + if err != nil { + return cmfsdk.FlinkApplication{}, err + } _, httpResponse, err := cmfClient.FlinkApplicationsApi.GetApplication(ctx, environment, applicationName).Execute() // check if the application exists by checking the status code if httpResponse != nil && httpResponse.StatusCode == http.StatusNotFound { diff --git a/pkg/flink/cmf_rest_client_test.go b/pkg/flink/cmf_rest_client_test.go index 689d2d5509..38790c7cd6 100644 --- a/pkg/flink/cmf_rest_client_test.go +++ b/pkg/flink/cmf_rest_client_test.go @@ -1,12 +1,89 @@ package flink import ( + "context" "fmt" "testing" "github.com/stretchr/testify/require" + + cmfsdk "github.com/confluentinc/cmf-sdk-go/v1" ) +func TestGetApplicationName(t *testing.T) { + tests := []struct { + name string + application cmfsdk.FlinkApplication + wantName string + wantErr string + }{ + { + name: "valid name", + application: cmfsdk.FlinkApplication{Metadata: map[string]interface{}{"name": "my-application"}}, + wantName: "my-application", + }, + { + name: "missing name", + application: cmfsdk.FlinkApplication{Metadata: map[string]interface{}{}}, + wantErr: `application name is required: ensure the resource file contains a non-empty "metadata.name" field`, + }, + { + name: "empty name", + application: cmfsdk.FlinkApplication{Metadata: map[string]interface{}{"name": ""}}, + wantErr: `application name is required: ensure the resource file contains a non-empty "metadata.name" field`, + }, + { + name: "non-string name", + application: cmfsdk.FlinkApplication{Metadata: map[string]interface{}{"name": 123}}, + wantErr: `application name is required: ensure the resource file contains a non-empty "metadata.name" field`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotName, err := getApplicationName(tt.application) + if tt.wantErr != "" { + require.EqualError(t, err, tt.wantErr) + require.Empty(t, gotName) + return + } + + require.NoError(t, err) + require.Equal(t, tt.wantName, gotName) + }) + } +} + +func TestApplicationMethodsRejectInvalidNames(t *testing.T) { + tests := []struct { + name string + invoke func(*CmfRestClient) error + }{ + { + name: "create", + invoke: func(client *CmfRestClient) error { + _, err := client.CreateApplication(context.Background(), "environment", cmfsdk.FlinkApplication{}) + return err + }, + }, + { + name: "update", + invoke: func(client *CmfRestClient) error { + _, err := client.UpdateApplication(context.Background(), "environment", cmfsdk.FlinkApplication{}) + return err + }, + }, + } + + wantErr := `application name is required: ensure the resource file contains a non-empty "metadata.name" field` + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := tt.invoke(&CmfRestClient{}) + require.EqualError(t, err, wantErr) + }) + } +} + func TestListAllPages(t *testing.T) { // pagedFetcher emulates a CMF endpoint that pages by zero-based index (offset = page * size). // It records the sizes it was asked for so tests can assert on the requested page size.