goes_test.go

// Copyright 2013 Belogik. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package goes

import (
	. "launchpad.net/gocheck"
	"net/http"
	"net/url"
	"os"
	"testing"
	"time"
)

var (
	ES_HOST = "localhost"
	ES_PORT = "9200"
)

// Hook up gocheck into the gotest runner.
func Test(t *testing.T) { TestingT(t) }

type GoesTestSuite struct{}

var _ = Suite(&GoesTestSuite{})

func (s *GoesTestSuite) SetUpTest(c *C) {
	h := os.Getenv("TEST_ELASTICSEARCH_HOST")
	if h != "" {
		ES_HOST = h
	}

	p := os.Getenv("TEST_ELASTICSEARCH_PORT")
	if p != "" {
		ES_PORT = p
	}
}

func (s *GoesTestSuite) TestNewConnection(c *C) {
	conn := NewConnection(ES_HOST, ES_PORT)
	c.Assert(conn, DeepEquals, &Connection{ES_HOST, ES_PORT, http.DefaultClient})
}

func (s *GoesTestSuite) TestWithClient(c *C) {
	tr := &http.Transport{
		DisableCompression:    true,
		ResponseHeaderTimeout: 1 * time.Second,
	}
	cl := &http.Client{
		Transport: tr,
	}
	conn := NewConnection(ES_HOST, ES_PORT).WithClient(cl)

	c.Assert(conn, DeepEquals, &Connection{ES_HOST, ES_PORT, cl})
	c.Assert(conn.Client.Transport.(*http.Transport).DisableCompression, Equals, true)
	c.Assert(conn.Client.Transport.(*http.Transport).ResponseHeaderTimeout, Equals, 1*time.Second)
}

func (s *GoesTestSuite) TestUrl(c *C) {
	conn := NewConnection(ES_HOST, ES_PORT)

	r := Request{
		Conn:      conn,
		Query:     "q",
		IndexList: []string{"i"},
		TypeList:  []string{},
		method:    "GET",
		api:       "_search",
	}

	c.Assert(r.Url(), Equals, "http://"+ES_HOST+":"+ES_PORT+"/i/_search")

	r.IndexList = []string{"a", "b"}
	c.Assert(r.Url(), Equals, "http://"+ES_HOST+":"+ES_PORT+"/a,b/_search")

	r.TypeList = []string{"c", "d"}
	c.Assert(r.Url(), Equals, "http://"+ES_HOST+":"+ES_PORT+"/a,b/c,d/_search")

	r.ExtraArgs = make(url.Values, 1)
	r.ExtraArgs.Set("version", "1")
	c.Assert(r.Url(), Equals, "http://"+ES_HOST+":"+ES_PORT+"/a,b/c,d/_search?version=1")

	r.id = "1234"
	r.api = ""
	c.Assert(r.Url(), Equals, "http://"+ES_HOST+":"+ES_PORT+"/a,b/c,d/1234/?version=1")
}

func (s *GoesTestSuite) TestEsDown(c *C) {
	conn := NewConnection("a.b.c.d", "1234")

	var query = map[string]interface{}{"query": "foo"}

	r := Request{
		Conn:      conn,
		Query:     query,
		IndexList: []string{"i"},
		method:    "GET",
		api:       "_search",
	}
	_, err := r.Run()

	c.Assert(err, ErrorMatches, "Get http://a.b.c.d:1234/i/_search:(.*)lookup a.b.c.d: no such host")
}

func (s *GoesTestSuite) TestRunMissingIndex(c *C) {
	conn := NewConnection(ES_HOST, ES_PORT)

	var query = map[string]interface{}{"query": "foo"}

	r := Request{
		Conn:      conn,
		Query:     query,
		IndexList: []string{"i"},
		method:    "GET",
		api:       "_search",
	}
	_, err := r.Run()

	c.Assert(err.Error(), Equals, "[404] IndexMissingException[[i] missing]")
}

func (s *GoesTestSuite) TestCreateIndex(c *C) {
	indexName := "testcreateindexgoes"

	conn := NewConnection(ES_HOST, ES_PORT)
	defer conn.DeleteIndex(indexName)

	mapping := map[string]interface{}{
		"settings": map[string]interface{}{
			"index.number_of_shards":   1,
			"index.number_of_replicas": 0,
		},
		"mappings": map[string]interface{}{
			"_default_": map[string]interface{}{
				"_source": map[string]interface{}{
					"enabled": false,
				},
				"_all": map[string]interface{}{
					"enabled": false,
				},
			},
		},
	}

	resp, err := conn.CreateIndex(indexName, mapping)

	c.Assert(err, IsNil)
	c.Assert(resp.Acknowledged, Equals, true)
}

func (s *GoesTestSuite) TestDeleteIndexInexistantIndex(c *C) {
	conn := NewConnection(ES_HOST, ES_PORT)
	resp, err := conn.DeleteIndex("foobar")

	c.Assert(err.Error(), Equals, "[404] IndexMissingException[[foobar] missing]")
	c.Assert(resp, DeepEquals, Response{})
}

func (s *GoesTestSuite) TestDeleteIndexExistingIndex(c *C) {
	conn := NewConnection(ES_HOST, ES_PORT)

	indexName := "testdeleteindexexistingindex"

	_, err := conn.CreateIndex(indexName, map[string]interface{}{})

	c.Assert(err, IsNil)

	resp, err := conn.DeleteIndex(indexName)
	c.Assert(err, IsNil)

	expectedResponse := Response{}
	expectedResponse.Acknowledged = true
	c.Assert(resp, DeepEquals, expectedResponse)
}

func (s *GoesTestSuite) TestRefreshIndex(c *C) {
	conn := NewConnection(ES_HOST, ES_PORT)
	indexName := "testrefreshindex"

	_, err := conn.CreateIndex(indexName, map[string]interface{}{})
	c.Assert(err, IsNil)

	_, err = conn.RefreshIndex(indexName)
	c.Assert(err, IsNil)

	_, err = conn.DeleteIndex(indexName)
	c.Assert(err, IsNil)
}

func (s *GoesTestSuite) TestBulkSend(c *C) {
	indexName := "testbulkadd"
	docType := "tweet"

	tweets := []Document{
		Document{
			Id:          "123",
			Index:       indexName,
			Type:        docType,
			BulkCommand: BULK_COMMAND_INDEX,
			Fields: map[string]interface{}{
				"user":    "foo",
				"message": "some foo message",
			},
		},

		Document{
			Id:          nil,
			Index:       indexName,
			Type:        docType,
			BulkCommand: BULK_COMMAND_INDEX,
			Fields: map[string]interface{}{
				"user":    "bar",
				"message": "some bar message",
			},
		},
	}

	conn := NewConnection(ES_HOST, ES_PORT)

	_, err := conn.CreateIndex(indexName, nil)
	c.Assert(err, IsNil)

	response, err := conn.BulkSend(tweets)
	i := Item{
		Id:      "123",
		Type:    docType,
		Version: 1,
		Index:   indexName,
	}
	c.Assert(response.Items[0][BULK_COMMAND_INDEX], Equals, i)
	c.Assert(err, IsNil)

	_, err = conn.RefreshIndex(indexName)
	c.Assert(err, IsNil)

	var query = map[string]interface{}{
		"query": map[string]interface{}{
			"match_all": map[string]interface{}{},
		},
	}

	searchResults, err := conn.Search(query, []string{indexName}, []string{}, url.Values{})
	c.Assert(err, IsNil)

	var expectedTotal uint64 = 2
	c.Assert(searchResults.Hits.Total, Equals, expectedTotal)

	extraDocId := ""
	checked := 0
	for _, hit := range searchResults.Hits.Hits {
		if hit.Source["user"] == "foo" {
			c.Assert(hit.Id, Equals, "123")
			checked++
		}

		if hit.Source["user"] == "bar" {
			c.Assert(len(hit.Id) > 0, Equals, true)
			extraDocId = hit.Id
			checked++
		}
	}
	c.Assert(checked, Equals, 2)

	docToDelete := []Document{
		Document{
			Id:          "123",
			Index:       indexName,
			Type:        docType,
			BulkCommand: BULK_COMMAND_DELETE,
		},
		Document{
			Id:          extraDocId,
			Index:       indexName,
			Type:        docType,
			BulkCommand: BULK_COMMAND_DELETE,
		},
	}

	response, err = conn.BulkSend(docToDelete)
	i = Item{
		Id:      "123",
		Type:    docType,
		Version: 2,
		Index:   indexName,
	}
	c.Assert(response.Items[0][BULK_COMMAND_DELETE], Equals, i)

	c.Assert(err, IsNil)

	_, err = conn.RefreshIndex(indexName)
	c.Assert(err, IsNil)

	searchResults, err = conn.Search(query, []string{indexName}, []string{}, url.Values{})
	c.Assert(err, IsNil)

	expectedTotal = 0
	c.Assert(searchResults.Hits.Total, Equals, expectedTotal)

	_, err = conn.DeleteIndex(indexName)
	c.Assert(err, IsNil)
}

func (s *GoesTestSuite) TestStats(c *C) {
	conn := NewConnection(ES_HOST, ES_PORT)
	indexName := "teststats"

	conn.DeleteIndex(indexName)
	_, err := conn.CreateIndex(indexName, map[string]interface{}{})
	c.Assert(err, IsNil)

	// we must wait for a bit otherwise ES crashes
	time.Sleep(1 * time.Second)

	response, err := conn.Stats([]string{indexName}, url.Values{})
	c.Assert(err, IsNil)

	c.Assert(response.All.Indices[indexName].Primaries["docs"].Count, Equals, 0)

	_, err = conn.DeleteIndex(indexName)
	c.Assert(err, IsNil)
}

func (s *GoesTestSuite) TestIndexIdDefined(c *C) {
	indexName := "testindexiddefined"
	docType := "tweet"
	docId := "1234"

	conn := NewConnection(ES_HOST, ES_PORT)
	// just in case
	conn.DeleteIndex(indexName)

	_, err := conn.CreateIndex(indexName, map[string]interface{}{})
	c.Assert(err, IsNil)
	defer conn.DeleteIndex(indexName)

	d := Document{
		Index: indexName,
		Type:  docType,
		Id:    docId,
		Fields: map[string]interface{}{
			"user":    "foo",
			"message": "bar",
		},
	}

	extraArgs := make(url.Values, 1)
	extraArgs.Set("ttl", "86400000")
	response, err := conn.Index(d, extraArgs)
	c.Assert(err, IsNil)

	expectedResponse := Response{
		Index:   indexName,
		Id:      docId,
		Type:    docType,
		Version: 1,
	}

	c.Assert(response, DeepEquals, expectedResponse)
}

func (s *GoesTestSuite) TestIndexIdNotDefined(c *C) {
	indexName := "testindexidnotdefined"
	docType := "tweet"

	conn := NewConnection(ES_HOST, ES_PORT)
	// just in case
	conn.DeleteIndex(indexName)

	_, err := conn.CreateIndex(indexName, map[string]interface{}{})
	c.Assert(err, IsNil)
	defer conn.DeleteIndex(indexName)

	d := Document{
		Index: indexName,
		Type:  docType,
		Fields: map[string]interface{}{
			"user":    "foo",
			"message": "bar",
		},
	}

	response, err := conn.Index(d, url.Values{})
	c.Assert(err, IsNil)

	c.Assert(response.Index, Equals, indexName)
	c.Assert(response.Type, Equals, docType)
	c.Assert(response.Version, Equals, 1)
	c.Assert(response.Id != "", Equals, true)
}

func (s *GoesTestSuite) TestDelete(c *C) {
	indexName := "testdelete"
	docType := "tweet"
	docId := "1234"

	conn := NewConnection(ES_HOST, ES_PORT)
	// just in case
	conn.DeleteIndex(indexName)

	_, err := conn.CreateIndex(indexName, map[string]interface{}{})
	c.Assert(err, IsNil)
	defer conn.DeleteIndex(indexName)

	d := Document{
		Index: indexName,
		Type:  docType,
		Id:    docId,
		Fields: map[string]interface{}{
			"user": "foo",
		},
	}

	_, err = conn.Index(d, url.Values{})
	c.Assert(err, IsNil)

	response, err := conn.Delete(d, url.Values{})
	c.Assert(err, IsNil)

	expectedResponse := Response{
		Found: true,
		Index: indexName,
		Type:  docType,
		Id:    docId,
		// XXX : even after a DELETE the version number seems to be incremented
		Version: 2,
	}
	c.Assert(response, DeepEquals, expectedResponse)

	response, err = conn.Delete(d, url.Values{})
	c.Assert(err, IsNil)

	expectedResponse = Response{
		Found: false,
		Index: indexName,
		Type:  docType,
		Id:    docId,
		// XXX : even after a DELETE the version number seems to be incremented
		Version: 3,
	}
	c.Assert(response, DeepEquals, expectedResponse)
}

func (s *GoesTestSuite) TestGet(c *C) {
	indexName := "testget"
	docType := "tweet"
	docId := "111"
	source := map[string]interface{}{
		"f1": "foo",
		"f2": "foo",
	}

	conn := NewConnection(ES_HOST, ES_PORT)
	conn.DeleteIndex(indexName)

	_, err := conn.CreateIndex(indexName, map[string]interface{}{})
	c.Assert(err, IsNil)
	defer conn.DeleteIndex(indexName)

	d := Document{
		Index:  indexName,
		Type:   docType,
		Id:     docId,
		Fields: source,
	}

	_, err = conn.Index(d, url.Values{})
	c.Assert(err, IsNil)

	response, err := conn.Get(indexName, docType, docId, url.Values{})
	c.Assert(err, IsNil)

	expectedResponse := Response{
		Index:   indexName,
		Type:    docType,
		Id:      docId,
		Version: 1,
		Found:   true,
		Source:  source,
	}

	c.Assert(response, DeepEquals, expectedResponse)

	fields := make(url.Values, 1)
	fields.Set("fields", "f1")
	response, err = conn.Get(indexName, docType, docId, fields)
	c.Assert(err, IsNil)

	expectedResponse = Response{
		Index:   indexName,
		Type:    docType,
		Id:      docId,
		Version: 1,
		Found:   true,
		Fields: map[string]interface{}{
			"f1": []interface{}{"foo"},
		},
	}

	c.Assert(response, DeepEquals, expectedResponse)
}

func (s *GoesTestSuite) TestSearch(c *C) {
	indexName := "testsearch"
	docType := "tweet"
	docId := "1234"
	source := map[string]interface{}{
		"user":    "foo",
		"message": "bar",
	}

	conn := NewConnection(ES_HOST, ES_PORT)
	conn.DeleteIndex(indexName)

	_, err := conn.CreateIndex(indexName, map[string]interface{}{})
	c.Assert(err, IsNil)
	defer conn.DeleteIndex(indexName)

	d := Document{
		Index:  indexName,
		Type:   docType,
		Id:     docId,
		Fields: source,
	}

	_, err = conn.Index(d, url.Values{})
	c.Assert(err, IsNil)

	_, err = conn.RefreshIndex(indexName)
	c.Assert(err, IsNil)

	// I can feel my eyes bleeding
	query := map[string]interface{}{
		"query": map[string]interface{}{
			"bool": map[string]interface{}{
				"must": []map[string]interface{}{
					map[string]interface{}{
						"match_all": map[string]interface{}{},
					},
				},
			},
		},
	}
	response, err := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})

	expectedHits := Hits{
		Total:    1,
		MaxScore: 1.0,
		Hits: []Hit{
			Hit{
				Index:  indexName,
				Type:   docType,
				Id:     docId,
				Score:  1.0,
				Source: source,
			},
		},
	}

	c.Assert(response.Hits, DeepEquals, expectedHits)
}

func (s *GoesTestSuite) TestIndexStatus(c *C) {
	indexName := "testindexstatus"
	conn := NewConnection(ES_HOST, ES_PORT)
	conn.DeleteIndex(indexName)

	mapping := map[string]interface{}{
		"settings": map[string]interface{}{
			"index.number_of_shards":   1,
			"index.number_of_replicas": 1,
		},
	}

	_, err := conn.CreateIndex(indexName, mapping)
	c.Assert(err, IsNil)
	defer conn.DeleteIndex(indexName)

	// gives ES some time to do its job
	time.Sleep(1 * time.Second)

	response, err := conn.IndexStatus([]string{"_all"})
	c.Assert(err, IsNil)

	expectedShards := Shard{Total: 2, Successful: 1, Failed: 0}
	c.Assert(response.Shards, Equals, expectedShards)

	primarySizeInBytes := response.Indices[indexName].Index["primary_size_in_bytes"].(float64)
	sizeInBytes := response.Indices[indexName].Index["size_in_bytes"].(float64)

	c.Assert(primarySizeInBytes > 0, Equals, true)
	c.Assert(sizeInBytes > 0, Equals, true)

	expectedIndices := map[string]IndexStatus{
		indexName: IndexStatus{
			Index: map[string]interface{}{
				"primary_size_in_bytes": primarySizeInBytes,
				"size_in_bytes":         sizeInBytes,
			},
			Translog: map[string]uint64{
				"operations": 0,
			},
			Docs: map[string]uint64{
				"num_docs":     0,
				"max_doc":      0,
				"deleted_docs": 0,
			},
			Merges: map[string]interface{}{
				"current":               float64(0),
				"current_docs":          float64(0),
				"current_size_in_bytes": float64(0),
				"total":                 float64(0),
				"total_time_in_millis":  float64(0),
				"total_docs":            float64(0),
				"total_size_in_bytes":   float64(0),
			},
			Refresh: map[string]interface{}{
				"total":                float64(1),
				"total_time_in_millis": float64(0),
			},
			Flush: map[string]interface{}{
				"total":                float64(0),
				"total_time_in_millis": float64(0),
			},
		},
	}

	c.Assert(response.Indices, DeepEquals, expectedIndices)
}

func (s *GoesTestSuite) TestScroll(c *C) {
	indexName := "testscroll"
	docType := "tweet"

	tweets := []Document{
		Document{
			Id:          nil,
			Index:       indexName,
			Type:        docType,
			BulkCommand: BULK_COMMAND_INDEX,
			Fields: map[string]interface{}{
				"user":    "foo",
				"message": "some foo message",
			},
		},

		Document{
			Id:          nil,
			Index:       indexName,
			Type:        docType,
			BulkCommand: BULK_COMMAND_INDEX,
			Fields: map[string]interface{}{
				"user":    "bar",
				"message": "some bar message",
			},
		},

		Document{
			Id:          nil,
			Index:       indexName,
			Type:        docType,
			BulkCommand: BULK_COMMAND_INDEX,
			Fields: map[string]interface{}{
				"user":    "foo",
				"message": "another foo message",
			},
		},
	}

	conn := NewConnection(ES_HOST, ES_PORT)

	mapping := map[string]interface{}{
		"settings": map[string]interface{}{
			"index.number_of_shards":   1,
			"index.number_of_replicas": 0,
		},
	}

	defer conn.DeleteIndex(indexName)

	_, err := conn.CreateIndex(indexName, mapping)
	c.Assert(err, IsNil)

	_, err = conn.BulkSend(tweets)
	c.Assert(err, IsNil)

	_, err = conn.RefreshIndex(indexName)
	c.Assert(err, IsNil)

	query := map[string]interface{}{
		"query": map[string]interface{}{
			"filtered": map[string]interface{}{
				"filter": map[string]interface{}{
					"term": map[string]interface{}{
						"user": "foo",
					},
				},
			},
		},
	}

	scan, err := conn.Scan(query, []string{indexName}, []string{docType}, "1m", 1)
	c.Assert(err, IsNil)
	c.Assert(len(scan.ScrollId) > 0, Equals, true)

	searchResults, err := conn.Scroll(scan.ScrollId, "1m")
	c.Assert(err, IsNil)

	// some data in first chunk
	c.Assert(searchResults.Hits.Total, Equals, uint64(2))
	c.Assert(len(searchResults.ScrollId) > 0, Equals, true)
	c.Assert(len(searchResults.Hits.Hits), Equals, 1)

	searchResults, err = conn.Scroll(searchResults.ScrollId, "1m")
	c.Assert(err, IsNil)

	// more data in second chunk
	c.Assert(searchResults.Hits.Total, Equals, uint64(2))
	c.Assert(len(searchResults.ScrollId) > 0, Equals, true)
	c.Assert(len(searchResults.Hits.Hits), Equals, 1)

	searchResults, err = conn.Scroll(searchResults.ScrollId, "1m")
	c.Assert(err, IsNil)

	// nothing in third chunk
	c.Assert(searchResults.Hits.Total, Equals, uint64(2))
	c.Assert(len(searchResults.ScrollId) > 0, Equals, true)
	c.Assert(len(searchResults.Hits.Hits), Equals, 0)
}

func (s *GoesTestSuite) TestAggregations(c *C) {
	indexName := "testaggs"
	docType := "tweet"

	tweets := []Document{
		Document{
			Id:          nil,
			Index:       indexName,
			Type:        docType,
			BulkCommand: BULK_COMMAND_INDEX,
			Fields: map[string]interface{}{
				"user":    "foo",
				"message": "some foo message",
				"age":     25,
			},
		},

		Document{
			Id:          nil,
			Index:       indexName,
			Type:        docType,
			BulkCommand: BULK_COMMAND_INDEX,
			Fields: map[string]interface{}{
				"user":    "bar",
				"message": "some bar message",
				"age":     30,
			},
		},

		Document{
			Id:          nil,
			Index:       indexName,
			Type:        docType,
			BulkCommand: BULK_COMMAND_INDEX,
			Fields: map[string]interface{}{
				"user":    "foo",
				"message": "another foo message",
			},
		},
	}

	conn := NewConnection(ES_HOST, ES_PORT)

	mapping := map[string]interface{}{
		"settings": map[string]interface{}{
			"index.number_of_shards":   1,
			"index.number_of_replicas": 0,
		},
	}

	defer conn.DeleteIndex(indexName)

	_, err := conn.CreateIndex(indexName, mapping)
	c.Assert(err, IsNil)

	_, err = conn.BulkSend(tweets)
	c.Assert(err, IsNil)

	_, err = conn.RefreshIndex(indexName)
	c.Assert(err, IsNil)

	query := map[string]interface{}{
		"aggs": map[string]interface{}{
			"user": map[string]interface{}{
				"terms": map[string]interface{}{
					"field": "user",
					"order": map[string]interface{}{
						"_term": "asc",
					},
				},
				"aggs": map[string]interface{}{
					"age": map[string]interface{}{
						"stats": map[string]interface{}{
							"field": "age",
						},
					},
				},
			},
			"age": map[string]interface{}{
				"stats": map[string]interface{}{
					"field": "age",
				},
			},
		},
	}

	resp, err := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})

	user, ok := resp.Aggregations["user"]
	c.Assert(ok, Equals, true)

	c.Assert(len(user.Buckets()), Equals, 2)
	c.Assert(user.Buckets()[0].Key(), Equals, "bar")
	c.Assert(user.Buckets()[1].Key(), Equals, "foo")

	barAge := user.Buckets()[0].Aggregation("age")
	c.Assert(barAge["count"], Equals, 1.0)
	c.Assert(barAge["sum"], Equals, 30.0)

	fooAge := user.Buckets()[1].Aggregation("age")
	c.Assert(fooAge["count"], Equals, 1.0)
	c.Assert(fooAge["sum"], Equals, 25.0)

	age, ok := resp.Aggregations["age"]
	c.Assert(ok, Equals, true)

	c.Assert(age["count"], Equals, 2.0)
	c.Assert(age["sum"], Equals, 25.0+30.0)
}