package main

import (
	"reflect"
	"testing"
	"time"

	"github.com/golang/protobuf/proto"

	"code.justin.tv/ear/kinesis/combiner"
	"code.justin.tv/release/trace/api"
	"code.justin.tv/release/trace/pbmsg"
)

var mockTx = &pbmsg.Transaction{
	TransactionId: []uint64{1, 2},
	Client: &pbmsg.Service{
		Name: proto.String("go"),
	},
	Root: &pbmsg.Call{
		Svc: &pbmsg.Service{
			Name: proto.String("test"),
		},
		Subcalls: []*pbmsg.Call{
			&pbmsg.Call{
				Path: []uint32{0},
				Svc: &pbmsg.Service{
					Name: proto.String("trace"),
					Host: proto.String("trace.notreal.justin.tv"),
					Pid:  proto.Int32(123),
				},
				ServiceTimestamps: []*pbmsg.Timestamp{
					&pbmsg.Timestamp{
						Time: proto.Int64(1),
						Kind: pbmsg.Event_REQUEST_HEAD_RECEIVED.Enum(),
					},
				},
				Params: &pbmsg.RPCParams{
					Http: &pbmsg.HTTPParams{
						Status: proto.Int32(200),
					},
					Sql: &pbmsg.SQLParams{
						Dbname: proto.String("alf"),
					},
					Grpc: &pbmsg.GRPCParams{
						Method: proto.String("do"),
					},
					Memcached: &pbmsg.MemcachedParams{
						NKeysRequest: proto.Uint32(7),
					},
				},
			},
		},
	},
}

func TestCombineTransactions(t *testing.T) {
	blobber := combiner.NewRecordCombiner(1, MAX_RECORD_SIZE, 1*time.Second)

	n := 1000
	txs := make(chan *pbmsg.Transaction, n)
	for i := 0; i < n; i++ {
		txs <- mockTx
	}
	close(txs)

	done := make(chan struct{})
	have := 0
	go func() {
		defer close(done)
		for msg := range blobber.Blobs() {
			ts := &api.TransactionSet{}
			proto.Unmarshal(msg.Data, ts)
			for _, tx := range ts.Transaction {
				have += 1
				if !reflect.DeepEqual(tx, convertTransaction(mockTx)) {
					t.Errorf("Unexpected message out of blobs")
					t.Logf("Have: %v", tx)
					t.Logf("Want: %v", mockTx)
				}
			}
		}
	}()
	combineTransactions(txs, blobber)

	<-done
	if have != n {
		t.Errorf("Expected %d events, only found %d", n, have)
	}

}

func TestConvertCallPath(t *testing.T) {
	old := &pbmsg.Call{
		Path: []uint32{0},
	}
	new := convertCall(old)

	if !reflect.DeepEqual(old.Path, new.Path) {
		t.Errorf("Failed to convert call path; have=%v want=%v", new.Path, old.Path)
	}
}

func TestConvertTransaction(t *testing.T) {
	have := convertTransaction(mockTx)
	want := new(api.Transaction)

	buf, err := proto.Marshal(mockTx)
	if err != nil {
		t.Fatalf("proto.Marshal; err = %q", err)
	}
	err = proto.Unmarshal(buf, want)
	if err != nil {
		t.Fatalf("proto.Unmarshal; err = %q", err)
	}

	if !reflect.DeepEqual(have, want) {
		t.Errorf("\n%s\n!=\n%s\n", proto.CompactTextString(have), proto.CompactTextString(want))
	}
}
