mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
move cluster Main test helpers to the test package
This commit is contained in:
Travis Turner
parent
2fdc8048c5
commit
1b0fcb0e5a
3 changed files with 120 additions and 237 deletions
|
|
@ -24,15 +24,16 @@ import (
|
|||
|
||||
"github.com/pilosa/pilosa"
|
||||
"github.com/pilosa/pilosa/gossip"
|
||||
"github.com/pilosa/pilosa/test"
|
||||
)
|
||||
|
||||
// Ensure program can send/receive broadcast messages.
|
||||
func TestMain_SendReceiveMessage(t *testing.T) {
|
||||
|
||||
m0 := MustRunMain()
|
||||
m0 := test.MustRunMain()
|
||||
defer m0.Close()
|
||||
|
||||
m1 := MustRunMain()
|
||||
m1 := test.MustRunMain()
|
||||
defer m1.Close()
|
||||
|
||||
// Update cluster config
|
||||
|
|
@ -205,7 +206,7 @@ func TestMain_SendReceiveMessage(t *testing.T) {
|
|||
|
||||
// Ensure that an empty node comes up in a NORMAL state.
|
||||
func TestClusterResize_EmptyNode(t *testing.T) {
|
||||
m0 := MustRunMain()
|
||||
m0 := test.MustRunMain()
|
||||
defer m0.Close()
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
|
|
@ -216,7 +217,7 @@ func TestClusterResize_EmptyNode(t *testing.T) {
|
|||
// Ensure that a cluster of empty nodes comes up in a NORMAL state.
|
||||
func TestClusterResize_EmptyNodes(t *testing.T) {
|
||||
// Configure node0
|
||||
m0 := NewMain()
|
||||
m0 := test.NewMain()
|
||||
defer m0.Close()
|
||||
|
||||
gossipHost := "localhost"
|
||||
|
|
@ -227,7 +228,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
|
|||
}
|
||||
|
||||
// Configure node1
|
||||
m1 := NewMain()
|
||||
m1 := test.NewMain()
|
||||
defer m1.Close()
|
||||
|
||||
seed, coord, err = m1.RunWithTransport(gossipHost, gossipPort, seed, &coord)
|
||||
|
|
@ -246,7 +247,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
|
|||
func TestClusterResize_AddNode(t *testing.T) {
|
||||
t.Run("NoData", func(t *testing.T) {
|
||||
// Configure node0
|
||||
m0 := NewMain()
|
||||
m0 := test.NewMain()
|
||||
defer m0.Close()
|
||||
|
||||
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
|
||||
|
|
@ -255,7 +256,7 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
}
|
||||
|
||||
// Configure node1
|
||||
m1 := NewMain()
|
||||
m1 := test.NewMain()
|
||||
defer m1.Close()
|
||||
|
||||
var eg errgroup.Group
|
||||
|
|
@ -280,7 +281,7 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
})
|
||||
t.Run("WithIndex", func(t *testing.T) {
|
||||
// Configure node0
|
||||
m0 := NewMain()
|
||||
m0 := test.NewMain()
|
||||
defer m0.Close()
|
||||
|
||||
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
|
||||
|
|
@ -299,7 +300,7 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
}
|
||||
|
||||
// Configure node1
|
||||
m1 := NewMain()
|
||||
m1 := test.NewMain()
|
||||
defer m1.Close()
|
||||
|
||||
var eg errgroup.Group
|
||||
|
|
@ -326,7 +327,7 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
t.Run("ContinuousSlices", func(t *testing.T) {
|
||||
|
||||
// Configure node0
|
||||
m0 := NewMain()
|
||||
m0 := test.NewMain()
|
||||
defer m0.Close()
|
||||
|
||||
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
|
||||
|
|
@ -354,7 +355,7 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
}
|
||||
|
||||
// Configure node1
|
||||
m1 := NewMain()
|
||||
m1 := test.NewMain()
|
||||
defer m1.Close()
|
||||
|
||||
var eg errgroup.Group
|
||||
|
|
@ -381,7 +382,7 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
t.Run("SkippedSlice", func(t *testing.T) {
|
||||
|
||||
// Configure node0
|
||||
m0 := NewMain()
|
||||
m0 := test.NewMain()
|
||||
defer m0.Close()
|
||||
|
||||
seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
|
||||
|
|
@ -409,7 +410,7 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
}
|
||||
|
||||
// Configure node1
|
||||
m1 := NewMain()
|
||||
m1 := test.NewMain()
|
||||
defer m1.Close()
|
||||
|
||||
var eg errgroup.Group
|
||||
|
|
|
|||
|
|
@ -15,26 +15,19 @@
|
|||
package server_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"math/rand"
|
||||
"net/http"
|
||||
"os"
|
||||
"reflect"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strings"
|
||||
"testing"
|
||||
"testing/quick"
|
||||
|
||||
"github.com/BurntSushi/toml"
|
||||
"github.com/pilosa/pilosa"
|
||||
"github.com/pilosa/pilosa/gossip"
|
||||
"github.com/pilosa/pilosa/server"
|
||||
"github.com/pilosa/pilosa/test"
|
||||
)
|
||||
|
||||
|
|
@ -45,7 +38,7 @@ func TestMain_Set_Quick(t *testing.T) {
|
|||
}
|
||||
|
||||
if err := quick.Check(func(cmds []SetCommand) bool {
|
||||
m := MustRunMain()
|
||||
m := test.MustRunMain()
|
||||
defer m.Close()
|
||||
|
||||
// Create client.
|
||||
|
|
@ -121,7 +114,7 @@ func TestMain_Set_Quick(t *testing.T) {
|
|||
|
||||
// Ensure program can set row attributes and retrieve them.
|
||||
func TestMain_SetRowAttrs(t *testing.T) {
|
||||
m := MustRunMain()
|
||||
m := test.MustRunMain()
|
||||
defer m.Close()
|
||||
|
||||
// Create frames.
|
||||
|
|
@ -198,7 +191,7 @@ func TestMain_SetRowAttrs(t *testing.T) {
|
|||
|
||||
// Ensure program can set column attributes and retrieve them.
|
||||
func TestMain_SetColumnAttrs(t *testing.T) {
|
||||
m := MustRunMain()
|
||||
m := test.MustRunMain()
|
||||
defer m.Close()
|
||||
|
||||
// Create frames.
|
||||
|
|
@ -242,7 +235,7 @@ func TestMain_SetColumnAttrs(t *testing.T) {
|
|||
|
||||
// Ensure program can set column attributes with columnLabel option.
|
||||
func TestMain_SetColumnAttrsWithColumnOption(t *testing.T) {
|
||||
m := MustRunMain()
|
||||
m := test.MustRunMain()
|
||||
defer m.Close()
|
||||
|
||||
// Create frames.
|
||||
|
|
@ -276,7 +269,7 @@ func TestMain_SetColumnAttrsWithColumnOption(t *testing.T) {
|
|||
|
||||
// Ensure program can set bits on one cluster and then restore to a second cluster.
|
||||
func TestMain_FrameRestore(t *testing.T) {
|
||||
mains1 := NewMainArrayWithCluster(2)
|
||||
mains1 := test.NewMainArrayWithCluster(2)
|
||||
m0 := mains1[0]
|
||||
|
||||
// Create frames.
|
||||
|
|
@ -309,7 +302,7 @@ func TestMain_FrameRestore(t *testing.T) {
|
|||
}
|
||||
|
||||
// Start second cluster.
|
||||
mains2 := NewMainArrayWithCluster(2)
|
||||
mains2 := test.NewMainArrayWithCluster(2)
|
||||
m2 := mains2[0]
|
||||
defer m2.Close()
|
||||
|
||||
|
|
@ -378,191 +371,6 @@ func TestCountOpenFiles(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// Main represents a test wrapper for main.Main.
|
||||
type Main struct {
|
||||
*server.Command
|
||||
|
||||
Stdin bytes.Buffer
|
||||
Stdout bytes.Buffer
|
||||
Stderr bytes.Buffer
|
||||
}
|
||||
|
||||
// NewMain returns a new instance of Main with a temporary data directory and random port.
|
||||
func NewMain() *Main {
|
||||
path, err := ioutil.TempDir("", "pilosa-")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr)}
|
||||
m.Server.Network = *test.Network
|
||||
m.Config.DataDir = path
|
||||
m.Config.Bind = "localhost:0"
|
||||
m.Config.Cluster.Type = "static"
|
||||
m.Command.Stdin = &m.Stdin
|
||||
m.Command.Stdout = &m.Stdout
|
||||
m.Command.Stderr = &m.Stderr
|
||||
|
||||
if testing.Verbose() {
|
||||
m.Command.Stdout = io.MultiWriter(os.Stdout, m.Command.Stdout)
|
||||
m.Command.Stderr = io.MultiWriter(os.Stderr, m.Command.Stderr)
|
||||
}
|
||||
|
||||
return m
|
||||
}
|
||||
|
||||
func NewMainArrayWithCluster(size int) []*Main {
|
||||
cluster, err := test.NewServerCluster(size)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
mainArray := make([]*Main, size)
|
||||
for i := 0; i < size; i++ {
|
||||
mainArray[i] = &Main{Command: cluster.Servers[i]}
|
||||
}
|
||||
return mainArray
|
||||
}
|
||||
|
||||
// MustRunMain returns a new, running Main. Panic on error.
|
||||
func MustRunMain() *Main {
|
||||
m := NewMain()
|
||||
m.Config.Metric.Diagnostics = false // Disable diagnostics.
|
||||
if err := m.Run(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// Close closes the program and removes the underlying data directory.
|
||||
func (m *Main) Close() error {
|
||||
defer os.RemoveAll(m.Config.DataDir)
|
||||
return m.Command.Close()
|
||||
}
|
||||
|
||||
// Reopen closes the program and reopens it.
|
||||
func (m *Main) Reopen() error {
|
||||
if err := m.Command.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Create new main with the same config.
|
||||
config := m.Config
|
||||
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr)
|
||||
m.Server.Network = *test.Network
|
||||
m.Config = config
|
||||
|
||||
// Run new program.
|
||||
if err := m.Run(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RunWithTransport runs Main and returns the dynamically allocated gossip port.
|
||||
func (m *Main) RunWithTransport(host string, bindPort int, joinSeed string, coordinator *pilosa.URI) (seed string, coord pilosa.URI, err error) {
|
||||
defer close(m.Started)
|
||||
|
||||
m.Config.Cluster.Type = "gossip"
|
||||
|
||||
/*
|
||||
TEST:
|
||||
- SetupServer (just static settings from config)
|
||||
- OpenListener (sets Server.Name to use in gossip)
|
||||
- NewTransport (gossip)
|
||||
- SetupNetworking (does the gossip or static stuff) - uses Server.Name
|
||||
- Open server
|
||||
|
||||
PRODUCTION:
|
||||
- SetupServer (just static settings from config)
|
||||
- SetupNetworking (does the gossip or static stuff) - calls NewTransport
|
||||
- Open server - calls OpenListener
|
||||
*/
|
||||
|
||||
// SetupServer
|
||||
err = m.SetupServer()
|
||||
if err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
|
||||
// Open server listener.
|
||||
// This is used to set Server.Name, which is used as the node
|
||||
// name for identifying a memberlist node.
|
||||
err = m.Server.OpenListener()
|
||||
if err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
|
||||
// Open gossip transport to use in SetupServer.
|
||||
transport, err := gossip.NewTransport(host, bindPort)
|
||||
if err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
m.GossipTransport = transport
|
||||
|
||||
if joinSeed != "" {
|
||||
m.Config.Gossip.Seed = joinSeed
|
||||
} else {
|
||||
m.Config.Gossip.Seed = transport.URI.String()
|
||||
}
|
||||
seed = m.Config.Gossip.Seed
|
||||
|
||||
// SetupNetworking
|
||||
err = m.SetupNetworking()
|
||||
if err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
|
||||
if err = m.Server.BroadcastReceiver.Start(m.Server); err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
|
||||
if coordinator != nil {
|
||||
coord = *coordinator
|
||||
} else {
|
||||
coord = m.Server.URI
|
||||
}
|
||||
m.Server.Cluster.Coordinator = coord
|
||||
m.Server.Cluster.Static = false
|
||||
|
||||
// Initialize server.
|
||||
err = m.Server.Open()
|
||||
if err != nil {
|
||||
return seed, coord, err
|
||||
}
|
||||
|
||||
return seed, coord, nil
|
||||
}
|
||||
|
||||
// URL returns the base URL string for accessing the running program.
|
||||
func (m *Main) URL() string { return "http://" + m.Server.Addr().String() }
|
||||
|
||||
// Client returns a client to connect to the program.
|
||||
func (m *Main) Client() *pilosa.InternalHTTPClient {
|
||||
client, err := pilosa.NewInternalHTTPClient(m.Server.URI.HostPort(), pilosa.GetHTTPClient(nil))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return client
|
||||
}
|
||||
|
||||
// Query executes a query against the program through the HTTP API.
|
||||
func (m *Main) Query(index, rawQuery, query string) (string, error) {
|
||||
resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/query?", index)+rawQuery, query)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
|
||||
}
|
||||
return resp.Body, nil
|
||||
}
|
||||
|
||||
// CreateDefinition.
|
||||
func (m *Main) CreateDefinition(index, def, query string) (string, error) {
|
||||
resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/input-definition/%s", index, def), query)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
|
||||
}
|
||||
return resp.Body, nil
|
||||
}
|
||||
|
||||
// SetCommand represents a command to set a bit.
|
||||
type SetCommand struct {
|
||||
ID uint64
|
||||
|
|
@ -619,32 +427,6 @@ func ParseConfig(s string) (pilosa.Config, error) {
|
|||
return c, err
|
||||
}
|
||||
|
||||
// MustDo executes http.Do() with an http.NewRequest(). Panic on error.
|
||||
func MustDo(method, urlStr string, body string) *httpResponse {
|
||||
req, err := http.NewRequest(method, urlStr, strings.NewReader(body))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
buf, err := ioutil.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
return &httpResponse{Response: resp, Body: string(buf)}
|
||||
}
|
||||
|
||||
// httpResponse is a wrapper for http.Response that holds the Body as a string.
|
||||
type httpResponse struct {
|
||||
*http.Response
|
||||
Body string
|
||||
}
|
||||
|
||||
// MustMarshalJSON marshals v into a string. Panic on error.
|
||||
func MustMarshalJSON(v interface{}) string {
|
||||
buf, err := json.Marshal(v)
|
||||
|
|
|
|||
100
test/pilosa.go
100
test/pilosa.go
|
|
@ -2,9 +2,12 @@ package test
|
|||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/pilosa/pilosa"
|
||||
|
|
@ -47,12 +50,53 @@ func NewMain() *Main {
|
|||
return m
|
||||
}
|
||||
|
||||
func NewMainArrayWithCluster(size int) []*Main {
|
||||
cluster, err := NewServerCluster(size)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
mainArray := make([]*Main, size)
|
||||
for i := 0; i < size; i++ {
|
||||
mainArray[i] = cluster.Servers[i]
|
||||
}
|
||||
return mainArray
|
||||
}
|
||||
|
||||
// MustRunMain returns a new, running Main. Panic on error.
|
||||
func MustRunMain() *Main {
|
||||
m := NewMain()
|
||||
m.Config.Metric.Diagnostics = false // Disable diagnostics.
|
||||
if err := m.Run(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// Close closes the program and removes the underlying data directory.
|
||||
func (m *Main) Close() error {
|
||||
defer os.RemoveAll(m.Config.DataDir)
|
||||
return m.Command.Close()
|
||||
}
|
||||
|
||||
// Reopen closes the program and reopens it.
|
||||
func (m *Main) Reopen() error {
|
||||
if err := m.Command.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Create new main with the same config.
|
||||
config := m.Config
|
||||
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr)
|
||||
m.Server.Network = *Network
|
||||
m.Config = config
|
||||
|
||||
// Run new program.
|
||||
if err := m.Run(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RunWithTransport runs Main and returns the dynamically allocated gossip port.
|
||||
func (m *Main) RunWithTransport(host string, bindPort int, joinSeed string, coordinator *pilosa.URI) (seed string, coord pilosa.URI, err error) {
|
||||
defer close(m.Started)
|
||||
|
|
@ -128,6 +172,36 @@ func (m *Main) RunWithTransport(host string, bindPort int, joinSeed string, coor
|
|||
return seed, coord, nil
|
||||
}
|
||||
|
||||
// URL returns the base URL string for accessing the running program.
|
||||
func (m *Main) URL() string { return "http://" + m.Server.Addr().String() }
|
||||
|
||||
// Client returns a client to connect to the program.
|
||||
func (m *Main) Client() *pilosa.InternalHTTPClient {
|
||||
client, err := pilosa.NewInternalHTTPClient(m.Server.URI.HostPort(), pilosa.GetHTTPClient(nil))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return client
|
||||
}
|
||||
|
||||
// Query executes a query against the program through the HTTP API.
|
||||
func (m *Main) Query(index, rawQuery, query string) (string, error) {
|
||||
resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/query?", index)+rawQuery, query)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
|
||||
}
|
||||
return resp.Body, nil
|
||||
}
|
||||
|
||||
// CreateDefinition.
|
||||
func (m *Main) CreateDefinition(index, def, query string) (string, error) {
|
||||
resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/input-definition/%s", index, def), query)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
|
||||
}
|
||||
return resp.Body, nil
|
||||
}
|
||||
|
||||
////////////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
type Cluster struct {
|
||||
|
|
@ -169,3 +243,29 @@ func NewServerCluster(size int) (cluster *Cluster, err error) {
|
|||
|
||||
return cluster, nil
|
||||
}
|
||||
|
||||
// MustDo executes http.Do() with an http.NewRequest(). Panic on error.
|
||||
func MustDo(method, urlStr string, body string) *httpResponse {
|
||||
req, err := http.NewRequest(method, urlStr, strings.NewReader(body))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
buf, err := ioutil.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
return &httpResponse{Response: resp, Body: string(buf)}
|
||||
}
|
||||
|
||||
// httpResponse is a wrapper for http.Response that holds the Body as a string.
|
||||
type httpResponse struct {
|
||||
*http.Response
|
||||
Body string
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue