[FAB-10591] Discovery Client Endorser Tests

Added integration tests for Discovery Client
endorser selection based on CC policy.

Also, refactored and fixed some of the existing
integration tests since they were failing
intermittently.

Change-Id: I9a35b5f9ab4037edcbcee8d1670b7d62e2fdcd59
Signed-off-by: Bob Stasyszyn <Bob.Stasyszyn@securekey.com>

Author: bstasyszyn

test/integration/base_test_setup.go

@@ -9,17 +9,23 @@ package integration
 import (
 	"os"
 	"path"
+	"time"
 
 	mspclient "github.com/hyperledger/fabric-sdk-go/pkg/client/msp"
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/resmgmt"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/errors/retry"
+	contextAPI "github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/core"
+	fabAPI "github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/msp"
+	contextImpl "github.com/hyperledger/fabric-sdk-go/pkg/context"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab"
 	packager "github.com/hyperledger/fabric-sdk-go/pkg/fab/ccpackager/gopackager"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/comm"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/resource"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fabsdk"
 	"github.com/hyperledger/fabric-sdk-go/pkg/util/test"
+	"github.com/hyperledger/fabric-sdk-go/test/metadata"
 	"github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/common/cauthdsl"
 	"github.com/pkg/errors"
 )
@@ -39,6 +45,7 @@ const (
 	ExampleCCInitB    = "200"
 	ExampleCCUpgradeB = "400"
 	AdminUser         = "Admin"
+	OrdererOrgName    = "ordererorg"
 )
 
 // ExampleCC query and transaction arguments
@@ -69,6 +76,20 @@ func ExampleCCUpgradeArgs() [][]byte {
 	return upgradeArgs
 }
 
+// IsJoinedChannel returns true if the given peer has joined the given channel
+func IsJoinedChannel(channelID string, resMgmtClient *resmgmt.Client, peer fabAPI.Peer) (bool, error) {
+	resp, err := resMgmtClient.QueryChannels(resmgmt.WithTargets(peer))
+	if err != nil {
+		return false, err
+	}
+	for _, chInfo := range resp.Channels {
+		if chInfo.ChannelId == channelID {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 // Initialize reads configuration from file and sets up client, channel and event hub
 func (setup *BaseSetupImpl) Initialize(sdk *fabsdk.FabricSDK) error {
 
@@ -168,3 +189,132 @@ func InstallAndInstantiateCC(sdk *fabsdk.FabricSDK, user fabsdk.ContextOption, o
 	ccPolicy := cauthdsl.SignedByMspMember(mspID)
 	return resMgmtClient.InstantiateCC("mychannel", resmgmt.InstantiateCCRequest{Name: ccName, Path: ccPath, Version: ccVersion, Args: ccArgs, Policy: ccPolicy}, resmgmt.WithRetry(retry.DefaultResMgmtOpts))
 }
+
+// OrgContext provides SDK client context for a given org
+type OrgContext struct {
+	OrgID                string
+	CtxProvider          contextAPI.ClientProvider
+	SigningIdentity      msp.SigningIdentity
+	ResMgmt              *resmgmt.Client
+	Peers                []fabAPI.Peer
+	AnchorPeerConfigFile string
+}
+
+// CreateChannelAndUpdateAnchorPeers creates the channel and updates all of the anchor peers for all orgs
+func CreateChannelAndUpdateAnchorPeers(sdk *fabsdk.FabricSDK, channelID string, channelConfigFile string, orgsContext []*OrgContext) error {
+	ordererCtx := sdk.Context(fabsdk.WithUser(AdminUser), fabsdk.WithOrg(OrdererOrgName))
+
+	// Channel management client is responsible for managing channels (create/update channel)
+	chMgmtClient, err := resmgmt.New(ordererCtx)
+	if err != nil {
+		return errors.New("failed to get a new resmgmt client for orderer")
+	}
+
+	var signingIdentities []msp.SigningIdentity
+	for _, orgCtx := range orgsContext {
+		signingIdentities = append(signingIdentities, orgCtx.SigningIdentity)
+	}
+
+	req := resmgmt.SaveChannelRequest{
+		ChannelID:         channelID,
+		ChannelConfigPath: path.Join("../../../", metadata.ChannelConfigPath, channelConfigFile),
+		SigningIdentities: signingIdentities,
+	}
+	_, err = chMgmtClient.SaveChannel(req, resmgmt.WithRetry(retry.DefaultResMgmtOpts), resmgmt.WithOrdererEndpoint("orderer.example.com"))
+	if err != nil {
+		return err
+	}
+
+	for _, orgCtx := range orgsContext {
+		req := resmgmt.SaveChannelRequest{
+			ChannelID:         channelID,
+			ChannelConfigPath: path.Join("../../../", metadata.ChannelConfigPath, orgCtx.AnchorPeerConfigFile),
+			SigningIdentities: []msp.SigningIdentity{orgCtx.SigningIdentity},
+		}
+		if _, err := orgCtx.ResMgmt.SaveChannel(req, resmgmt.WithRetry(retry.DefaultResMgmtOpts), resmgmt.WithOrdererEndpoint("orderer.example.com")); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// JoinPeersToChannel joins all peers in all of the given orgs to the given channel
+func JoinPeersToChannel(channelID string, orgsContext []*OrgContext) error {
+	for _, orgCtx := range orgsContext {
+		err := orgCtx.ResMgmt.JoinChannel(
+			channelID,
+			resmgmt.WithRetry(retry.DefaultResMgmtOpts),
+			resmgmt.WithOrdererEndpoint("orderer.example.com"),
+			resmgmt.WithTargets(orgCtx.Peers...),
+		)
+		if err != nil {
+			return errors.Wrapf(err, "failed to join peers in org [%s] to channel [%s]", orgCtx.OrgID, channelID)
+		}
+	}
+	return nil
+}
+
+// InstallAndInstantiateChaincode installs the given chaincode to all peers in the given orgs and instantiates it on the given channel
+func InstallAndInstantiateChaincode(channelID string, ccPkg *resource.CCPackage, ccID, ccVersion, ccPolicy string, orgs []*OrgContext) error {
+	for _, orgCtx := range orgs {
+		if err := InstallChaincode(orgCtx.ResMgmt, orgCtx.CtxProvider, ccPkg, ccID, ccVersion, orgCtx.Peers); err != nil {
+			return errors.Wrapf(err, "failed to install chaincode to peers in org [%s]", orgCtx.OrgID)
+		}
+	}
+	_, err := InstantiateChaincode(orgs[0].ResMgmt, channelID, ccID, ccVersion, ccPolicy)
+	return err
+}
+
+// InstallChaincode installs the given chaincode to the given peers
+func InstallChaincode(resMgmt *resmgmt.Client, ctxProvider contextAPI.ClientProvider, ccPkg *resource.CCPackage, ccName, ccVersion string, localPeers []fabAPI.Peer) error {
+	installCCReq := resmgmt.InstallCCRequest{Name: ccName, Path: "github.com/example_cc", Version: ccVersion, Package: ccPkg}
+	_, err := resMgmt.InstallCC(installCCReq, resmgmt.WithRetry(retry.DefaultResMgmtOpts))
+	return err
+}
+
+// InstantiateChaincode instantiates the given chaincode to the given channel
+func InstantiateChaincode(resMgmt *resmgmt.Client, channelID, ccName, ccVersion string, ccPolicyStr string) (resmgmt.InstantiateCCResponse, error) {
+	ccPolicy, err := cauthdsl.FromString(ccPolicyStr)
+	if err != nil {
+		return resmgmt.InstantiateCCResponse{}, errors.Wrapf(err, "error creating CC policy [%s]", ccPolicyStr)
+	}
+
+	return resMgmt.InstantiateCC(
+		channelID,
+		resmgmt.InstantiateCCRequest{
+			Name:    ccName,
+			Path:    "github.com/example_cc",
+			Version: ccVersion,
+			Args:    ExampleCCInitArgs(),
+			Policy:  ccPolicy,
+		},
+		resmgmt.WithRetry(retry.DefaultResMgmtOpts),
+	)
+}
+
+// DiscoverLocalPeers queries the local peers for the given MSP context and returns all of the peers. If
+// the number of peers does not match the expected number then an error is returned.
+func DiscoverLocalPeers(ctxProvider contextAPI.ClientProvider, expectedPeers int) ([]fabAPI.Peer, error) {
+	ctx, err := contextImpl.NewLocal(ctxProvider)
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating local context")
+	}
+
+	var peers []fabAPI.Peer
+	for i := 0; i < 10; i++ {
+		peers, err = ctx.LocalDiscoveryService().GetPeers()
+		if err != nil {
+			return nil, errors.Wrapf(err, "error getting peers for MSP [%s]", ctx.Identifier().MSPID)
+		}
+		if len(peers) >= expectedPeers {
+			break
+		}
+		// wait some time to allow the gossip to propagate the peers discovery
+		time.Sleep(3 * time.Second)
+	}
+	if expectedPeers != len(peers) {
+		return nil, errors.Errorf("Expecting %d peers but got %d", expectedPeers, len(peers))
+	}
+	return peers, nil
+}

test/integration/e2e/end_to_end.go

@@ -9,10 +9,12 @@ package e2e
 import (
 	"path"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/errors/retry"
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/msp"
 	"github.com/stretchr/testify/require"
 
@@ -96,15 +98,18 @@ func setupAndRun(t *testing.T, doSetup bool, configOpt core.ConfigProvider, sdkO
 	// Move funds
 	executeCC(client, t)
 
+	var ccEvent *fab.CCEvent
 	select {
-	case ccEvent := <-notifier:
+	case ccEvent = <-notifier:
 		t.Logf("Received CC event: %#v\n", ccEvent)
 	case <-time.After(time.Second * 20):
 		t.Fatalf("Did NOT receive CC event for eventId(%s)\n", eventID)
 	}
 
-	// Verify move funds transaction result
-	verifyFundsIsMoved(client, t, value)
+	i := strings.Index(ccEvent.SourceURL, ":")
+
+	// Verify move funds transaction result on the same peer where the event came from.
+	verifyFundsIsMoved(client, t, value, ccEvent.SourceURL[0:i])
 
 }
 
@@ -143,8 +148,8 @@ func createChannelAndCC(t *testing.T, sdk *fabsdk.FabricSDK) {
 	createCC(t, orgResMgmt)
 }
 
-func verifyFundsIsMoved(client *channel.Client, t *testing.T, value []byte) {
-	newValue := queryCC(client, t)
+func verifyFundsIsMoved(client *channel.Client, t *testing.T, value []byte, targetEndpoints ...string) {
+	newValue := queryCC(client, t, targetEndpoints...)
 	valueInt, err := strconv.Atoi(string(value))
 	if err != nil {
 		t.Fatal(err.Error())
@@ -166,9 +171,11 @@ func executeCC(client *channel.Client, t *testing.T) {
 	}
 }
 
-func queryCC(client *channel.Client, t *testing.T) []byte {
+func queryCC(client *channel.Client, t *testing.T, targetEndpoints ...string) []byte {
 	response, err := client.Query(channel.Request{ChaincodeID: ccID, Fcn: "invoke", Args: integration.ExampleCCQueryArgs()},
-		channel.WithRetry(retry.DefaultChannelOpts))
+		channel.WithRetry(retry.DefaultChannelOpts),
+		channel.WithTargetEndpoints(targetEndpoints...),
+	)
 	if err != nil {
 		t.Fatalf("Failed to query funds: %s", err)
 	}