services/mgmt/binary/impl/impl_test.go

package impl_test

import (
	"bytes"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"net"
	"net/http"
	"reflect"
	"testing"

	"v.io/core/veyron2"
	"v.io/core/veyron2/context"
	"v.io/core/veyron2/naming"
	"v.io/core/veyron2/services/mgmt/repository"
	verror "v.io/core/veyron2/verror2"
	"v.io/core/veyron2/vlog"

	"v.io/core/veyron/lib/testutil"
	_ "v.io/core/veyron/profiles/static"
	"v.io/core/veyron/services/mgmt/binary/impl"
	mgmttest "v.io/core/veyron/services/mgmt/lib/testutil"
)

const (
	veyronPrefix = "veyron_binary_repository"
)

var gctx *context.T
var globalCancel context.CancelFunc

// startServer starts the binary repository server.
func startServer(t *testing.T, depth int) (repository.BinaryClientMethods, string, string, func()) {
	// Setup the root of the binary repository.
	rootDir, cleanup := mgmttest.SetupRootDir(t, "bindir")
	prepDirectory(t, rootDir)

	// Setup and start the binary repository server.
	server, endpoint := mgmttest.NewServer(gctx)

	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	state, err := impl.NewState(rootDir, listener.Addr().String(), depth)
	if err != nil {
		t.Fatalf("NewState(%v, %v) failed: %v", rootDir, listener.Addr().String(), depth, err)
	}
	go func() {
		if err := http.Serve(listener, http.FileServer(impl.NewHTTPRoot(state))); err != nil {
			vlog.Fatalf("Serve() failed: %v", err)
		}
	}()
	dispatcher, err := impl.NewDispatcher(veyron2.GetPrincipal(gctx), state)
	if err != nil {
		t.Fatalf("NewDispatcher failed: %v", err)
	}
	dontPublishName := ""
	if err := server.ServeDispatcher(dontPublishName, dispatcher); err != nil {
		t.Fatalf("Serve(%q) failed: %v", dontPublishName, err)
	}
	name := naming.JoinAddressName(endpoint, "test")
	binary := repository.BinaryClient(name)
	return binary, endpoint, fmt.Sprintf("http://%s/test", listener.Addr()), func() {
		// Shutdown the binary repository server.
		if err := server.Stop(); err != nil {
			t.Fatalf("Stop() failed: %v", err)
		}
		cleanup()
	}
}

// TestHierarchy checks that the binary repository works correctly for
// all possible valid values of the depth used for the directory
// hierarchy that stores binary objects in the local file system.
func TestHierarchy(t *testing.T) {
	for i := 0; i < md5.Size; i++ {
		binary, ep, _, cleanup := startServer(t, i)
		defer cleanup()
		data := testData()

		// Test the binary repository interface.
		if err := binary.Create(gctx, 1, repository.MediaInfo{Type: "application/octet-stream"}); err != nil {
			t.Fatalf("Create() failed: %v", err)
		}
		if streamErr, err := invokeUpload(t, gctx, binary, data, 0); streamErr != nil || err != nil {
			t.FailNow()
		}
		parts, _, err := binary.Stat(gctx)
		if err != nil {
			t.Fatalf("Stat() failed: %v", err)
		}
		h := md5.New()
		h.Write(data)
		checksum := hex.EncodeToString(h.Sum(nil))
		if expected, got := checksum, parts[0].Checksum; expected != got {
			t.Fatalf("Unexpected checksum: expected %v, got %v", expected, got)
		}
		if expected, got := len(data), int(parts[0].Size); expected != got {
			t.Fatalf("Unexpected size: expected %v, got %v", expected, got)
		}
		output, streamErr, err := invokeDownload(t, gctx, binary, 0)
		if streamErr != nil || err != nil {
			t.FailNow()
		}
		if bytes.Compare(output, data) != 0 {
			t.Fatalf("Unexpected output: expected %v, got %v", data, output)
		}
		results, err := testutil.GlobName(gctx, naming.JoinAddressName(ep, ""), "...")
		if err != nil {
			t.Fatalf("GlobName failed: %v", err)
		}
		if expected := []string{"", "test"}; !reflect.DeepEqual(results, expected) {
			t.Errorf("Unexpected results: expected %q, got %q", expected, results)
		}
		if err := binary.Delete(gctx); err != nil {
			t.Fatalf("Delete() failed: %v", err)
		}
	}
}

// TestMultiPart checks that the binary repository supports multi-part
// uploads and downloads ranging the number of parts the test binary
// consists of.
func TestMultiPart(t *testing.T) {
	for length := 2; length < 5; length++ {
		binary, _, _, cleanup := startServer(t, 2)
		defer cleanup()
		// Create <length> chunks of up to 4MB of random bytes.
		data := make([][]byte, length)
		for i := 0; i < length; i++ {
			data[i] = testData()
		}
		// Test the binary repository interface.
		if err := binary.Create(gctx, int32(length), repository.MediaInfo{Type: "application/octet-stream"}); err != nil {
			t.Fatalf("Create() failed: %v", err)
		}
		for i := 0; i < length; i++ {
			if streamErr, err := invokeUpload(t, gctx, binary, data[i], int32(i)); streamErr != nil || err != nil {
				t.FailNow()
			}
		}
		parts, _, err := binary.Stat(gctx)
		if err != nil {
			t.Fatalf("Stat() failed: %v", err)
		}
		for i := 0; i < length; i++ {
			hpart := md5.New()
			output, streamErr, err := invokeDownload(t, gctx, binary, int32(i))
			if streamErr != nil || err != nil {
				t.FailNow()
			}
			if bytes.Compare(output, data[i]) != 0 {
				t.Fatalf("Unexpected output: expected %v, got %v", data[i], output)
			}
			hpart.Write(data[i])
			checksum := hex.EncodeToString(hpart.Sum(nil))
			if expected, got := checksum, parts[i].Checksum; expected != got {
				t.Fatalf("Unexpected checksum: expected %v, got %v", expected, got)
			}
			if expected, got := len(data[i]), int(parts[i].Size); expected != got {
				t.Fatalf("Unexpected size: expected %v, got %v", expected, got)
			}
		}
		if err := binary.Delete(gctx); err != nil {
			t.Fatalf("Delete() failed: %v", err)
		}
	}
}

// TestResumption checks that the binary interface supports upload
// resumption ranging the number of parts the uploaded binary consists
// of.
func TestResumption(t *testing.T) {
	for length := 2; length < 5; length++ {
		binary, _, _, cleanup := startServer(t, 2)
		defer cleanup()
		// Create <length> chunks of up to 4MB of random bytes.
		data := make([][]byte, length)
		for i := 0; i < length; i++ {
			data[i] = testData()
		}
		if err := binary.Create(gctx, int32(length), repository.MediaInfo{Type: "application/octet-stream"}); err != nil {
			t.Fatalf("Create() failed: %v", err)
		}
		// Simulate a flaky upload client that keeps uploading parts until
		// finished.
		for {
			parts, _, err := binary.Stat(gctx)
			if err != nil {
				t.Fatalf("Stat() failed: %v", err)
			}
			finished := true
			for _, part := range parts {
				finished = finished && (part != impl.MissingPart)
			}
			if finished {
				break
			}
			for i := 0; i < length; i++ {
				fail := testutil.Rand.Intn(2)
				if parts[i] == impl.MissingPart && fail != 0 {
					if streamErr, err := invokeUpload(t, gctx, binary, data[i], int32(i)); streamErr != nil || err != nil {
						t.FailNow()
					}
				}
			}
		}
		if err := binary.Delete(gctx); err != nil {
			t.Fatalf("Delete() failed: %v", err)
		}
	}
}

// TestErrors checks that the binary interface correctly reports errors.
func TestErrors(t *testing.T) {
	binary, _, _, cleanup := startServer(t, 2)
	defer cleanup()
	const length = 2
	data := make([][]byte, length)
	for i := 0; i < length; i++ {
		data[i] = testData()
		for j := 0; j < len(data[i]); j++ {
			data[i][j] = byte(testutil.Rand.Int())
		}
	}
	if err := binary.Create(gctx, int32(length), repository.MediaInfo{Type: "application/octet-stream"}); err != nil {
		t.Fatalf("Create() failed: %v", err)
	}
	if err := binary.Create(gctx, int32(length), repository.MediaInfo{Type: "application/octet-stream"}); err == nil {
		t.Fatalf("Create() did not fail when it should have")
	} else if want := verror.Exist.ID; !verror.Is(err, want) {
		t.Fatalf("Unexpected error: %v, expected error id %v", err, want)
	}
	if streamErr, err := invokeUpload(t, gctx, binary, data[0], 0); streamErr != nil || err != nil {
		t.Fatalf("Upload() failed: %v", err)
	}
	if _, err := invokeUpload(t, gctx, binary, data[0], 0); err == nil {
		t.Fatalf("Upload() did not fail when it should have")
	} else if want := verror.Exist.ID; !verror.Is(err, want) {
		t.Fatalf("Unexpected error: %v, expected error id %v", err, want)
	}
	if _, _, err := invokeDownload(t, gctx, binary, 1); err == nil {
		t.Fatalf("Download() did not fail when it should have")
	} else if want := verror.NoExist.ID; !verror.Is(err, want) {
		t.Fatalf("Unexpected error: %v, expected error id %v", err, want)
	}
	if streamErr, err := invokeUpload(t, gctx, binary, data[1], 1); streamErr != nil || err != nil {
		t.Fatalf("Upload() failed: %v", err)
	}
	if _, streamErr, err := invokeDownload(t, gctx, binary, 0); streamErr != nil || err != nil {
		t.Fatalf("Download() failed: %v", err)
	}
	// Upload/Download on a part number that's outside the range set forth in
	// Create should fail.
	for _, part := range []int32{-1, length} {
		if _, err := invokeUpload(t, gctx, binary, []byte("dummy"), part); err == nil {
			t.Fatalf("Upload() did not fail when it should have")
		} else if want := impl.ErrInvalidPart.ID; !verror.Is(err, want) {
			t.Fatalf("Unexpected error: %v, expected error id %v", err, want)
		}
		if _, _, err := invokeDownload(t, gctx, binary, part); err == nil {
			t.Fatalf("Download() did not fail when it should have")
		} else if want := impl.ErrInvalidPart.ID; !verror.Is(err, want) {
			t.Fatalf("Unexpected error: %v, expected error id %v", err, want)
		}
	}
	if err := binary.Delete(gctx); err != nil {
		t.Fatalf("Delete() failed: %v", err)
	}
	if err := binary.Delete(gctx); err == nil {
		t.Fatalf("Delete() did not fail when it should have")
	} else if want := verror.NoExist.ID; !verror.Is(err, want) {
		t.Fatalf("Unexpected error: %v, expected error id %v", err, want)
	}
}

func TestGlob(t *testing.T) {
	_, ep, _, cleanup := startServer(t, 2)
	defer cleanup()
	data := testData()

	objects := []string{"foo", "bar", "hello world", "a/b/c"}
	for _, obj := range objects {
		name := naming.JoinAddressName(ep, obj)
		binary := repository.BinaryClient(name)

		if err := binary.Create(gctx, 1, repository.MediaInfo{Type: "application/octet-stream"}); err != nil {
			t.Fatalf("Create() failed: %v", err)
		}
		if streamErr, err := invokeUpload(t, gctx, binary, data, 0); streamErr != nil || err != nil {
			t.FailNow()
		}
	}
	results, err := testutil.GlobName(gctx, naming.JoinAddressName(ep, ""), "...")
	if err != nil {
		t.Fatalf("GlobName failed: %v", err)
	}
	expected := []string{"", "a", "a/b", "a/b/c", "bar", "foo", "hello world"}
	if !reflect.DeepEqual(results, expected) {
		t.Errorf("Unexpected results: expected %q, got %q", expected, results)
	}
}