* Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
* KubernetesPlugin is licensed under Mulan PSL v2.
* You can use this software according to the terms and conditions of the Mulan PSL v2.
* You may obtain a copy of Mulan PSL v2 at:
* http://license.coscl.org.cn/MulanPSL2
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PSL v2 for more details.
*/
package main
import (
"errors"
"os"
"reflect"
"testing"
"time"
"bou.ke/monkey"
"github.com/container-storage-interface/spec/lib/go/csi"
"github.com/stretchr/testify/assert"
"k8s.io/kubelet/pkg/apis/pluginregistration/v1"
"matrixplugin/pkg/communication"
"matrixplugin/pkg/driver/shm"
"matrixplugin/pkg/utils/fs"
)
var runValue = false
const sleepDuration = 200 * time.Millisecond
func TestStartCentralNode(t *testing.T) {
runValue = false
monkeyNewRegistrationServer := monkey.Patch(shm.NewRegistrationServer, func(driverName string,
endpoint string, versions []string, errChan chan<- error) v1.RegistrationServer {
return &shm.RegistrationServer{}
})
defer monkeyNewRegistrationServer.Unpatch()
monkeyNewDefaultCSIDriver := monkey.Patch(shm.NewDefaultCSIDriver, func(name string,
nodeID string, version string) *shm.DefaultCSIDriver {
return &shm.DefaultCSIDriver{}
})
defer monkeyNewDefaultCSIDriver.Unpatch()
monkey.PatchInstanceMethod(reflect.TypeOf(&communication.GRPCServer{}), "Start", func(_ *communication.GRPCServer,
endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer,
rs v1.RegistrationServer,
) {
runValue = true
})
blockChan := make(chan error, 1)
go func() {
time.Sleep(1 * time.Millisecond)
blockChan <- errors.New("test")
}()
StartCentralNode(blockChan)
time.Sleep(sleepDuration)
assert.Equal(t, runValue, true)
}
func TestPrepareIpcChmodUdsDirFail(t *testing.T) {
err := PrepareIpc()
assert.Error(t, err)
assert.Contains(t, err.Error(), "chmod uds dir")
}
func TestPrepareIpcSuccess(t *testing.T) {
defer monkey.UnpatchAll()
monkey.Patch(os.Chmod, func(string, os.FileMode) error { return nil })
monkey.Patch(os.Chown, func(string, int, int) error { return nil })
monkey.Patch(initInnerUdsDir, func() error { return nil })
monkey.Patch(initShmAgentPath, func() error { return nil })
monkey.Patch(initCsiDriverDir, func() error { return nil })
err := PrepareIpc()
assert.NoError(t, err)
}
func TestInitInnerUdsDirChmodFail(t *testing.T) {
defer monkey.UnpatchAll()
p := monkey.Patch(os.Chmod, func(string, os.FileMode) error { return errors.New("chmod fail") })
defer p.Unpatch()
err := initInnerUdsDir()
assert.Error(t, err)
assert.Contains(t, err.Error(), "chmod inner uds dir")
}
func TestInitInnerUdsDirSetAclFail(t *testing.T) {
defer monkey.UnpatchAll()
monkey.Patch(os.Chmod, func(string, os.FileMode) error { return nil })
monkey.Patch(os.Chown, func(string, int, int) error { return nil })
p := monkey.Patch(fs.SetUserAcl, func(string, string, uint16) error { return errors.New("setacl fail") })
defer p.Unpatch()
err := initInnerUdsDir()
assert.Error(t, err)
assert.Contains(t, err.Error(), "setAcl inner uds dir")
}
func TestInitInnerUdsDirSuccess(t *testing.T) {
defer monkey.UnpatchAll()
monkey.Patch(os.Chmod, func(string, os.FileMode) error { return nil })
monkey.Patch(os.Chown, func(string, int, int) error { return nil })
monkey.Patch(fs.SetUserAcl, func(string, string, uint16) error { return nil })
err := initInnerUdsDir()
assert.NoError(t, err)
}
func TestInitShmAgentPathNormalizationFail(t *testing.T) {
defer monkey.UnpatchAll()
p := monkey.Patch(fs.PathNormalization, func(path string) (string, error) {
return "", errors.New("normalize fail")
})
defer p.Unpatch()
err := initShmAgentPath()
assert.Error(t, err)
assert.Contains(t, err.Error(), "normalization")
}
func TestInitShmAgentPathChmodFail(t *testing.T) {
defer monkey.UnpatchAll()
monkey.Patch(fs.PathNormalization, func(path string) (string, error) {
return "/tmp/shm", nil
})
p := monkey.Patch(os.Chmod, func(string, os.FileMode) error { return errors.New("chmod fail") })
defer p.Unpatch()
err := initShmAgentPath()
assert.Error(t, err)
assert.Contains(t, err.Error(), "chmod shmagent")
}
func TestInitShmAgentPathChownFail(t *testing.T) {
defer monkey.UnpatchAll()
monkey.Patch(fs.PathNormalization, func(path string) (string, error) {
return "/tmp/shm", nil
})
monkey.Patch(os.Chmod, func(string, os.FileMode) error { return nil })
p := monkey.Patch(os.Chown, func(string, int, int) error { return errors.New("chown fail") })
defer p.Unpatch()
err := initShmAgentPath()
assert.Error(t, err)
assert.Contains(t, err.Error(), "chown shmagent")
}
func TestInitShmAgentPathSuccess(t *testing.T) {
defer monkey.UnpatchAll()
monkey.Patch(fs.PathNormalization, func(path string) (string, error) { return "/tmp/shm", nil })
monkey.Patch(os.Chmod, func(string, os.FileMode) error { return nil })
monkey.Patch(os.Chown, func(string, int, int) error { return nil })
err := initShmAgentPath()
assert.NoError(t, err)
}
func TestInitCsiDriverDirNormalizationFail(t *testing.T) {
defer monkey.UnpatchAll()
p := monkey.Patch(fs.PathNormalization, func(path string) (string, error) {
return "", errors.New("normalize fail")
})
defer p.Unpatch()
err := initCsiDriverDir()
assert.Error(t, err)
assert.Contains(t, err.Error(), "normalization")
}
func TestInitCsiDriverDirChmodFail(t *testing.T) {
defer monkey.UnpatchAll()
monkey.Patch(fs.PathNormalization, func(string) (string, error) { return "/tmp/csi", nil })
p := monkey.Patch(os.Chmod, func(string, os.FileMode) error { return errors.New("chmod fail") })
defer p.Unpatch()
err := initCsiDriverDir()
assert.Error(t, err)
assert.Contains(t, err.Error(), "chmod csidriver")
}
func TestInitCsiDriverDirChownFail(t *testing.T) {
defer monkey.UnpatchAll()
monkey.Patch(fs.PathNormalization, func(string) (string, error) { return "/tmp/csi", nil })
monkey.Patch(os.Chmod, func(string, os.FileMode) error { return nil })
p := monkey.Patch(os.Chown, func(string, int, int) error { return errors.New("chown fail") })
defer p.Unpatch()
err := initCsiDriverDir()
assert.Error(t, err)
assert.Contains(t, err.Error(), "chown csidriver")
}
func TestInitCsiDriverDirSuccess(t *testing.T) {
defer monkey.UnpatchAll()
monkey.Patch(fs.PathNormalization, func(path string) (string, error) { return "/tmp/csi", nil })
monkey.Patch(os.Chmod, func(string, os.FileMode) error { return nil })
monkey.Patch(os.Chown, func(string, int, int) error { return nil })
err := initCsiDriverDir()
assert.NoError(t, err)
}