Dedup go errors

.gitignore

@@ -7,3 +7,5 @@
 coverage.txt
 pkg/tests/coverage.txt
 testing/coverage.txt
+go.work
+go.work.sum

testing/x/replay.go

@@ -181,8 +181,11 @@ func replay[Req protoreflect.ProtoMessage, Resp protoreflect.ProtoMessage](
 	assert.NoError(t, err)
 
 	resp, err := serve(ctx, req)
+	if err != nil && entry.Errors != nil {
+		assert.Equal(t, *entry.Errors, err.Error())
+		return
+	}
 	require.NoError(t, err)
-
 	bytes, err := jsonpb.Marshal(resp)
 	assert.NoError(t, err)
 
@@ -236,4 +239,5 @@ type jsonLogEntry struct {
 	Method   string          `json:"method"`
 	Request  json.RawMessage `json:"request,omitempty"`
 	Response json.RawMessage `json:"response,omitempty"`
+	Errors   *string         `json:"errors,omitempty"`
 }

x/muxer/muxer.go

@@ -87,8 +87,10 @@ type GetMappingResponse struct {
 	Data     []byte
 }
 
-type getMappingHandler = map[string]MultiMappingHandler
-type MultiMappingHandler = func(GetMappingArgs) (GetMappingResponse, error)
+type (
+	getMappingHandler   = map[string]MultiMappingHandler
+	MultiMappingHandler = func(GetMappingArgs) (GetMappingResponse, error)
+)
 
 func (m *muxer) getFunction(token string) server {
 	i, ok := m.dispatchTable.Functions[token]
@@ -128,8 +130,15 @@ func (m *muxer) CheckConfig(ctx context.Context, req *rpc.CheckRequest) (*rpc.Ch
 	failures := []*rpc.CheckFailure{}
 	uniqueFailures := map[string]struct{}{}
 	var errs multierror.Error
+	uniqueErrors := map[string]struct{}{}
 	for i, r := range asyncJoin(subs) {
 		if err := r.B; err != nil {
+			errString := err.Error()
+			// De-duplicate errors
+			if _, has := uniqueErrors[errString]; has {
+				continue
+			}
+			uniqueErrors[errString] = struct{}{}
 			errs.Errors = append(errs.Errors, err)
 			continue
 		}
@@ -149,7 +158,7 @@ func (m *muxer) CheckConfig(ctx context.Context, req *rpc.CheckRequest) (*rpc.Ch
 			}
 		}
 
-		// Here we de-duplicate errors.
+		// Here we de-duplicate rpc failures.
 		for _, e := range r.A.GetFailures() {
 			s := e.GetProperty() + ":" + e.GetReason()
 			if _, has := uniqueFailures[s]; has {
@@ -392,7 +401,6 @@ func (m *muxer) Cancel(ctx context.Context, e *emptypb.Empty) (*emptypb.Empty, e
 		}
 	}
 	return e, m.muxedErrors(errs)
-
 }
 
 func (m *muxer) GetPluginInfo(ctx context.Context, e *emptypb.Empty) (*rpc.PluginInfo, error) {
@@ -437,7 +445,6 @@ func (a *getMappingArgs) Fetch() []GetMappingResponse {
 }
 
 func (m *muxer) GetMapping(ctx context.Context, req *rpc.GetMappingRequest) (*rpc.GetMappingResponse, error) {
-
 	// We need to merge multiple mappings
 	combineMapping, found := m.getMappingByKey[req.Key]
 	if !found {

x/muxer/tests/go.mod

@@ -4,6 +4,8 @@ go 1.20
 
 replace github.com/pulumi/pulumi-terraform-bridge/x/muxer => ../
 
+replace github.com/pulumi/pulumi-terraform-bridge/testing => ../../../testing/
+
 require (
 	github.com/pulumi/pulumi-terraform-bridge/testing v0.0.0-20230406212415-0b560771908d
 	github.com/pulumi/pulumi-terraform-bridge/x/muxer v0.0.4

x/muxer/tests/go.sum

@@ -142,8 +142,6 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pulumi/esc v0.5.7-0.20231030195049-f71961c0d5fa h1:5y6/zZsPQW8xNgfyWVMTnuSl8gH2wrYkvTOAqPwhM9Q=
 github.com/pulumi/esc v0.5.7-0.20231030195049-f71961c0d5fa/go.mod h1:Y6W21yUukvxS2NnS5ae1beMSPhMvj0xNAYcDqDHVj/g=
-github.com/pulumi/pulumi-terraform-bridge/testing v0.0.0-20230406212415-0b560771908d h1:OjHMocIBmWuY13luNE7okcyRu1bxNdXP6kPSW0p6x84=
-github.com/pulumi/pulumi-terraform-bridge/testing v0.0.0-20230406212415-0b560771908d/go.mod h1:slt4v0erbPg9FCxmbQFc6SA6toGtqfdSsQPihRMktBU=
 github.com/pulumi/pulumi/pkg/v3 v3.93.0 h1:ryb8a691MffcAv8KAgAIRvpZOjVW1xCpSDH10x9ZYUg=
 github.com/pulumi/pulumi/pkg/v3 v3.93.0/go.mod h1:cGnwvoPZ8fjR0cUvOt1skqLl8E4fIN708lFITPhY/K8=
 github.com/pulumi/pulumi/sdk/v3 v3.93.0 h1:5InTUxulGuPMXQbnrruIbOEJylRkRWfoDyQIj6dtsYA=

x/muxer/tests/muxer_test.go

@@ -17,6 +17,7 @@ package muxer_test
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"reflect"
 	"testing"
@@ -68,6 +69,54 @@ func TestSimpleDispatch(t *testing.T) {
 	)
 }
 
+func TestCheckConfigErrorNotDuplicated(t *testing.T) {
+	var m muxer.DispatchTable
+	m.Resources = map[string]int{
+		"test:mod:A": 0,
+		"test:mod:B": 1,
+	}
+	errString := "myerr"
+	mux(t, m).replay(
+		exchange("/pulumirpc.ResourceProvider/CheckConfig", "{}", "{}", &errString,
+			part(0, "{}", "{}", &errString),
+			part(1, "{}", "{}", &errString),
+		))
+}
+
+func TestCheckConfigDifferentErrorsNotDropped(t *testing.T) {
+	var m muxer.DispatchTable
+	m.Resources = map[string]int{
+		"test:mod:A": 0,
+		"test:mod:B": 1,
+	}
+	firstErr := "myerr"
+	secondErr := "othererr"
+	expectedErr := "2 errors occurred:\n\t* myerr\n\t* othererr\n\n"
+	mux(t, m).replay(
+		exchange(
+			"/pulumirpc.ResourceProvider/CheckConfig",
+			"{}",
+			"{}",
+			&expectedErr,
+			part(0, "{}", "{}", &firstErr),
+			part(1, "{}", "{}", &secondErr),
+		))
+}
+
+func TestCheckConfigOneErrorReturned(t *testing.T) {
+	var m muxer.DispatchTable
+	m.Resources = map[string]int{
+		"test:mod:A": 0,
+		"test:mod:B": 1,
+	}
+	err := "myerr"
+	mux(t, m).replay(
+		exchange("/pulumirpc.ResourceProvider/CheckConfig", "{}", "{}", &err,
+			part(0, "{}", `{"inputs": {"myurn":"urn"}}`, nil),
+			part(1, "{}", "{}", &err),
+		))
+}
+
 func TestConfigure(t *testing.T) {
 	var m muxer.DispatchTable
 	m.Resources = map[string]int{
@@ -84,7 +133,7 @@ func TestConfigure(t *testing.T) {
       }
     }`, `{
       "supportsPreview": true
-  }`,
+  }`, nil,
 			part(0, `{
   "args": {
     "a": "1",
@@ -94,7 +143,7 @@ func TestConfigure(t *testing.T) {
 }`, `{
   "acceptSecrets": true,
   "supportsPreview": true
-}`),
+}`, nil),
 			part(1, `{
   "args": {
     "a": "1",
@@ -104,7 +153,7 @@ func TestConfigure(t *testing.T) {
 }`, `{
   "supportsPreview": true,
   "acceptResources": true
-}`),
+}`, nil),
 		))
 }
 
@@ -139,9 +188,9 @@ func TestDivergentCheckConfig(t *testing.T) {
 	  }
 	}`
 	muxedResp := resp0
-	e := exchange("/pulumirpc.ResourceProvider/CheckConfig", req, muxedResp,
-		part(0, req, resp0),
-		part(1, req, resp1))
+	e := exchange("/pulumirpc.ResourceProvider/CheckConfig", req, muxedResp, nil,
+		part(0, req, resp0, nil),
+		part(1, req, resp1, nil))
 
 	m := muxer.DispatchTable{}
 	m.Resources = map[string]int{}
@@ -162,17 +211,17 @@ func TestGetMapping(t *testing.T) {
 }`, `{
   "provider": "p1",
   "data": "dw=="`+ /* the base64 encoding of d1 */ `
-}`, part(0, `{
+}`, nil, part(0, `{
   "key": "k1"
 }`, `{
   "provider": "p1",
   "data": "d1"
-}`), part(1, `{
+}`, nil), part(1, `{
   "key": "k1"
 }`, `{
   "provider": "",
   "data": ""
-}`)))
+}`, nil)))
 	})
 	t.Run("merged-responding-server", func(t *testing.T) {
 		var m muxer.DispatchTable
@@ -199,17 +248,17 @@ func TestGetMapping(t *testing.T) {
 }`, `{
   "provider": "p1",
   "data": "cjE="`+ /* the base64 encoding of r1 */ `
-}`, part(0, `{
+}`, nil, part(0, `{
   "key": "k"
 }`, `{
   "provider": "p1",
   "data": "ZDE="`+ /* the base64 encoding of d1*/ `
-}`), part(1, `{
+}`, nil), part(1, `{
   "key": "k"
 }`, `{
   "provider": "p1",
   "data": "ZDI="`+ /* the base64 encoding of d2*/ `
-}`)))
+}`, nil)))
 	})
 }
 
@@ -242,6 +291,7 @@ func (m testMuxer) replay(exchanges ...Exchange) {
 				call{
 					incoming: part.Request,
 					response: part.Response,
+					errors:   part.Errors,
 				})
 		}
 	}
@@ -260,13 +310,15 @@ type Exchange struct {
 	Method   string          `json:"method"`
 	Request  json.RawMessage `json:"request"`
 	Response json.RawMessage `json:"response"`
-	Parts    []ExchangePart  `json:"-"`
+	Errors   *string
+	Parts    []ExchangePart `json:"-"`
 }
 
 type ExchangePart struct {
 	Provider int
 	Request  string `json:"request"`
 	Response string `json:"response"`
+	Errors   *string
 }
 
 // A simple exchange is one where only one sub-server is used
@@ -285,20 +337,22 @@ func simpleExchange(provider int, method, request, response string) Exchange {
 	}
 }
 
-func exchange(method, request, response string, parts ...ExchangePart) Exchange {
+func exchange(method, request, response string, errors *string, parts ...ExchangePart) Exchange {
 	return Exchange{
 		Method:   method,
 		Request:  json.RawMessage(request),
 		Response: json.RawMessage(response),
+		Errors:   errors,
 		Parts:    parts,
 	}
 }
 
-func part(provider int, request, response string) ExchangePart {
+func part(provider int, request, response string, errors *string) ExchangePart {
 	return ExchangePart{
 		provider,
 		request,
 		response,
+		errors,
 	}
 }
 
@@ -340,6 +394,7 @@ type server struct {
 type call struct {
 	incoming string
 	response string
+	errors   *string
 }
 
 // Assert that a gRPC call matches the next expected call, then rehydrate and return the
@@ -353,6 +408,10 @@ func handleMethod[T proto.Message, R proto.Message](m *server, req T) (R, error)
 	var r R
 	reflect.ValueOf(&r).Elem().Set(reflect.New(reflect.TypeOf(r).Elem()))
 
+	if next.errors != nil {
+		return r, errors.New(*next.errors)
+	}
+
 	marshalled, err := protojson.MarshalOptions{Multiline: true}.Marshal(req)
 	require.NoError(m.t, err)