pkg/grpc/client_test.go - voltha-lib-go - Gitiles

 /*
  * Copyright 2021-present Open Networking Foundation

  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at

  * http://www.apache.org/licenses/LICENSE-2.0

  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package grpc

 import (
 	"context"
 	"fmt"
 	"os"
 	"strconv"
 	"sync"
 	"testing"
 	"time"

 	"github.com/opencord/voltha-lib-go/v7/pkg/log"
 	"github.com/opencord/voltha-lib-go/v7/pkg/probe"
 	"github.com/opencord/voltha-protos/v5/go/common"
 	"github.com/opencord/voltha-protos/v5/go/core_service"
 	"github.com/opencord/voltha-protos/v5/go/health"
 	"github.com/opencord/voltha-protos/v5/go/voltha"
 	"github.com/phayes/freeport"
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/grpc"
 )

 const (
 	testGrpcServer  = "test-grpc-server"
 	initialInterval = 100 * time.Millisecond
 	maxInterval     = 5000 * time.Millisecond
 	maxElapsedTime  = 0 * time.Millisecond
 	monitorInterval = 2 * time.Second
 	timeout         = 10 * time.Second
 )

 var testForNoActivityCh = make(chan time.Time, 10)

 type testCoreServer struct {
 	apiEndPoint string
 	server      *GrpcServer
 	probe       *probe.Probe
 }

 func newTestCoreServer(apiEndpoint string) *testCoreServer {
 	return &testCoreServer{
 		apiEndPoint: apiEndpoint,
 		probe:       &probe.Probe{},
 	}
 }

 func (s *testCoreServer) registerService(ctx context.Context, t *testing.T) {
 	assert.NotEqual(t, "", s.apiEndPoint)

 	probePort, err := freeport.GetFreePort()
 	assert.Nil(t, err)
 	probeEndpoint := "127.0.0.1:" + strconv.Itoa(probePort)
 	go s.probe.ListenAndServe(ctx, probeEndpoint)
 	s.probe.RegisterService(ctx, testGrpcServer)

 	s.server = NewGrpcServer(s.apiEndPoint, nil, false, s.probe)

 	s.server.AddService(func(server *grpc.Server) {
 		core_service.RegisterCoreServiceServer(server, &MockCoreServiceHandler{})
 	})
 }

 func (s *testCoreServer) start(ctx context.Context, t *testing.T) {
 	assert.NotNil(t, s.server)
 	assert.NotEqual(t, "", s.apiEndPoint)

 	s.probe.UpdateStatus(ctx, testGrpcServer, probe.ServiceStatusRunning)
 	s.server.Start(ctx)
 	s.probe.UpdateStatus(ctx, testGrpcServer, probe.ServiceStatusStopped)
 }

 func (s *testCoreServer) stop() {
 	if s.server != nil {
 		s.server.Stop()
 	}
 }

 type testClient struct {
 	apiEndPoint string
 	probe       *probe.Probe
 	client      *Client
 }

 func serverRestarted(ctx context.Context, endPoint string) error {
 	logger.Infow(ctx, "remote-restarted", log.Fields{"endpoint": endPoint})
 	return nil
 }

 func newTestClient(apiEndpoint string, handler RestartedHandler) *testClient {
 	tc := &testClient{
 		apiEndPoint: apiEndpoint,
 		probe:       &probe.Probe{},
 	}
 	// Set the environment variables that this client will use
 	var err error
 	err = os.Setenv(grpcBackoffInitialInterval, initialInterval.String())
 	if err != nil {
 		logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcBackoffInitialInterval})
 		return nil
 	}
 	err = os.Setenv(grpcBackoffInitialInterval, maxInterval.String())
 	if err != nil {
 		logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcBackoffInitialInterval})
 		return nil
 	}
 	err = os.Setenv(grpcBackoffMaxElapsedTime, maxElapsedTime.String())
 	if err != nil {
 		logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcBackoffMaxElapsedTime})
 		return nil
 	}

 	err = os.Setenv(grpcMonitorInterval, monitorInterval.String())
 	if err != nil {
 		logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcMonitorInterval})
 		return nil
 	}

 	tc.client, err = NewClient(
 		"test-endpoint",
 		apiEndpoint,
 		handler)
 	if err != nil {
 		return nil
 	}
 	return tc
 }

 func setAndTestCoreServiceHandler(ctx context.Context, conn *grpc.ClientConn, clientConn *common.Connection) interface{} {
 	if conn == nil {
 		return nil
 	}
 	svc := core_service.NewCoreServiceClient(conn)
 	if h, err := svc.GetHealthStatus(ctx, clientConn); err != nil || h.State != health.HealthStatus_HEALTHY {
 		return nil
 	}
 	return svc
 }

 func idleConnectionTest(ctx context.Context, conn *grpc.ClientConn, clientConn *common.Connection) interface{} {
 	if conn == nil {
 		return nil
 	}
 	svc := core_service.NewCoreServiceClient(conn)
 	if h, err := svc.GetHealthStatus(ctx, clientConn); err != nil || h.State != health.HealthStatus_HEALTHY {
 		return nil
 	}
 	testForNoActivityCh <- time.Now()
 	return svc
 }

 func (c *testClient) start(ctx context.Context, t *testing.T, handler SetAndTestServiceHandler) {
 	assert.NotNil(t, c.client)

 	probePort, err := freeport.GetFreePort()
 	assert.Nil(t, err)
 	probeEndpoint := "127.0.0.1:" + strconv.Itoa(probePort)
 	go c.probe.ListenAndServe(ctx, probeEndpoint)

 	probeCtx := context.WithValue(ctx, probe.ProbeContextKey, c.probe)
 	c.client.Start(probeCtx, handler)
 }

 func (c *testClient) getClient(t *testing.T) core_service.CoreServiceClient {
 	gc, err := c.client.GetClient()
 	assert.Nil(t, err)
 	coreClient, ok := gc.(core_service.CoreServiceClient)
 	assert.True(t, ok)
 	return coreClient
 }

 func serverStartsFirstTest(t *testing.T) {
 	// Setup
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

 	// Create and start the test server
 	grpcPort, err := freeport.GetFreePort()
 	assert.Nil(t, err)
 	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
 	ts := newTestCoreServer(apiEndpoint)
 	ts.registerService(ctx, t)
 	go ts.start(ctx, t)

 	// Create the test client and start it
 	tc := newTestClient(apiEndpoint, serverRestarted)
 	assert.NotNil(t, tc)
 	go tc.start(ctx, t, setAndTestCoreServiceHandler)

 	// Test 1: Verify that probe status shows ready eventually
 	var servicesReady isConditionSatisfied = func() bool {
 		return ts.probe.IsReady() && tc.probe.IsReady()
 	}
 	err = waitUntilCondition(timeout, servicesReady)
 	assert.Nil(t, err)

 	// Test 2: Verify we get a valid client and can make grpc requests with it
 	coreClient := tc.getClient(t)
 	assert.NotNil(t, coreClient)

 	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
 	assert.Nil(t, err)
 	assert.NotNil(t, device)
 	assert.Equal(t, "test-1234", device.Type)
 }

 func clientStartsFirstTest(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

 	// Create a grpc endpoint for the server
 	grpcPort, err := freeport.GetFreePort()
 	assert.Nil(t, err)
 	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)

 	// Create the test client and start it
 	tc := newTestClient(apiEndpoint, serverRestarted)
 	assert.NotNil(t, tc)
 	go tc.start(ctx, t, setAndTestCoreServiceHandler)

 	// Verify client is not ready
 	var clientNotReady isConditionSatisfied = func() bool {
 		serviceStatus := tc.probe.GetStatus(apiEndpoint)
 		return serviceStatus == probe.ServiceStatusNotReady ||
 			serviceStatus == probe.ServiceStatusPreparing ||
 			serviceStatus == probe.ServiceStatusFailed
 	}
 	err = waitUntilCondition(timeout, clientNotReady)
 	assert.Nil(t, err)

 	// Create and start the test server
 	ts := newTestCoreServer(apiEndpoint)
 	ts.registerService(ctx, t)
 	go ts.start(ctx, t)

 	// Test 1: Verify that probe status shows ready eventually
 	var servicesReady isConditionSatisfied = func() bool {
 		return ts.probe.IsReady() && tc.probe.IsReady()
 	}
 	err = waitUntilCondition(timeout, servicesReady)
 	assert.Nil(t, err)

 	// Test 2: Verify we get a valid client and can make grpc requests with it
 	coreClient := tc.getClient(t)
 	assert.NotNil(t, coreClient)

 	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
 	assert.Nil(t, err)
 	assert.NotNil(t, device)
 	assert.Equal(t, "test-1234", device.Type)
 }

 // Liveness function
 func livessness(timestamp time.Time) {
 	logger.Debugw(context.Background(), "received-liveness", log.Fields{"timestamp": timestamp})
 }

 func serverRestarts(t *testing.T, numRestartRuns int) {
 	// Setup
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

 	// Create and start the test server
 	grpcPort, err := freeport.GetFreePort()
 	assert.Nil(t, err)
 	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
 	ts := newTestCoreServer(apiEndpoint)
 	ts.registerService(ctx, t)
 	go ts.start(ctx, t)

 	// Create the test client and start it
 	tc := newTestClient(apiEndpoint, serverRestarted)
 	assert.NotNil(t, tc)

 	// Subscribe for liveness
 	tc.client.SubscribeForLiveness(livessness)
 	go tc.start(ctx, t, setAndTestCoreServiceHandler)

 	// Test 1: Verify that probe status shows ready eventually
 	var servicesReady isConditionSatisfied = func() bool {
 		return ts.probe.IsReady() && tc.probe.IsReady()
 	}
 	err = waitUntilCondition(timeout, servicesReady)
 	assert.Nil(t, err)

 	// Test 2: Verify we get a valid client and can make grpc requests with it
 	coreClient := tc.getClient(t)
 	assert.NotNil(t, coreClient)

 	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
 	assert.Nil(t, err)
 	assert.NotNil(t, device)
 	assert.Equal(t, "test-1234", device.Type)

 	for i := 1; i <= numRestartRuns; i++ {
 		//Test 3: Stop server and verify server status
 		ts.stop()
 		var serverDown isConditionSatisfied = func() bool {
 			return ts.probe.GetStatus(testGrpcServer) == probe.ServiceStatusStopped
 		}
 		err = waitUntilCondition(timeout, serverDown)
 		assert.Nil(t, err)

 		// Make a grpc request - this will detect the server being down and automatically trigger the grpc client
 		// to reconnect
 		_, err = coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
 		assert.NotNil(t, err)

 		// Wait until the client service shows as not ready. A wait is not needed.  It's just to verify that the
 		// client changes connection state.
 		var clientNotReady isConditionSatisfied = func() bool {
 			serviceStatus := tc.probe.GetStatus(apiEndpoint)
 			return serviceStatus == probe.ServiceStatusNotReady ||
 				serviceStatus == probe.ServiceStatusPreparing ||
 				serviceStatus == probe.ServiceStatusFailed
 		}
 		err = waitUntilCondition(timeout, clientNotReady)

 		assert.Nil(t, err)

 		// Keep the server down for 1/2 second
 		time.Sleep(500 * time.Millisecond)

 		// Test 4: Restart the server and verify the server is back online
 		go ts.start(ctx, t)
 		err = waitUntilCondition(timeout, servicesReady)
 		assert.Nil(t, err)

 		// Test 5: verify we can pull new device with a new client instance
 		coreClient = tc.getClient(t)
 		assert.NotNil(t, coreClient)
 		device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
 		assert.Nil(t, err)
 		assert.Equal(t, "test-1234", device.Type)
 	}
 	// Stop the server
 	ts.stop()
 }

 func testNoActivity(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

 	// Create a grpc endpoint for the server
 	grpcPort, err := freeport.GetFreePort()
 	assert.Nil(t, err)
 	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)

 	// Create the test client and start it
 	tc := newTestClient(apiEndpoint, serverRestarted)
 	assert.NotNil(t, tc)
 	go tc.start(ctx, t, idleConnectionTest)

 	// Create and start the test server
 	ts := newTestCoreServer(apiEndpoint)
 	ts.registerService(ctx, t)
 	go ts.start(ctx, t)

 	// Test 1: Verify that probe status shows ready eventually
 	var servicesReady isConditionSatisfied = func() bool {
 		return ts.probe.IsReady() && tc.probe.IsReady()
 	}
 	err = waitUntilCondition(timeout, servicesReady)
 	assert.Nil(t, err)

 	// Test 2: Verify we get a valid client and can make grpc requests with it
 	coreClient := tc.getClient(t)
 	assert.NotNil(t, coreClient)

 	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
 	assert.Nil(t, err)
 	assert.NotNil(t, device)
 	assert.Equal(t, "test-1234", device.Type)

 	start := time.Now()
 	numChecks := 3 // Test for 3 checks
 	// Wait on the the idle channel - on no activity a connection probe will be attempted by the client
 	timer := time.NewTimer((monitorInterval + 1*time.Second) * time.Duration(numChecks))
 	defer timer.Stop()
 	count := 0
 loop:
 	for {
 		select {
 		case timestamp := <-testForNoActivityCh:
 			if timestamp.After(start) {
 				count += 1
 				if count > numChecks {
 					break loop
 				}
 			}
 		case <-timer.C:
 			t.Fatal("no activity on the idle channel")
 		}
 	}
 }

 func testClientFailure(t *testing.T, numClientRestarts int) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	// Create a grpc endpoint for the server
 	grpcPort, err := freeport.GetFreePort()
 	assert.Nil(t, err)
 	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
 	// Create the test client and start it
 	tc := newTestClient(apiEndpoint, serverRestarted)
 	assert.NotNil(t, tc)
 	go tc.start(ctx, t, idleConnectionTest)
 	// Create and start the test server
 	ts := newTestCoreServer(apiEndpoint)
 	ts.registerService(ctx, t)
 	go ts.start(ctx, t)
 	defer ts.stop()
 	// Test 1: Verify that probe status shows ready eventually
 	var servicesReady isConditionSatisfied = func() bool {
 		return ts.probe.IsReady() && tc.probe.IsReady()
 	}
 	err = waitUntilCondition(timeout, servicesReady)
 	assert.Nil(t, err)
 	// Test 2: Verify we get a valid client and can make grpc requests with it
 	coreClient := tc.getClient(t)
 	assert.NotNil(t, coreClient)
 	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
 	assert.Nil(t, err)
 	assert.NotNil(t, device)
 	assert.Equal(t, "test-1234", device.Type)
 	for i := 1; i <= numClientRestarts; i++ {
 		// Kill grpc client
 		tc.client.Stop(context.Background())
 		var clientNotReady isConditionSatisfied = func() bool {
 			return !tc.probe.IsReady()
 		}
 		err = waitUntilCondition(timeout, clientNotReady)
 		assert.Nil(t, err)
 		// Create a new client
 		tc.client, err = NewClient(
 			"test-ednpoint",
 			apiEndpoint,
 			serverRestarted)
 		assert.Nil(t, err)
 		probeCtx := context.WithValue(ctx, probe.ProbeContextKey, tc.probe)
 		go tc.client.Start(probeCtx, idleConnectionTest)
 		//Verify that probe status shows ready eventually
 		err = waitUntilCondition(timeout, servicesReady)
 		assert.Nil(t, err)
 		// Verify we get a valid client and can make grpc requests with it
 		coreClient = tc.getClient(t)
 		assert.NotNil(t, coreClient)
 		device, err = coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
 		assert.Nil(t, err)
 		assert.NotNil(t, device)
 		assert.Equal(t, "test-1234", device.Type)
 	}
 	tc.client.Stop(context.Background())
 }

 func testServerLimit(t *testing.T) {
 	t.Skip() // Not needed for regular unit tests

 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

 	// Create a grpc endpoint for the server
 	grpcPort, err := freeport.GetFreePort()
 	assert.Nil(t, err)
 	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)

 	// Create the test client and start it
 	tc := newTestClient(apiEndpoint, serverRestarted)
 	assert.NotNil(t, tc)
 	go tc.start(ctx, t, idleConnectionTest)

 	// Create and start the test server
 	ts := newTestCoreServer(apiEndpoint)
 	ts.registerService(ctx, t)
 	go ts.start(ctx, t)

 	// Test 1: Verify that probe status shows ready eventually
 	var servicesReady isConditionSatisfied = func() bool {
 		return ts.probe.IsReady() && tc.probe.IsReady()
 	}
 	err = waitUntilCondition(timeout, servicesReady)
 	assert.Nil(t, err)

 	// Test 2: Verify we get a valid client and can make grpc requests with it
 	coreClient := tc.getClient(t)
 	assert.NotNil(t, coreClient)

 	var lock sync.RWMutex
 	bad := []time.Duration{}
 	bad_err := []string{}
 	good := []time.Duration{}
 	var wg sync.WaitGroup
 	numRPCs := 10
 	total_good := time.Duration(0)
 	max_good := time.Duration(0)
 	total_bad := time.Duration(0)
 	max_bad := time.Duration(0)
 	order_received := []uint32{}
 	for i := 1; i <= numRPCs; i++ {
 		wg.Add(1)
 		go func(seq int) {
 			local := time.Now()
 			ctx, cancel := context.WithTimeout(context.Background(), 10000*time.Millisecond)
 			defer cancel()
 			var err error
 			var d *voltha.Device
 			d, err = coreClient.GetDevice(ctx, &common.ID{Id: strconv.Itoa(seq)})
 			if err != nil {
 				lock.Lock()
 				bad = append(bad, time.Since(local))
 				bad_err = append(bad_err, err.Error())
 				total_bad += time.Since(local)
 				if time.Since(local) > max_bad {
 					max_bad = time.Since(local)
 				}
 				logger.Errorw(ctx, "error produced", log.Fields{"error": err})
 				lock.Unlock()
 			} else {
 				lock.Lock()
 				good = append(good, time.Since(local))
 				total_good += time.Since(local)
 				if time.Since(local) > max_good {
 					max_good = time.Since(local)
 				}
 				if d != nil {
 					order_received = append(order_received, d.Vlan)
 				}
 				lock.Unlock()
 			}
 			wg.Done()
 		}(i)
 	}
 	wg.Wait()
 	assert.Equal(t, 0, len(bad))
 	assert.Equal(t, numRPCs, len(good))
 	//fmt.Println("Bad:", bad[:10])
 	if len(bad_err) > 0 {
 		fmt.Println("Bad Err Last:", bad_err[len(bad_err)-1:])
 		fmt.Println("Bad Err First:", bad_err[:1])
 	}
 	fmt.Println("Good:", good[len(good)-10:])
 	fmt.Println("Good average time:", total_good.Milliseconds()/int64(numRPCs))
 	fmt.Println("Bad average time:", total_bad.Milliseconds()/int64(numRPCs))
 	fmt.Println("Bad Max:", max_bad)
 	fmt.Println("Good Max:", max_good)
 	//fmt.Println("Order received:", order_received)

 	prev := order_received[0]

 	for i := 1; i < len(order_received); i++ {
 		if order_received[i] < prev {
 			fmt.Println("Prev:", prev, " curr:", order_received[i])
 		}
 		prev = order_received[i]
 	}
 }

 func TestSuiteClient3(t *testing.T) {
 	// Setup
 	log.SetAllLogLevel(volthaTestLogLevel)

 	// Test starting server before client
 	serverStartsFirstTest(t)

 	// Test starting client before server
 	clientStartsFirstTest(t)

 	// Test server restarts
 	serverRestarts(t, 1)

 	//Test that the client test the grpc connection on no activity
 	testNoActivity(t)

 	// Test client queueing with server limit
 	testServerLimit(t)

 	// Test the scenario where a client restarts
 	testClientFailure(t, 10)
 }
	/*
	* Copyright 2021-present Open Networking Foundation

	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at

	* http://www.apache.org/licenses/LICENSE-2.0

	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package grpc

	import (
	"context"
	"fmt"
	"os"
	"strconv"
	"sync"
	"testing"
	"time"

	"github.com/opencord/voltha-lib-go/v7/pkg/log"
	"github.com/opencord/voltha-lib-go/v7/pkg/probe"
	"github.com/opencord/voltha-protos/v5/go/common"
	"github.com/opencord/voltha-protos/v5/go/core_service"
	"github.com/opencord/voltha-protos/v5/go/health"
	"github.com/opencord/voltha-protos/v5/go/voltha"
	"github.com/phayes/freeport"
	"github.com/stretchr/testify/assert"
	"google.golang.org/grpc"
	)

	const (
	testGrpcServer = "test-grpc-server"
	initialInterval = 100 * time.Millisecond
	maxInterval = 5000 * time.Millisecond
	maxElapsedTime = 0 * time.Millisecond
	monitorInterval = 2 * time.Second
	timeout = 10 * time.Second
	)

	var testForNoActivityCh = make(chan time.Time, 10)

	type testCoreServer struct {
	apiEndPoint string
	server *GrpcServer
	probe *probe.Probe
	}

	func newTestCoreServer(apiEndpoint string) *testCoreServer {
	return &testCoreServer{
	apiEndPoint: apiEndpoint,
	probe: &probe.Probe{},
	}
	}

	func (s testCoreServer) registerService(ctx context.Context, t testing.T) {
	assert.NotEqual(t, "", s.apiEndPoint)

	probePort, err := freeport.GetFreePort()
	assert.Nil(t, err)
	probeEndpoint := "127.0.0.1:" + strconv.Itoa(probePort)
	go s.probe.ListenAndServe(ctx, probeEndpoint)
	s.probe.RegisterService(ctx, testGrpcServer)

	s.server = NewGrpcServer(s.apiEndPoint, nil, false, s.probe)

	s.server.AddService(func(server *grpc.Server) {
	core_service.RegisterCoreServiceServer(server, &MockCoreServiceHandler{})
	})
	}

	func (s testCoreServer) start(ctx context.Context, t testing.T) {
	assert.NotNil(t, s.server)
	assert.NotEqual(t, "", s.apiEndPoint)

	s.probe.UpdateStatus(ctx, testGrpcServer, probe.ServiceStatusRunning)
	s.server.Start(ctx)
	s.probe.UpdateStatus(ctx, testGrpcServer, probe.ServiceStatusStopped)
	}

	func (s *testCoreServer) stop() {
	if s.server != nil {
	s.server.Stop()
	}
	}

	type testClient struct {
	apiEndPoint string
	probe *probe.Probe
	client *Client
	}

	func serverRestarted(ctx context.Context, endPoint string) error {
	logger.Infow(ctx, "remote-restarted", log.Fields{"endpoint": endPoint})
	return nil
	}

	func newTestClient(apiEndpoint string, handler RestartedHandler) *testClient {
	tc := &testClient{
	apiEndPoint: apiEndpoint,
	probe: &probe.Probe{},
	}
	// Set the environment variables that this client will use
	var err error
	err = os.Setenv(grpcBackoffInitialInterval, initialInterval.String())
	if err != nil {
	logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcBackoffInitialInterval})
	return nil
	}
	err = os.Setenv(grpcBackoffInitialInterval, maxInterval.String())
	if err != nil {
	logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcBackoffInitialInterval})
	return nil
	}
	err = os.Setenv(grpcBackoffMaxElapsedTime, maxElapsedTime.String())
	if err != nil {
	logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcBackoffMaxElapsedTime})
	return nil
	}

	err = os.Setenv(grpcMonitorInterval, monitorInterval.String())
	if err != nil {
	logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcMonitorInterval})
	return nil
	}

	tc.client, err = NewClient(
	"test-endpoint",
	apiEndpoint,
	handler)
	if err != nil {
	return nil
	}
	return tc
	}

	func setAndTestCoreServiceHandler(ctx context.Context, conn grpc.ClientConn, clientConn common.Connection) interface{} {
	if conn == nil {
	return nil
	}
	svc := core_service.NewCoreServiceClient(conn)
	if h, err := svc.GetHealthStatus(ctx, clientConn); err != nil \|\| h.State != health.HealthStatus_HEALTHY {
	return nil
	}
	return svc
	}

	func idleConnectionTest(ctx context.Context, conn grpc.ClientConn, clientConn common.Connection) interface{} {
	if conn == nil {
	return nil
	}
	svc := core_service.NewCoreServiceClient(conn)
	if h, err := svc.GetHealthStatus(ctx, clientConn); err != nil \|\| h.State != health.HealthStatus_HEALTHY {
	return nil
	}
	testForNoActivityCh <- time.Now()
	return svc
	}

	func (c testClient) start(ctx context.Context, t testing.T, handler SetAndTestServiceHandler) {
	assert.NotNil(t, c.client)

	probePort, err := freeport.GetFreePort()
	assert.Nil(t, err)
	probeEndpoint := "127.0.0.1:" + strconv.Itoa(probePort)
	go c.probe.ListenAndServe(ctx, probeEndpoint)

	probeCtx := context.WithValue(ctx, probe.ProbeContextKey, c.probe)
	c.client.Start(probeCtx, handler)
	}

	func (c testClient) getClient(t testing.T) core_service.CoreServiceClient {
	gc, err := c.client.GetClient()
	assert.Nil(t, err)
	coreClient, ok := gc.(core_service.CoreServiceClient)
	assert.True(t, ok)
	return coreClient
	}

	func serverStartsFirstTest(t *testing.T) {
	// Setup
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Create and start the test server
	grpcPort, err := freeport.GetFreePort()
	assert.Nil(t, err)
	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
	ts := newTestCoreServer(apiEndpoint)
	ts.registerService(ctx, t)
	go ts.start(ctx, t)

	// Create the test client and start it
	tc := newTestClient(apiEndpoint, serverRestarted)
	assert.NotNil(t, tc)
	go tc.start(ctx, t, setAndTestCoreServiceHandler)

	// Test 1: Verify that probe status shows ready eventually
	var servicesReady isConditionSatisfied = func() bool {
	return ts.probe.IsReady() && tc.probe.IsReady()
	}
	err = waitUntilCondition(timeout, servicesReady)
	assert.Nil(t, err)

	// Test 2: Verify we get a valid client and can make grpc requests with it
	coreClient := tc.getClient(t)
	assert.NotNil(t, coreClient)

	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
	assert.Nil(t, err)
	assert.NotNil(t, device)
	assert.Equal(t, "test-1234", device.Type)
	}

	func clientStartsFirstTest(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Create a grpc endpoint for the server
	grpcPort, err := freeport.GetFreePort()
	assert.Nil(t, err)
	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)

	// Create the test client and start it
	tc := newTestClient(apiEndpoint, serverRestarted)
	assert.NotNil(t, tc)
	go tc.start(ctx, t, setAndTestCoreServiceHandler)

	// Verify client is not ready
	var clientNotReady isConditionSatisfied = func() bool {
	serviceStatus := tc.probe.GetStatus(apiEndpoint)
	return serviceStatus == probe.ServiceStatusNotReady \|\|
	serviceStatus == probe.ServiceStatusPreparing \|\|
	serviceStatus == probe.ServiceStatusFailed
	}
	err = waitUntilCondition(timeout, clientNotReady)
	assert.Nil(t, err)

	// Create and start the test server
	ts := newTestCoreServer(apiEndpoint)
	ts.registerService(ctx, t)
	go ts.start(ctx, t)

	// Test 1: Verify that probe status shows ready eventually
	var servicesReady isConditionSatisfied = func() bool {
	return ts.probe.IsReady() && tc.probe.IsReady()
	}
	err = waitUntilCondition(timeout, servicesReady)
	assert.Nil(t, err)

	// Test 2: Verify we get a valid client and can make grpc requests with it
	coreClient := tc.getClient(t)
	assert.NotNil(t, coreClient)

	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
	assert.Nil(t, err)
	assert.NotNil(t, device)
	assert.Equal(t, "test-1234", device.Type)
	}

	// Liveness function
	func livessness(timestamp time.Time) {
	logger.Debugw(context.Background(), "received-liveness", log.Fields{"timestamp": timestamp})
	}

	func serverRestarts(t *testing.T, numRestartRuns int) {
	// Setup
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Create and start the test server
	grpcPort, err := freeport.GetFreePort()
	assert.Nil(t, err)
	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
	ts := newTestCoreServer(apiEndpoint)
	ts.registerService(ctx, t)
	go ts.start(ctx, t)

	// Create the test client and start it
	tc := newTestClient(apiEndpoint, serverRestarted)
	assert.NotNil(t, tc)

	// Subscribe for liveness
	tc.client.SubscribeForLiveness(livessness)
	go tc.start(ctx, t, setAndTestCoreServiceHandler)

	// Test 1: Verify that probe status shows ready eventually
	var servicesReady isConditionSatisfied = func() bool {
	return ts.probe.IsReady() && tc.probe.IsReady()
	}
	err = waitUntilCondition(timeout, servicesReady)
	assert.Nil(t, err)

	// Test 2: Verify we get a valid client and can make grpc requests with it
	coreClient := tc.getClient(t)
	assert.NotNil(t, coreClient)

	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
	assert.Nil(t, err)
	assert.NotNil(t, device)
	assert.Equal(t, "test-1234", device.Type)

	for i := 1; i <= numRestartRuns; i++ {
	//Test 3: Stop server and verify server status
	ts.stop()
	var serverDown isConditionSatisfied = func() bool {
	return ts.probe.GetStatus(testGrpcServer) == probe.ServiceStatusStopped
	}
	err = waitUntilCondition(timeout, serverDown)
	assert.Nil(t, err)

	// Make a grpc request - this will detect the server being down and automatically trigger the grpc client
	// to reconnect
	_, err = coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
	assert.NotNil(t, err)

	// Wait until the client service shows as not ready. A wait is not needed. It's just to verify that the
	// client changes connection state.
	var clientNotReady isConditionSatisfied = func() bool {
	serviceStatus := tc.probe.GetStatus(apiEndpoint)
	return serviceStatus == probe.ServiceStatusNotReady \|\|
	serviceStatus == probe.ServiceStatusPreparing \|\|
	serviceStatus == probe.ServiceStatusFailed
	}
	err = waitUntilCondition(timeout, clientNotReady)

	assert.Nil(t, err)

	// Keep the server down for 1/2 second
	time.Sleep(500 * time.Millisecond)

	// Test 4: Restart the server and verify the server is back online
	go ts.start(ctx, t)
	err = waitUntilCondition(timeout, servicesReady)
	assert.Nil(t, err)

	// Test 5: verify we can pull new device with a new client instance
	coreClient = tc.getClient(t)
	assert.NotNil(t, coreClient)
	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
	assert.Nil(t, err)
	assert.Equal(t, "test-1234", device.Type)
	}
	// Stop the server
	ts.stop()
	}

	func testNoActivity(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Create a grpc endpoint for the server
	grpcPort, err := freeport.GetFreePort()
	assert.Nil(t, err)
	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)

	// Create the test client and start it
	tc := newTestClient(apiEndpoint, serverRestarted)
	assert.NotNil(t, tc)
	go tc.start(ctx, t, idleConnectionTest)

	// Create and start the test server
	ts := newTestCoreServer(apiEndpoint)
	ts.registerService(ctx, t)
	go ts.start(ctx, t)

	// Test 1: Verify that probe status shows ready eventually
	var servicesReady isConditionSatisfied = func() bool {
	return ts.probe.IsReady() && tc.probe.IsReady()
	}
	err = waitUntilCondition(timeout, servicesReady)
	assert.Nil(t, err)

	// Test 2: Verify we get a valid client and can make grpc requests with it
	coreClient := tc.getClient(t)
	assert.NotNil(t, coreClient)

	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
	assert.Nil(t, err)
	assert.NotNil(t, device)
	assert.Equal(t, "test-1234", device.Type)

	start := time.Now()
	numChecks := 3 // Test for 3 checks
	// Wait on the the idle channel - on no activity a connection probe will be attempted by the client
	timer := time.NewTimer((monitorInterval + 1time.Second) time.Duration(numChecks))
	defer timer.Stop()
	count := 0
	loop:
	for {
	select {
	case timestamp := <-testForNoActivityCh:
	if timestamp.After(start) {
	count += 1
	if count > numChecks {
	break loop
	}
	}
	case <-timer.C:
	t.Fatal("no activity on the idle channel")
	}
	}
	}

	func testClientFailure(t *testing.T, numClientRestarts int) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	// Create a grpc endpoint for the server
	grpcPort, err := freeport.GetFreePort()
	assert.Nil(t, err)
	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
	// Create the test client and start it
	tc := newTestClient(apiEndpoint, serverRestarted)
	assert.NotNil(t, tc)
	go tc.start(ctx, t, idleConnectionTest)
	// Create and start the test server
	ts := newTestCoreServer(apiEndpoint)
	ts.registerService(ctx, t)
	go ts.start(ctx, t)
	defer ts.stop()
	// Test 1: Verify that probe status shows ready eventually
	var servicesReady isConditionSatisfied = func() bool {
	return ts.probe.IsReady() && tc.probe.IsReady()
	}
	err = waitUntilCondition(timeout, servicesReady)
	assert.Nil(t, err)
	// Test 2: Verify we get a valid client and can make grpc requests with it
	coreClient := tc.getClient(t)
	assert.NotNil(t, coreClient)
	device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
	assert.Nil(t, err)
	assert.NotNil(t, device)
	assert.Equal(t, "test-1234", device.Type)
	for i := 1; i <= numClientRestarts; i++ {
	// Kill grpc client
	tc.client.Stop(context.Background())
	var clientNotReady isConditionSatisfied = func() bool {
	return !tc.probe.IsReady()
	}
	err = waitUntilCondition(timeout, clientNotReady)
	assert.Nil(t, err)
	// Create a new client
	tc.client, err = NewClient(
	"test-ednpoint",
	apiEndpoint,
	serverRestarted)
	assert.Nil(t, err)
	probeCtx := context.WithValue(ctx, probe.ProbeContextKey, tc.probe)
	go tc.client.Start(probeCtx, idleConnectionTest)
	//Verify that probe status shows ready eventually
	err = waitUntilCondition(timeout, servicesReady)
	assert.Nil(t, err)
	// Verify we get a valid client and can make grpc requests with it
	coreClient = tc.getClient(t)
	assert.NotNil(t, coreClient)
	device, err = coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
	assert.Nil(t, err)
	assert.NotNil(t, device)
	assert.Equal(t, "test-1234", device.Type)
	}
	tc.client.Stop(context.Background())
	}

	func testServerLimit(t *testing.T) {
	t.Skip() // Not needed for regular unit tests

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Create a grpc endpoint for the server
	grpcPort, err := freeport.GetFreePort()
	assert.Nil(t, err)
	apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)

	// Create the test client and start it
	tc := newTestClient(apiEndpoint, serverRestarted)
	assert.NotNil(t, tc)
	go tc.start(ctx, t, idleConnectionTest)

	// Create and start the test server
	ts := newTestCoreServer(apiEndpoint)
	ts.registerService(ctx, t)
	go ts.start(ctx, t)

	// Test 1: Verify that probe status shows ready eventually
	var servicesReady isConditionSatisfied = func() bool {
	return ts.probe.IsReady() && tc.probe.IsReady()
	}
	err = waitUntilCondition(timeout, servicesReady)
	assert.Nil(t, err)

	// Test 2: Verify we get a valid client and can make grpc requests with it
	coreClient := tc.getClient(t)
	assert.NotNil(t, coreClient)

	var lock sync.RWMutex
	bad := []time.Duration{}
	bad_err := []string{}
	good := []time.Duration{}
	var wg sync.WaitGroup
	numRPCs := 10
	total_good := time.Duration(0)
	max_good := time.Duration(0)
	total_bad := time.Duration(0)
	max_bad := time.Duration(0)
	order_received := []uint32{}
	for i := 1; i <= numRPCs; i++ {
	wg.Add(1)
	go func(seq int) {
	local := time.Now()
	ctx, cancel := context.WithTimeout(context.Background(), 10000*time.Millisecond)
	defer cancel()
	var err error
	var d *voltha.Device
	d, err = coreClient.GetDevice(ctx, &common.ID{Id: strconv.Itoa(seq)})
	if err != nil {
	lock.Lock()
	bad = append(bad, time.Since(local))
	bad_err = append(bad_err, err.Error())
	total_bad += time.Since(local)
	if time.Since(local) > max_bad {
	max_bad = time.Since(local)
	}
	logger.Errorw(ctx, "error produced", log.Fields{"error": err})
	lock.Unlock()
	} else {
	lock.Lock()
	good = append(good, time.Since(local))
	total_good += time.Since(local)
	if time.Since(local) > max_good {
	max_good = time.Since(local)
	}
	if d != nil {
	order_received = append(order_received, d.Vlan)
	}
	lock.Unlock()
	}
	wg.Done()
	}(i)
	}
	wg.Wait()
	assert.Equal(t, 0, len(bad))
	assert.Equal(t, numRPCs, len(good))
	//fmt.Println("Bad:", bad[:10])
	if len(bad_err) > 0 {
	fmt.Println("Bad Err Last:", bad_err[len(bad_err)-1:])
	fmt.Println("Bad Err First:", bad_err[:1])
	}
	fmt.Println("Good:", good[len(good)-10:])
	fmt.Println("Good average time:", total_good.Milliseconds()/int64(numRPCs))
	fmt.Println("Bad average time:", total_bad.Milliseconds()/int64(numRPCs))
	fmt.Println("Bad Max:", max_bad)
	fmt.Println("Good Max:", max_good)
	//fmt.Println("Order received:", order_received)

	prev := order_received[0]

	for i := 1; i < len(order_received); i++ {
	if order_received[i] < prev {
	fmt.Println("Prev:", prev, " curr:", order_received[i])
	}
	prev = order_received[i]
	}
	}

	func TestSuiteClient3(t *testing.T) {
	// Setup
	log.SetAllLogLevel(volthaTestLogLevel)

	// Test starting server before client
	serverStartsFirstTest(t)

	// Test starting client before server
	clientStartsFirstTest(t)

	// Test server restarts
	serverRestarts(t, 1)

	//Test that the client test the grpc connection on no activity
	testNoActivity(t)

	// Test client queueing with server limit
	testServerLimit(t)

	// Test the scenario where a client restarts
	testClientFailure(t, 10)
	}