kra-new/internal/integration/storage/compose_test.go

package storage

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"errors"
	"io"
	"kra/internal/biz/system"
	"strings"
	"testing"
	"time"
)

func TestComposeStreams(t *testing.T) {
	objects := map[string]string{"a": "hello ", "b": "world"}
	var stored string
	hash, err := composeStreams(context.Background(), []string{"a", "b"}, func(_ context.Context, name string) (io.ReadCloser, error) {
		value, ok := objects[name]
		if !ok {
			return nil, errors.New("missing object")
		}
		return io.NopCloser(strings.NewReader(value)), nil
	}, func(_ context.Context, _ string, reader io.Reader) error {
		value, err := io.ReadAll(reader)
		stored = string(value)
		return err
	}, func(context.Context, string) error { return nil }, "out")
	if err != nil || stored != "hello world" {
		t.Fatalf("compose = %q, %v", stored, err)
	}
	want := md5.Sum([]byte(stored))
	if hash != hex.EncodeToString(want[:]) {
		t.Fatalf("hash = %q", hash)
	}
}

func TestComposeStreamsRemovesPartialDestination(t *testing.T) {
	removed := false
	_, err := composeStreams(context.Background(), []string{"missing"}, func(context.Context, string) (io.ReadCloser, error) {
		return nil, errors.New("open failed")
	}, func(_ context.Context, _ string, reader io.Reader) error {
		_, _ = io.ReadAll(reader)
		return nil
	}, func(context.Context, string) error {
		removed = true
		return nil
	}, "out")
	if err == nil || !removed {
		t.Fatalf("compose error = %v, removed = %v", err, removed)
	}
}

func TestComposeStreamsUnblocksProducerWhenDestinationFails(t *testing.T) {
	putErr := errors.New("destination failed")
	done := make(chan error, 1)
	go func() {
		_, err := composeStreams(context.Background(), []string{"large"}, func(context.Context, string) (io.ReadCloser, error) {
			return io.NopCloser(strings.NewReader(strings.Repeat("x", 1<<20))), nil
		}, func(context.Context, string, io.Reader) error {
			return putErr
		}, func(context.Context, string) error { return nil }, "out")
		done <- err
	}()

	select {
	case err := <-done:
		if !errors.Is(err, putErr) {
			t.Fatalf("compose error = %v, want %v", err, putErr)
		}
	case <-time.After(time.Second):
		t.Fatal("compose remained blocked after destination failure")
	}
}

func TestNormalizeDeletePrefix(t *testing.T) {
	tests := []struct {
		name    string
		value   string
		valid   string
		wantErr bool
	}{
		{name: "canonicalizes", value: "/uploads/chunks/", valid: "uploads/chunks"},
		{name: "cleans duplicate separators", value: "uploads//chunks", valid: "uploads/chunks"},
		{name: "rejects empty", value: " / ", wantErr: true},
		{name: "rejects traversal", value: "uploads/../", wantErr: true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := normalizeDeletePrefix(tt.value)
			if tt.wantErr {
				if err == nil {
					t.Fatalf("normalizeDeletePrefix(%q) succeeded with %q", tt.value, got)
				}
				return
			}
			if err != nil || got != tt.valid {
				t.Fatalf("normalizeDeletePrefix(%q) = %q, %v; want %q", tt.value, got, err, tt.valid)
			}
		})
	}
}

func TestBoundedPrefixKeepsDirectoryBoundary(t *testing.T) {
	if got := boundedPrefix("uploads/chunks/1", "uploads/chunks/1/"); got != "uploads/chunks/1/" {
		t.Fatalf("boundedPrefix() = %q", got)
	}
	if got := boundedPrefix("uploads/chunks/1", "uploads/chunks/1"); got != "uploads/chunks/1" {
		t.Fatalf("boundedPrefix() changed ordinary prefix to %q", got)
	}
}

func TestAdvanceDeletePrefixCursor(t *testing.T) {
	if _, err := advanceDeletePrefixCursor("cursor", "cursor", true); err == nil {
		t.Fatal("same cursor should fail")
	}
	if next, err := advanceDeletePrefixCursor("", "next", true); err != nil || next != "next" {
		t.Fatalf("advance cursor = %q, %v", next, err)
	}
}

func TestDeletePrefixViaListDeletesAllPages(t *testing.T) {
	pages := map[string][]*system.StoredFile{
		"":     {{Path: "uploads/a"}, {Path: "uploads/b"}},
		"next": {{Path: "uploads/c"}},
	}
	removed := []string{}
	if err := deletePrefixViaList(context.Background(), "uploads", func(_ context.Context, prefix, cursor string, limit int) ([]*system.StoredFile, string, bool, error) {
		if prefix != "uploads/" || limit != 1000 {
			t.Fatalf("list arguments = %q, %q, %d", prefix, cursor, limit)
		}
		items := pages[cursor]
		if cursor == "" {
			return items, "next", true, nil
		}
		return items, "", false, nil
	}, func(_ context.Context, path string) error {
		removed = append(removed, path)
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if len(removed) != 3 || removed[2] != "uploads/c" {
		t.Fatalf("removed paths = %#v", removed)
	}
}

func TestDeletePrefixViaListRejectsStalledCursor(t *testing.T) {
	err := deletePrefixViaList(context.Background(), "uploads", func(_ context.Context, _, _ string, _ int) ([]*system.StoredFile, string, bool, error) {
		return nil, "", true, nil
	}, func(context.Context, string) error { return nil })
	if err == nil || !strings.Contains(err.Error(), "no progress") {
		t.Fatalf("stalled pagination error = %v", err)
	}
}