X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/1a0a58c4f22af82e0a37440af3b0948771bca5e1..241f0bcdacbf83b587bff9ff45985e720bde9f0b:/services/keepstore/volume_unix_test.go

diff --git a/services/keepstore/volume_unix_test.go b/services/keepstore/volume_unix_test.go
index 278e656066..7fea560d5a 100644
--- a/services/keepstore/volume_unix_test.go
+++ b/services/keepstore/volume_unix_test.go
@@ -2,64 +2,84 @@ package main
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
+	"strings"
+	"sync"
+	"syscall"
 	"testing"
 	"time"
 )
 
-func TempUnixVolume(t *testing.T, serialize bool) UnixVolume {
+type TestableUnixVolume struct {
+	UnixVolume
+	t *testing.T
+}
+
+func NewTestableUnixVolume(t *testing.T, serialize bool, readonly bool) *TestableUnixVolume {
 	d, err := ioutil.TempDir("", "volume_test")
 	if err != nil {
 		t.Fatal(err)
 	}
-	return MakeUnixVolume(d, serialize)
+	var locker sync.Locker
+	if serialize {
+		locker = &sync.Mutex{}
+	}
+	return &TestableUnixVolume{
+		UnixVolume: UnixVolume{
+			root:     d,
+			locker:   locker,
+			readonly: readonly,
+		},
+		t: t,
+	}
 }
 
-func _teardown(v UnixVolume) {
-	if v.queue != nil {
-		close(v.queue)
+// PutRaw writes a Keep block directly into a UnixVolume, even if
+// the volume is readonly.
+func (v *TestableUnixVolume) PutRaw(locator string, data []byte) {
+	defer func(orig bool) {
+		v.readonly = orig
+	}(v.readonly)
+	v.readonly = false
+	err := v.Put(locator, data)
+	if err != nil {
+		v.t.Fatal(err)
 	}
-	os.RemoveAll(v.root)
 }
 
-// store writes a Keep block directly into a UnixVolume, for testing
-// UnixVolume methods.
-//
-func _store(t *testing.T, vol UnixVolume, filename string, block []byte) {
-	blockdir := fmt.Sprintf("%s/%s", vol.root, filename[:3])
-	if err := os.MkdirAll(blockdir, 0755); err != nil {
-		t.Fatal(err)
+func (v *TestableUnixVolume) TouchWithDate(locator string, lastPut time.Time) {
+	err := syscall.Utime(v.blockPath(locator), &syscall.Utimbuf{lastPut.Unix(), lastPut.Unix()})
+	if err != nil {
+		v.t.Fatal(err)
 	}
+}
 
-	blockpath := fmt.Sprintf("%s/%s", blockdir, filename)
-	if f, err := os.Create(blockpath); err == nil {
-		f.Write(block)
-		f.Close()
-	} else {
-		t.Fatal(err)
+func (v *TestableUnixVolume) Teardown() {
+	if err := os.RemoveAll(v.root); err != nil {
+		v.t.Fatal(err)
 	}
 }
 
-func TestGet(t *testing.T) {
-	v := TempUnixVolume(t, false)
-	defer _teardown(v)
-	_store(t, v, TEST_HASH, TEST_BLOCK)
+func TestUnixVolumeWithGenericTests(t *testing.T) {
+	DoGenericVolumeTests(t, func(t *testing.T) TestableVolume {
+		return NewTestableUnixVolume(t, false, false)
+	})
+}
 
-	buf, err := v.Get(TEST_HASH)
-	if err != nil {
-		t.Error(err)
-	}
-	if bytes.Compare(buf, TEST_BLOCK) != 0 {
-		t.Errorf("expected %s, got %s", string(TEST_BLOCK), string(buf))
-	}
+func TestUnixReadOnlyVolumeWithGenericTests(t *testing.T) {
+	DoGenericReadOnlyVolumeTests(t, func(t *testing.T) TestableVolume {
+		return NewTestableUnixVolume(t, false, true)
+	})
 }
 
 func TestGetNotFound(t *testing.T) {
-	v := TempUnixVolume(t, false)
-	defer _teardown(v)
-	_store(t, v, TEST_HASH, TEST_BLOCK)
+	v := NewTestableUnixVolume(t, false, false)
+	defer v.Teardown()
+	v.Put(TEST_HASH, TEST_BLOCK)
 
 	buf, err := v.Get(TEST_HASH_2)
 	switch {
@@ -73,8 +93,8 @@ func TestGetNotFound(t *testing.T) {
 }
 
 func TestPut(t *testing.T) {
-	v := TempUnixVolume(t, false)
-	defer _teardown(v)
+	v := NewTestableUnixVolume(t, false, false)
+	defer v.Teardown()
 
 	err := v.Put(TEST_HASH, TEST_BLOCK)
 	if err != nil {
@@ -90,8 +110,8 @@ func TestPut(t *testing.T) {
 }
 
 func TestPutBadVolume(t *testing.T) {
-	v := TempUnixVolume(t, false)
-	defer _teardown(v)
+	v := NewTestableUnixVolume(t, false, false)
+	defer v.Teardown()
 
 	os.Chmod(v.root, 000)
 	err := v.Put(TEST_HASH, TEST_BLOCK)
@@ -100,133 +120,36 @@ func TestPutBadVolume(t *testing.T) {
 	}
 }
 
-// Serialization tests: launch a bunch of concurrent
-//
-// TODO(twp): show that the underlying Read/Write operations executed
-// serially and not concurrently. The easiest way to do this is
-// probably to activate verbose or debug logging, capture log output
-// and examine it to confirm that Reads and Writes did not overlap.
-//
-// TODO(twp): a proper test of I/O serialization requires that a
-// second request start while the first one is still underway.
-// Guaranteeing that the test behaves this way requires some tricky
-// synchronization and mocking.  For now we'll just launch a bunch of
-// requests simultaenously in goroutines and demonstrate that they
-// return accurate results.
-//
-func TestGetSerialized(t *testing.T) {
-	// Create a volume with I/O serialization enabled.
-	v := TempUnixVolume(t, true)
-	defer _teardown(v)
-
-	_store(t, v, TEST_HASH, TEST_BLOCK)
-	_store(t, v, TEST_HASH_2, TEST_BLOCK_2)
-	_store(t, v, TEST_HASH_3, TEST_BLOCK_3)
-
-	sem := make(chan int)
-	go func(sem chan int) {
-		buf, err := v.Get(TEST_HASH)
-		if err != nil {
-			t.Errorf("err1: %v", err)
-		}
-		if bytes.Compare(buf, TEST_BLOCK) != 0 {
-			t.Errorf("buf should be %s, is %s", string(TEST_BLOCK), string(buf))
-		}
-		sem <- 1
-	}(sem)
-
-	go func(sem chan int) {
-		buf, err := v.Get(TEST_HASH_2)
-		if err != nil {
-			t.Errorf("err2: %v", err)
-		}
-		if bytes.Compare(buf, TEST_BLOCK_2) != 0 {
-			t.Errorf("buf should be %s, is %s", string(TEST_BLOCK_2), string(buf))
-		}
-		sem <- 1
-	}(sem)
-
-	go func(sem chan int) {
-		buf, err := v.Get(TEST_HASH_3)
-		if err != nil {
-			t.Errorf("err3: %v", err)
-		}
-		if bytes.Compare(buf, TEST_BLOCK_3) != 0 {
-			t.Errorf("buf should be %s, is %s", string(TEST_BLOCK_3), string(buf))
-		}
-		sem <- 1
-	}(sem)
-
-	// Wait for all goroutines to finish
-	for done := 0; done < 3; {
-		done += <-sem
-	}
-}
-
-func TestPutSerialized(t *testing.T) {
-	// Create a volume with I/O serialization enabled.
-	v := TempUnixVolume(t, true)
-	defer _teardown(v)
-
-	sem := make(chan int)
-	go func(sem chan int) {
-		err := v.Put(TEST_HASH, TEST_BLOCK)
-		if err != nil {
-			t.Errorf("err1: %v", err)
-		}
-		sem <- 1
-	}(sem)
-
-	go func(sem chan int) {
-		err := v.Put(TEST_HASH_2, TEST_BLOCK_2)
-		if err != nil {
-			t.Errorf("err2: %v", err)
-		}
-		sem <- 1
-	}(sem)
-
-	go func(sem chan int) {
-		err := v.Put(TEST_HASH_3, TEST_BLOCK_3)
-		if err != nil {
-			t.Errorf("err3: %v", err)
-		}
-		sem <- 1
-	}(sem)
-
-	// Wait for all goroutines to finish
-	for done := 0; done < 2; {
-		done += <-sem
-	}
-
-	// Double check that we actually wrote the blocks we expected to write.
-	buf, err := v.Get(TEST_HASH)
-	if err != nil {
-		t.Errorf("Get #1: %v", err)
-	}
-	if bytes.Compare(buf, TEST_BLOCK) != 0 {
-		t.Errorf("Get #1: expected %s, got %s", string(TEST_BLOCK), string(buf))
-	}
+func TestUnixVolumeReadonly(t *testing.T) {
+	v := NewTestableUnixVolume(t, false, true)
+	defer v.Teardown()
 
-	buf, err = v.Get(TEST_HASH_2)
+	v.PutRaw(TEST_HASH, TEST_BLOCK)
+
+	_, err := v.Get(TEST_HASH)
 	if err != nil {
-		t.Errorf("Get #2: %v", err)
+		t.Errorf("got err %v, expected nil", err)
 	}
-	if bytes.Compare(buf, TEST_BLOCK_2) != 0 {
-		t.Errorf("Get #2: expected %s, got %s", string(TEST_BLOCK_2), string(buf))
+
+	err = v.Put(TEST_HASH, TEST_BLOCK)
+	if err != MethodDisabledError {
+		t.Errorf("got err %v, expected MethodDisabledError", err)
 	}
 
-	buf, err = v.Get(TEST_HASH_3)
-	if err != nil {
-		t.Errorf("Get #3: %v", err)
+	err = v.Touch(TEST_HASH)
+	if err != MethodDisabledError {
+		t.Errorf("got err %v, expected MethodDisabledError", err)
 	}
-	if bytes.Compare(buf, TEST_BLOCK_3) != 0 {
-		t.Errorf("Get #3: expected %s, got %s", string(TEST_BLOCK_3), string(buf))
+
+	err = v.Delete(TEST_HASH)
+	if err != MethodDisabledError {
+		t.Errorf("got err %v, expected MethodDisabledError", err)
 	}
 }
 
 func TestIsFull(t *testing.T) {
-	v := TempUnixVolume(t, false)
-	defer _teardown(v)
+	v := NewTestableUnixVolume(t, false, false)
+	defer v.Teardown()
 
 	full_path := v.root + "/full"
 	now := fmt.Sprintf("%d", time.Now().Unix())
@@ -243,3 +166,118 @@ func TestIsFull(t *testing.T) {
 		t.Errorf("%s: should no longer be full", v)
 	}
 }
+
+func TestNodeStatus(t *testing.T) {
+	v := NewTestableUnixVolume(t, false, false)
+	defer v.Teardown()
+
+	// Get node status and make a basic sanity check.
+	volinfo := v.Status()
+	if volinfo.MountPoint != v.root {
+		t.Errorf("GetNodeStatus mount_point %s, expected %s", volinfo.MountPoint, v.root)
+	}
+	if volinfo.DeviceNum == 0 {
+		t.Errorf("uninitialized device_num in %v", volinfo)
+	}
+	if volinfo.BytesFree == 0 {
+		t.Errorf("uninitialized bytes_free in %v", volinfo)
+	}
+	if volinfo.BytesUsed == 0 {
+		t.Errorf("uninitialized bytes_used in %v", volinfo)
+	}
+}
+
+func TestUnixVolumeGetFuncWorkerError(t *testing.T) {
+	v := NewTestableUnixVolume(t, false, false)
+	defer v.Teardown()
+
+	v.Put(TEST_HASH, TEST_BLOCK)
+	mockErr := errors.New("Mock error")
+	err := v.getFunc(v.blockPath(TEST_HASH), func(rdr io.Reader) error {
+		return mockErr
+	})
+	if err != mockErr {
+		t.Errorf("Got %v, expected %v", err, mockErr)
+	}
+}
+
+func TestUnixVolumeGetFuncFileError(t *testing.T) {
+	v := NewTestableUnixVolume(t, false, false)
+	defer v.Teardown()
+
+	funcCalled := false
+	err := v.getFunc(v.blockPath(TEST_HASH), func(rdr io.Reader) error {
+		funcCalled = true
+		return nil
+	})
+	if err == nil {
+		t.Errorf("Expected error opening non-existent file")
+	}
+	if funcCalled {
+		t.Errorf("Worker func should not have been called")
+	}
+}
+
+func TestUnixVolumeGetFuncWorkerWaitsOnMutex(t *testing.T) {
+	v := NewTestableUnixVolume(t, false, false)
+	defer v.Teardown()
+
+	v.Put(TEST_HASH, TEST_BLOCK)
+
+	mtx := NewMockMutex()
+	v.locker = mtx
+
+	funcCalled := make(chan struct{})
+	go v.getFunc(v.blockPath(TEST_HASH), func(rdr io.Reader) error {
+		funcCalled <- struct{}{}
+		return nil
+	})
+	select {
+	case mtx.AllowLock <- struct{}{}:
+	case <-funcCalled:
+		t.Fatal("Function was called before mutex was acquired")
+	case <-time.After(5 * time.Second):
+		t.Fatal("Timed out before mutex was acquired")
+	}
+	select {
+	case <-funcCalled:
+	case mtx.AllowUnlock <- struct{}{}:
+		t.Fatal("Mutex was released before function was called")
+	case <-time.After(5 * time.Second):
+		t.Fatal("Timed out waiting for funcCalled")
+	}
+	select {
+	case mtx.AllowUnlock <- struct{}{}:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Timed out waiting for getFunc() to release mutex")
+	}
+}
+
+func TestUnixVolumeCompare(t *testing.T) {
+	v := NewTestableUnixVolume(t, false, false)
+	defer v.Teardown()
+
+	v.Put(TEST_HASH, TEST_BLOCK)
+	err := v.Compare(TEST_HASH, TEST_BLOCK)
+	if err != nil {
+		t.Errorf("Got err %q, expected nil", err)
+	}
+
+	err = v.Compare(TEST_HASH, []byte("baddata"))
+	if err != CollisionError {
+		t.Errorf("Got err %q, expected %q", err, CollisionError)
+	}
+
+	v.Put(TEST_HASH, []byte("baddata"))
+	err = v.Compare(TEST_HASH, TEST_BLOCK)
+	if err != DiskHashError {
+		t.Errorf("Got err %q, expected %q", err, DiskHashError)
+	}
+
+	p := fmt.Sprintf("%s/%s/%s", v.root, TEST_HASH[:3], TEST_HASH)
+	os.Chmod(p, 000)
+	err = v.Compare(TEST_HASH, TEST_BLOCK)
+	if err == nil || strings.Index(err.Error(), "permission denied") < 0 {
+		t.Errorf("Got err %q, expected %q", err, "permission denied")
+	}
+}