<commit_before>package broker

import (
	"io/ioutil"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/travisjeffery/jocko/cluster"
	"github.com/travisjeffery/simplelog"
)

func TestStoreOpen(t *testing.T) {
	DataDir, _ := ioutil.TempDir("", "storetest")
	defer os.RemoveAll(DataDir)

	bind0 := "127.0.0.1:4000"
	bind1 := "127.0.0.1:4001"
	bind2 := "127.0.0.1:4002"

	b0 := &cluster.Broker{
		Host: "127.0.0.1",
		Port: "4000",
		ID:   0,
	}
	b1 := &cluster.Broker{
		Host: "127.0.0.1",
		Port: "4001",
		ID:   1,
	}
	b2 := &cluster.Broker{
		Host: "127.0.0.1",
		Port: "4002",
		ID:   2,
	}

	logger := simplelog.New(os.Stdout, simplelog.DEBUG, "jocko/brokertest")
	s0 := New(Options{
		DataDir:              filepath.Join(DataDir, "0"),
		RaftAddr:             bind0,
		Logger:               logger,
		TCPAddr:              bind0,
		ID:                   0,
		DefaultNumPartitions: 2,
		Brokers:              []*cluster.Broker{b1, b2},
	})
	assert.NotNil(t, s0)

	err := s0.Open()
	assert.NoError(t, err)
	defer s0.Close()

	s1 := New(Options{
		DataDir:              filepath.Join(DataDir, "1"),
		RaftAddr:             bind1,
		Logger:               logger,
		TCPAddr:              bind1,
		ID:                   1,
		DefaultNumPartitions: 2,
		Brokers:              []*cluster.Broker{b0, b2},
	})
	err = s1.Open()
	assert.NoError(t, err)
	defer s1.Close()

	s2 := New(Options{
		DataDir:              filepath.Join(DataDir, "2"),
		RaftAddr:             bind2,
		TCPAddr:              bind2,
		Logger:               logger,
		ID:                   2,
		DefaultNumPartitions: 2,
		Brokers:              []*cluster.Broker{b0, b1},
	})
	err = s2.Open()
	assert.NoError(t, err)
	defer s2.Close()

	l, err := s0.WaitForLeader(10 * time.Second)
	assert.NoError(t, err)

	tp := cluster.TopicPartition{
		Topic:           "test",
		Partition:       0,
		Leader:          b0,
		PreferredLeader: b0,
		Replicas:        nil,
	}

	var peer, leader *Broker
	bs := []*Broker{s0, s1, s2}<commit_msg>fix test<commit_after>package broker

import (
	"io/ioutil"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/travisjeffery/jocko/cluster"
	"github.com/travisjeffery/simplelog"
)

func TestStoreOpen(t *testing.T) {
	dataDir, _ := ioutil.TempDir("", "broker_test")
	defer os.RemoveAll(dataDir)

	raft0 := "127.0.0.1:5000"
	raft1 := "127.0.0.1:5001"
	raft2 := "127.0.0.1:5002"

	b0 := &cluster.Broker{
		Host:     "127.0.0.1",
		Port:     "3001",
		RaftAddr: raft0,
		ID:       0,
	}
	b1 := &cluster.Broker{
		Host:     "127.0.0.1",
		Port:     "5001",
		RaftAddr: raft1,
		ID:       1,
	}
	b2 := &cluster.Broker{
		Host:     "127.0.0.1",
		Port:     "5002",
		RaftAddr: raft2,
		ID:       2,
	}

	logger := simplelog.New(os.Stdout, simplelog.INFO, "jocko/broker_test")
	s0 := New(Options{
		DataDir:              filepath.Join(dataDir, "0"),
		RaftAddr:             raft0,
		Logger:               logger,
		TCPAddr:              "127.0.0.1:3001",
		ID:                   0,
		DefaultNumPartitions: 2,
		Brokers:              []*cluster.Broker{b1, b2},
	})
	assert.NotNil(t, s0)

	err := s0.Open()
	assert.NoError(t, err)
	defer s0.Close()

	s1 := New(Options{
		DataDir:              filepath.Join(dataDir, "1"),
		RaftAddr:             raft1,
		Logger:               logger,
		TCPAddr:              raft1,
		ID:                   1,
		DefaultNumPartitions: 2,
		Brokers:              []*cluster.Broker{b0, b2},
	})
	err = s1.Open()
	assert.NoError(t, err)
	defer s1.Close()

	s2 := New(Options{
		DataDir:              filepath.Join(dataDir, "2"),
		RaftAddr:             raft2,
		TCPAddr:              raft2,
		Logger:               logger,
		ID:                   2,
		DefaultNumPartitions: 2,
		Brokers:              []*cluster.Broker{b0, b1},
	})
	err = s2.Open()
	assert.NoError(t, err)
	defer s2.Close()

	l, err := s0.WaitForLeader(10 * time.Second)
	assert.NoError(t, err)

	tp := cluster.TopicPartition{
		Topic:           "test",
		Partition:       0,
		Leader:          b0,
		PreferredLeader: b0,
		Replicas:        nil,
	}

	var peer, leader *Broker
	bs := []*Broker{s0, s1, s2}
{'fraction_non_alphanumeric': 0.15790736741060715, 'fraction_numerical': 0.05519558435325174, 'mean_word_length': 2.066961000735835, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 13, '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': '4187371', 'n_tokens_mistral': 1867, 'n_tokens_neox': 1701, 'n_words': 312}