fix(bigtable/bttest): Emulator too lenient for empty RowMutation (#4359)

* fix(bigtable/bttest): Emulator too lenient for empty RowMutation

* gofmt

Co-authored-by: Christopher Wilcox <crwilcox@google.com>

bigtable/bttest/inmem.go

@@ -809,6 +809,12 @@ func newRegexp(pat []byte) (*binaryregexp.Regexp, error) {
 }
 
 func (s *server) MutateRow(ctx context.Context, req *btpb.MutateRowRequest) (*btpb.MutateRowResponse, error) {
+	if len(req.Mutations) == 0 {
+		return nil, status.Errorf(
+			codes.InvalidArgument,
+			"No mutations provided",
+		)
+	}
 	s.mu.Lock()
 	tbl, ok := s.tables[req.TableName]
 	s.mu.Unlock()
@@ -826,6 +832,16 @@ func (s *server) MutateRow(ctx context.Context, req *btpb.MutateRowRequest) (*bt
 }
 
 func (s *server) MutateRows(req *btpb.MutateRowsRequest, stream btpb.Bigtable_MutateRowsServer) error {
+	nMutations := 0
+	for _, entry := range req.Entries {
+		nMutations += len(entry.Mutations)
+	}
+	if nMutations == 0 {
+		return status.Errorf(
+			codes.InvalidArgument,
+			"No mutations provided",
+		)
+	}
 	s.mu.Lock()
 	tbl, ok := s.tables[req.TableName]
 	s.mu.Unlock()

bigtable/bttest/inmem_test.go

@@ -2034,6 +2034,50 @@ func TestValueFilterRowWithAlternationInRegex(t *testing.T) {
 	}
 }
 
+func TestMutateRowEmptyMutationErrors(t *testing.T) {
+	srv := &server{tables: make(map[string]*table)}
+	ctx := context.Background()
+	req := &btpb.MutateRowRequest{
+		TableName: "mytable",
+		RowKey:    []byte("r"),
+		Mutations: []*btpb.Mutation{},
+	}
+
+	resp, err := srv.MutateRow(ctx, req)
+	if resp != nil ||
+		fmt.Sprint(err) !=
+			"rpc error: code = InvalidArgument"+
+				" desc = No mutations provided" {
+		t.Fatalf("Failed to error %s", err)
+	}
+}
+
+type bigtableTestingMutateRowsServer struct {
+	grpc.ServerStream
+}
+
+func (x *bigtableTestingMutateRowsServer) Send(m *btpb.MutateRowsResponse) error {
+	return nil
+}
+
+func TestMutateRowsEmptyMutationErrors(t *testing.T) {
+	srv := &server{tables: make(map[string]*table)}
+	req := &btpb.MutateRowsRequest{
+		TableName: "mytable",
+		Entries: []*btpb.MutateRowsRequest_Entry{
+			{Mutations: []*btpb.Mutation{}},
+			{Mutations: []*btpb.Mutation{}},
+		},
+	}
+
+	err := srv.MutateRows(req, &bigtableTestingMutateRowsServer{})
+	if fmt.Sprint(err) !=
+		"rpc error: code = InvalidArgument "+
+			"desc = No mutations provided" {
+		t.Fatalf("Failed to error %s", err)
+	}
+}
+
 func TestFilterRowCellsPerRowLimitFilterTruthiness(t *testing.T) {
 	row := &row{
 		key: "row",