|
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391 |
- // 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 (
- "net/http"
- "net/url"
- "os"
- "strings"
- "testing"
- "time"
-
- . "gopkg.in/check.v1"
- )
-
- 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
- resp.Raw = nil
- 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) TestOptimize(c *C) {
- conn := NewConnection(ES_HOST, ES_PORT)
- indexName := "testoptimize"
-
- 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.Optimize([]string{indexName}, url.Values{"max_num_segments": []string{"1"}})
- 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) TestBulkSend(c *C) {
- indexName := "testbulkadd"
- docType := "tweet"
-
- tweets := []Document{
- {
- Id: "123",
- Index: indexName,
- Type: docType,
- BulkCommand: BULK_COMMAND_INDEX,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "some foo message",
- },
- },
-
- {
- 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)
-
- conn.DeleteIndex(indexName)
- _, err := conn.CreateIndex(indexName, nil)
- c.Assert(err, IsNil)
-
- response, err := conn.BulkSend(tweets)
- i := Item{
- Id: "123",
- Type: docType,
- Version: 1,
- Index: indexName,
- Status: 201, //201 for indexing ( https://issues.apache.org/jira/browse/CONNECTORS-634 )
- }
- 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{
- {
- Id: "123",
- Index: indexName,
- Type: docType,
- BulkCommand: BULK_COMMAND_DELETE,
- },
- {
- 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,
- Status: 200, //200 for updates
- }
- 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) TestIndexWithFieldsInStruct(c *C) {
- indexName := "testindexwithfieldsinstruct"
- 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: struct {
- user string
- message string
- }{
- "foo",
- "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,
- }
-
- response.Raw = nil
- c.Assert(response, DeepEquals, expectedResponse)
- }
-
- func (s *GoesTestSuite) TestIndexWithFieldsNotInMapOrStruct(c *C) {
- indexName := "testindexwithfieldsnotinmaporstruct"
- 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: "test",
- }
-
- extraArgs := make(url.Values, 1)
- extraArgs.Set("ttl", "86400000")
- _, err = conn.Index(d, extraArgs)
- c.Assert(err, Not(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,
- }
-
- response.Raw = nil
- 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,
- }
- response.Raw = nil
- 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,
- }
- response.Raw = nil
- c.Assert(response, DeepEquals, expectedResponse)
- }
-
- func (s *GoesTestSuite) TestDeleteByQuery(c *C) {
- indexName := "testdeletebyquery"
- 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)
-
- _, err = conn.RefreshIndex(indexName)
- c.Assert(err, IsNil)
-
- query := map[string]interface{}{
- "query": map[string]interface{}{
- "bool": map[string]interface{}{
- "must": []map[string]interface{}{
- {
- "match_all": map[string]interface{}{},
- },
- },
- },
- },
- }
-
- //should be 1 doc before delete by query
- response, err := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
- c.Assert(err, IsNil)
- c.Assert(response.Hits.Total, Equals, uint64(1))
-
- response, err = conn.Query(query, []string{indexName}, []string{docType}, "DELETE", url.Values{})
-
- c.Assert(err, IsNil)
-
- expectedResponse := &Response{
- Found: false,
- Index: "",
- Type: "",
- Id: "",
- Version: 0,
- }
- response.Raw = nil
- c.Assert(response, DeepEquals, expectedResponse)
-
- //should be 0 docs after delete by query
- response, err = conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
- c.Assert(err, IsNil)
- c.Assert(response.Hits.Total, Equals, uint64(0))
- }
-
- 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,
- }
-
- response.Raw = nil
- 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"},
- },
- }
-
- response.Raw = nil
- 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{}{
- {
- "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{
- {
- Index: indexName,
- Type: docType,
- Id: docId,
- Score: 1.0,
- Source: source,
- },
- },
- }
-
- c.Assert(response.Hits, DeepEquals, expectedHits)
- }
-
- func (s *GoesTestSuite) TestCount(c *C) {
- indexName := "testcount"
- 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{}{
- {
- "match_all": map[string]interface{}{},
- },
- },
- },
- },
- }
- response, err := conn.Count(query, []string{indexName}, []string{docType}, url.Values{})
-
- c.Assert(response.Count, Equals, 1)
- }
-
- 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{"testindexstatus"})
- 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: {
- 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{
- {
- Id: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BULK_COMMAND_INDEX,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "some foo message",
- },
- },
-
- {
- Id: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BULK_COMMAND_INDEX,
- Fields: map[string]interface{}{
- "user": "bar",
- "message": "some bar message",
- },
- },
-
- {
- 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{
- {
- Id: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BULK_COMMAND_INDEX,
- Fields: map[string]interface{}{
- "user": "foo",
- "message": "some foo message",
- "age": 25,
- },
- },
-
- {
- Id: nil,
- Index: indexName,
- Type: docType,
- BulkCommand: BULK_COMMAND_INDEX,
- Fields: map[string]interface{}{
- "user": "bar",
- "message": "some bar message",
- "age": 30,
- },
- },
-
- {
- 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)
- }
-
- func (s *GoesTestSuite) TestPutMapping(c *C) {
- indexName := "testputmapping"
- 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)
-
- mapping := map[string]interface{}{
- "tweet": map[string]interface{}{
- "properties": map[string]interface{}{
- "count": map[string]interface{}{
- "type": "integer",
- "index": "not_analyzed",
- "store": true,
- },
- },
- },
- }
- response, err = conn.PutMapping("tweet", mapping, []string{indexName})
- c.Assert(err, IsNil)
-
- c.Assert(response.Acknowledged, Equals, true)
- c.Assert(response.TimedOut, Equals, false)
- }
-
- func (s *GoesTestSuite) TestIndicesExist(c *C) {
- indices := []string{"testindicesexist"}
-
- conn := NewConnection(ES_HOST, ES_PORT)
- // just in case
- conn.DeleteIndex(indices[0])
-
- exists, err := conn.IndicesExist(indices)
- c.Assert(exists, Equals, false)
-
- _, err = conn.CreateIndex(indices[0], map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(indices[0])
- time.Sleep(200 * time.Millisecond)
-
- exists, err = conn.IndicesExist(indices)
- c.Assert(exists, Equals, true)
-
- indices = append(indices, "nonexistent")
- exists, err = conn.IndicesExist(indices)
- c.Assert(exists, Equals, false)
- }
-
- func (s *GoesTestSuite) TestUpdate(c *C) {
- indexName := "testupdate"
- 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",
- "counter": 1,
- },
- }
-
- extraArgs := make(url.Values, 1)
- response, err := conn.Index(d, extraArgs)
- c.Assert(err, IsNil)
- time.Sleep(200 * time.Millisecond)
-
- expectedResponse := &Response{
- Index: indexName,
- Id: docId,
- Type: docType,
- Version: 1,
- }
-
- response.Raw = nil
- c.Assert(response, DeepEquals, expectedResponse)
-
- // Now that we have an ordinary document indexed, try updating it
- query := map[string]interface{}{
- "script": "ctx._source.counter += count",
- "params": map[string]interface{}{
- "count": 5,
- },
- "upsert": map[string]interface{}{
- "message": "candybar",
- "user": "admin",
- "counter": 1,
- },
- }
-
- response, err = conn.Update(d, query, extraArgs)
- if err != nil && strings.Contains(err.(*SearchError).Msg, "dynamic scripting disabled") {
- c.Skip("Scripting is disabled on server, skipping this test")
- return
- }
- time.Sleep(200 * time.Millisecond)
-
- c.Assert(err, Equals, nil)
-
- response, err = conn.Get(indexName, docType, docId, url.Values{})
- c.Assert(err, Equals, nil)
- c.Assert(response.Source["counter"], Equals, float64(6))
- c.Assert(response.Source["user"], Equals, "foo")
- c.Assert(response.Source["message"], Equals, "bar")
-
- // Test another document, non-existent
- docId = "555"
- d.Id = docId
- response, err = conn.Update(d, query, extraArgs)
- c.Assert(err, Equals, nil)
- time.Sleep(200 * time.Millisecond)
-
- response, err = conn.Get(indexName, docType, docId, url.Values{})
- c.Assert(err, Equals, nil)
- c.Assert(response.Source["user"], Equals, "admin")
- c.Assert(response.Source["message"], Equals, "candybar")
- }
-
- func (s *GoesTestSuite) TestGetMapping(c *C) {
- indexName := "testmapping"
- 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)
-
- time.Sleep(300 * time.Millisecond)
-
- response, err := conn.GetMapping([]string{docType}, []string{indexName})
- c.Assert(err, Equals, nil)
- c.Assert(len(response.Raw), Equals, 0)
-
- 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)
- time.Sleep(200 * time.Millisecond)
-
- response, err = conn.GetMapping([]string{docType}, []string{indexName})
- c.Assert(err, Equals, nil)
- c.Assert(len(response.Raw), Not(Equals), 0)
- }
-
- func (s *GoesTestSuite) TestDeleteMapping(c *C) {
- indexName := "testdeletemapping"
- 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)
-
- mapping := map[string]interface{}{
- "tweet": map[string]interface{}{
- "properties": map[string]interface{}{
- "count": map[string]interface{}{
- "type": "integer",
- "index": "not_analyzed",
- "store": true,
- },
- },
- },
- }
- response, err = conn.PutMapping("tweet", mapping, []string{indexName})
- c.Assert(err, IsNil)
- time.Sleep(200 * time.Millisecond)
-
- response, err = conn.DeleteMapping("tweet", []string{indexName})
- c.Assert(err, IsNil)
-
- c.Assert(response.Acknowledged, Equals, true)
- c.Assert(response.TimedOut, Equals, false)
- }
-
- func (s *GoesTestSuite) TestAddAlias(c *C) {
- aliasName := "testAlias"
- indexName := "testalias_1"
- docType := "testDoc"
- docId := "1234"
- source := map[string]interface{}{
- "user": "foo",
- "message": "bar",
- }
-
- conn := NewConnection(ES_HOST, ES_PORT)
- defer 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,
- }
-
- // Index data
- _, err = conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- // Add alias
- _, err = conn.AddAlias(aliasName, []string{indexName})
- c.Assert(err, IsNil)
-
- // Get document via alias
- response, err := conn.Get(aliasName, docType, docId, url.Values{})
- c.Assert(err, IsNil)
-
- expectedResponse := &Response{
- Index: indexName,
- Type: docType,
- Id: docId,
- Version: 1,
- Found: true,
- Source: source,
- }
-
- response.Raw = nil
- c.Assert(response, DeepEquals, expectedResponse)
- }
-
- func (s *GoesTestSuite) TestRemoveAlias(c *C) {
- aliasName := "testAlias"
- indexName := "testalias_1"
- docType := "testDoc"
- docId := "1234"
- source := map[string]interface{}{
- "user": "foo",
- "message": "bar",
- }
-
- conn := NewConnection(ES_HOST, ES_PORT)
- defer 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,
- }
-
- // Index data
- _, err = conn.Index(d, url.Values{})
- c.Assert(err, IsNil)
-
- // Add alias
- _, err = conn.AddAlias(aliasName, []string{indexName})
- c.Assert(err, IsNil)
-
- // Remove alias
- _, err = conn.RemoveAlias(aliasName, []string{indexName})
- c.Assert(err, IsNil)
-
- // Get document via alias
- _, err = conn.Get(aliasName, docType, docId, url.Values{})
- c.Assert(err.Error(), Equals, "[404] IndexMissingException[["+aliasName+"] missing]")
- }
-
- func (s *GoesTestSuite) TestAliasExists(c *C) {
- index := "testaliasexist_1"
- alias := "testaliasexists"
-
- conn := NewConnection(ES_HOST, ES_PORT)
- // just in case
- conn.DeleteIndex(index)
-
- exists, err := conn.AliasExists(alias)
- c.Assert(exists, Equals, false)
-
- _, err = conn.CreateIndex(index, map[string]interface{}{})
- c.Assert(err, IsNil)
- defer conn.DeleteIndex(index)
- time.Sleep(200 * time.Millisecond)
-
- _, err = conn.AddAlias(alias, []string{index})
- c.Assert(err, IsNil)
- time.Sleep(200 * time.Millisecond)
- defer conn.RemoveAlias(alias, []string{index})
-
- exists, err = conn.AliasExists(alias)
- c.Assert(exists, Equals, true)
- }
|