// Licensed to the Apache Software Foundation (ASF) under one or more
// contributor license agreements.  See the NOTICE file distributed with
// this work for additional information regarding copyright ownership.
// The ASF licenses this file to You 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 dataflowlib

import (
	"encoding/json"
	"fmt"
	"testing"

	"github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
	"github.com/google/go-cmp/cmp"
	df "google.golang.org/api/dataflow/v1b3"
)

func TestFromMetricUpdates_Counters(t *testing.T) {
	want := metrics.CounterResult{
		Attempted: 15,
		Committed: 15,
		Key: metrics.StepKey{
			Step:      "main.customDoFn",
			Name:      "customCounter",
			Namespace: "customDoFn",
		}}
	cName := newMetricStructuredName("customCounter", "customDoFn", false)
	committed := df.MetricUpdate{Name: &cName, Scalar: 15.0}

	aName := newMetricStructuredName("customCounter", "customDoFn", true)
	attempted := df.MetricUpdate{Name: &aName, Scalar: 15.0}

	job, err := newJob("main.customDoFn")
	if err != nil {
		t.Fatal(err)
	}

	got := FromMetricUpdates([]*df.MetricUpdate{&attempted, &committed}, &job).AllMetrics().Counters()
	size := len(got)
	if size < 1 {
		t.Fatalf("Invalid array's size: got: %v, want: %v", size, 1)
	}
	if d := cmp.Diff(want, got[0]); d != "" {
		t.Fatalf("Invalid counter: got: %v, want: %v, diff(-want,+got):\n %v",
			got[0], want, d)
	}
}

func TestFromMetricUpdates_Distributions(t *testing.T) {
	want := metrics.DistributionResult{
		Attempted: metrics.DistributionValue{
			Count: 100,
			Sum:   5,
			Min:   -12,
			Max:   30,
		},
		Committed: metrics.DistributionValue{
			Count: 100,
			Sum:   5,
			Min:   -12,
			Max:   30,
		},
		Key: metrics.StepKey{
			Step:      "main.customDoFn",
			Name:      "customDist",
			Namespace: "customDoFn",
		}}
	distribution := map[string]interface{}{
		"count": 100.0,
		"sum":   5.0,
		"min":   -12.0,
		"max":   30.0,
	}
	cName := newMetricStructuredName("customDist", "customDoFn", false)
	committed := df.MetricUpdate{Name: &cName, Distribution: distribution}

	aName := newMetricStructuredName("customDist", "customDoFn", true)
	attempted := df.MetricUpdate{Name: &aName, Distribution: distribution}

	job, err := newJob("main.customDoFn")
	if err != nil {
		t.Fatal(err)
	}

	got := FromMetricUpdates([]*df.MetricUpdate{&attempted, &committed}, &job).AllMetrics().Distributions()
	size := len(got)
	if size < 1 {
		t.Fatalf("Invalid array's size: got: %v, want: %v", size, 1)
	}
	if d := cmp.Diff(want, got[0]); d != "" {
		t.Fatalf("Invalid distribution: got: %v, want: %v, diff(-want,+got):\n %v",
			got[0], want, d)
	}
}

func newMetricStructuredName(name, namespace string, attempted bool) df.MetricStructuredName {
	context := map[string]string{
		"step":      "e5",
		"namespace": namespace,
	}
	if attempted {
		context["tentative"] = "true"
	}
	return df.MetricStructuredName{Context: context, Name: name}
}

func newJob(stepName string) (df.Job, error) {
	stepRepr := map[string]interface{}{
		"name": "e5",
		"properties": map[string]string{
			"user_name": stepName,
		},
	}
	stepJson, err := json.Marshal(&stepRepr)
	if err != nil {
		return df.Job{}, fmt.Errorf("Could not create Step object: %v", err)
	}
	step := df.Step{}
	if err := json.Unmarshal(stepJson, &step); err != nil {
		return df.Job{}, fmt.Errorf("Could not create Step object: %v", err)
	}
	return df.Job{Steps: []*df.Step{&step}}, nil
}
