// 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 internal

import (
	"testing"

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

func TestConvertStringMap(t *testing.T) {
	m := make(map[string]string)
	m["a"] = "1"
	m["b"] = "2"

	pbm := ConvertFromStringMap(m)

	assert.Equal(t, 2, len(pbm))

	m2 := ConvertToStringMap(pbm)
	assert.Equal(t, 2, len(m2))
	assert.Equal(t, "1", m2["a"])
	assert.Equal(t, "2", m2["b"])
}

func TestReadMessageMetadata(t *testing.T) {
	// read old style message (not batched)
	reader := NewMessageReaderFromArray(rawCompatSingleMessage)
	meta, err := reader.ReadMessageMetadata()
	if err != nil {
		t.Fatal(err)
	}

	props := meta.GetProperties()
	assert.Equal(t, len(props), 2)
	assert.Equal(t, "a", props[0].GetKey())
	assert.Equal(t, "1", props[0].GetValue())
	assert.Equal(t, "b", props[1].GetKey())
	assert.Equal(t, "2", props[1].GetValue())

	// read message with batch of 1
	reader = NewMessageReaderFromArray(rawBatchMessage1)
	meta, err = reader.ReadMessageMetadata()
	if err != nil {
		t.Fatal(err)
	}
	assert.Equal(t, 1, int(meta.GetNumMessagesInBatch()))

	// read message with batch of 10
	reader = NewMessageReaderFromArray(rawBatchMessage10)
	meta, err = reader.ReadMessageMetadata()
	if err != nil {
		t.Fatal(err)
	}
	assert.Equal(t, 10, int(meta.GetNumMessagesInBatch()))
}

func TestReadMessageOldFormat(t *testing.T) {
	reader := NewMessageReaderFromArray(rawCompatSingleMessage)
	_, err := reader.ReadMessageMetadata()
	if err != nil {
		t.Fatal(err)
	}

	ssm, payload, err := reader.ReadMessage()
	if err != nil {
		t.Fatal(err)
	}
	// old message format does not have a single message metadata
	assert.Equal(t, true, ssm == nil)
	assert.Equal(t, "hello", string(payload))

	_, _, err = reader.ReadMessage()
	assert.Equal(t, ErrEOM, err)
}

func TestReadMessagesBatchSize1(t *testing.T) {
	reader := NewMessageReaderFromArray(rawBatchMessage1)
	meta, err := reader.ReadMessageMetadata()
	if err != nil {
		t.Fatal(err)
	}

	assert.Equal(t, 1, int(meta.GetNumMessagesInBatch()))
	for i := 0; i < int(meta.GetNumMessagesInBatch()); i++ {
		ssm, payload, err := reader.ReadMessage()
		if err != nil {
			t.Fatal(err)
		}
		assert.Equal(t, true, ssm != nil)
		assert.Equal(t, "hello", string(payload))
	}

	_, _, err = reader.ReadMessage()
	assert.Equal(t, ErrEOM, err)
}

func TestReadMessagesBatchSize10(t *testing.T) {
	reader := NewMessageReaderFromArray(rawBatchMessage10)
	meta, err := reader.ReadMessageMetadata()
	if err != nil {
		t.Fatal(err)
	}

	assert.Equal(t, 10, int(meta.GetNumMessagesInBatch()))
	for i := 0; i < int(meta.GetNumMessagesInBatch()); i++ {
		ssm, payload, err := reader.ReadMessage()
		if err != nil {
			t.Fatal(err)
		}
		assert.Equal(t, true, ssm != nil)
		assert.Equal(t, "hello", string(payload))
	}

	_, _, err = reader.ReadMessage()
	assert.Equal(t, ErrEOM, err)
}

// Raw single message in old format
// metadata properties:<key:"a" value:"1" > properties:<key:"b" value:"2" >
// payload = "hello"
var rawCompatSingleMessage = []byte{
	0x0e, 0x01, 0x08, 0x36, 0xb4, 0x66, 0x00, 0x00,
	0x00, 0x31, 0x0a, 0x0f, 0x73, 0x74, 0x61, 0x6e,
	0x64, 0x61, 0x6c, 0x6f, 0x6e, 0x65, 0x2d, 0x37,
	0x34, 0x2d, 0x30, 0x10, 0x00, 0x18, 0xac, 0xef,
	0xe8, 0xa0, 0xe2, 0x2d, 0x22, 0x06, 0x0a, 0x01,
	0x61, 0x12, 0x01, 0x31, 0x22, 0x06, 0x0a, 0x01,
	0x62, 0x12, 0x01, 0x32, 0x48, 0x05, 0x60, 0x05,
	0x82, 0x01, 0x00, 0x68, 0x65, 0x6c, 0x6c, 0x6f,
}

// Message with batch of 1
// singe message metadata properties:<key:"a" value:"1" > properties:<key:"b" value:"2" >
// payload = "hello"
var rawBatchMessage1 = []byte{
	0x0e, 0x01, 0x1f, 0x80, 0x09, 0x68, 0x00, 0x00,
	0x00, 0x1f, 0x0a, 0x0f, 0x73, 0x74, 0x61, 0x6e,
	0x64, 0x61, 0x6c, 0x6f, 0x6e, 0x65, 0x2d, 0x37,
	0x34, 0x2d, 0x31, 0x10, 0x00, 0x18, 0xdb, 0x80,
	0xf4, 0xa0, 0xe2, 0x2d, 0x58, 0x01, 0x82, 0x01,
	0x00, 0x00, 0x00, 0x00, 0x16, 0x0a, 0x06, 0x0a,
	0x01, 0x61, 0x12, 0x01, 0x31, 0x0a, 0x06, 0x0a,
	0x01, 0x62, 0x12, 0x01, 0x32, 0x18, 0x05, 0x28,
	0x05, 0x40, 0x00, 0x68, 0x65, 0x6c, 0x6c, 0x6f,
}

// Message with batch of 10
// singe message metadata properties:<key:"a" value:"1" > properties:<key:"b" value:"2" >
// payload = "hello"
var rawBatchMessage10 = []byte{
	0x0e, 0x01, 0x7b, 0x28, 0x8c, 0x08,
	0x00, 0x00, 0x00, 0x1f, 0x0a, 0x0f, 0x73, 0x74,
	0x61, 0x6e, 0x64, 0x61, 0x6c, 0x6f, 0x6e, 0x65,
	0x2d, 0x37, 0x34, 0x2d, 0x32, 0x10, 0x00, 0x18,
	0xd0, 0xc2, 0xfa, 0xa0, 0xe2, 0x2d, 0x58, 0x0a,
	0x82, 0x01, 0x00, 0x00, 0x00, 0x00, 0x16, 0x0a,
	0x06, 0x0a, 0x01, 0x61, 0x12, 0x01, 0x31, 0x0a,
	0x06, 0x0a, 0x01, 0x62, 0x12, 0x01, 0x32, 0x18,
	0x05, 0x28, 0x05, 0x40, 0x00, 0x68, 0x65, 0x6c,
	0x6c, 0x6f, 0x00, 0x00, 0x00, 0x16, 0x0a, 0x06,
	0x0a, 0x01, 0x61, 0x12, 0x01, 0x31, 0x0a, 0x06,
	0x0a, 0x01, 0x62, 0x12, 0x01, 0x32, 0x18, 0x05,
	0x28, 0x05, 0x40, 0x01, 0x68, 0x65, 0x6c, 0x6c,
	0x6f, 0x00, 0x00, 0x00, 0x16, 0x0a, 0x06, 0x0a,
	0x01, 0x61, 0x12, 0x01, 0x31, 0x0a, 0x06, 0x0a,
	0x01, 0x62, 0x12, 0x01, 0x32, 0x18, 0x05, 0x28,
	0x05, 0x40, 0x02, 0x68, 0x65, 0x6c, 0x6c, 0x6f,
	0x00, 0x00, 0x00, 0x16, 0x0a, 0x06, 0x0a, 0x01,
	0x61, 0x12, 0x01, 0x31, 0x0a, 0x06, 0x0a, 0x01,
	0x62, 0x12, 0x01, 0x32, 0x18, 0x05, 0x28, 0x05,
	0x40, 0x03, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x00,
	0x00, 0x00, 0x16, 0x0a, 0x06, 0x0a, 0x01, 0x61,
	0x12, 0x01, 0x31, 0x0a, 0x06, 0x0a, 0x01, 0x62,
	0x12, 0x01, 0x32, 0x18, 0x05, 0x28, 0x05, 0x40,
	0x04, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x00, 0x00,
	0x00, 0x16, 0x0a, 0x06, 0x0a, 0x01, 0x61, 0x12,
	0x01, 0x31, 0x0a, 0x06, 0x0a, 0x01, 0x62, 0x12,
	0x01, 0x32, 0x18, 0x05, 0x28, 0x05, 0x40, 0x05,
	0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x00, 0x00, 0x00,
	0x16, 0x0a, 0x06, 0x0a, 0x01, 0x61, 0x12, 0x01,
	0x31, 0x0a, 0x06, 0x0a, 0x01, 0x62, 0x12, 0x01,
	0x32, 0x18, 0x05, 0x28, 0x05, 0x40, 0x06, 0x68,
	0x65, 0x6c, 0x6c, 0x6f, 0x00, 0x00, 0x00, 0x16,
	0x0a, 0x06, 0x0a, 0x01, 0x61, 0x12, 0x01, 0x31,
	0x0a, 0x06, 0x0a, 0x01, 0x62, 0x12, 0x01, 0x32,
	0x18, 0x05, 0x28, 0x05, 0x40, 0x07, 0x68, 0x65,
	0x6c, 0x6c, 0x6f, 0x00, 0x00, 0x00, 0x16, 0x0a,
	0x06, 0x0a, 0x01, 0x61, 0x12, 0x01, 0x31, 0x0a,
	0x06, 0x0a, 0x01, 0x62, 0x12, 0x01, 0x32, 0x18,
	0x05, 0x28, 0x05, 0x40, 0x08, 0x68, 0x65, 0x6c,
	0x6c, 0x6f, 0x00, 0x00, 0x00, 0x16, 0x0a, 0x06,
	0x0a, 0x01, 0x61, 0x12, 0x01, 0x31, 0x0a, 0x06,
	0x0a, 0x01, 0x62, 0x12, 0x01, 0x32, 0x18, 0x05,
	0x28, 0x05, 0x40, 0x09, 0x68, 0x65, 0x6c, 0x6c,
	0x6f,
}