diff --git a/test/integration_test/common_test.go b/test/integration_test/common_test.go
index a4694bfb44..8d0af30df1 100644
--- a/test/integration_test/common_test.go
+++ b/test/integration_test/common_test.go
@@ -20,8 +20,6 @@ import (
 	snapv1 "github.com/kubernetes-incubator/external-storage/snapshot/pkg/apis/crd/v1"
 	oputils "github.com/libopenstorage/operator/drivers/storage/portworx/util"
 	opcorev1 "github.com/libopenstorage/operator/pkg/apis/core/v1"
-	"github.com/libopenstorage/stork/pkg/aetosutils"
-	aetosLogger "github.com/libopenstorage/stork/pkg/log"
 	"github.com/portworx/sched-ops/k8s/apps"
 	"github.com/portworx/sched-ops/k8s/batch"
 	"github.com/portworx/sched-ops/k8s/core"
@@ -68,7 +66,9 @@ import (
 	_ "github.com/libopenstorage/stork/drivers/volume/gcp"
 	_ "github.com/libopenstorage/stork/drivers/volume/linstor"
 	_ "github.com/libopenstorage/stork/drivers/volume/portworx"
+	"github.com/libopenstorage/stork/pkg/aetosutils"
 	storkv1 "github.com/libopenstorage/stork/pkg/apis/stork/v1alpha1"
+	aetosLogger "github.com/libopenstorage/stork/pkg/log"
 	"github.com/libopenstorage/stork/pkg/schedule"
 	"github.com/libopenstorage/stork/pkg/storkctl"
 	"github.com/libopenstorage/stork/pkg/version"
@@ -834,7 +834,7 @@ func scheduleClusterPair(ctx *scheduler.Context, skipStorage, resetConfig bool,
 }
 
 // Create a cluster pair from source to destination and another cluster pair from destination to source
-func scheduleBidirectionalClusterPair(cpName, cpNamespace, projectMappings string, objectStoreType storkv1.BackupLocationType, secretName string) error {
+func scheduleBidirectionalClusterPair(cpName, cpNamespace, projectMappings string, objectStoreType storkv1.BackupLocationType, secretName string, args map[string]string) error {
 	//var token string
 	// Setting kubeconfig to source because we will create bidirectional cluster pair based on source as reference
 	err := setSourceKubeConfig()
@@ -937,13 +937,23 @@ func scheduleBidirectionalClusterPair(cpName, cpNamespace, projectMappings strin
 		cmdArgs = append(cmdArgs, projectMappings)
 	}
 
-	// Get external object store details and append to the command accordingily
-	objectStoreArgs, err := getObjectStoreArgs(objectStoreType, secretName)
-	if err != nil {
-		return fmt.Errorf("failed to get  %s secret in configmap secret-config in default namespace", objectStoreType)
+	if objectStoreType != "" {
+		// Get external object store details and append to the command accordingily
+		objectStoreArgs, err := getObjectStoreArgs(objectStoreType, secretName)
+		if err != nil {
+			return fmt.Errorf("failed to get  %s secret in configmap secret-config in default namespace", objectStoreType)
+		}
+		cmdArgs = append(cmdArgs, objectStoreArgs...)
+	}
+
+	if args != nil {
+		// Get external object store details and append to the command accordingily
+		for k, v := range args {
+			cmdArgs = append(cmdArgs, fmt.Sprintf("--%v", k))
+			cmdArgs = append(cmdArgs, fmt.Sprintf("%v", v))
+		}
 	}
 
-	cmdArgs = append(cmdArgs, objectStoreArgs...)
 	cmd.SetArgs(cmdArgs)
 	logrus.Infof("Following is the bidirectional command: %v", cmdArgs)
 	if err := cmd.Execute(); err != nil {
diff --git a/test/integration_test/migration_features_test.go b/test/integration_test/migration_features_test.go
index ab031921c1..f0fdcb1924 100644
--- a/test/integration_test/migration_features_test.go
+++ b/test/integration_test/migration_features_test.go
@@ -9,8 +9,6 @@ import (
 	"testing"
 	"time"
 
-	storkapi "github.com/libopenstorage/stork/pkg/apis/stork/v1alpha1"
-	"github.com/libopenstorage/stork/pkg/resourcecollector"
 	"github.com/portworx/sched-ops/k8s/core"
 	storkops "github.com/portworx/sched-ops/k8s/stork"
 	"github.com/portworx/torpedo/pkg/asyncdr"
@@ -18,6 +16,9 @@ import (
 	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/clientcmd"
+
+	storkapi "github.com/libopenstorage/stork/pkg/apis/stork/v1alpha1"
+	"github.com/libopenstorage/stork/pkg/resourcecollector"
 )
 
 var (
@@ -92,7 +93,7 @@ func migrationStashStrategy(t *testing.T, appName string, appPath string) {
 	err = setSourceKubeConfig()
 	require.NoError(t, err, "Error setting source kubeconfig")
 
-	err = scheduleBidirectionalClusterPair(clusterPairName, appData.Ns, "", storkapi.BackupLocationType(backupLocation), backupSecret)
+	err = scheduleBidirectionalClusterPair(clusterPairName, appData.Ns, "", storkapi.BackupLocationType(backupLocation), backupSecret, nil)
 	require.NoError(t, err, "Error creating cluster pair")
 
 	err = setSourceKubeConfig()
@@ -294,7 +295,7 @@ func testMigrationStashStrategyWithStartApplication(t *testing.T) {
 	err = setSourceKubeConfig()
 	require.NoError(t, err, "Error setting source kubeconfig")
 
-	err = scheduleBidirectionalClusterPair(clusterPairName, appData.Ns, "", storkapi.BackupLocationType(backupLocation), backupSecret)
+	err = scheduleBidirectionalClusterPair(clusterPairName, appData.Ns, "", storkapi.BackupLocationType(backupLocation), backupSecret, nil)
 	require.NoError(t, err, "Error creating cluster pair")
 
 	// set stashstrategy
@@ -393,7 +394,7 @@ func testMultipleTimesMigrationsWithStashStrategy(t *testing.T) {
 	err = setSourceKubeConfig()
 	require.NoError(t, err, "Error setting source kubeconfig")
 
-	err = scheduleBidirectionalClusterPair(clusterPairName, appData.Ns, "", storkapi.BackupLocationType(backupLocation), backupSecret)
+	err = scheduleBidirectionalClusterPair(clusterPairName, appData.Ns, "", storkapi.BackupLocationType(backupLocation), backupSecret, nil)
 	require.NoError(t, err, "Error creating cluster pair")
 
 	// set stashstrategy
@@ -535,7 +536,7 @@ func testFailbackWithStashStrategy(t *testing.T) {
 
 	// creating migration and clusterpair in kube-system namespace as we have to delete the ns in source for failback
 	migrationNamespace := "kube-system"
-	err = scheduleBidirectionalClusterPair(clusterPairName, migrationNamespace, "", storkapi.BackupLocationType(backupLocation), backupSecret)
+	err = scheduleBidirectionalClusterPair(clusterPairName, migrationNamespace, "", storkapi.BackupLocationType(backupLocation), backupSecret, nil)
 	require.NoError(t, err, "Error creating cluster pair")
 
 	// set stashstrategy
diff --git a/test/integration_test/migration_test.go b/test/integration_test/migration_test.go
index d68eae20ca..e41a75ac14 100644
--- a/test/integration_test/migration_test.go
+++ b/test/integration_test/migration_test.go
@@ -1509,7 +1509,7 @@ func bidirectionalClusterPairTest(t *testing.T) {
 	// Scheduler cluster pairs: source cluster --> destination cluster and destination cluster --> source cluster
 	for location, secret := range cmData {
 		logrus.Infof("Creating a bidirectional-pair using %s as objectstore.", location)
-		err := scheduleBidirectionalClusterPair(clusterPairName, clusterPairNamespace, "", storkv1.BackupLocationType(location), secret)
+		err := scheduleBidirectionalClusterPair(clusterPairName, clusterPairNamespace, "", storkv1.BackupLocationType(location), secret, nil)
 		require.NoError(t, err, "failed to set bidirectional cluster pair: %v", err)
 
 		err = setSourceKubeConfig()
@@ -2692,7 +2692,7 @@ func scheduleClusterPairGeneric(t *testing.T, ctxs []*scheduler.Context,
 		logrus.Infof("Namespace: %s", clusterPairNamespace)
 		logrus.Infof("Backuplocation: %s", defaultBackupLocation)
 		logrus.Infof("Secret name: %s", defaultSecretName)
-		err = scheduleBidirectionalClusterPair(remotePairName, clusterPairNamespace, projectIDMappings, defaultBackupLocation, defaultSecretName)
+		err = scheduleBidirectionalClusterPair(remotePairName, clusterPairNamespace, projectIDMappings, defaultBackupLocation, defaultSecretName, nil)
 		require.NoError(t, err, "failed to set bidirectional cluster pair: %v", err)
 		err = setSourceKubeConfig()
 		require.NoError(t, err, "failed to set kubeconfig to source cluster: %v", err)
diff --git a/test/integration_test/operator_test.go b/test/integration_test/operator_test.go
index f35e027d30..94bc2fa420 100644
--- a/test/integration_test/operator_test.go
+++ b/test/integration_test/operator_test.go
@@ -9,12 +9,13 @@ import (
 	"testing"
 	"time"
 
-	storkv1 "github.com/libopenstorage/stork/pkg/apis/stork/v1alpha1"
 	"github.com/portworx/sched-ops/k8s/core"
 	"github.com/portworx/torpedo/pkg/asyncdr"
 	"github.com/portworx/torpedo/pkg/log"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/require"
+
+	storkv1 "github.com/libopenstorage/stork/pkg/apis/stork/v1alpha1"
 )
 
 var (
@@ -83,7 +84,7 @@ func validateAndDestroyCrMigration(t *testing.T, appName string, appPath string)
 	err = asyncdr.ValidateCRD(appData.ExpectedCrdList, sourceClusterConfigPath)
 	require.NoError(t, err, "Error validating source crds")
 
-	err = scheduleBidirectionalClusterPair(clusterPairName, appData.Ns, "", storkv1.BackupLocationType(backupLocation), backupSecret)
+	err = scheduleBidirectionalClusterPair(clusterPairName, appData.Ns, "", storkv1.BackupLocationType(backupLocation), backupSecret, nil)
 	require.NoError(t, err, "Error creating cluster pair")
 
 	logrus.Infof("Migration Started")
diff --git a/test/integration_test/storkctl_clusterpair_test.go b/test/integration_test/storkctl_clusterpair_test.go
new file mode 100644
index 0000000000..666898b991
--- /dev/null
+++ b/test/integration_test/storkctl_clusterpair_test.go
@@ -0,0 +1,279 @@
+//go:build integrationtest
+// +build integrationtest
+
+package integrationtest
+
+import (
+	"fmt"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/portworx/sched-ops/k8s/core"
+	storkops "github.com/portworx/sched-ops/k8s/stork"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+
+	storkv1 "github.com/libopenstorage/stork/pkg/apis/stork/v1alpha1"
+)
+
+const (
+	clusterPairTestDir = "clusterpair-test"
+	cpNamespace        = "automation-clusterpair-ns"
+	defaultOptionPort  = "9001"
+	disableSSL         = "disableSSL"
+	ip                 = "ip"
+	port               = "port"
+	token              = "token"
+	CPbackuplocation   = "backuplocation"
+)
+
+func TestStorkCtlClusterPair(t *testing.T) {
+	err := setSourceKubeConfig()
+	require.NoError(t, err, "Failed to set kubeconfig to source cluster: %v", err)
+	currentTestSuite = t.Name()
+	t.Run("createClusterPairWithEncKeydisableSSLTest", createClusterPairWithEncKeydisableSSLTest)
+	t.Run("createClusterPairWithoutEncKeydisableSSLTest", createClusterPairWithoutEncKeydisableSSLTest)
+	t.Run("createClusterPairWithEpTest", createClusterPairWithEpTest)
+	t.Run("createClusterPairSyncDrTest", createClusterPairSyncDrTest)
+}
+
+func createClusterPairWithEncKeydisableSSLTest(t *testing.T) {
+	var testrailID, testResult = 297068, testResultFail
+	runID := testrailSetupForTest(testrailID, &testResult)
+	defer updateTestRail(&testResult, testrailID, runID)
+	defer updateDashStats(t.Name(), &testResult)
+
+	expectedBlData := map[string]interface{}{
+		disableSSL: true,
+		"encryptionKey": "testKey",
+	}
+    cpName := validateBlData(t, expectedBlData, nil)
+	logrus.Infof("CP is %v", cpName)
+	expectedCPOptionData := map[string]string{
+		"backuplocation": cpName,
+		"port": defaultOptionPort,
+		"ip": "",
+		"token": "",
+	}
+	validateClusterPairOptions(t, cpName, expectedCPOptionData, false)
+	cleanupNamespaces()
+	testResult = testResultPass
+	logrus.Infof("Test status at end of %s test: %s", t.Name(), testResult)
+}
+
+func createClusterPairWithoutEncKeydisableSSLTest(t *testing.T) {
+	var testrailID, testResult = 297069, testResultFail
+	runID := testrailSetupForTest(testrailID, &testResult)
+	defer updateTestRail(&testResult, testrailID, runID)
+	defer updateDashStats(t.Name(), &testResult)
+
+	expectedBlData := map[string]interface{}{
+		disableSSL: false,
+		"encryptionKey": "",
+	}
+    cpName := validateBlData(t, expectedBlData, nil)
+	logrus.Infof("CP is %v", cpName)
+	expectedCPOptionData := map[string]string{
+		"backuplocation": cpName,
+		"port": defaultOptionPort,
+		"ip": "",
+		"token": "",
+	}
+	validateClusterPairOptions(t, cpName, expectedCPOptionData, false)
+	cleanupNamespaces()
+	testResult = testResultPass
+	logrus.Infof("Test status at end of %s test: %s", t.Name(), testResult)
+}
+
+func createClusterPairWithEpTest(t *testing.T) {
+	var testrailID, testResult = 297070, testResultFail
+	runID := testrailSetupForTest(testrailID, &testResult)
+	defer updateTestRail(&testResult, testrailID, runID)
+	defer updateDashStats(t.Name(), &testResult)
+
+    srcIps := getNodeIps(t, true)
+	err := setDestinationKubeConfig()
+	require.NoError(t, err, "failed to set destination kubeconfig: %v", err)
+	destIps := getNodeIps(t, false)
+	extraAgrs := map[string]string{
+		"src-ep": fmt.Sprintf("%s:%s", srcIps[0], defaultOptionPort),
+		"dest-ep": fmt.Sprintf("%s:%s", destIps[0], defaultOptionPort),
+	}
+	expectedBlData := map[string]interface{}{
+		disableSSL: true,
+		"encryptionKey": "testKey",
+	}
+	err = setSourceKubeConfig()
+	require.NoError(t, err, "failed to set source kubeconfig: %v", err)
+    cpName := validateBlData(t, expectedBlData, extraAgrs)
+	logrus.Infof("CP is %v", cpName)
+	expectedCPOptionDataSrc := map[string]string{
+		"backuplocation": cpName,
+		"port": defaultOptionPort,
+		"ip": destIps[0],
+		"token": "",
+	}
+	validateClusterPairOptions(t, cpName, expectedCPOptionDataSrc, false)
+	err = setDestinationKubeConfig()
+	require.NoError(t, err, "failed to set kubeconfig: %v", err)
+	expectedCPOptionDataDest := map[string]string{
+		"backuplocation": cpName,
+		"port": defaultOptionPort,
+		"ip": srcIps[0],
+		"token": "",
+	}
+	validateClusterPairOptions(t, cpName, expectedCPOptionDataDest, true)
+	cleanupNamespaces()
+	testResult = testResultPass
+	logrus.Infof("Test status at end of %s test: %s", t.Name(), testResult)
+}
+
+func createClusterPairSyncDrTest(t *testing.T) {
+	var testrailID, testResult = 297071, testResultFail
+	runID := testrailSetupForTest(testrailID, &testResult)
+	defer updateTestRail(&testResult, testrailID, runID)
+	defer updateDashStats(t.Name(), &testResult)
+
+	extraAgrs := map[string]string{
+		"mode": "sync-dr",
+	}
+    cpName := validateBlData(t, nil, extraAgrs)
+	logrus.Infof("CP is %v", cpName)
+	validateClusterPairOptions(t, cpName, nil, false)
+	cleanupNamespaces()
+	testResult = testResultPass
+	logrus.Infof("Test status at end of %s test: %s", t.Name(), testResult)
+}
+
+func updateSecretData(t *testing.T, secretName string, requiredMap map[string]interface{}) {
+	secret, err := core.Instance().GetSecret(secretName, "default")
+	require.NoError(t, err, "Error getting secret data")
+	for key, newValue := range requiredMap {
+		var newValueByte []byte
+		switch val := newValue.(type) {
+		case string:
+			newValueByte = []byte(val)
+		case bool:
+			newValueByte = []byte(strconv.FormatBool(val))
+		default:
+			fmt.Printf("Key: %s, Value: %v\n", key, val)
+		}
+		secret.Data[key] = newValueByte
+	}
+	_, err = core.Instance().UpdateSecretData(secretName, "default", secret.Data)
+}
+
+func validateBlData(t *testing.T, expectedData map[string]interface{}, extraAgrs map[string]string) string {
+	cpName := fmt.Sprintf("automation-cluster-pair-%v", time.Now().Format("2006-01-02-150405"))
+	if expectedData == nil {
+		scheduleBidirectionalClusterPair(cpName, cpNamespace, "", "", "", extraAgrs)
+		err := setSourceKubeConfig()
+		require.NoError(t, err, "failed to set source kubeconfig: %v", err)
+		_, err = storkops.Instance().GetBackupLocation(cpName, cpNamespace)
+		require.Error(t, err, "Backuplocation should not be created")
+	} else {
+		configMap, err := core.Instance().GetConfigMap("secret-config", "default")
+		if err != nil {
+			require.NoError(t, err, "Error getting configmap secret-config in defaule namespace: %v")
+		}
+		cmData := configMap.Data
+		for location, secret := range cmData {
+			logrus.Infof("Creating a cluster-pair using %v as objectstore.", location)
+			updateSecretData(t, secret, expectedData)
+			scheduleBidirectionalClusterPair(cpName, cpNamespace, "", storkv1.BackupLocationType(location), secret, extraAgrs)
+			err := setSourceKubeConfig()
+		    require.NoError(t, err, "failed to set source kubeconfig: %v", err)
+			bl, err := storkops.Instance().GetBackupLocation(cpName, cpNamespace)
+			require.NoError(t, err, "Not able to get backup Location for %v clusterpair", cpName)
+			for key, value := range expectedData {
+				switch key {
+				case "encryptionKey":
+					logrus.Infof("EncryptionKey is %v", bl.Location.EncryptionV2Key)
+					require.Equal(t, value, bl.Location.EncryptionV2Key)
+				case disableSSL:
+					logrus.Infof("SSL is %v", bl.Location.S3Config.DisableSSL)
+					require.Equal(t, value, bl.Location.S3Config.DisableSSL)
+				}
+			}		
+		}
+    }
+	return cpName
+}
+
+func validateClusterPairOptions(t *testing.T, name string, expectedData map[string]string, reverse bool) {
+	cp, err := storkops.Instance().GetClusterPair(name, cpNamespace)
+	cpOptions := cp.Spec.Options
+	require.NoError(t, err, "failed to get clusterpair: %v", err)
+	if expectedData == nil {
+		require.Equal(t, 0, len(cpOptions))
+	} else {
+		for key, value := range expectedData {
+			switch key {
+			case CPbackuplocation:
+				logrus.Infof("backuplocation is %v", cpOptions[CPbackuplocation])
+				require.Equal(t, value, cpOptions[CPbackuplocation])
+			case port:
+				logrus.Infof("port is %v", cpOptions[port])
+				require.Equal(t, value, cpOptions[port])
+			case token:
+				_, exists := expectedData[key]
+				require.Equal(t, exists, true)
+			case ip:
+				if key == ip {
+					if value == "" {
+						ips := getNodeIps(t, reverse)
+						require.Contains(t, ips, cpOptions[ip])
+					} else {
+						logrus.Infof("ip is %v", cpOptions[ip])
+						require.Equal(t, value, cpOptions[ip])
+					}
+				}
+			}
+		}
+	}
+}
+
+func getNodeIps(t *testing.T, reverse bool) []string {
+	err := setDestinationKubeConfig()
+	if reverse {
+		err = setSourceKubeConfig()
+	}
+	require.NoError(t, err, "failed to set kubeconfig: %v", err)
+	var nodeIps []string
+	pxNodes, err := core.Instance().GetNodes()
+	require.NoError(t, err, "failed to get PX nodes: %v", err)
+	for _, node := range pxNodes.Items {
+		if !core.Instance().IsNodeMaster(node) {
+			for _, addr := range node.Status.Addresses {
+				if addr.Type == corev1.NodeExternalIP || addr.Type == corev1.NodeInternalIP {
+					nodeIps = append(nodeIps, addr.Address)
+				}
+			}
+		}
+	}
+	logrus.Infof("Node IPs are %v", nodeIps)
+	err = setSourceKubeConfig()
+	require.NoError(t, err, "failed to set kubeconfig to source cluster: %v", err)
+	return nodeIps
+}
+
+func cleanupNamespaces() {
+	err := core.Instance().DeleteNamespace(cpNamespace)
+	if err != nil {
+		logrus.Infof("Error deleting namespace %s: %v\n", cpNamespace, err)
+	}
+	err = setDestinationKubeConfig()
+	if err != nil {
+		logrus.Infof("Error setting kubeconfig on destination cluster")
+	}
+	err = core.Instance().DeleteNamespace(cpNamespace)
+	if err != nil {
+		logrus.Infof("Error deleting namespace %s: %v\n", cpNamespace, err)
+	}
+	err = setSourceKubeConfig()
+	if err != nil {
+		logrus.Infof("Error setting kubeconfig on source cluster")
+	}
+}