Repository: lf-edge/ekuiper
Pull request #2913 (2024-06-11 03:59:38)

test: add test for plugin manager

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2913?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@9402a87`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.


<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff              @@
##             master-v2    #2913   +/-   ##
============================================
  Coverage             ?   68.80%           
============================================
  Files                ?      297           
  Lines                ?    28668           
  Branches             ?        0           
============================================
  Hits                 ?    19725           
  Misses               ?     7223           
  Partials             ?     1720           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2913?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/internal/conf/path.go b/internal/conf/path.go
index a1c7bb44..b5831477 100644
--- a/internal/conf/path.go
+++ b/internal/conf/path.go
@@ -15,11 +15,14 @@
 package conf
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"path"
 	"path/filepath"
 	"strings"
+
+	"github.com/pingcap/failpoint"
 )
 
 func init() {
@@ -58,7 +61,12 @@ func GetLogLoc() (string, error) {
 	return GetLoc(logDir)
 }
 
-func GetDataLoc() (string, error) {
+func GetDataLoc() (s string, err error) {
+	defer func() {
+		failpoint.Inject("GetDataLocErr", func() {
+			err = errors.New("GetDataLocErr")
+		})
+	}()
 	if IsTesting {
 		dataDir, err := GetLoc(dataDir)
 		if err != nil {
@@ -80,7 +88,12 @@ func GetDataLoc() (string, error) {
 	return GetLoc(dataDir)
 }
 
-func GetPluginsLoc() (string, error) {
+func GetPluginsLoc() (s string, err error) {
+	defer func() {
+		failpoint.Inject("GetPluginsLocErr", func() {
+			err = errors.New("GetPluginsLocErr")
+		})
+	}()
 	return GetLoc(pluginsDir)
 }
 
diff --git a/internal/pkg/filex/zip.go b/internal/pkg/filex/zip.go
index 52da7dc4..b1c16111 100644
--- a/internal/pkg/filex/zip.go
+++ b/internal/pkg/filex/zip.go
@@ -16,14 +16,22 @@ package filex
 
 import (
 	"archive/zip"
+	"errors"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
+
+	"github.com/pingcap/failpoint"
 )
 
-func UnzipTo(f *zip.File, fpath string) error {
-	_, err := os.Stat(fpath)
+func UnzipTo(f *zip.File, fpath string) (err error) {
+	defer func() {
+		failpoint.Inject("UnzipToErr", func() {
+			err = errors.New("UnzipToErr")
+		})
+	}()
+	_, err = os.Stat(fpath)
 
 	if f.FileInfo().IsDir() {
 		// Make Folder
diff --git a/internal/pkg/httpx/http.go b/internal/pkg/httpx/http.go
index 9644cd7c..b2f458a1 100644
--- a/internal/pkg/httpx/http.go
+++ b/internal/pkg/httpx/http.go
@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"crypto/tls"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -26,6 +27,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pingcap/failpoint"
+
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 )
@@ -204,7 +207,12 @@ func ReadFile(uri string) (io.ReadCloser, error) {
 	return src, nil
 }
 
-func DownloadFile(filepath string, uri string) error {
+func DownloadFile(filepath string, uri string) (err error) {
+	defer func() {
+		failpoint.Inject("DownloadFileErr", func() {
+			err = errors.New("DownloadFileErr")
+		})
+	}()
 	src, err := ReadFile(uri)
 	if err != nil {
 		return err
diff --git a/internal/plugin/portable/manager.go b/internal/plugin/portable/manager.go
index dc542351..fb54c421 100644
--- a/internal/plugin/portable/manager.go
+++ b/internal/plugin/portable/manager.go
@@ -19,6 +19,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -28,6 +29,8 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/pingcap/failpoint"
+
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/filex"
@@ -80,10 +83,16 @@ func InitManager() (*Manager, error) {
 		return nil, err
 	}
 	plgDb, err := store.GetKV("portablePlugin")
+	failpoint.Inject("plgDBErr", func() {
+		err = errors.New("plgDBErr")
+	})
 	if err != nil {
 		return nil, fmt.Errorf("error when opening portablePlugin: %v", err)
 	}
 	plgStatusDb, err := store.GetKV("portablePluginStatus")
+	failpoint.Inject("plgStatusDbErr", func() {
+		err = errors.New("plgStatusDbErr")
+	})
 	if err != nil {
 		return nil, fmt.Errorf("error when opening portablePluginStatus: %v", err)
 	}
@@ -93,30 +102,17 @@ func InitManager() (*Manager, error) {
 	return m, nil
 }
 
-func GetManager() *Manager {
-	return manager
-}
-
-func MockManager(plugins map[string]*PluginInfo) (*Manager, error) {
-	reg := &registry{
-		RWMutex:   sync.RWMutex{},
-		plugins:   make(map[string]*PluginInfo),
-		sources:   make(map[string]string),
-		sinks:     make(map[string]string),
-		functions: make(map[string]string),
-	}
-	for name, pi := range plugins {
-		err := pi.Validate(name)
-		if err != nil {
-			return nil, err
-		}
-		reg.Set(name, pi)
-	}
-	return &Manager{reg: reg}, nil
-}
+func (m *Manager) syncRegistry() (err error) {
+	defer func() {
+		failpoint.Inject("syncRegistryErr", func() {
+			err = errors.New("syncRegistryErr")
+		})
+	}()
 
-func (m *Manager) syncRegistry() error {
 	files, err := os.ReadDir(m.pluginDir)
+	failpoint.Inject("syncRegistryReadDirErr", func() {
+		err = errors.New("syncRegistryReadDirErr")
+	})
 	if err != nil {
 		return fmt.Errorf("read path '%s' error: %v", m.pluginDir, err)
 	}
@@ -143,7 +139,8 @@ func (m *Manager) parsePlugin(name string) error {
 
 func (m *Manager) doRegister(name string, pi *PluginInfo, isInit bool) error {
 	exeAbs := filepath.Clean(filepath.Join(m.pluginDir, name, pi.Executable))
-	if _, err := os.Stat(exeAbs); err != nil {
+	_, err := os.Stat(exeAbs)
+	if err != nil {
 		return fmt.Errorf("cannot find executable `%s` when loading portable plugins: %v", exeAbs, err)
 	}
 	pi.Executable = exeAbs
@@ -166,10 +163,18 @@ func (m *Manager) doRegister(name string, pi *PluginInfo, isInit bool) error {
 	return nil
 }
 
-func (m *Manager) parsePluginJson(name string) (*PluginInfo, error) {
+func (m *Manager) parsePluginJson(name string) (info *PluginInfo, err error) {
+	defer func() {
+		failpoint.Inject("parsePluginJsonErr", func() {
+			err = errors.New("parsePluginJsonErr")
+		})
+	}()
 	jsonPath := filepath.Join(m.pluginDir, name, name+".json")
 	pi := &PluginInfo{PluginMeta: runtime.PluginMeta{Name: name}}
-	err := filex.ReadJsonUnmarshal(jsonPath, pi)
+	err = filex.ReadJsonUnmarshal(jsonPath, pi)
+	failpoint.Inject("parsePluginJsonReadJsonUnmarshalErr", func() {
+		err = errors.New("parsePluginJsonReadJsonUnmarshalErr")
+	})
 	if err != nil {
 		return nil, fmt.Errorf("cannot read json file `%s` when loading portable plugins: %v", jsonPath, err)
 	}
@@ -240,6 +245,9 @@ func (m *Manager) install(name, src string, shellParas []string) (resultErr erro
 		}
 	}()
 	r, err := zip.OpenReader(src)
+	failpoint.Inject("installOpenReaderErr", func() {
+		err = errors.New("installOpenReaderErr")
+	})
 	if err != nil {
 		return err
 	}
@@ -251,16 +259,25 @@ func (m *Manager) install(name, src string, shellParas []string) (resultErr erro
 		filesNumber++
 		if file.Name == jsonName {
 			jf, err := file.Open()
+			failpoint.Inject("installFileOpenErr", func() {
+				err = errors.New("installFileOpenErr")
+			})
 			if err != nil {
 				err = fmt.Errorf("invalid json file %s: %s", jsonName, err)
 				return err
 			}
 			pi = &PluginInfo{PluginMeta: runtime.PluginMeta{Name: name}}
 			allBytes, err := io.ReadAll(jf)
+			failpoint.Inject("installReadErr", func() {
+				err = errors.New("installReadErr")
+			})
 			if err != nil {
 				return err
 			}
 			err = json.Unmarshal(allBytes, pi)
+			failpoint.Inject("installJsonMarshalErr", func() {
+				err = errors.New("installJsonMarshalErr")
+			})
 			if err != nil {
 				return err
 			}
@@ -285,6 +302,9 @@ func (m *Manager) install(name, src string, shellParas []string) (resultErr erro
 	d := filepath.Clean(pluginTarget)
 	if _, err := os.Stat(d); os.IsNotExist(err) {
 		err = os.MkdirAll(d, 0o755)
+		failpoint.Inject("installMkdirErr", func() {
+			err = errors.New("installMkdirErr")
+		})
 		if err != nil {
 			return err
 		}
diff --git a/internal/plugin/portable/model.go b/internal/plugin/portable/model.go
index c0322d65..423307f5 100644
--- a/internal/plugin/portable/model.go
+++ b/internal/plugin/portable/model.go
@@ -15,8 +15,11 @@
 package portable
 
 import (
+	"errors"
 	"fmt"
 
+	"github.com/pingcap/failpoint"
+
 	"github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime"
 )
 
@@ -33,7 +36,12 @@ var langMap = map[string]bool{
 }
 
 // Validate TODO validate duplication of source, sink and functions
-func (p *PluginInfo) Validate(expectedName string) error {
+func (p *PluginInfo) Validate(expectedName string) (err error) {
+	defer func() {
+		failpoint.Inject("PluginInfoValidateErr", func() {
+			err = errors.New("PluginInfoValidateErr")
+		})
+	}()
 	if p.Name != expectedName {
 		return fmt.Errorf("invalid plugin, expect name '%s' but got '%s'", expectedName, p.Name)
 	}


Test patch:
diff --git a/internal/plugin/portable/manager_test.go b/internal/plugin/portable/manager_test.go
index 40294b7e..77a89f52 100644
--- a/internal/plugin/portable/manager_test.go
+++ b/internal/plugin/portable/manager_test.go
@@ -26,6 +26,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/require"
+
 	"github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/plugin"
 	"github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime"
@@ -87,19 +90,19 @@ func TestManager_Install(t *testing.T) {
 	}
 
 	fmt.Printf("The test bucket size is %d.\n\n", len(data))
-	for i, tt := range data {
+	for _, tt := range data {
 		p := &plugin.IOPlugin{
 			Name: tt.n,
 			File: tt.u,
 		}
-		err := manager.Register(p)
-		if !reflect.DeepEqual(tt.err, err) {
-			t.Errorf("%d: error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.err, err)
-		} else if tt.err == nil {
-			err := checkFileForMirror(manager.pluginDir, manager.pluginConfDir, true)
-			if err != nil {
-				t.Errorf("%d: error : %s\n\n", i, err)
-			}
+		m, err := InitManager()
+		require.NoError(t, err)
+		err = m.Register(p)
+		if err != nil {
+			require.Equal(t, tt.err, err)
+		} else {
+			err := checkFileForMirror(m.pluginDir, m.pluginConfDir, true)
+			require.NoError(t, err)
 		}
 	}
 }
@@ -202,3 +205,120 @@ func checkFileForMirror(pluginDir, etcDir string, exist bool) error {
 	}
 	return nil
 }
+
+func TestManagerErr(t *testing.T) {
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/conf/GetPluginsLocErr", "return(true)")
+	_, err := InitManager()
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/conf/GetPluginsLocErr")
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/conf/GetDataLocErr", "return(true)")
+	_, err = InitManager()
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/conf/GetDataLocErr")
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/syncRegistryErr", "return(true)")
+	_, err = InitManager()
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/syncRegistryErr")
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/plgDBErr", "return(true)")
+	_, err = InitManager()
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/plgDBErr")
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/plgStatusDbErr", "return(true)")
+	_, err = InitManager()
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/plgStatusDbErr")
+
+	m, err := InitManager()
+	require.NoError(t, err)
+	require.NotNil(t, m)
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/syncRegistryReadDirErr", "return(true)")
+	err = m.syncRegistry()
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/syncRegistryReadDirErr")
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/parsePluginJsonErr", "return(true)")
+	err = m.parsePlugin("mock")
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/parsePluginJsonErr")
+
+	err = m.doRegister("mock", &PluginInfo{}, true)
+	require.Error(t, err)
+}
+
+func TestParsePluginJson(t *testing.T) {
+	m, err := InitManager()
+	require.NoError(t, err)
+	require.NotNil(t, m)
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/parsePluginJsonReadJsonUnmarshalErr", "return(true)")
+	_, err = m.parsePluginJson("mock")
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/parsePluginJsonReadJsonUnmarshalErr")
+
+	_, err = m.parsePluginJson("mock")
+	require.Error(t, err)
+	m.reg.Set("mirror", &PluginInfo{})
+	_, err = m.parsePluginJson("mirror")
+	require.Error(t, err)
+}
+
+func TestRegisterErr(t *testing.T) {
+	s := httptest.NewServer(
+		http.FileServer(http.Dir("../testzips")),
+	)
+	defer s.Close()
+	endpoint := s.URL
+	p := &plugin.IOPlugin{
+		Name: "mirror2",
+		File: endpoint + "/portables/mirror.zip",
+	}
+
+	manager.reg.Set("mirror2", &PluginInfo{})
+	err := manager.Register(p)
+	require.Error(t, err)
+	manager.reg.Delete("mirror2")
+
+	testcases := []struct {
+		mockErr string
+	}{
+		{
+			mockErr: "github.com/lf-edge/ekuiper/v2/internal/pkg/httpx/DownloadFileErr",
+		},
+		{
+			mockErr: "github.com/lf-edge/ekuiper/v2/internal/plugin/portable/installOpenReaderErr",
+		},
+		{
+			mockErr: "github.com/lf-edge/ekuiper/v2/internal/plugin/portable/installFileOpenErr",
+		},
+		{
+			mockErr: "github.com/lf-edge/ekuiper/v2/internal/plugin/portable/installReadErr",
+		},
+		{
+			mockErr: "github.com/lf-edge/ekuiper/v2/internal/plugin/portable/installJsonMarshalErr",
+		},
+		{
+			mockErr: "github.com/lf-edge/ekuiper/v2/internal/plugin/portable/PluginInfoValidateErr",
+		},
+		{
+			mockErr: "github.com/lf-edge/ekuiper/v2/internal/plugin/portable/PluginInfoValidateErr",
+		},
+		{
+			mockErr: "github.com/lf-edge/ekuiper/v2/internal/plugin/portable/installMkdirErr",
+		},
+		{
+			mockErr: "github.com/lf-edge/ekuiper/v2/internal/pkg/filex/UnzipToErr",
+		},
+	}
+
+	for _, testcase := range testcases {
+		failpoint.Enable(testcase.mockErr, "return(true)")
+		err = manager.Register(p)
+		require.Error(t, err, testcase.mockErr)
+		failpoint.Disable(testcase.mockErr)
+	}
+}