[FAB-10549] gracefully stop mock servers

Change-Id: I1e271af290c1d96dc8e999a1ef443bbecc429f3f
Signed-off-by: Troy Ronda <troy@troyronda.com>

Author: troyronda

pkg/client/resmgmt/resmgmt_test.go

@@ -8,7 +8,6 @@ package resmgmt
 
 import (
 	"fmt"
-	"net"
 	"net/http"
 	"os"
 	"strings"
@@ -18,10 +17,6 @@ import (
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/errors/retry"
 
 	"github.com/golang/protobuf/proto"
-	"github.com/pkg/errors"
-	"github.com/stretchr/testify/assert"
-	"google.golang.org/grpc"
-
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/errors/status"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/core"
@@ -40,12 +35,15 @@ import (
 	"github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/common/cauthdsl"
 	"github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/common"
 	pb "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/peer"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
 )
 
 const (
 	channelConfig = "../../../test/fixtures/fabric/v1.0/channel/mychannel.tx"
 	networkCfg    = "../../../test/fixtures/config/config_test.yaml"
 	configPath    = "../../core/config/testdata/config_test.yaml"
+	testAddress   = "127.0.0.1:0"
 )
 
 func withLocalContextProvider(provider context.LocalProvider) ClientOption {
@@ -57,10 +55,10 @@ func withLocalContextProvider(provider context.LocalProvider) ClientOption {
 
 func TestJoinChannelFail(t *testing.T) {
 
-	grpcServer := grpc.NewServer()
-	defer grpcServer.Stop()
+	srv := &fcmocks.MockEndorserServer{}
+	addr := srv.Start(testAddress)
+	defer srv.Stop()
 
-	endorserServer, addr := startEndorserServer(t, grpcServer)
 	ctx := setupTestContext("test", "Org1MSP")
 
 	// Create mock orderer with simple mock block
@@ -83,7 +81,7 @@ func TestJoinChannelFail(t *testing.T) {
 	peer1, _ := peer.New(fcmocks.NewMockEndpointConfig(), peer.WithURL("grpc://"+addr))
 
 	// Test fail with send proposal error
-	endorserServer.ProposalError = errors.New("Test Error")
+	srv.ProposalError = errors.New("Test Error")
 	err = rc.JoinChannel("mychannel", WithTargets(peer1))
 
 	if err == nil || !strings.Contains(err.Error(), "Test Error") {
@@ -93,10 +91,10 @@ func TestJoinChannelFail(t *testing.T) {
 }
 
 func TestJoinChannelSuccess(t *testing.T) {
-	grpcServer := grpc.NewServer()
-	defer grpcServer.Stop()
+	srv := &fcmocks.MockEndorserServer{}
+	addr := srv.Start(testAddress)
+	defer srv.Stop()
 
-	_, addr := startEndorserServer(t, grpcServer)
 	ctx := setupTestContext("test", "Org1MSP")
 
 	// Create mock orderer with simple mock block
@@ -129,10 +127,10 @@ func TestWithFilterOption(t *testing.T) {
 }
 
 func TestJoinChannelWithFilter(t *testing.T) {
-	grpcServer := grpc.NewServer()
-	defer grpcServer.Stop()
+	srv := &fcmocks.MockEndorserServer{}
+	addr := srv.Start(testAddress)
+	defer srv.Stop()
 
-	_, addr := startEndorserServer(t, grpcServer)
 	ctx := setupTestContext("test", "Org1MSP")
 
 	// Create mock orderer with simple mock block
@@ -216,10 +214,9 @@ func TestJoinChannelRequiredParameters(t *testing.T) {
 
 func TestJoinChannelWithOptsRequiredParameters(t *testing.T) {
 
-	grpcServer := grpc.NewServer()
-	defer grpcServer.Stop()
-
-	_, addr := startEndorserServer(t, grpcServer)
+	srv := &fcmocks.MockEndorserServer{}
+	addr := srv.Start(testAddress)
+	defer srv.Stop()
 
 	ctx := setupTestContext("test", "Org1MSP")
 	network := getNetworkConfig(t)
@@ -1003,21 +1000,6 @@ func setupCustomOrderer(ctx *fcmocks.MockContext, mockOrderer fab.Orderer) *fcmo
 	return ctx
 }
 
-func startEndorserServer(t *testing.T, grpcServer *grpc.Server) (*fcmocks.MockEndorserServer, string) {
-	lis, err := net.Listen("tcp", "127.0.0.1:0")
-	addr := lis.Addr().String()
-
-	endorserServer := &fcmocks.MockEndorserServer{}
-	pb.RegisterEndorserServer(grpcServer, endorserServer)
-	if err != nil {
-		t.Logf("Error starting test server %s", err)
-		t.FailNow()
-	}
-	t.Logf("Starting test server on %s\n", addr)
-	go grpcServer.Serve(lis)
-	return endorserServer, addr
-}
-
 func getNetworkConfig(t *testing.T) fab.EndpointConfig {
 	configBackend, err := configImpl.FromFile(networkCfg)()
 	if err != nil {
@@ -1034,9 +1016,9 @@ func getNetworkConfig(t *testing.T) fab.EndpointConfig {
 
 func TestSaveChannelSuccess(t *testing.T) {
 
-	grpcServer := grpc.NewServer()
-	defer grpcServer.Stop()
-	_, addr := fcmocks.StartMockBroadcastServer("127.0.0.1:0", grpcServer)
+	mb := fcmocks.MockBroadcastServer{}
+	addr := mb.Start("127.0.0.1:0")
+	defer mb.Stop()
 
 	ctx := setupTestContext("test", "Org1MSP")
 
@@ -1137,9 +1119,9 @@ func TestSaveChannelFailure(t *testing.T) {
 
 func TestSaveChannelWithOpts(t *testing.T) {
 
-	grpcServer := grpc.NewServer()
-	defer grpcServer.Stop()
-	_, addr := fcmocks.StartMockBroadcastServer("127.0.0.1:0", grpcServer)
+	mb := fcmocks.MockBroadcastServer{}
+	addr := mb.Start("127.0.0.1:0")
+	defer mb.Stop()
 
 	ctx := setupTestContext("test", "Org1MSP")
 
@@ -1210,9 +1192,10 @@ func TestJoinChannelWithInvalidOpts(t *testing.T) {
 }
 
 func TestSaveChannelWithMultipleSigningIdenities(t *testing.T) {
-	grpcServer := grpc.NewServer()
-	defer grpcServer.Stop()
-	_, addr := fcmocks.StartMockBroadcastServer("127.0.0.1:0", grpcServer)
+	mb := fcmocks.MockBroadcastServer{}
+	addr := mb.Start("127.0.0.1:0")
+	defer mb.Stop()
+
 	ctx := setupTestContext("test", "Org1MSP")
 
 	mockConfig := &fcmocks.MockConfig{}

pkg/fab/comm/comm_test.go

@@ -11,13 +11,10 @@ import (
 	"net"
 	"os"
 	"testing"
-	"time"
 
 	eventmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/mocks"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
-	"github.com/hyperledger/fabric-sdk-go/pkg/util/test"
 	pb "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/peer"
-	"github.com/pkg/errors"
 	"google.golang.org/grpc"
 )
 
@@ -51,37 +48,18 @@ func TestMain(m *testing.M) {
 	}
 	endorserAddr = addrs
 
-	time.Sleep(2 * time.Second)
 	os.Exit(m.Run())
 }
 
-func startEndorsers(count int, address string) ([]*grpc.Server, []string, error) {
-	srvs := make([]*grpc.Server, 0, count)
+func startEndorsers(count int, address string) ([]*mocks.MockEndorserServer, []string, error) {
+	srvs := make([]*mocks.MockEndorserServer, 0, count)
 	addrs := make([]string, 0, count)
 
 	for i := 0; i < count; i++ {
-		srv := grpc.NewServer()
-		_, addr, ok := startEndorserServer(srv, address)
-		if !ok {
-			return nil, nil, errors.New("unable to start GRPC server")
-		}
+		srv := &mocks.MockEndorserServer{}
+		addr := srv.Start(address)
 		srvs = append(srvs, srv)
 		addrs = append(addrs, addr)
 	}
 	return srvs, addrs, nil
 }
-
-func startEndorserServer(grpcServer *grpc.Server, address string) (*mocks.MockEndorserServer, string, bool) {
-	lis, err := net.Listen("tcp", address)
-	if err != nil {
-		test.Logf("Error starting test server [%s]", err)
-		return nil, "", false
-	}
-	addr := lis.Addr().String()
-
-	endorserServer := &mocks.MockEndorserServer{}
-	pb.RegisterEndorserServer(grpcServer, endorserServer)
-	test.Logf("Starting test server [%s]", addr)
-	go grpcServer.Serve(lis)
-	return endorserServer, addr, true
-}

pkg/fab/events/endpoint/endpoint_test.go

@@ -135,7 +135,7 @@ func newMockConfig(channelPeers ...fab.ChannelPeer) *mockConfig {
 }
 
 func (c *mockConfig) ChannelPeers(name string) ([]fab.ChannelPeer, bool) {
-	test.Logf("mockConfig.ChannelPeers [%#n]", c.channelPeers)
+	test.Logf("mockConfig.ChannelPeers [%#v]", c.channelPeers)
 	return c.channelPeers, true
 }
 

pkg/fab/mocks/mockbroadcastserver.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"sync"
 
 	po "github.com/hyperledger/fabric-sdk-go/internal/github.com/hyperledger/fabric/protos/orderer"
 	"github.com/hyperledger/fabric-sdk-go/pkg/util/test"
@@ -38,6 +39,8 @@ type MockBroadcastServer struct {
 	DeliverResponse              *po.DeliverResponse
 	BroadcastError               error
 	BroadcastCustomResponse      *po.BroadcastResponse
+	srv                          *grpc.Server
+	wg                           sync.WaitGroup
 }
 
 // Broadcast mock broadcast
@@ -90,22 +93,39 @@ func (m *MockBroadcastServer) Deliver(server po.AtomicBroadcast_DeliverServer) e
 	return nil
 }
 
-//StartMockBroadcastServer starts mock server for unit testing purpose
-func StartMockBroadcastServer(broadcastTestURL string, grpcServer *grpc.Server) (*MockBroadcastServer, string) {
-	lis, err := net.Listen("tcp", broadcastTestURL)
+// Start the mock broadcast server
+func (m *MockBroadcastServer) Start(address string) string {
+	if m.srv != nil {
+		panic("MockBroadcastServer already started")
+	}
+	m.srv = grpc.NewServer()
+
+	lis, err := net.Listen("tcp", address)
 	if err != nil {
 		panic(fmt.Sprintf("Error starting BroadcastServer %s", err))
 	}
 	addr := lis.Addr().String()
 
 	test.Logf("Starting MockEventServer [%s]", addr)
-	broadcastServer := new(MockBroadcastServer)
-	po.RegisterAtomicBroadcastServer(grpcServer, broadcastServer)
+	po.RegisterAtomicBroadcastServer(m.srv, m)
+	m.wg.Add(1)
 	go func() {
-		if err := grpcServer.Serve(lis); err != nil {
+		defer m.wg.Done()
+		if err := m.srv.Serve(lis); err != nil {
 			test.Logf("StartMockBroadcastServer failed [%s]", err)
 		}
 	}()
 
-	return broadcastServer, addr
+	return addr
+}
+
+// Stop the mock broadcast server and wait for completion.
+func (m *MockBroadcastServer) Stop() {
+	if m.srv == nil {
+		panic("MockBroadcastServer not started")
+	}
+
+	m.srv.Stop()
+	m.wg.Wait()
+	m.srv = nil
 }

pkg/fab/mocks/mockendorserserver.go

@@ -9,20 +9,23 @@ package mocks
 import (
 	"fmt"
 	"net"
+	"sync"
 
 	"github.com/golang/protobuf/proto"
+	"github.com/hyperledger/fabric-sdk-go/pkg/util/test"
+	"github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/rwsetutil"
+	"github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/ledger/rwset/kvrwset"
+	pb "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/peer"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
-
-	rwsetutil "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/rwsetutil"
-	kvrwset "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/ledger/rwset/kvrwset"
-	pb "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/peer"
 )
 
 // MockEndorserServer mock endoreser server to process endorsement proposals
 type MockEndorserServer struct {
 	ProposalError error
 	AddkvWrite    bool
+	srv           *grpc.Server
+	wg            sync.WaitGroup
 }
 
 // ProcessProposal mock implementation that returns success if error is not set
@@ -72,20 +75,39 @@ func (m *MockEndorserServer) createProposalResponsePayload() []byte {
 	return prpBytes
 }
 
-//StartEndorserServer starts mock server for unit testing purpose
-func StartEndorserServer(endorserTestURL string) *MockEndorserServer {
-	grpcServer := grpc.NewServer()
-	lis, err := net.Listen("tcp", endorserTestURL)
+// Start the mock broadcast server
+func (m *MockEndorserServer) Start(address string) string {
+	if m.srv != nil {
+		panic("MockBroadcastServer already started")
+	}
+	m.srv = grpc.NewServer()
+
+	lis, err := net.Listen("tcp", address)
 	if err != nil {
-		panic(fmt.Sprintf("Error starting endorser server: %s", err))
+		panic(fmt.Sprintf("Error starting BroadcastServer %s", err))
 	}
-	endorserServer := &MockEndorserServer{}
-	pb.RegisterEndorserServer(grpcServer, endorserServer)
-	fmt.Print("Test endorser server started\n")
+	addr := lis.Addr().String()
+
+	test.Logf("Starting MockEventServer [%s]", addr)
+	pb.RegisterEndorserServer(m.srv, m)
+	m.wg.Add(1)
 	go func() {
-		if err := grpcServer.Serve(lis); err != nil {
-			panic(err.Error())
+		defer m.wg.Done()
+		if err := m.srv.Serve(lis); err != nil {
+			test.Logf("StartMockBroadcastServer failed [%s]", err)
 		}
 	}()
-	return endorserServer
+
+	return addr
+}
+
+// Stop the mock broadcast server and wait for completion.
+func (m *MockEndorserServer) Stop() {
+	if m.srv == nil {
+		panic("MockBroadcastServer not started")
+	}
+
+	m.srv.Stop()
+	m.wg.Wait()
+	m.srv = nil
 }