package util

import (
	"testing"

	"github.com/stretchr/testify/require"
)

func TestParseFlinkApplicationOutput(t *testing.T) {
	assert := require.New(t)
	out1 := `2021-05-27 15:43:28,854 INFO  org.apache.flink.kubernetes.utils.KubernetesUtils            [] - Kubernetes deployment requires a fixed port. Configuration blob.server.port will be set to 6124
2021-05-27 15:43:28,854 INFO  org.apache.flink.kubernetes.utils.KubernetesUtils            [] - Kubernetes deployment requires a fixed port. Configuration taskmanager.rpc.port will be set to 6122
2021-05-27 15:43:30,226 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Create flink application cluster foo successfully, JobManager Web Interface: http://150.158.223.52:8081
`

	out2 := `2021-05-27 13:24:45,271 INFO  org.apache.flink.kubernetes.utils.KubernetesUtils            [] - Kubernetes deployment requires a fixed port. Configuration blob.server.port will be set to 6124
2021-05-27 13:24:45,271 INFO  org.apache.flink.kubernetes.utils.KubernetesUtils            [] - Kubernetes deployment requires a fixed port. Configuration taskmanager.rpc.port will be set to 6122
2021-05-27 13:24:46,434 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Create flink application cluster flink-cluster-cdf78016dc6ed5102fa01e78df26322 successfully, JobManager Web Interface: http://172.17.16.16:8081
`
	tt := []struct {
		out             string
		expectClusterID string
		expectJmAddress string
	}{
		{out1, "foo", "http://150.158.223.52:8081"},
		{out2, "flink-cluster-cdf78016dc6ed5102fa01e78df26322", "http://172.17.16.16:8081"},
	}
	for _, tc := range tt {
		clusterID, jmAddress := ParseFlinkRunApplicationOutput([]byte(tc.out))
		assert.Equal(tc.expectClusterID, clusterID)
		assert.Equal(tc.expectJmAddress, jmAddress)
	}
}