add unit-test for file:// and sftp:// (#68)

Co-authored-by: Nchnliyong <liyong@juicedata.io>

.gitignore

 dist
 juicesync
 .idea
+.DS_Store

object/filesystem_test.go

+package object
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+)
+
+func testKeysEqual(objs []*Object, expectedKeys []string) error {
+	gottenKeys := make([]string, len(objs))
+	for idx, obj := range objs {
+		gottenKeys[idx] = obj.Key
+	}
+	if len(gottenKeys) != len(expectedKeys) {
+		return fmt.Errorf("Expected {%s}, got {%s}", strings.Join(expectedKeys, ", "),
+			strings.Join(gottenKeys, ", "))
+	}
+
+	for idx, key := range gottenKeys {
+		if key != expectedKeys[idx] {
+			return fmt.Errorf("Expected {%s}, got {%s}", strings.Join(expectedKeys, ", "),
+				strings.Join(gottenKeys, ", "))
+		}
+	}
+	return nil
+}
+
+func TestFsFile(t *testing.T) {
+	keys := []string{
+		"x/",
+		"x/x.txt",
+		"xy.txt",
+		"xyz/",
+		"xyz/xyz.txt",
+	}
+	s0 := newDisk("/tmp/abc/unit-test/", "", "")
+	// initialize directory tree
+	for _, key := range keys {
+		if err := s0.Put(key, bytes.NewReader([]byte{})); err != nil {
+			t.Fatalf("PUT object `%s` failed: %q", key, err)
+		}
+	}
+	// cleanup
+	defer func() {
+		// delete reversely, directory only can be deleted when it's empty
+		idx := len(keys) - 1
+		for ; idx >= 0; idx-- {
+			if err := s0.Delete(keys[idx]); err != nil {
+				t.Fatalf("DELETE object `%s` failed: %q", keys[idx], err)
+			}
+		}
+	}()
+
+	s := newDisk("/tmp/abc/unit-test/x/", "", "")
+	objs, err := listAll(s, "", "", 100)
+	if err != nil {
+		t.Fatalf("list failed: %s", err)
+	}
+	expectedKeys := []string{"", "x.txt"}
+	if err = testKeysEqual(objs, expectedKeys); err != nil {
+		t.Fatalf("testKeysEqual fail: %s", err)
+	}
+
+	s = newDisk("/tmp/abc/unit-test/x", "", "")
+	objs, err = listAll(s, "", "", 100)
+	if err != nil {
+		t.Fatalf("list failed: %s", err)
+	}
+	expectedKeys = []string{"/", "/x.txt", "y.txt", "yz/", "yz/xyz.txt"}
+	if err = testKeysEqual(objs, expectedKeys); err != nil {
+		t.Fatalf("testKeysEqual fail: %s", err)
+	}
+
+	s = newDisk("/tmp/abc/unit-test/xy", "", "")
+	objs, err = listAll(s, "", "", 100)
+	if err != nil {
+		t.Fatalf("list failed: %s", err)
+	}
+	expectedKeys = []string{".txt", "z/", "z/xyz.txt"}
+	if err = testKeysEqual(objs, expectedKeys); err != nil {
+		t.Fatalf("testKeysEqual fail: %s", err)
+	}
+}
+
+func TestFsSftp(t *testing.T) {
+	// utils.SetLogLevel(logrus.DebugLevel)
+	sftpHost := os.Getenv("SFTP_HOST")
+	if sftpHost == "" {
+		t.SkipNow()
+	}
+	sftpUser, sftpPass := os.Getenv("SFTP_USER"), os.Getenv("SFTP_PASS")
+	s0 := newSftp(sftpHost, sftpUser, sftpPass)
+
+	keys := []string{
+		"x/",
+		"x/x.txt",
+		"xy.txt",
+		"xyz/",
+		"xyz/xyz.txt",
+	}
+	// initialize directory tree
+	for _, key := range keys {
+		if err := s0.Put(key, bytes.NewReader([]byte{})); err != nil {
+			t.Fatalf("PUT object `%s` failed: %q", key, err)
+		}
+	}
+	// cleanup
+	defer func() {
+		// delete reversely, directory only can be deleted when it's empty
+		idx := len(keys) - 1
+		for ; idx >= 0; idx-- {
+			if err := s0.Delete(keys[idx]); err != nil {
+				t.Fatalf("DELETE object `%s` failed: %q", keys[idx], err)
+			}
+		}
+	}()
+
+	s := newSftp(sftpHost+"x/", sftpUser, sftpPass)
+	objs, err := listAll(s, "", "", 100)
+	if err != nil {
+		t.Fatalf("list failed: %s", err)
+	}
+	expectedKeys := []string{"", "x.txt"}
+	if err = testKeysEqual(objs, expectedKeys); err != nil {
+		t.Fatalf("testKeysEqual fail: %s", err)
+	}
+
+	s = newSftp(sftpHost+"x", sftpUser, sftpPass)
+	objs, err = listAll(s, "", "", 100)
+	if err != nil {
+		t.Fatalf("list failed: %s", err)
+	}
+	expectedKeys = []string{"/", "/x.txt", "y.txt", "yz/", "yz/xyz.txt"}
+	if err = testKeysEqual(objs, expectedKeys); err != nil {
+		t.Fatalf("testKeysEqual fail: %s", err)
+	}
+
+	s = newSftp(sftpHost+"xy", sftpUser, sftpPass)
+	objs, err = listAll(s, "", "", 100)
+	if err != nil {
+		t.Fatalf("list failed: %s", err)
+	}
+	expectedKeys = []string{".txt", "z/", "z/xyz.txt"}
+	if err = testKeysEqual(objs, expectedKeys); err != nil {
+		t.Fatalf("testKeysEqual fail: %s", err)
+	}
+}

object/object_storage_test.go

@@ -66,7 +66,6 @@ func testStorage(t *testing.T, s ObjectStorage) {
 		t.Fatalf("expect ll, but got %v, error:%s", d, e)
 	}
 	if d, e := get(s, "/test", 4, 2); d != "o" {
-		// OSS fail
 		t.Errorf("out-of-range get: 'o', but got %v, error:%s", len(d), e)
 	}
 

object/sftp.go

@@ -144,7 +144,14 @@ func (f *sftpStore) String() string {
 	return fmt.Sprintf("%s@%s:%s", f.config.User, f.host, f.root)
 }
 
+// always preserve suffix `/` for directory key
 func (f *sftpStore) path(key string) string {
+	if key == "" {
+		return f.root
+	}
+	if strings.HasSuffix(key, dirSuffix) {
+		return filepath.Join(f.root, key) + dirSuffix
+	}
 	return filepath.Join(f.root, key)
 }
 
@@ -285,7 +292,7 @@ func (f *sftpStore) doFind(c *sftp.Client, path, marker string, out chan *Object
 
 	sort.Sort(sortFI(infos))
 	for _, fi := range infos {
-		p := path + "/" + fi.Name()
+		p := path + fi.Name()
 		key := p[len(f.root):]
 		if key > marker {
 			if fi.IsDir() {
@@ -295,22 +302,18 @@ func (f *sftpStore) doFind(c *sftp.Client, path, marker string, out chan *Object
 			}
 		}
 		if fi.IsDir() && (key > marker || strings.HasPrefix(marker, key)) {
-			f.doFind(c, p, marker, out)
+			f.doFind(c, p+dirSuffix, marker, out)
 		}
 	}
 }
 
 func (f *sftpStore) find(c *sftp.Client, path, marker string, out chan *Object) {
-	if path == "" {
-		path = "."
-	}
-
-	fi, err := c.Stat(path)
-	if err != nil {
-		logger.Errorf("Stat %s error: %q", path, err)
-		return
-	}
-	if fi.IsDir() && strings.HasSuffix(path, "/") {
+	if strings.HasSuffix(path, dirSuffix) {
+		fi, err := c.Stat(path)
+		if err != nil {
+			logger.Errorf("Stat %s error: %q", path, err)
+			return
+		}
 		if marker == "" {
 			out <- &Object{"", 0, fi.ModTime(), true}
 		}
@@ -318,7 +321,7 @@ func (f *sftpStore) find(c *sftp.Client, path, marker string, out chan *Object)
 	} else {
 		// As files or dirs in the same directory of file `path` resides
 		// may have prefix `path`, we should list the directory.
-		dir := filepath.Dir(path)
+		dir := filepath.Dir(path) + dirSuffix
 		infos, err := c.ReadDir(dir)
 		if err != nil {
 			logger.Errorf("readdir %s: %s", dir, err)
@@ -327,7 +330,7 @@ func (f *sftpStore) find(c *sftp.Client, path, marker string, out chan *Object)
 
 		sort.Sort(sortFI(infos))
 		for _, fi := range infos {
-			p := dir + "/" + fi.Name()
+			p := dir + fi.Name()
 			if !strings.HasPrefix(p, f.root) {
 				if p > f.root {
 					break
@@ -344,7 +347,7 @@ func (f *sftpStore) find(c *sftp.Client, path, marker string, out chan *Object)
 				}
 			}
 			if fi.IsDir() && (key > marker || strings.HasPrefix(marker, key)) {
-				f.doFind(c, p, marker, out)
+				f.doFind(c, p+dirSuffix, marker, out)
 			}
 		}
 	}
@@ -362,7 +365,8 @@ func (f *sftpStore) ListAll(prefix, marker string) (<-chan *Object, error) {
 	listed := make(chan *Object, 10240)
 	go func() {
 		defer f.putSftpConnection(&c, nil)
-		f.find(c.sftpClient, filepath.Join(f.root, prefix), marker, listed)
+
+		f.find(c.sftpClient, f.path(prefix), marker, listed)
 		close(listed)
 	}()
 	return listed, nil
@@ -370,7 +374,12 @@ func (f *sftpStore) ListAll(prefix, marker string) (<-chan *Object, error) {
 
 func newSftp(endpoint, user, pass string) ObjectStorage {
 	parts := strings.Split(endpoint, ":")
-	root := parts[1]
+	root := filepath.Clean(parts[1])
+	// append suffix `/` removed by filepath.Clean()
+	// `.` is a directory, add `/`
+	if strings.HasSuffix(parts[1], dirSuffix) || root == "." {
+		root = root + dirSuffix
+	}
 
 	config := &ssh.ClientConfig{
 		User:            user,
@@ -396,11 +405,33 @@ func newSftp(endpoint, user, pass string) ObjectStorage {
 		config.Auth = append(config.Auth, ssh.PublicKeys(signer))
 	}
 
-	return &sftpStore{
+	f := &sftpStore{
 		host:   parts[0],
 		root:   root,
 		config: config,
 	}
+
+	c, err := f.getSftpConnection()
+	if err != nil {
+		logger.Errorf("getSftpConnection failed: %s", err)
+		return nil
+	}
+	defer f.putSftpConnection(&c, err)
+
+	if strings.HasSuffix(root, dirSuffix) {
+		logger.Debugf("Ensure dicectory %s", root)
+		if err := c.sftpClient.MkdirAll(root); err != nil {
+			logger.Fatalf("Creating directory %s failed: %q", root, err)
+		}
+	} else {
+		dir := filepath.Dir(root)
+		logger.Debugf("Ensure dicectory %s", dir)
+		if err := c.sftpClient.MkdirAll(dir); err != nil {
+			logger.Fatalf("Creating directory %s failed: %q", dir, err)
+		}
+	}
+
+	return f
 }
 
 func init() {