From 6d12478e6bc9409eabb32c34151a03872407824e Mon Sep 17 00:00:00 2001
From: Linh Vo <lvo@umbel.com>
Date: Tue, 20 Jun 2017 12:12:49 -0500
Subject: [PATCH] add validation for input definition fields

---
 input_definition.go      | 50 ++++++++++++++++++++++++++++++++++++++++
 input_definition_test.go | 48 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 98 insertions(+)

diff --git a/input_definition.go b/input_definition.go
index e6bb5e972..cb15732ca 100644
--- a/input_definition.go
+++ b/input_definition.go
@@ -19,10 +19,14 @@ import (
 	"os"
 	"path/filepath"
 
+	"errors"
+	"fmt"
 	"github.com/gogo/protobuf/proto"
 	"github.com/pilosa/pilosa/internal"
 )
 
+var ValidValueDestination = []string{"map", "valueToRow", "stringToBool"}
+
 // InputDefinition represents a container for the data input definition.
 type InputDefinition struct {
 	name        string
@@ -89,9 +93,22 @@ func (i *InputDefinition) LoadDefinition(pb *internal.InputDefinition) error {
 		i.frames = append(i.frames, inputFrame)
 	}
 
+	numPrimaryKey := 0
+	countRowID := make(map[uint64]bool)
 	for _, field := range pb.Fields {
 		var actions []Action
 		for _, action := range field.Actions {
+			if err := i.ValidateAction(action); err != nil {
+				return err
+			}
+			if action.RowID != 0 {
+				_, ok := countRowID[action.RowID]
+				if !ok {
+					countRowID[action.RowID] = true
+				} else {
+					return fmt.Errorf("duplicate rowID with other field: %s", action.RowID)
+				}
+			}
 			actions = append(actions, Action{
 				Frame:            action.Frame,
 				ValueDestination: action.ValueDestination,
@@ -100,6 +117,14 @@ func (i *InputDefinition) LoadDefinition(pb *internal.InputDefinition) error {
 			})
 		}
 
+		if field.PrimaryKey {
+			numPrimaryKey += 1
+		}
+
+		if numPrimaryKey > 1 {
+			return errors.New("duplicate primaryKey with other field")
+		}
+
 		inputField := Field{
 			Name:       field.Name,
 			PrimaryKey: field.PrimaryKey,
@@ -246,3 +271,28 @@ func (i *InputDefinition) AddFrame(frame InputFrame) error {
 	}
 	return nil
 }
+
+func (i *InputDefinition) ValidateAction(action *internal.Action) error {
+	validValues := make(map[string]bool)
+	for _, val := range ValidValueDestination {
+		validValues[val] = true
+	}
+	if _, ok := validValues[action.ValueDestination]; !ok {
+		return fmt.Errorf("invalid ValueDestination: %s", action.ValueDestination)
+	}
+
+	switch action.ValueDestination {
+	case "map":
+		if len(action.ValueMap) == 0 {
+			return errors.New("valueMap required for map")
+		}
+	case "stringToBool":
+		if action.RowID == 0 {
+			return errors.New("rowID required for stringToBool")
+		}
+	default:
+		return nil
+	}
+
+	return nil
+}
diff --git a/input_definition_test.go b/input_definition_test.go
index 037a97c28..178b30ea7 100644
--- a/input_definition_test.go
+++ b/input_definition_test.go
@@ -20,6 +20,7 @@ import (
 
 	"github.com/pilosa/pilosa"
 	"github.com/pilosa/pilosa/internal"
+	"strings"
 )
 
 func TestInputDefinition_Open(t *testing.T) {
@@ -93,3 +94,50 @@ func TestInputDefinition_Encoding(t *testing.T) {
 		t.Fatalf("unexpected ValueDestination: %v", internalDef.Fields[1].Actions[0])
 	}
 }
+
+func TestInputDefinition_LoadDefinition(t *testing.T) {
+	index := MustOpenIndex()
+	defer index.Close()
+
+	// Create Input Definition.
+	input := pilosa.InputDefinition{}
+	frames := internal.Frame{Name: "f", Meta: &internal.FrameMeta{RowLabel: "row"}}
+	action := internal.Action{Frame: "f", ValueDestination: "ValueToRow", ValueMap: map[string]uint64{"Green": 1}}
+	field := internal.InputDefinitionField{Name: "id", PrimaryKey: true, Actions: []*internal.Action{&action}}
+	def := &internal.InputDefinition{Name: "test", Frames: []*internal.Frame{&frames}, Fields: []*internal.InputDefinitionField{&field}}
+	err := input.LoadDefinition(def)
+	if !strings.Contains(err.Error(), "invalid ValueDestination") {
+		t.Fatalf("Expected invalid ValueDestination error, actual error: %s", err)
+	}
+
+	action = internal.Action{Frame: "f", ValueDestination: "stringToBool", ValueMap: map[string]uint64{"Green": 1}}
+	def = &internal.InputDefinition{Name: "test", Frames: []*internal.Frame{&frames}, Fields: []*internal.InputDefinitionField{&field}}
+	err = input.LoadDefinition(def)
+	if !strings.Contains(err.Error(), "rowID required for stringToBool") {
+		t.Fatalf("Expected rowID required for stringToBool error, actual error: %s", err)
+	}
+
+	action = internal.Action{Frame: "f", ValueDestination: "map", RowID: 100}
+	def = &internal.InputDefinition{Name: "test", Frames: []*internal.Frame{&frames}, Fields: []*internal.InputDefinitionField{&field}}
+	err = input.LoadDefinition(def)
+	if !strings.Contains(err.Error(), "valueMap required for map") {
+		t.Fatalf("Expected valueMap required for map error, actual error: %s", err)
+	}
+
+	action = internal.Action{Frame: "f", ValueDestination: "stringToBool", RowID: 100}
+	action1 := internal.Action{Frame: "f", ValueDestination: "stringToBool", RowID: 101}
+	field1 := internal.InputDefinitionField{Name: "newID", PrimaryKey: true, Actions: []*internal.Action{&action1}}
+	def = &internal.InputDefinition{Name: "test", Frames: []*internal.Frame{&frames}, Fields: []*internal.InputDefinitionField{&field, &field1}}
+	err = input.LoadDefinition(def)
+	if !strings.Contains(err.Error(), "duplicate primaryKey with other field") {
+		t.Fatalf("Expected duplicate primaryKey error, actual error: %s", err)
+	}
+
+	action1 = internal.Action{Frame: "f", ValueDestination: "stringToBool", RowID: 100}
+	field1 = internal.InputDefinitionField{Name: "id", PrimaryKey: true, Actions: []*internal.Action{&action1}}
+	def = &internal.InputDefinition{Name: "test", Frames: []*internal.Frame{&frames}, Fields: []*internal.InputDefinitionField{&field, &field1}}
+	err = input.LoadDefinition(def)
+	if !strings.Contains(err.Error(), "duplicate rowID with other field") {
+		t.Fatalf("Expected duplicate rowID with other field error, actual error: %s", err)
+	}
+}