// +build integration
package zk
import (
"bytes"
"log"
"os"
"testing"
"time"
stdzk "github.com/samuel/go-zookeeper/zk"
)
var (
host []string
)
func TestMain(m *testing.M) {
zkAddr := os.Getenv("ZK_ADDR")
if zkAddr == "" {
log.Fatal("ZK_ADDR is not set")
}
host = []string{zkAddr}
}
func TestCreateParentNodesOnServer(t *testing.T) {
payload := [][]byte{[]byte("Payload"), []byte("Test")}
c1, err := NewClient(host, logger, Payload(payload))
if err != nil {
t.Fatalf("Connect returned error: %v", err)
}
if c1 == nil {
t.Fatal("Expected pointer to client, got nil")
}
defer c1.Stop()
instancer, err := NewInstancer(c1, path, logger)
if err != nil {
t.Fatalf("Unable to create Subscriber: %v", err)
}
defer instancer.Stop()
state := instancer.state()
if state.Err != nil {
t.Fatal(err)
}
if want, have := 0, len(state.Instances); want != have {
t.Errorf("want %d, have %d", want, have)
}
c2, err := NewClient(host, logger)
if err != nil {
t.Fatalf("Connect returned error: %v", err)
}
defer c2.Stop()
data, _, err := c2.(*client).Get(path)
if err != nil {
t.Fatal(err)
}
// test Client implementation of CreateParentNodes. It should have created
// our payload
if bytes.Compare(data, payload[1]) != 0 {
t.Errorf("want %s, have %s", payload[1], data)
}
}
func TestCreateBadParentNodesOnServer(t *testing.T) {
c, _ := NewClient(host, logger)
defer c.Stop()
_, err := NewInstancer(c, "invalid/path", logger)
if want, have := stdzk.ErrInvalidPath, err; want != have {
t.Errorf("want %v, have %v", want, have)
}
}
func TestCredentials1(t *testing.T) {
acl := stdzk.DigestACL(stdzk.PermAll, "user", "secret")
c, _ := NewClient(host, logger, ACL(acl), Credentials("user", "secret"))
defer c.Stop()
_, err := NewInstancer(c, "/acl-issue-test", logger)
if err != nil {
t.Fatal(err)
}
}
func TestCredentials2(t *testing.T) {
acl := stdzk.DigestACL(stdzk.PermAll, "user", "secret")
c, _ := NewClient(host, logger, ACL(acl))
defer c.Stop()
_, err := NewInstancer(c, "/acl-issue-test", logger)
if err != stdzk.ErrNoAuth {
t.Errorf("want %v, have %v", stdzk.ErrNoAuth, err)
}
}
func TestConnection(t *testing.T) {
c, _ := NewClient(host, logger)
c.Stop()
_, err := NewInstancer(c, "/acl-issue-test", logger)
if err != ErrClientClosed {
t.Errorf("want %v, have %v", ErrClientClosed, err)
}
}
func TestGetEntriesOnServer(t *testing.T) {
var instancePayload = "10.0.3.204:8002"
c1, err := NewClient(host, logger)
if err != nil {
t.Fatalf("Connect returned error: %v", err)
}
defer c1.Stop()
c2, err := NewClient(host, logger)
s, err := NewInstancer(c2, path, logger)
if err != nil {
t.Fatal(err)
}
defer c2.Stop()
instance1 := &Service{
Path: path,
Name: "instance1",
Data: []byte(instancePayload),
}
if err = c2.Register(instance1); err != nil {
t.Fatalf("Unable to create test ephemeral znode 1: %+v", err)
}
instance2 := &Service{
Path: path,
Name: "instance2",
Data: []byte(instancePayload),
}
if err = c2.Register(instance2); err != nil {
t.Fatalf("Unable to create test ephemeral znode 2: %+v", err)
}
time.Sleep(50 * time.Millisecond)
state := s.state()
if state.Err != nil {
t.Fatal(state.Err)
}
if want, have := 2, len(state.Instances); want != have {
t.Errorf("want %d, have %d", want, have)
}
}
func TestGetEntriesPayloadOnServer(t *testing.T) {
c, err := NewClient(host, logger)
if err != nil {
t.Fatalf("Connect returned error: %v", err)
}
_, eventc, err := c.GetEntries(path)
if err != nil {
t.Fatal(err)
}
instance3 := Service{
Path: path,
Name: "instance3",
Data: []byte("just some payload"),
}
registrar := NewRegistrar(c, instance3, logger)
registrar.Register()
select {
case event := <-eventc:
if want, have := stdzk.EventNodeChildrenChanged.String(), event.Type.String(); want != have {
t.Errorf("want %s, have %s", want, have)
}
case <-time.After(100 * time.Millisecond):
t.Errorf("expected incoming watch event, timeout occurred")
}
_, eventc, err = c.GetEntries(path)
if err != nil {
t.Fatal(err)
}
registrar.Deregister()
select {
case event := <-eventc:
if want, have := stdzk.EventNodeChildrenChanged.String(), event.Type.String(); want != have {
t.Errorf("want %s, have %s", want, have)
}
case <-time.After(100 * time.Millisecond):
t.Errorf("expected incoming watch event, timeout occurred")
}
}