mirror of
https://github.com/hanzoai/s3.git
synced 2026-10-30 14:59:09 +00:00
Full internal rename — build-verified, every module green (root + s3-rdma-sidecar
+ telemetry/server + all test modules):
- Dirs: weed/→s3/ (main tree); cmd/weed-{db,sql}→cmd/s3-{db,sql};
seaweed-volume→s3-volume; seaweedfs-rdma-sidecar→s3-rdma-sidecar;
s3/mount/weedfs_*.go→s3fs_*.go (~113 path renames).
- Imports: github.com/hanzoai/s3/weed/ → /s3/ across 1700+ files.
- gRPC proto services Seaweed/SeaweedFiler/SeaweedMessaging/... → Hanzo*
(8 services) via full protoc regeneration (correct descriptors, no sed-corruption).
- Brand SeaweedFS/Seaweed/seaweed → Hanzo/hanzo; internal X-Seaweedfs-* headers
→ X-Hanzo-*; .seaweedfs config dir → .s3.
- CLI is `s3`: go install github.com/hanzoai/s3/s3@latest | make install | Docker.
install.sh rewritten as a Hanzo S3 installer.
- External forks github.com/seaweedfs/{raft,go-fuse,goexif,cockroachdb-parser}
kept verbatim — immutable third-party module paths; NOTICE retained.
Residual (cosmetic, non-build): kafka loadtest test-proto package com.seaweedfs.loadtest.
423 lines
11 KiB
Go
423 lines
11 KiB
Go
package sftp
|
|
|
|
import (
|
|
"fmt"
|
|
"net"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"runtime"
|
|
"syscall"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/pkg/sftp"
|
|
"github.com/stretchr/testify/require"
|
|
"golang.org/x/crypto/ssh"
|
|
)
|
|
|
|
// SftpTestFramework provides utilities for SFTP integration testing
|
|
type SftpTestFramework struct {
|
|
t *testing.T
|
|
tempDir string
|
|
dataDir string
|
|
masterProcess *os.Process
|
|
volumeProcess *os.Process
|
|
filerProcess *os.Process
|
|
sftpProcess *os.Process
|
|
masterAddr string
|
|
volumeAddr string
|
|
filerAddr string
|
|
sftpAddr string
|
|
s3Binary string
|
|
userStoreFile string
|
|
hostKeyFile string
|
|
isSetup bool
|
|
skipCleanup bool
|
|
}
|
|
|
|
// TestConfig holds configuration for SFTP tests
|
|
type TestConfig struct {
|
|
NumVolumes int
|
|
EnableDebug bool
|
|
SkipCleanup bool // for debugging failed tests
|
|
UserStoreFile string
|
|
}
|
|
|
|
// DefaultTestConfig returns a default configuration for SFTP tests
|
|
func DefaultTestConfig() *TestConfig {
|
|
return &TestConfig{
|
|
NumVolumes: 3,
|
|
EnableDebug: false,
|
|
SkipCleanup: false,
|
|
UserStoreFile: "",
|
|
}
|
|
}
|
|
|
|
// NewSftpTestFramework creates a new SFTP testing framework
|
|
func NewSftpTestFramework(t *testing.T, config *TestConfig) *SftpTestFramework {
|
|
if config == nil {
|
|
config = DefaultTestConfig()
|
|
}
|
|
|
|
tempDir, err := os.MkdirTemp("", "hanzo_sftp_test_")
|
|
require.NoError(t, err)
|
|
|
|
// Generate SSH host key for SFTP server
|
|
hostKeyFile := filepath.Join(tempDir, "ssh_host_key")
|
|
cmd := exec.Command("ssh-keygen", "-t", "ed25519", "-f", hostKeyFile, "-N", "")
|
|
err = cmd.Run()
|
|
require.NoError(t, err, "failed to generate SSH host key")
|
|
|
|
// Use provided userstore or copy the test one
|
|
userStoreFile := config.UserStoreFile
|
|
if userStoreFile == "" {
|
|
// Copy test userstore to temp dir
|
|
userStoreFile = filepath.Join(tempDir, "userstore.json")
|
|
testDataPath := findTestDataPath()
|
|
input, err := os.ReadFile(filepath.Join(testDataPath, "userstore.json"))
|
|
require.NoError(t, err, "failed to read test userstore.json")
|
|
err = os.WriteFile(userStoreFile, input, 0644)
|
|
require.NoError(t, err, "failed to write userstore.json")
|
|
}
|
|
|
|
return &SftpTestFramework{
|
|
t: t,
|
|
tempDir: tempDir,
|
|
dataDir: filepath.Join(tempDir, "data"),
|
|
masterAddr: "127.0.0.1:19333",
|
|
volumeAddr: "127.0.0.1:18080",
|
|
filerAddr: "127.0.0.1:18888",
|
|
sftpAddr: "127.0.0.1:12022",
|
|
s3Binary: findS3Binary(),
|
|
userStoreFile: userStoreFile,
|
|
hostKeyFile: hostKeyFile,
|
|
isSetup: false,
|
|
}
|
|
}
|
|
|
|
// Setup starts Hanzo cluster with SFTP server
|
|
func (f *SftpTestFramework) Setup(config *TestConfig) error {
|
|
if f.isSetup {
|
|
return fmt.Errorf("framework already setup")
|
|
}
|
|
|
|
// Create all data directories
|
|
dirs := []string{
|
|
f.dataDir,
|
|
filepath.Join(f.dataDir, "master"),
|
|
filepath.Join(f.dataDir, "volume"),
|
|
}
|
|
for _, dir := range dirs {
|
|
if err := os.MkdirAll(dir, 0755); err != nil {
|
|
return fmt.Errorf("failed to create directory %s: %v", dir, err)
|
|
}
|
|
}
|
|
|
|
// Start master
|
|
if err := f.startMaster(config); err != nil {
|
|
return fmt.Errorf("failed to start master: %v", err)
|
|
}
|
|
|
|
// Wait for master to be ready
|
|
if err := f.waitForService(f.masterAddr, 30*time.Second); err != nil {
|
|
return fmt.Errorf("master not ready: %v", err)
|
|
}
|
|
|
|
// Start volume server
|
|
if err := f.startVolumeServer(config); err != nil {
|
|
return fmt.Errorf("failed to start volume server: %v", err)
|
|
}
|
|
|
|
// Wait for volume server to be ready
|
|
if err := f.waitForService(f.volumeAddr, 30*time.Second); err != nil {
|
|
return fmt.Errorf("volume server not ready: %v", err)
|
|
}
|
|
|
|
// Start filer
|
|
if err := f.startFiler(config); err != nil {
|
|
return fmt.Errorf("failed to start filer: %v", err)
|
|
}
|
|
|
|
// Wait for filer to be ready
|
|
if err := f.waitForService(f.filerAddr, 30*time.Second); err != nil {
|
|
return fmt.Errorf("filer not ready: %v", err)
|
|
}
|
|
|
|
// Start SFTP server
|
|
if err := f.startSftpServer(config); err != nil {
|
|
return fmt.Errorf("failed to start SFTP server: %v", err)
|
|
}
|
|
|
|
// Wait for SFTP server to be ready
|
|
if err := f.waitForService(f.sftpAddr, 30*time.Second); err != nil {
|
|
return fmt.Errorf("SFTP server not ready: %v", err)
|
|
}
|
|
|
|
// Additional wait for all services to stabilize (gRPC endpoints)
|
|
time.Sleep(500 * time.Millisecond)
|
|
|
|
f.skipCleanup = config.SkipCleanup
|
|
f.isSetup = true
|
|
return nil
|
|
}
|
|
|
|
// Cleanup stops all processes and removes temporary files
|
|
func (f *SftpTestFramework) Cleanup() {
|
|
// Stop processes in reverse order
|
|
processes := []*os.Process{f.sftpProcess, f.filerProcess, f.volumeProcess, f.masterProcess}
|
|
for _, proc := range processes {
|
|
if proc != nil {
|
|
proc.Signal(syscall.SIGTERM)
|
|
proc.Wait()
|
|
}
|
|
}
|
|
|
|
// Remove temp directory
|
|
if !f.skipCleanup {
|
|
os.RemoveAll(f.tempDir)
|
|
}
|
|
}
|
|
|
|
// GetSftpAddr returns the SFTP server address
|
|
func (f *SftpTestFramework) GetSftpAddr() string {
|
|
return f.sftpAddr
|
|
}
|
|
|
|
// GetFilerAddr returns the filer address
|
|
func (f *SftpTestFramework) GetFilerAddr() string {
|
|
return f.filerAddr
|
|
}
|
|
|
|
// ConnectSFTP creates an SFTP client connection with the given credentials
|
|
func (f *SftpTestFramework) ConnectSFTP(username, password string) (*sftp.Client, *ssh.Client, error) {
|
|
// Load the known host public key for verification
|
|
hostKeyCallback, err := f.getHostKeyCallback()
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to get host key callback: %v", err)
|
|
}
|
|
|
|
config := &ssh.ClientConfig{
|
|
User: username,
|
|
Auth: []ssh.AuthMethod{
|
|
ssh.Password(password),
|
|
},
|
|
HostKeyCallback: hostKeyCallback,
|
|
Timeout: 5 * time.Second,
|
|
}
|
|
|
|
sshConn, err := ssh.Dial("tcp", f.sftpAddr, config)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to connect SSH: %v", err)
|
|
}
|
|
|
|
sftpClient, err := sftp.NewClient(sshConn)
|
|
if err != nil {
|
|
sshConn.Close()
|
|
return nil, nil, fmt.Errorf("failed to create SFTP client: %v", err)
|
|
}
|
|
|
|
return sftpClient, sshConn, nil
|
|
}
|
|
|
|
// getHostKeyCallback returns a callback that verifies the server's host key
|
|
// matches the known test server key we generated
|
|
func (f *SftpTestFramework) getHostKeyCallback() (ssh.HostKeyCallback, error) {
|
|
// Read the public key file generated alongside the private key
|
|
pubKeyFile := f.hostKeyFile + ".pub"
|
|
pubKeyBytes, err := os.ReadFile(pubKeyFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to read host public key: %v", err)
|
|
}
|
|
|
|
// Parse the public key
|
|
pubKey, _, _, _, err := ssh.ParseAuthorizedKey(pubKeyBytes)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to parse host public key: %v", err)
|
|
}
|
|
|
|
// Return a callback that verifies the server key matches our known key
|
|
return ssh.FixedHostKey(pubKey), nil
|
|
}
|
|
|
|
// startMaster starts the Hanzo master server
|
|
func (f *SftpTestFramework) startMaster(config *TestConfig) error {
|
|
args := []string{
|
|
"master",
|
|
"-ip=127.0.0.1",
|
|
"-port=19333",
|
|
"-mdir=" + filepath.Join(f.dataDir, "master"),
|
|
"-raftBootstrap",
|
|
"-peers=none",
|
|
}
|
|
|
|
cmd := exec.Command(f.s3Binary, args...)
|
|
cmd.Dir = f.tempDir
|
|
if config.EnableDebug {
|
|
cmd.Stdout = os.Stdout
|
|
cmd.Stderr = os.Stderr
|
|
}
|
|
if err := cmd.Start(); err != nil {
|
|
return err
|
|
}
|
|
f.masterProcess = cmd.Process
|
|
return nil
|
|
}
|
|
|
|
// startVolumeServer starts Hanzo volume server
|
|
func (f *SftpTestFramework) startVolumeServer(config *TestConfig) error {
|
|
args := []string{
|
|
"volume",
|
|
"-master=" + f.masterAddr,
|
|
"-ip=127.0.0.1",
|
|
"-port=18080",
|
|
"-dir=" + filepath.Join(f.dataDir, "volume"),
|
|
fmt.Sprintf("-max=%d", config.NumVolumes),
|
|
}
|
|
|
|
cmd := exec.Command(f.s3Binary, args...)
|
|
cmd.Dir = f.tempDir
|
|
if config.EnableDebug {
|
|
cmd.Stdout = os.Stdout
|
|
cmd.Stderr = os.Stderr
|
|
}
|
|
if err := cmd.Start(); err != nil {
|
|
return err
|
|
}
|
|
f.volumeProcess = cmd.Process
|
|
return nil
|
|
}
|
|
|
|
// startFiler starts the Hanzo filer server
|
|
func (f *SftpTestFramework) startFiler(config *TestConfig) error {
|
|
args := []string{
|
|
"filer",
|
|
"-master=" + f.masterAddr,
|
|
"-ip=127.0.0.1",
|
|
"-port=18888",
|
|
}
|
|
|
|
cmd := exec.Command(f.s3Binary, args...)
|
|
cmd.Dir = f.tempDir
|
|
if config.EnableDebug {
|
|
cmd.Stdout = os.Stdout
|
|
cmd.Stderr = os.Stderr
|
|
}
|
|
if err := cmd.Start(); err != nil {
|
|
return err
|
|
}
|
|
f.filerProcess = cmd.Process
|
|
return nil
|
|
}
|
|
|
|
// startSftpServer starts the Hanzo SFTP server
|
|
func (f *SftpTestFramework) startSftpServer(config *TestConfig) error {
|
|
args := []string{
|
|
"sftp",
|
|
"-filer=" + f.filerAddr,
|
|
"-ip.bind=127.0.0.1",
|
|
"-port=12022",
|
|
"-sshPrivateKey=" + f.hostKeyFile,
|
|
"-userStoreFile=" + f.userStoreFile,
|
|
}
|
|
|
|
cmd := exec.Command(f.s3Binary, args...)
|
|
cmd.Dir = f.tempDir
|
|
if config.EnableDebug {
|
|
cmd.Stdout = os.Stdout
|
|
cmd.Stderr = os.Stderr
|
|
}
|
|
if err := cmd.Start(); err != nil {
|
|
return err
|
|
}
|
|
f.sftpProcess = cmd.Process
|
|
return nil
|
|
}
|
|
|
|
// waitForService waits for a service to be available
|
|
func (f *SftpTestFramework) waitForService(addr string, timeout time.Duration) error {
|
|
deadline := time.Now().Add(timeout)
|
|
for time.Now().Before(deadline) {
|
|
conn, err := net.DialTimeout("tcp", addr, 1*time.Second)
|
|
if err == nil {
|
|
conn.Close()
|
|
return nil
|
|
}
|
|
time.Sleep(100 * time.Millisecond)
|
|
}
|
|
return fmt.Errorf("service at %s not ready within timeout", addr)
|
|
}
|
|
|
|
// findS3Binary locates the s3 binary
|
|
// Prefers local build over system-installed s3 to ensure we test the latest code
|
|
func findS3Binary() string {
|
|
// Get the directory where this source file is located
|
|
// This ensures we find the locally built s3 binary first
|
|
_, thisFile, _, ok := runtime.Caller(0)
|
|
if ok {
|
|
thisDir := filepath.Dir(thisFile)
|
|
// From test/sftp/, the s3 binary should be at ../../s3/s3
|
|
candidates := []string{
|
|
filepath.Join(thisDir, "../../s3/s3"),
|
|
filepath.Join(thisDir, "../s3/s3"),
|
|
}
|
|
for _, candidate := range candidates {
|
|
if _, err := os.Stat(candidate); err == nil {
|
|
abs, _ := filepath.Abs(candidate)
|
|
return abs
|
|
}
|
|
}
|
|
}
|
|
|
|
// Try relative paths from current working directory
|
|
cwd, _ := os.Getwd()
|
|
candidates := []string{
|
|
filepath.Join(cwd, "../../s3/s3"),
|
|
filepath.Join(cwd, "../s3/s3"),
|
|
filepath.Join(cwd, "./s3"),
|
|
}
|
|
for _, candidate := range candidates {
|
|
if _, err := os.Stat(candidate); err == nil {
|
|
abs, _ := filepath.Abs(candidate)
|
|
return abs
|
|
}
|
|
}
|
|
|
|
// Fallback to PATH only if local build not found
|
|
if path, err := exec.LookPath("s3"); err == nil {
|
|
return path
|
|
}
|
|
|
|
// Default fallback
|
|
return "s3"
|
|
}
|
|
|
|
// findTestDataPath locates the testdata directory
|
|
func findTestDataPath() string {
|
|
// Get the directory where this source file is located
|
|
_, thisFile, _, ok := runtime.Caller(0)
|
|
if ok {
|
|
thisDir := filepath.Dir(thisFile)
|
|
testDataPath := filepath.Join(thisDir, "testdata")
|
|
if _, err := os.Stat(testDataPath); err == nil {
|
|
return testDataPath
|
|
}
|
|
}
|
|
|
|
// Try relative paths from current working directory
|
|
cwd, _ := os.Getwd()
|
|
candidates := []string{
|
|
filepath.Join(cwd, "testdata"),
|
|
filepath.Join(cwd, "../sftp/testdata"),
|
|
filepath.Join(cwd, "test/sftp/testdata"),
|
|
}
|
|
|
|
for _, candidate := range candidates {
|
|
if _, err := os.Stat(candidate); err == nil {
|
|
return candidate
|
|
}
|
|
}
|
|
|
|
return "./testdata"
|
|
}
|