feat(flink): add command for listing flink jobs

cmd/meroxa/root/flink/flink.go

@@ -47,6 +47,7 @@ func (*Job) SubCommands() []*cobra.Command {
 	return []*cobra.Command{
 		builder.BuildCobraCommand(&Deploy{}),
 		builder.BuildCobraCommand(&Remove{}),
+		builder.BuildCobraCommand(&List{}),
 	}
 }
 

cmd/meroxa/root/flink/list.go

@@ -0,0 +1,79 @@
+/*
+Copyright © 2023 Meroxa Inc
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package flink
+
+import (
+	"context"
+
+	"github.com/meroxa/cli/cmd/meroxa/builder"
+	"github.com/meroxa/cli/log"
+	"github.com/meroxa/cli/utils/display"
+	"github.com/meroxa/meroxa-go/pkg/meroxa"
+)
+
+var (
+	_ builder.CommandWithDocs    = (*List)(nil)
+	_ builder.CommandWithClient  = (*List)(nil)
+	_ builder.CommandWithLogger  = (*List)(nil)
+	_ builder.CommandWithExecute = (*List)(nil)
+	_ builder.CommandWithAliases = (*List)(nil)
+)
+
+type listJobsClient interface {
+	ListFlinkJobs(ctx context.Context) ([]*meroxa.FlinkJob, error)
+}
+
+type List struct {
+	client listJobsClient
+	logger log.Logger
+}
+
+func (l *List) Usage() string {
+	return "list"
+}
+
+func (l *List) Docs() builder.Docs {
+	return builder.Docs{
+		Short: "List Flink jobs",
+	}
+}
+
+func (l *List) Aliases() []string {
+	return []string{"ls"}
+}
+
+func (l *List) Execute(ctx context.Context) error {
+	flinkJobs, err := l.client.ListFlinkJobs(ctx)
+	if err != nil {
+		return err
+	}
+
+	l.logger.JSON(ctx, flinkJobs)
+	l.logger.Info(ctx, display.FlinkJobsTable(flinkJobs))
+	output := "\n ✨ To view your Flink Jobs, visit https://dashboard.meroxa.io/apps"
+	l.logger.Info(ctx, output)
+
+	return nil
+}
+
+func (l *List) Logger(logger log.Logger) {
+	l.logger = logger
+}
+
+func (l *List) Client(client meroxa.Client) {
+	l.client = client
+}

cmd/meroxa/root/flink/list_test.go

@@ -0,0 +1,118 @@
+/*
+Copyright © 2023 Meroxa Inc
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package flink
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/meroxa/cli/log"
+	"github.com/meroxa/cli/utils"
+	"github.com/meroxa/cli/utils/display"
+	"github.com/meroxa/meroxa-go/pkg/meroxa"
+	"github.com/meroxa/meroxa-go/pkg/mock"
+)
+
+func getFlinkJobs() []*meroxa.FlinkJob {
+	var flinkJobs []*meroxa.FlinkJob
+	f := utils.GenerateFlinkJob()
+	return append(flinkJobs, &f)
+}
+
+func TestListFlinkJobsExecution(t *testing.T) {
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	client := mock.NewMockClient(ctrl)
+	logger := log.NewTestLogger()
+
+	flinkJobs := getFlinkJobs()
+
+	client.
+		EXPECT().
+		ListFlinkJobs(ctx).
+		Return(flinkJobs, nil)
+
+	l := &List{
+		client: client,
+		logger: logger,
+	}
+
+	err := l.Execute(ctx)
+	if err != nil {
+		t.Fatalf("not expected error, got \"%s\"", err.Error())
+	}
+
+	gotLeveledOutput := logger.LeveledOutput()
+	wantLeveledOutput := display.FlinkJobsTable(flinkJobs)
+
+	if !strings.Contains(gotLeveledOutput, wantLeveledOutput) {
+		t.Fatalf("expected output:\n%s\ngot:\n%s", wantLeveledOutput, gotLeveledOutput)
+	}
+
+	gotJSONOutput := logger.JSONOutput()
+	var gotJobs []meroxa.FlinkJob
+	err = json.Unmarshal([]byte(gotJSONOutput), &gotJobs)
+
+	var lf []meroxa.FlinkJob
+
+	for _, f := range flinkJobs {
+		lf = append(lf, *f)
+	}
+
+	if err != nil {
+		t.Fatalf("not expected error, got %q", err.Error())
+	}
+
+	if !reflect.DeepEqual(gotJobs, lf) {
+		t.Fatalf("expected \"%v\", got \"%v\"", lf, gotJobs)
+	}
+}
+
+func TestListFlinkJobsErrorHandling(t *testing.T) {
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	client := mock.NewMockClient(ctrl)
+	logger := log.NewTestLogger()
+
+	errMsg := "some API error"
+
+	client.
+		EXPECT().
+		ListFlinkJobs(ctx).
+		Return(nil, errors.New(errMsg))
+
+	l := &List{
+		client: client,
+		logger: logger,
+	}
+
+	err := l.Execute(ctx)
+	if err == nil {
+		t.Fatalf("expected error, got %q", err.Error())
+	}
+
+	require.Error(t, err)
+	assert.ErrorContains(t, err, errMsg)
+}

utils/display/flink_jobs.go

@@ -0,0 +1,53 @@
+package display
+
+import (
+	"fmt"
+
+	"github.com/alexeyco/simpletable"
+
+	"github.com/meroxa/meroxa-go/pkg/meroxa"
+)
+
+func FlinkJobsTable(flinkJobs []*meroxa.FlinkJob) string {
+	if len(flinkJobs) == 0 {
+		return ""
+	}
+
+	table := simpletable.New()
+
+	table.Header = &simpletable.Header{
+		Cells: []*simpletable.Cell{
+			{Align: simpletable.AlignCenter, Text: "UUID"},
+			{Align: simpletable.AlignCenter, Text: "NAME"},
+			{Align: simpletable.AlignCenter, Text: "ENVIRONMENT"},
+			{Align: simpletable.AlignCenter, Text: "STATE"},
+			{Align: simpletable.AlignCenter, Text: "DEPLOYMENT STATE"},
+		},
+	}
+
+	for _, flinkJob := range flinkJobs {
+		var env string
+
+		if flinkJob.Environment.Name != "" {
+			env = flinkJob.Environment.Name
+		} else {
+			env = string(meroxa.EnvironmentTypeCommon)
+		}
+
+		r := []*simpletable.Cell{
+			{Align: simpletable.AlignRight, Text: flinkJob.UUID},
+			{Text: flinkJob.Name},
+			{Text: env},
+			{Text: string(flinkJob.Status.State)},
+			{Text: string(flinkJob.Status.ManagerDeploymentState)},
+		}
+
+		table.Body.Cells = append(table.Body.Cells, r)
+	}
+	table.SetStyle(simpletable.StyleCompact)
+	return table.String()
+}
+
+func PrintFlinkJobsTable(jobs []*meroxa.FlinkJob) {
+	fmt.Println(FlinkJobsTable(jobs))
+}

utils/display/flink_jobs_test.go

@@ -0,0 +1,113 @@
+package display
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/meroxa/cli/utils"
+	"github.com/meroxa/meroxa-go/pkg/meroxa"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFlinkJobsTable(t *testing.T) {
+	fOne := &meroxa.FlinkJob{
+		UUID:          "424ec647-9f0f-45a5-8e4b-3e0441f12555",
+		Name:          "my-flink-job",
+		InputStreams:  []string{"inputstream_one", "inputstream_two"},
+		OutputStreams: []string{"outtt"},
+		Status: meroxa.FlinkJobStatus{
+			State:                  "running",
+			LifecycleState:         "success",
+			ReconciliationState:    "deployed",
+			ManagerDeploymentState: "ready",
+		},
+		CreatedAt: time.Time{},
+		UpdatedAt: time.Time{},
+	}
+
+	fTwo := &meroxa.FlinkJob{
+		UUID:          "123d4da3-9f0f-45a5-8e4b-77777777",
+		Name:          "squirrel-app",
+		InputStreams:  []string{"inputstream_one"},
+		OutputStreams: []string{"outtt", "anotheroutt"},
+		Status: meroxa.FlinkJobStatus{
+			State:                  "failed",
+			LifecycleState:         "suspended",
+			ReconciliationState:    "rolling back",
+			ManagerDeploymentState: "error",
+		},
+		CreatedAt: time.Time{},
+		UpdatedAt: time.Time{},
+	}
+
+	tests := map[string][]*meroxa.FlinkJob{
+		"Base":         {fOne},
+		"ID_Alignment": {fOne, fTwo},
+		"Empty":        {},
+	}
+
+	for name, flinkJobs := range tests {
+		t.Run(name, func(t *testing.T) {
+			out := utils.CaptureOutput(func() {
+				PrintFlinkJobsTable(flinkJobs)
+			})
+
+			switch name {
+			case "Base":
+				verifyPrintFlinkJobsOutput(t, out, fOne)
+			case "ID_Alignment":
+				verifyPrintFlinkJobsOutput(t, out, fOne)
+				verifyPrintFlinkJobsOutput(t, out, fTwo)
+			case "Empty":
+				assert.Equal(t, out, "\n")
+			}
+			fmt.Println(out)
+		})
+	}
+}
+
+func verifyPrintFlinkJobsOutput(t *testing.T, out string, flinkJob *meroxa.FlinkJob) {
+	// verify header fields
+	tableHeaders := []string{"UUID", "NAME", "STATE", "DEPLOYMENT STATE"}
+
+	for _, header := range tableHeaders {
+		if !strings.Contains(out, header) {
+			t.Errorf("%s header is missing", header)
+		}
+	}
+
+	// verify fields that are supposed to be included in the output
+	if !strings.Contains(out, flinkJob.Name) {
+		t.Errorf("%s, not found", flinkJob.Name)
+	}
+	if !strings.Contains(out, flinkJob.UUID) {
+		t.Errorf("%s, not found", flinkJob.UUID)
+	}
+	if !strings.Contains(out, string(flinkJob.Status.State)) {
+		t.Errorf("state %s, not found", flinkJob.Status.State)
+	}
+	if !strings.Contains(out, string(meroxa.EnvironmentTypeCommon)) {
+		t.Errorf("state %s, not found", string(meroxa.EnvironmentTypeCommon))
+	}
+	// verify fields that are supposed to be excluded from the output
+	if strings.Contains(out, fmt.Sprintf("%v", flinkJob.InputStreams)) {
+		t.Errorf("found unwanted output: %s", flinkJob.InputStreams)
+	}
+	if strings.Contains(out, fmt.Sprintf("%v", flinkJob.OutputStreams)) {
+		t.Errorf("found unwanted output: %s", flinkJob.OutputStreams)
+	}
+	if strings.Contains(out, string(flinkJob.Status.LifecycleState)) {
+		t.Errorf("found unwanted output: %s", string(flinkJob.Status.LifecycleState))
+	}
+	if strings.Contains(out, string(flinkJob.Status.ReconciliationState)) {
+		t.Errorf("found unwanted output: %s", string(flinkJob.Status.ReconciliationState))
+	}
+	if strings.Contains(out, flinkJob.CreatedAt.String()) {
+		t.Errorf("found unwanted output: %s", flinkJob.CreatedAt.String())
+	}
+	if strings.Contains(out, flinkJob.UpdatedAt.String()) {
+		t.Errorf("found unwanted output: %s", flinkJob.UpdatedAt.String())
+	}
+}