Add additional sanity checks and increase test coverage fo ES bulk re…

…sponse parsing
driskell · Nov 22, 2021 · d18ac50 · d18ac50
1 parent 1d2a12c
commit d18ac50
Show file tree

Hide file tree

Showing 2 changed files with 92 additions and 17 deletions.
diff --git a/lc-lib/transports/es/bulkresponse.go b/lc-lib/transports/es/bulkresponse.go
@@ -69,6 +69,8 @@ func (b *bulkResponse) parse() error {
 		return err
 	}
 
+	var hasTook, hasItems bool
+
 	for {
 		// Expect } or key:value
 		key, err := b.parseKeyOrEnd()
@@ -81,10 +83,18 @@ func (b *bulkResponse) parse() error {
 
 		switch *key {
 		case "took":
+			if hasTook {
+				return errors.New("unexpected duplicate key \"took\"")
+			}
+			hasTook = true
 			if err := b.decoder.Decode(&b.Took); err != nil {
 				return err
 			}
 		case "items":
+			if hasItems {
+				return errors.New("unexpected duplicate key \"items\"")
+			}
+			hasItems = true
 			if err := b.parseItems(); err != nil {
 				return err
 			}
@@ -99,6 +109,13 @@ func (b *bulkResponse) parse() error {
 		return errors.New("unexpected tokens at end of stream")
 	}
 
+	if !hasTook {
+		return errors.New("response is missing \"took\" key")
+	}
+	if !hasItems {
+		return errors.New("response is missing \"items\" key")
+	}
+
 	return nil
 }
 
@@ -260,8 +277,9 @@ func (b *bulkResponse) parseArrayNextOrEnd() (json.Token, error) {
 		if delim == json.Delim(']') {
 			return nil, nil
 		}
+		return delim, nil
 	}
-	return delim, nil
+	return token, nil
 }
 
 // consumeValue takes away an entire value

diff --git a/lc-lib/transports/es/bulkresponse_test.go b/lc-lib/transports/es/bulkresponse_test.go
@@ -25,7 +25,7 @@ func TestBulkResponseParse(t *testing.T) {
 	responseString := `{"ignore":1,"again":"test","took":444,"more":false,"items":[{"index":{"in":5,"result":"created"}},{"index":{"error":{"type": "failed"},"status":404}},{"index":{"result":"created"}}],"last":"gone"}`
 	response, err := newBulkResponse([]byte(responseString), request)
 	if err != nil {
-		t.Errorf("Unexpected response parse error: %s", err)
+		t.Fatalf("Unexpected response parse error: %s", err)
 	}
 	if request.Remaining() != 1 {
 		t.Errorf("Unexpected remaining count: %d", request.Remaining())
@@ -40,7 +40,7 @@ func TestBulkResponseParseSkip(t *testing.T) {
 	responseString := `{"ignore":1,"again":"test","took":222,"more":false,"items":[{"index":{"in":5,"result":"created"}},{"index":{"error":{"type": "failed"},"status":400}},{"index":{"result":"created"}}],"last":"gone"}`
 	response, err := newBulkResponse([]byte(responseString), request)
 	if err != nil {
-		t.Errorf("Unexpected response parse error: %s", err)
+		t.Fatalf("Unexpected response parse error: %s", err)
 	}
 	if request.Remaining() != 0 {
 		t.Errorf("Unexpected remaining count: %d", request.Remaining())
@@ -52,25 +52,28 @@ func TestBulkResponseParseSkip(t *testing.T) {
 
 func TestBulkResponseParseTooFew(t *testing.T) {
 	request := createTestBulkRequest(3, "2020-08-03", "2020-08-03")
-	responseString := `{"more":false,"items":[{"index":{"result":"created"}}]}`
+	responseString := `{"more":false,"took":444,"items":[{"index":{"result":"created"}}]}`
 	_, err := newBulkResponse([]byte(responseString), request)
 	if err == nil {
-		t.Errorf("Unexpected response parse success")
+		t.Error("Unexpected response parse success")
 	}
 }
 
 func TestBulkResponseParseTookMissing(t *testing.T) {
 	request := createTestBulkRequest(3, "2020-08-03", "2020-08-03")
 	responseString := `{"more":false,"items":[{"index":{"result":"created"}},{"index":{"status":500,"error":{"type":"error"}}},{"index":{"result":"created"}}]}`
-	response, err := newBulkResponse([]byte(responseString), request)
-	if err != nil {
-		t.Errorf("Unexpected response parse error: %s", err)
-	}
-	if request.Remaining() != 1 {
-		t.Errorf("Unexpected remaining count: %d", request.Remaining())
+	_, err := newBulkResponse([]byte(responseString), request)
+	if err == nil {
+		t.Error("Expected missing took key error")
 	}
-	if response.Took != 0 {
-		t.Errorf("Unexpected took value: %d", response.Took)
+}
+
+func TestBulkResponseParseItemsMissing(t *testing.T) {
+	request := createTestBulkRequest(3, "2020-08-03", "2020-08-03")
+	responseString := `{"more":false,"took":123}`
+	_, err := newBulkResponse([]byte(responseString), request)
+	if err == nil {
+		t.Error("Expected missing items key error")
 	}
 }
 
@@ -79,20 +82,74 @@ func TestBulkResponseParseTooMany(t *testing.T) {
 	responseString := `{"more":false,"items":[{"index":{"result":"created"}},{"index":{"result":"failed"}},{"index":{"result":"created"}}]}`
 	_, err := newBulkResponse([]byte(responseString), request)
 	if err == nil {
-		t.Errorf("Unexpected response parse success")
+		t.Error("Expected too many error")
 	}
 }
 
 func TestBulkResponseParseSyntax(t *testing.T) {
 	request := createTestBulkRequest(1, "2020-08-03", "2020-08-03")
-	responseString := `{"more":false,"items":[{"index":{"result":"created"}}]}`
+	responseString := `{"more":false,"took":123,"items":[{"index":{"result":"created"}}]}`
 	_, err := newBulkResponse([]byte(responseString), request)
 	if err != nil {
 		t.Errorf("Unexpected response parse error: %s", err)
 	}
-	responseString = `{"more":false,"items":["index":{"result":"created"}}]}`
+	responseString = `{"more":false,"took":123,"items":["index":{"result":"created"}}]}`
 	_, err = newBulkResponse([]byte(responseString), request)
 	if err == nil {
-		t.Errorf("Unexpected response parse success")
+		t.Error("Expected parse error")
+	}
+}
+
+func TestBulkResponseDuplicateTook(t *testing.T) {
+	request := createTestBulkRequest(1, "2020-08-03", "2020-08-03")
+	responseString := `{"more":false,"took":123,"took":444,"items":[{"index":{"result":"created"}}]}`
+	_, err := newBulkResponse([]byte(responseString), request)
+	if err == nil {
+		t.Error("Expected duplicate took error")
+	}
+}
+
+func TestBulkResponseDuplicateItems(t *testing.T) {
+	request := createTestBulkRequest(1, "2020-08-03", "2020-08-03")
+	responseString := `{"more":false,"took":123,"items":[{"index":{"result":"created"}}],"items":[{"index":{"result":"created"}}]}`
+	_, err := newBulkResponse([]byte(responseString), request)
+	if err == nil {
+		t.Error("Expected duplicate items error")
+	}
+}
+
+func TestBulkResponseExtra(t *testing.T) {
+	request := createTestBulkRequest(1, "2020-08-03", "2020-08-03")
+	responseString := `{"more":false,"took":123,"items":[{"index":{"result":"created"}}]}ExtraStuff`
+	_, err := newBulkResponse([]byte(responseString), request)
+	if err == nil {
+		t.Error("Expected extra data error")
+	}
+}
+
+func TestBulkResponseIgnoreUnknown(t *testing.T) {
+	request := createTestBulkRequest(1, "2020-08-03", "2020-08-03")
+	responseString := `{"more":false,"took":123,"items":[{"index":{"result":"created"}}],"ignored":123,"ignored":{"test":""},"moreignore":["ignore",{"more":1.2}]}`
+	_, err := newBulkResponse([]byte(responseString), request)
+	if err != nil {
+		t.Errorf("Unexpected parse error with data that should be ignored: %s", err)
+	}
+}
+
+func TestBulkResponseError(t *testing.T) {
+	request := createTestBulkRequest(1, "2020-08-03", "2020-08-03")
+	responseString := `{"more":false,"took":123,"items":[{"index":{"error":{"type":"Type","reason":"Error","caused_by":{"type":"Inner","reason":"message"}}}}]}`
+	response, err := newBulkResponse([]byte(responseString), request)
+	if err != nil {
+		t.Fatalf("Unexpected parse error with error response: %s", err)
+	}
+	if len(response.Errors) != 1 {
+		t.Fatalf("Unexpected number of errors: %d", len(response.Errors))
+	}
+	if response.Errors[0].Type != "Type" || response.Errors[0].Reason != "Error" || response.Errors[0].CausedBy == nil {
+		t.Fatal("Unexpected error format")
+	}
+	if response.Errors[0].CausedBy.Type != "Inner" || response.Errors[0].CausedBy.Reason != "message" || response.Errors[0].CausedBy.CausedBy != nil {
+		t.Fatal("Unexpected inner error format")
 	}
 }