internal/store/postgres/blob_storage_integration_test.go

Rama main · réplica de lectura del repositorio original.

package postgres

import (
	"context"
	"os"
	"testing"
	"time"

	"telesrv/internal/domain"
)

func TestBlobStorageAdvisoryLock(t *testing.T) {
	dsn := os.Getenv("TELESRV_TEST_POSTGRES_DSN")
	if dsn == "" {
		t.Skip("set TELESRV_TEST_POSTGRES_DSN to run postgres integration test")
	}
	ctx := context.Background()
	runtimeLock, err := AcquireBlobRuntimeLock(ctx, dsn)
	if err != nil {
		t.Fatalf("acquire runtime lock: %v", err)
	}
	if _, err := AcquireBlobMigrationLock(ctx, dsn); err == nil {
		t.Fatal("exclusive migration lock acquired while runtime shared lock was held")
	}
	if err := runtimeLock.Close(); err != nil {
		t.Fatalf("close runtime lock: %v", err)
	}
	migrationLock, err := AcquireBlobMigrationLock(ctx, dsn)
	if err != nil {
		t.Fatalf("acquire migration lock after runtime stopped: %v", err)
	}
	if err := migrationLock.Close(); err != nil {
		t.Fatalf("close migration lock: %v", err)
	}
}

func TestBlobMigrationMetadataRoundTrip(t *testing.T) {
	pool := testPool(t)
	ctx := context.Background()
	media := NewMediaStore(pool)
	uniqueBefore, err := media.UniqueFileBlobBytes(ctx, domain.MediaBackendLocalFS)
	if err != nil {
		t.Fatalf("unique blob bytes before insert: %v", err)
	}
	suffix := time.Now().UnixNano()
	first := postgresTestBlob("blob-migration:first:"+time.Unix(0, suffix).Format("150405.000000000"), "shared-migration", 4096, "application/octet-stream")
	second := first
	second.LocationKey = "blob-migration:second:" + time.Unix(0, suffix).Format("150405.000000000")
	for _, blob := range []domain.FileBlob{first, second} {
		if err := media.PutFileBlob(ctx, blob); err != nil {
			t.Fatalf("put %s: %v", blob.LocationKey, err)
		}
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), "DELETE FROM file_blobs WHERE location_key = ANY($1::text[])", []string{first.LocationKey, second.LocationKey})
	})

	counts, err := media.FileBlobBackendCounts(ctx)
	if err != nil || counts[domain.MediaBackendLocalFS] < 2 {
		t.Fatalf("backend counts=%v err=%v", counts, err)
	}
	uniqueBytes, err := media.UniqueFileBlobBytes(ctx, domain.MediaBackendLocalFS)
	if err != nil {
		t.Fatalf("unique blob bytes: %v", err)
	}
	if uniqueBytes-uniqueBefore != first.Size {
		t.Fatalf("unique blob byte delta=%d, want shared object counted once as %d", uniqueBytes-uniqueBefore, first.Size)
	}
	objects, err := media.ListBlobMigrationObjects(ctx, domain.MediaBackendLocalFS, first.ObjectKey[:len(first.ObjectKey)-1], 10)
	if err != nil {
		t.Fatalf("list migration objects: %v", err)
	}
	var found *BlobMigrationObject
	for i := range objects {
		if objects[i].ObjectKey == first.ObjectKey {
			found = &objects[i]
			break
		}
	}
	if found == nil || found.LocationRows != 2 || found.Size != first.Size {
		t.Fatalf("migration object=%+v", found)
	}
	if err := media.MoveFileBlobBackendForObject(
		ctx, domain.MediaBackendLocalFS, domain.MediaBackendS3,
		found.ObjectKey, found.LocationRows,
	); err != nil {
		t.Fatalf("move backend: %v", err)
	}
	for _, key := range []string{first.LocationKey, second.LocationKey} {
		blob, ok, err := media.GetFileBlob(ctx, key)
		if err != nil || !ok || blob.Backend != domain.MediaBackendS3 {
			t.Fatalf("get %s backend=%q ok=%v err=%v", key, blob.Backend, ok, err)
		}
	}
}