<commit_before>package kinesis_connector_go_test

import (
	"bytes"
	"fmt"

	"code.google.com/p/go-uuid/uuid"

	. "github.com/runtakun/kinesis-connector-go/kcl"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
)

type testProcessor struct {
}

func (p *testProcessor) Initialize(shardID string) error {
	return nil
}
func (p *testProcessor) ProcessRecords(records []*Record, cp *CheckPointer) error {
	return nil
}

func (p *testProcessor) Shutdown(reason string, cp *CheckPointer) error {
	return nil
}

type ginkgoLogger struct {
}

func (l *ginkgoLogger) Printf(format string, v ...interface{}) {
	GinkgoWriter.Write([]byte(fmt.Sprintf(format, v...)))
	GinkgoWriter.Write([]byte{0xa}) //new line
}

var _ = Describe("Message", func() {

	BeforeEach(func() {
		SetLogger(&ginkgoLogger{})
	})

	Context("receive intialize message", func() {

		var stdout *bytes.Buffer
		var processor RecordProcessor

		BeforeEach(func() {
			stdout = new(bytes.Buffer)
			processor = &testProcessor{}
		})

		It("should be initialize processor", func() {
			stdin := bytes.NewBufferString(fmt.Sprintf(`{"action": "initialize", "shardId": "%s"}`, uuid.NewRandom().String()))
			Expect(RunFile(processor, stdin, stdout)).ShouldNot(HaveOccurred())
			Expect(stdout.Bytes()).Should(MatchJSON(`{"action": "status", "responseFor": "initialize"}`))
		})
	})
})
<commit_msg>add processRecords and shutdown action test<commit_after>package kinesis_connector_go_test

import (
	"bytes"
	"fmt"

	"code.google.com/p/go-uuid/uuid"

	. "github.com/runtakun/kinesis-connector-go/kcl"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
)

type testProcessor struct {
}

func (p *testProcessor) Initialize(shardID string) error {
	return nil
}
func (p *testProcessor) ProcessRecords(records []*Record, cp *CheckPointer) error {
	return nil
}

func (p *testProcessor) Shutdown(reason string, cp *CheckPointer) error {
	return nil
}

type ginkgoLogger struct {
}

func (l *ginkgoLogger) Printf(format string, v ...interface{}) {
	GinkgoWriter.Write([]byte(fmt.Sprintf(format, v...)))
	GinkgoWriter.Write([]byte{0xa}) //new line
}

var _ = Describe("Message", func() {
	BeforeEach(func() {
		SetLogger(&ginkgoLogger{})
	})

	Context("receive normal message", func() {

		var stdout *bytes.Buffer
		var processor RecordProcessor

		var shardID string

		BeforeEach(func() {
			stdout = new(bytes.Buffer)
			processor = &testProcessor{}

			shardID = uuid.NewRandom().String()
		})

		It("received intialize action", func() {
			stdin := bytes.NewBufferString(fmt.Sprintf(`{"action": "initialize", "shardId": "%s"}`, shardID))
			err := RunFile(processor, stdin, stdout)
			Expect(err).ShouldNot(HaveOccurred())
			Expect(stdout.Bytes()).Should(MatchJSON(`{"action": "status", "responseFor": "initialize"}`))
		})

		It("received processRecord action", func() {
			stdin := bytes.NewBufferString(fmt.Sprintf(`{"action": "processRecords", "data": "data", "partitionKey": "pk1", "sequenceNumber": "001"}`))
			err := RunFile(processor, stdin, stdout)
			Expect(err).ShouldNot(HaveOccurred())
			Expect(stdout.Bytes()).Should(MatchJSON(`{"action": "status", "responseFor": "processRecords"}`))
		})

		It("received shutdown action", func() {
			stdin := bytes.NewBufferString(fmt.Sprintf(`{"action": "shutdown", "reason": "TERMINATE"}`))
			err := RunFile(processor, stdin, stdout)
			Expect(err).ShouldNot(HaveOccurred())
			Expect(stdout.Bytes()).Should(MatchJSON(`{"action": "status", "responseFor": "shutdown"}`))
		})
	})
})

{'fraction_non_alphanumeric': 0.17432239657631954, 'fraction_numerical': 0.0017118402282453639, 'mean_word_length': 5.205309734513274, 'pattern_counts': {'":': 18, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4843985', 'n_tokens_mistral': 1266, 'n_tokens_neox': 1178, 'n_words': 237}