Add an unit test example for install command

Signed-off-by: Shaohui Liu <liushaohui@xiaomi.com>

pkg/cluster/install/install_test.go

@@ -0,0 +1,121 @@
+/*
+Copyright 2020 The KubeSphere Authors.
+
+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 install
+
+import (
+	"fmt"
+	kubekeyapiv1alpha1 "github.com/kubesphere/kubekey/apis/kubekey/v1alpha1"
+	"github.com/kubesphere/kubekey/pkg/connector"
+	"github.com/kubesphere/kubekey/pkg/util"
+	"github.com/kubesphere/kubekey/pkg/util/executor"
+	"runtime"
+	"testing"
+)
+
+var logger = util.InitLogger(true)
+
+type MockConnector struct {
+	connections map[string]*MockConnection
+}
+
+func (connector *MockConnector) Connect(host kubekeyapiv1alpha1.HostCfg) (connector.Connection, error) {
+	if val, ok := connector.connections[host.Address]; ok {
+		logger.Infof("Connection exist for host: %s", host.Address)
+		return val, nil
+	}
+
+	logger.Infof("New connection on host: %s", host.Address)
+	operations := make([]string, 0)
+	connection := &MockConnection{operations: operations}
+	connector.connections[host.Address] = connection
+	return connection, nil
+}
+
+type MockConnection struct {
+	operations []string
+}
+
+func (mock *MockConnection) Exec(cmd string, host *kubekeyapiv1alpha1.HostCfg) (stdout string, err error) {
+	logger.Infof("run cmd: %s on host: %s", cmd, host.Address)
+	mock.operations = append(mock.operations, cmd)
+	return "OK", nil
+}
+
+func (mock *MockConnection) Scp(src, dst string) error {
+	return nil
+}
+
+func (mock *MockConnection) Close() {
+}
+
+func GenTestClusterCfg(name string) (*kubekeyapiv1alpha1.Cluster, string) {
+	cfg := kubekeyapiv1alpha1.Cluster{}
+	cfg.Spec.Hosts = append(cfg.Spec.Hosts, kubekeyapiv1alpha1.HostCfg{
+		Name:            name,
+		Address:         util.LocalIP(),
+		InternalAddress: util.LocalIP(),
+		Port:            kubekeyapiv1alpha1.DefaultSSHPort,
+		User:            "user",
+		Password:        "",
+		PrivateKeyPath:  fmt.Sprintf("%s/.ssh/id_rsa", "/home/user"),
+		Arch:            runtime.GOARCH,
+	})
+
+	cfg.Spec.RoleGroups = kubekeyapiv1alpha1.RoleGroups{
+		Etcd:   []string{name},
+		Master: []string{name},
+		Worker: []string{name},
+	}
+	cfg.Spec.Kubernetes = kubekeyapiv1alpha1.Kubernetes{
+		Version: kubekeyapiv1alpha1.DefaultKubeVersion,
+	}
+	return &cfg, name
+}
+
+func assertEqual(t *testing.T, actual interface{}, expected interface{}) {
+	if actual != expected {
+		t.Fatalf("actual: %s != expected: %s", actual, expected)
+	}
+}
+
+func Test_install(t *testing.T) {
+	cfg, objName := GenTestClusterCfg("Test_install")
+	executor := executor.NewExecutor(&cfg.Spec, objName, logger, "", true, true, true, false, false, nil)
+
+	connections := make(map[string]*MockConnection)
+	executor.Connector = &MockConnector{connections: connections}
+
+	executor.DownloadCommand = func(path, url string) string {
+		// this is an extension point for downloading tools, for example users can set the timeout, proxy or retry under
+		// some poor network environment. Or users even can choose another cli, it might be wget.
+		// perhaps we should have a build-in download function instead of totally rely on the external one
+		return fmt.Sprintf("curl -L -o %s %s", path, url)
+	}
+
+	mgr, err := executor.CreateManager()
+	if err != nil {
+		t.Errorf("Create executor manager failure: %s", err)
+	}
+	ExecTasks(mgr)
+	assertEqual(t, len(connections), 1)
+	operations := connections[util.LocalIP()].operations
+	assertEqual(t, len(operations), 50)
+
+	// first command
+	assertEqual(t, operations[0][20:27], "useradd")
+	// TODO
+}

pkg/connector/connector.go

@@ -0,0 +1,28 @@
+/*
+Copyright 2020 The KubeSphere Authors.
+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 connector
+
+import (
+	kubekeyapiv1alpha1 "github.com/kubesphere/kubekey/apis/kubekey/v1alpha1"
+)
+
+type Connection interface {
+	Exec(cmd string, host *kubekeyapiv1alpha1.HostCfg) (stdout string, err error)
+	Scp(src, dst string) error
+	Close()
+}
+
+type Connector interface {
+	Connect(host kubekeyapiv1alpha1.HostCfg) (Connection, error)
+}

pkg/connector/ssh/dialer.go

@@ -17,6 +17,7 @@ limitations under the License.
 package ssh
 
 import (
+	"github.com/kubesphere/kubekey/pkg/connector"
 	"sync"
 	"time"
 
@@ -25,16 +26,16 @@ import (
 
 type Dialer struct {
 	lock        sync.Mutex
-	connections map[int]Connection
+	connections map[int]connector.Connection
 }
 
 func NewDialer() *Dialer {
 	return &Dialer{
-		connections: make(map[int]Connection),
+		connections: make(map[int]connector.Connection),
 	}
 }
 
-func (dialer *Dialer) Connect(host kubekeyapiv1alpha1.HostCfg) (Connection, error) {
+func (dialer *Dialer) Connect(host kubekeyapiv1alpha1.HostCfg) (connector.Connection, error) {
 	var err error
 
 	dialer.lock.Lock()

pkg/connector/ssh/ssh.go

@@ -20,6 +20,7 @@ import (
 	"bufio"
 	"context"
 	"fmt"
+	"github.com/kubesphere/kubekey/pkg/connector"
 	"io/ioutil"
 	"net"
 	"os"
@@ -39,15 +40,9 @@ import (
 const socketEnvPrefix = "env:"
 
 var (
-	_ Connection = &connection{}
+	_ connector.Connection = &connection{}
 )
 
-type Connection interface {
-	Exec(cmd string, host *kubekeyapiv1alpha1.HostCfg) (stdout string, err error)
-	Scp(src, dst string) error
-	Close()
-}
-
 type Cfg struct {
 	Username    string
 	Password    string
@@ -123,7 +118,7 @@ func validateOptions(cfg Cfg) (Cfg, error) {
 	return cfg, nil
 }
 
-func NewConnection(cfg Cfg) (Connection, error) {
+func NewConnection(cfg Cfg) (connector.Connection, error) {
 	cfg, err := validateOptions(cfg)
 	if err != nil {
 		return nil, errors.Wrap(err, "Failed to validate ssh connection parameters")

pkg/util/executor/executor.go

@@ -18,13 +18,14 @@ package executor
 
 import (
 	"fmt"
+	"github.com/kubesphere/kubekey/pkg/connector/ssh"
 	"os"
 	"path/filepath"
 
 	kubekeyapiv1alpha1 "github.com/kubesphere/kubekey/apis/kubekey/v1alpha1"
 	kubekeyclientset "github.com/kubesphere/kubekey/clients/clientset/versioned"
+	"github.com/kubesphere/kubekey/pkg/connector"
 	"github.com/kubesphere/kubekey/pkg/util/manager"
-	"github.com/kubesphere/kubekey/pkg/util/ssh"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 )
@@ -42,6 +43,7 @@ type Executor struct {
 	InCluster          bool
 	ClientSet          *kubekeyclientset.Clientset
 	DownloadCommand    func(path, url string) string
+	Connector          connector.Connector
 }
 
 func NewExecutor(cluster *kubekeyapiv1alpha1.ClusterSpec, objName string, logger *log.Logger, sourcesDir string, debug, skipCheck, skipPullImages, addImagesRepo, inCluster bool, clientset *kubekeyclientset.Clientset) *Executor {
@@ -56,6 +58,7 @@ func NewExecutor(cluster *kubekeyapiv1alpha1.ClusterSpec, objName string, logger
 		AddImagesRepo:  addImagesRepo,
 		InCluster:      inCluster,
 		ClientSet:      clientset,
+		Connector:      ssh.NewDialer(),
 	}
 }
 
@@ -72,7 +75,7 @@ func (executor *Executor) CreateManager() (*manager.Manager, error) {
 	mgr.K8sNodes = hostGroups.K8s
 	mgr.Cluster = defaultCluster
 	mgr.ClusterHosts = GenerateHosts(hostGroups, defaultCluster)
-	mgr.Connector = ssh.NewDialer()
+	mgr.Connector = executor.Connector
 	mgr.WorkDir = GenerateWorkDir(executor.Logger)
 	mgr.KsEnable = executor.Cluster.KubeSphere.Enabled
 	mgr.KsVersion = executor.Cluster.KubeSphere.Version

pkg/util/manager/manager.go

@@ -19,8 +19,8 @@ package manager
 import (
 	kubekeyapiv1alpha1 "github.com/kubesphere/kubekey/apis/kubekey/v1alpha1"
 	kubekeyclientset "github.com/kubesphere/kubekey/clients/clientset/versioned"
+	"github.com/kubesphere/kubekey/pkg/connector"
 	"github.com/kubesphere/kubekey/pkg/util/runner"
-	"github.com/kubesphere/kubekey/pkg/util/ssh"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -29,7 +29,7 @@ type Manager struct {
 	ObjName            string
 	Cluster            *kubekeyapiv1alpha1.ClusterSpec
 	Logger             log.FieldLogger
-	Connector          *ssh.Dialer
+	Connector          connector.Connector
 	Runner             *runner.Runner
 	AllNodes           []kubekeyapiv1alpha1.HostCfg
 	EtcdNodes          []kubekeyapiv1alpha1.HostCfg

pkg/util/manager/run_task.go

@@ -21,7 +21,7 @@ import (
 	"time"
 
 	kubekeyapiv1alpha1 "github.com/kubesphere/kubekey/apis/kubekey/v1alpha1"
-	"github.com/kubesphere/kubekey/pkg/util/ssh"
+	"github.com/kubesphere/kubekey/pkg/connector"
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/pkg/errors"
@@ -75,7 +75,7 @@ func (t *Task) Run(mgr *Manager) error {
 func (mgr *Manager) runTask(node *kubekeyapiv1alpha1.HostCfg, task NodeTask, index int) error {
 	var (
 		err  error
-		conn ssh.Connection
+		conn connector.Connection
 	)
 
 	conn, err = mgr.Connector.Connect(*node)

pkg/util/runner/runner.go

@@ -19,13 +19,13 @@ package runner
 import (
 	"fmt"
 	kubekeyapiv1alpha1 "github.com/kubesphere/kubekey/apis/kubekey/v1alpha1"
-	"github.com/kubesphere/kubekey/pkg/util/ssh"
+	"github.com/kubesphere/kubekey/pkg/connector"
 	"github.com/pkg/errors"
 	"time"
 )
 
 type Runner struct {
-	Conn  ssh.Connection
+	Conn  connector.Connection
 	Debug bool
 	Host  *kubekeyapiv1alpha1.HostCfg
 	Index int