fix(flink): reject application resources without metadata.name

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`)
+}

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 {

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.