distributor/internal/app/backends_test.go

package app

import (
	"bytes"
	"context"
	"fmt"
	"strings"
	"testing"

	"gitea.maximumdirect.net/eric/distributor/internal/config"
	"gitea.maximumdirect.net/eric/distributor/internal/storage"
	"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
)

func TestBackendFactoryOpensLocalSource(t *testing.T) {
	factory := newBackendFactory()
	backend, err := factory.openSource(context.Background(), config.Backend{
		Backend: config.BackendLocal,
		Path:    t.TempDir(),
	})
	if err != nil {
		t.Fatalf("openSource() error = %v", err)
	}
	if backend == nil {
		t.Fatal("openSource() backend = nil")
	}
}

func TestBackendFactoryOpensLocalDestination(t *testing.T) {
	factory := newBackendFactory()
	backend, err := factory.openDestination(context.Background(), config.Destination{
		Backend: config.BackendLocal,
		Path:    t.TempDir(),
	})
	if err != nil {
		t.Fatalf("openDestination() error = %v", err)
	}
	if backend == nil {
		t.Fatal("openDestination() backend = nil")
	}
}

func TestBackendFactoryOpensDirectLocalPath(t *testing.T) {
	factory := newBackendFactory()
	backend, err := factory.openLocalPath(context.Background(), t.TempDir())
	if err != nil {
		t.Fatalf("openLocalPath() error = %v", err)
	}
	if backend == nil {
		t.Fatal("openLocalPath() backend = nil")
	}
}

func TestBackendFactoryOpensSSHSourceWithRegisteredOpener(t *testing.T) {
	factory := &backendFactory{registry: storage.NewRegistry()}
	var got storage.OpenConfig
	if err := factory.registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
		got = cfg
		return fake.New(), nil
	}); err != nil {
		t.Fatalf("Register() error = %v", err)
	}

	backend, err := factory.openSource(context.Background(), config.Backend{
		Backend: config.BackendSSH,
		Host:    "source.example.com",
		User:    "reports",
		Port:    22,
		Path:    "/reports",
		SSH:     config.SSH{HostKeyPolicy: config.HostKeyPolicyAcceptNew},
	})
	if err != nil {
		t.Fatalf("openSource() error = %v", err)
	}
	if backend == nil {
		t.Fatal("openSource() backend = nil")
	}
	if got[sshHostKey] != "source.example.com" || got[storagePathKey] != "/reports" {
		t.Fatalf("open config = %#v, want SSH source fields", got)
	}
}

func TestBackendFactoryOpensSSHDestinationWithRegisteredOpener(t *testing.T) {
	factory := &backendFactory{registry: storage.NewRegistry()}
	var got storage.OpenConfig
	if err := factory.registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
		got = cfg
		return fake.New(), nil
	}); err != nil {
		t.Fatalf("Register() error = %v", err)
	}

	backend, err := factory.openDestination(context.Background(), config.Destination{
		Backend: config.BackendSSH,
		Host:    "destination.example.com",
		User:    "deploy",
		Port:    2222,
		Path:    "/archive",
		SSH:     config.SSH{HostKeyPolicy: config.HostKeyPolicyStrict},
	})
	if err != nil {
		t.Fatalf("openDestination() error = %v", err)
	}
	if backend == nil {
		t.Fatal("openDestination() backend = nil")
	}
	if got[sshHostKey] != "destination.example.com" || got[sshPortKey] != "2222" || got[sshHostKeyPolicyKey] != "strict" {
		t.Fatalf("open config = %#v, want SSH destination fields", got)
	}
}

func TestBackendFactorySetsReadOnlyKnownHostsForDryRunSSH(t *testing.T) {
	factory := &backendFactory{
		registry:           storage.NewRegistry(),
		readOnlyKnownHosts: true,
	}
	var got storage.OpenConfig
	if err := factory.registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
		got = cfg
		return fake.New(), nil
	}); err != nil {
		t.Fatalf("Register() error = %v", err)
	}

	_, err := factory.openDestination(context.Background(), config.Destination{
		Backend: config.BackendSSH,
		Host:    "destination.example.com",
		User:    "deploy",
		Port:    22,
		Path:    "/archive",
		SSH:     config.SSH{HostKeyPolicy: config.HostKeyPolicyAcceptNew},
	})
	if err != nil {
		t.Fatalf("openDestination() error = %v", err)
	}
	if got[sshReadOnlyHostsKey] != "true" {
		t.Fatalf("open config %s = %q, want true", sshReadOnlyHostsKey, got[sshReadOnlyHostsKey])
	}
}

func TestBackendFactoryUsesPersistentKnownHostsByDefault(t *testing.T) {
	factory := &backendFactory{registry: storage.NewRegistry()}
	var got storage.OpenConfig
	if err := factory.registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
		got = cfg
		return fake.New(), nil
	}); err != nil {
		t.Fatalf("Register() error = %v", err)
	}

	_, err := factory.openDestination(context.Background(), config.Destination{
		Backend: config.BackendSSH,
		Host:    "destination.example.com",
		User:    "deploy",
		Port:    22,
		Path:    "/archive",
		SSH:     config.SSH{HostKeyPolicy: config.HostKeyPolicyAcceptNew},
	})
	if err != nil {
		t.Fatalf("openDestination() error = %v", err)
	}
	if got[sshReadOnlyHostsKey] != "false" {
		t.Fatalf("open config %s = %q, want false", sshReadOnlyHostsKey, got[sshReadOnlyHostsKey])
	}
}

func TestBackendFactoryRejectsUnsupportedSource(t *testing.T) {
	factory := newBackendFactory()
	_, err := factory.openSource(context.Background(), config.Backend{
		Backend: "ftp",
	})
	if err == nil || !strings.Contains(err.Error(), "source backend ftp is not implemented for execution") {
		t.Fatalf("openSource() error = %v, want not implemented", err)
	}
}

func TestBackendFactoryRejectsUnsupportedDestination(t *testing.T) {
	factory := newBackendFactory()
	_, err := factory.openDestination(context.Background(), config.Destination{
		Backend: "ftp",
	})
	if err == nil || !strings.Contains(err.Error(), "backend ftp is not implemented for execution") {
		t.Fatalf("openDestination() error = %v, want not implemented", err)
	}
}

func TestBackendFactoryOpensS3DestinationWithRegisteredOpener(t *testing.T) {
	factory := &backendFactory{
		registry:    storage.NewRegistry(),
		environment: config.NewEnvironment(nil, func(string) (string, bool) { return "", false }),
	}
	var got storage.OpenConfig
	if err := factory.registry.Register(config.BackendS3, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
		got = cfg
		return fake.New(), nil
	}); err != nil {
		t.Fatalf("Register() error = %v", err)
	}
	forcePathStyle := false
	backend, err := factory.openDestination(context.Background(), config.Destination{
		Backend:   config.BackendS3,
		Endpoint:  "https://s3.example.com",
		Bucket:    "reports",
		Prefix:    "archive",
		Region:    config.DefaultS3Region,
		ForcePath: &forcePathStyle,
	})
	if err != nil {
		t.Fatalf("openDestination() error = %v", err)
	}
	if backend == nil {
		t.Fatal("openDestination() backend = nil")
	}
	assertOpenConfig(t, got, map[string]string{
		s3EndpointKey:       "https://s3.example.com",
		s3BucketKey:         "reports",
		s3PrefixKey:         "archive",
		s3RegionKey:         config.DefaultS3Region,
		s3ForcePathStyleKey: "false",
	})
}

func TestBackendFactoryResolvesS3CredentialsThroughSecretsAwareEnvironment(t *testing.T) {
	factory := &backendFactory{
		registry: storage.NewRegistry(),
		environment: config.NewEnvironment(map[string]string{
			"ACCESS_KEY_ID":     "secret-access",
			"SECRET_ACCESS_KEY": "secret-secret",
		}, func(string) (string, bool) { return "", false }),
	}
	var got storage.OpenConfig
	if err := factory.registry.Register(config.BackendS3, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
		got = cfg
		return fake.New(), nil
	}); err != nil {
		t.Fatalf("Register() error = %v", err)
	}
	forcePathStyle := true
	_, err := factory.openSource(context.Background(), config.Backend{
		Backend:   config.BackendS3,
		Endpoint:  "https://s3.example.com",
		Bucket:    "reports",
		Region:    config.DefaultS3Region,
		ForcePath: &forcePathStyle,
		Creds: config.Credentials{
			AccessKeyIDEnv:     "ACCESS_KEY_ID",
			SecretAccessKeyEnv: "SECRET_ACCESS_KEY",
		},
	})
	if err != nil {
		t.Fatalf("openSource() error = %v", err)
	}
	assertOpenConfig(t, got, map[string]string{
		s3AccessKeyIDKey:  "secret-access",
		s3SecretAccessKey: "secret-secret",
	})
}

func TestBackendFactoryResolvesCredentialsThroughEnvironment(t *testing.T) {
	factory := newBackendFactoryWithEnvironment(config.NewEnvironment(map[string]string{
		"ACCESS_KEY_ID":     "secret-access",
		"SECRET_ACCESS_KEY": "secret-secret",
	}, func(string) (string, bool) {
		return "", false
	}))

	creds, err := factory.resolveCredentials(config.Credentials{
		AccessKeyIDEnv:     "ACCESS_KEY_ID",
		SecretAccessKeyEnv: "SECRET_ACCESS_KEY",
	})
	if err != nil {
		t.Fatalf("resolveCredentials() error = %v", err)
	}
	if creds.AccessKeyID != "secret-access" || creds.SecretAccessKey != "secret-secret" {
		t.Fatalf("resolved credentials = %#v", creds)
	}
}

func TestBackendFactoryBuildsSSHSourceOpenConfig(t *testing.T) {
	factory := &backendFactory{environment: config.NewEnvironment(nil, nil)}
	cfg, err := factory.openConfig(backendOpenSpecFromSource(config.Backend{
		Backend: config.BackendSSH,
		Host:    "source.example.com",
		User:    "reports",
		Port:    2222,
		Path:    "/reports",
		SSH: config.SSH{
			KeyFile:       "/home/reports/.ssh/id_ed25519",
			KnownHosts:    "/home/reports/.ssh/known_hosts",
			HostKeyPolicy: config.HostKeyPolicyStrict,
		},
	}))
	if err != nil {
		t.Fatalf("openConfig() error = %v", err)
	}

	assertOpenConfig(t, cfg, map[string]string{
		storagePathKey:      "/reports",
		sshHostKey:          "source.example.com",
		sshUserKey:          "reports",
		sshPortKey:          "2222",
		sshKeyFileKey:       "/home/reports/.ssh/id_ed25519",
		sshKnownHostsKey:    "/home/reports/.ssh/known_hosts",
		sshHostKeyPolicyKey: "strict",
	})
}

func TestBackendFactoryBuildsSSHDestinationOpenConfig(t *testing.T) {
	factory := &backendFactory{environment: config.NewEnvironment(nil, nil)}
	cfg, err := factory.openConfig(backendOpenSpecFromDestination(config.Destination{
		Backend: config.BackendSSH,
		Host:    "destination.example.com",
		User:    "deploy",
		Port:    22,
		Path:    "/srv/archive",
		SSH: config.SSH{
			HostKeyPolicy: config.HostKeyPolicyAcceptNew,
		},
	}))
	if err != nil {
		t.Fatalf("openConfig() error = %v", err)
	}

	assertOpenConfig(t, cfg, map[string]string{
		storagePathKey:      "/srv/archive",
		sshHostKey:          "destination.example.com",
		sshUserKey:          "deploy",
		sshPortKey:          "22",
		sshHostKeyPolicyKey: "accept-new",
	})
}

func TestBackendFactoryBuildsEquivalentSourceAndDestinationOpenConfig(t *testing.T) {
	forcePathStyle := false
	tests := []struct {
		name        string
		source      config.Backend
		destination config.Destination
	}{
		{
			name:        "local",
			source:      config.Backend{Backend: config.BackendLocal, Path: "/reports"},
			destination: config.Destination{Backend: config.BackendLocal, Path: "/reports"},
		},
		{
			name: "ssh",
			source: config.Backend{
				Backend: config.BackendSSH,
				Host:    "reports.example.com",
				User:    "reports",
				Port:    2222,
				Path:    "/reports",
				SSH: config.SSH{
					KeyFile:       "/home/reports/.ssh/id_ed25519",
					KnownHosts:    "/home/reports/.ssh/known_hosts",
					HostKeyPolicy: config.HostKeyPolicyStrict,
				},
			},
			destination: config.Destination{
				Backend: config.BackendSSH,
				Host:    "reports.example.com",
				User:    "reports",
				Port:    2222,
				Path:    "/reports",
				SSH: config.SSH{
					KeyFile:       "/home/reports/.ssh/id_ed25519",
					KnownHosts:    "/home/reports/.ssh/known_hosts",
					HostKeyPolicy: config.HostKeyPolicyStrict,
				},
			},
		},
		{
			name: "s3",
			source: config.Backend{
				Backend:   config.BackendS3,
				Endpoint:  "https://s3.example.com",
				Bucket:    "reports",
				Prefix:    "archive",
				Region:    "us-west-2",
				ForcePath: &forcePathStyle,
			},
			destination: config.Destination{
				Backend:   config.BackendS3,
				Endpoint:  "https://s3.example.com",
				Bucket:    "reports",
				Prefix:    "archive",
				Region:    "us-west-2",
				ForcePath: &forcePathStyle,
			},
		},
		{
			name: "s3 explicit credentials",
			source: config.Backend{
				Backend:  config.BackendS3,
				Endpoint: "https://s3.example.com",
				Bucket:   "reports",
				Region:   config.DefaultS3Region,
				Creds: config.Credentials{
					AccessKeyIDEnv:     "ACCESS_KEY_ID",
					SecretAccessKeyEnv: "SECRET_ACCESS_KEY",
				},
			},
			destination: config.Destination{
				Backend:  config.BackendS3,
				Endpoint: "https://s3.example.com",
				Bucket:   "reports",
				Region:   config.DefaultS3Region,
				Creds: config.Credentials{
					AccessKeyIDEnv:     "ACCESS_KEY_ID",
					SecretAccessKeyEnv: "SECRET_ACCESS_KEY",
				},
			},
		},
	}

	factory := &backendFactory{
		environment: config.NewEnvironment(map[string]string{
			"ACCESS_KEY_ID":     "secret-access",
			"SECRET_ACCESS_KEY": "secret-secret",
		}, func(string) (string, bool) { return "", false }),
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sourceConfig, err := factory.openConfig(backendOpenSpecFromSource(tt.source))
			if err != nil {
				t.Fatalf("source openConfig() error = %v", err)
			}
			destinationConfig, err := factory.openConfig(backendOpenSpecFromDestination(tt.destination))
			if err != nil {
				t.Fatalf("destination openConfig() error = %v", err)
			}
			if !openConfigEqual(sourceConfig, destinationConfig) {
				t.Fatalf("source open config = %#v, destination open config = %#v, want equivalent", sourceConfig, destinationConfig)
			}
		})
	}
}

func TestBackendFactoryBuildsEquivalentDryRunSSHOpenConfig(t *testing.T) {
	factory := &backendFactory{readOnlyKnownHosts: true}
	sourceConfig, err := factory.openConfig(backendOpenSpecFromSource(config.Backend{
		Backend: config.BackendSSH,
		Host:    "reports.example.com",
		Port:    22,
		Path:    "/reports",
		SSH:     config.SSH{HostKeyPolicy: config.HostKeyPolicyAcceptNew},
	}))
	if err != nil {
		t.Fatalf("source openConfig() error = %v", err)
	}
	destinationConfig, err := factory.openConfig(backendOpenSpecFromDestination(config.Destination{
		Backend: config.BackendSSH,
		Host:    "reports.example.com",
		Port:    22,
		Path:    "/reports",
		SSH:     config.SSH{HostKeyPolicy: config.HostKeyPolicyAcceptNew},
	}))
	if err != nil {
		t.Fatalf("destination openConfig() error = %v", err)
	}
	if !openConfigEqual(sourceConfig, destinationConfig) {
		t.Fatalf("source open config = %#v, destination open config = %#v, want equivalent", sourceConfig, destinationConfig)
	}
	if sourceConfig[sshReadOnlyHostsKey] != "true" {
		t.Fatalf("open config %s = %q, want true", sshReadOnlyHostsKey, sourceConfig[sshReadOnlyHostsKey])
	}
}

func TestConfiguredSourceValidationAndRunUseEquivalentSourceOpenConfig(t *testing.T) {
	tests := []struct {
		name       string
		source     config.Backend
		sourceKey  string
		dest       config.Destination
		destKey    string
		wantFields map[string]string
	}{
		{
			name: "s3",
			source: config.Backend{
				Backend:  config.BackendS3,
				Endpoint: "https://s3.example.com",
				Bucket:   "source-bucket",
				Prefix:   "source-prefix",
				Region:   config.DefaultS3Region,
			},
			sourceKey: "s3:source-bucket",
			dest: config.Destination{
				ID:       "archive",
				Backend:  config.BackendS3,
				Endpoint: "https://s3.example.com",
				Bucket:   "destination-bucket",
				Region:   config.DefaultS3Region,
			},
			destKey: "s3:destination-bucket",
			wantFields: map[string]string{
				s3EndpointKey:       "https://s3.example.com",
				s3BucketKey:         "source-bucket",
				s3PrefixKey:         "source-prefix",
				s3RegionKey:         config.DefaultS3Region,
				s3ForcePathStyleKey: "true",
			},
		},
		{
			name: "ssh",
			source: config.Backend{
				Backend: config.BackendSSH,
				Host:    "ssh.example.com",
				User:    "reports",
				Port:    2222,
				Path:    "/source",
				SSH: config.SSH{
					KeyFile:       "/home/reports/.ssh/id_ed25519",
					KnownHosts:    "/home/reports/.ssh/known_hosts",
					HostKeyPolicy: config.HostKeyPolicyStrict,
				},
			},
			sourceKey: "ssh:/source",
			dest: config.Destination{
				ID:      "archive",
				Backend: config.BackendSSH,
				Host:    "ssh.example.com",
				Port:    2222,
				Path:    "/destination",
				SSH:     config.SSH{HostKeyPolicy: config.HostKeyPolicyStrict},
			},
			destKey: "ssh:/destination",
			wantFields: map[string]string{
				storagePathKey:      "/source",
				sshHostKey:          "ssh.example.com",
				sshUserKey:          "reports",
				sshPortKey:          "2222",
				sshKeyFileKey:       "/home/reports/.ssh/id_ed25519",
				sshKnownHostsKey:    "/home/reports/.ssh/known_hosts",
				sshHostKeyPolicyKey: "strict",
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sourceBackend := fake.New()
			testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{ID: "reports.source"})
			destinationBackend := fake.New()
			var validateSourceConfig storage.OpenConfig
			var runSourceConfig storage.OpenConfig
			validateProvider := recordingBackendFactoryProvider(t, map[string]storage.Backend{
				tt.sourceKey: sourceBackend,
				tt.destKey:   destinationBackend,
			}, func(cfg storage.OpenConfig) {
				validateSourceConfig = cfg
			})
			runProvider := recordingBackendFactoryProvider(t, map[string]storage.Backend{
				tt.sourceKey: sourceBackend,
				tt.destKey:   destinationBackend,
			}, func(cfg storage.OpenConfig) {
				runSourceConfig = cfg
			})
			cfg := config.Config{Pipelines: []config.Pipeline{{
				ID:           "reports",
				Source:       tt.source,
				Destinations: []config.Destination{tt.dest},
			}}}
			config.ApplyDefaults(&cfg)

			var validateOutput bytes.Buffer
			if err := validateConfigWithBackendFactory(context.Background(), cfg, ValidateOptions{
				PipelineID: "reports",
				Stdout:     &validateOutput,
			}, validateProvider); err != nil {
				t.Fatalf("validateConfigWithBackendFactory() error = %v", err)
			}
			if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{}, runProvider); err != nil {
				t.Fatalf("runConfigWithBackendFactory() error = %v", err)
			}
			if !openConfigEqual(validateSourceConfig, runSourceConfig) {
				t.Fatalf("validate source config = %#v, run source config = %#v, want equivalent", validateSourceConfig, runSourceConfig)
			}
			assertOpenConfig(t, runSourceConfig, tt.wantFields)
		})
	}
}

func assertOpenConfig(t *testing.T, got map[string]string, want map[string]string) {
	t.Helper()
	for key, wantValue := range want {
		if gotValue := got[key]; gotValue != wantValue {
			t.Fatalf("open config %s = %q, want %q", key, gotValue, wantValue)
		}
	}
}

func openConfigEqual(left, right storage.OpenConfig) bool {
	if len(left) != len(right) {
		return false
	}
	for key, leftValue := range left {
		if right[key] != leftValue {
			return false
		}
	}
	return true
}

func recordingBackendFactoryProvider(t *testing.T, remoteBackends map[string]storage.Backend, recordSource func(storage.OpenConfig)) backendFactoryProvider {
	t.Helper()
	return func(environment config.Environment) *backendFactory {
		registry := storage.NewRegistry()
		if err := registry.Register(config.BackendS3, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
			if cfg[s3BucketKey] == "source-bucket" {
				recordSource(cfg)
			}
			key := "s3:" + cfg[s3BucketKey]
			backend := remoteBackends[key]
			if backend == nil {
				return nil, fmt.Errorf("missing fake backend for %s", key)
			}
			return backend, nil
		}); err != nil {
			t.Fatalf("register s3 backend: %v", err)
		}
		if err := registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
			if cfg[storagePathKey] == "/source" {
				recordSource(cfg)
			}
			key := "ssh:" + cfg[storagePathKey]
			backend := remoteBackends[key]
			if backend == nil {
				return nil, fmt.Errorf("missing fake backend for %s", key)
			}
			return backend, nil
		}); err != nil {
			t.Fatalf("register ssh backend: %v", err)
		}
		return &backendFactory{registry: registry, environment: environment}
	}
}