X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/f8af0c6c331d5b52deab50abf7afa8c7881cddfb..0dc0c5650ddcd8376aea84d32d2b81b1cdba0946:/services/keepstore/handler_test.go

diff --git a/services/keepstore/handler_test.go b/services/keepstore/handler_test.go
index 05b410c97d..33d585ae1e 100644
--- a/services/keepstore/handler_test.go
+++ b/services/keepstore/handler_test.go
@@ -25,10 +25,10 @@ import (
 // A RequestTester represents the parameters for an HTTP request to
 // be issued on behalf of a unit test.
 type RequestTester struct {
-	uri          string
-	api_token    string
-	method       string
-	request_body []byte
+	uri         string
+	apiToken    string
+	method      string
+	requestBody []byte
 }
 
 // Test GetBlockHandler on the following situations:
@@ -43,79 +43,79 @@ func TestGetHandler(t *testing.T) {
 
 	// Prepare two test Keep volumes. Our block is stored on the second volume.
 	KeepVM = MakeTestVolumeManager(2)
-	defer KeepVM.Quit()
+	defer KeepVM.Close()
 
-	vols := KeepVM.Volumes()
-	if err := vols[0].Put(TEST_HASH, TEST_BLOCK); err != nil {
+	vols := KeepVM.AllWritable()
+	if err := vols[0].Put(TestHash, TestBlock); err != nil {
 		t.Error(err)
 	}
 
 	// Create locators for testing.
 	// Turn on permission settings so we can generate signed locators.
-	enforce_permissions = true
-	PermissionSecret = []byte(known_key)
-	permission_ttl = time.Duration(300) * time.Second
+	enforcePermissions = true
+	PermissionSecret = []byte(knownKey)
+	blobSignatureTTL = 300 * time.Second
 
 	var (
-		unsigned_locator  = "/" + TEST_HASH
-		valid_timestamp   = time.Now().Add(permission_ttl)
-		expired_timestamp = time.Now().Add(-time.Hour)
-		signed_locator    = "/" + SignLocator(TEST_HASH, known_token, valid_timestamp)
-		expired_locator   = "/" + SignLocator(TEST_HASH, known_token, expired_timestamp)
+		unsignedLocator  = "/" + TestHash
+		validTimestamp   = time.Now().Add(blobSignatureTTL)
+		expiredTimestamp = time.Now().Add(-time.Hour)
+		signedLocator    = "/" + SignLocator(TestHash, knownToken, validTimestamp)
+		expiredLocator   = "/" + SignLocator(TestHash, knownToken, expiredTimestamp)
 	)
 
 	// -----------------
 	// Test unauthenticated request with permissions off.
-	enforce_permissions = false
+	enforcePermissions = false
 
 	// Unauthenticated request, unsigned locator
 	// => OK
 	response := IssueRequest(
 		&RequestTester{
 			method: "GET",
-			uri:    unsigned_locator,
+			uri:    unsignedLocator,
 		})
 	ExpectStatusCode(t,
 		"Unauthenticated request, unsigned locator", http.StatusOK, response)
 	ExpectBody(t,
 		"Unauthenticated request, unsigned locator",
-		string(TEST_BLOCK),
+		string(TestBlock),
 		response)
 
-	received_cl := response.Header().Get("Content-Length")
-	expected_cl := fmt.Sprintf("%d", len(TEST_BLOCK))
-	if received_cl != expected_cl {
-		t.Errorf("expected Content-Length %s, got %s", expected_cl, received_cl)
+	receivedLen := response.Header().Get("Content-Length")
+	expectedLen := fmt.Sprintf("%d", len(TestBlock))
+	if receivedLen != expectedLen {
+		t.Errorf("expected Content-Length %s, got %s", expectedLen, receivedLen)
 	}
 
 	// ----------------
 	// Permissions: on.
-	enforce_permissions = true
+	enforcePermissions = true
 
 	// Authenticated request, signed locator
 	// => OK
 	response = IssueRequest(&RequestTester{
-		method:    "GET",
-		uri:       signed_locator,
-		api_token: known_token,
+		method:   "GET",
+		uri:      signedLocator,
+		apiToken: knownToken,
 	})
 	ExpectStatusCode(t,
 		"Authenticated request, signed locator", http.StatusOK, response)
 	ExpectBody(t,
-		"Authenticated request, signed locator", string(TEST_BLOCK), response)
+		"Authenticated request, signed locator", string(TestBlock), response)
 
-	received_cl = response.Header().Get("Content-Length")
-	expected_cl = fmt.Sprintf("%d", len(TEST_BLOCK))
-	if received_cl != expected_cl {
-		t.Errorf("expected Content-Length %s, got %s", expected_cl, received_cl)
+	receivedLen = response.Header().Get("Content-Length")
+	expectedLen = fmt.Sprintf("%d", len(TestBlock))
+	if receivedLen != expectedLen {
+		t.Errorf("expected Content-Length %s, got %s", expectedLen, receivedLen)
 	}
 
 	// Authenticated request, unsigned locator
 	// => PermissionError
 	response = IssueRequest(&RequestTester{
-		method:    "GET",
-		uri:       unsigned_locator,
-		api_token: known_token,
+		method:   "GET",
+		uri:      unsignedLocator,
+		apiToken: knownToken,
 	})
 	ExpectStatusCode(t, "unsigned locator", PermissionError.HTTPCode, response)
 
@@ -123,7 +123,7 @@ func TestGetHandler(t *testing.T) {
 	// => PermissionError
 	response = IssueRequest(&RequestTester{
 		method: "GET",
-		uri:    signed_locator,
+		uri:    signedLocator,
 	})
 	ExpectStatusCode(t,
 		"Unauthenticated request, signed locator",
@@ -132,9 +132,9 @@ func TestGetHandler(t *testing.T) {
 	// Authenticated request, expired locator
 	// => ExpiredError
 	response = IssueRequest(&RequestTester{
-		method:    "GET",
-		uri:       expired_locator,
-		api_token: known_token,
+		method:   "GET",
+		uri:      expiredLocator,
+		apiToken: knownToken,
 	})
 	ExpectStatusCode(t,
 		"Authenticated request, expired locator",
@@ -151,32 +151,32 @@ func TestPutHandler(t *testing.T) {
 
 	// Prepare two test Keep volumes.
 	KeepVM = MakeTestVolumeManager(2)
-	defer KeepVM.Quit()
+	defer KeepVM.Close()
 
 	// --------------
 	// No server key.
 
 	// Unauthenticated request, no server key
 	// => OK (unsigned response)
-	unsigned_locator := "/" + TEST_HASH
+	unsignedLocator := "/" + TestHash
 	response := IssueRequest(
 		&RequestTester{
-			method:       "PUT",
-			uri:          unsigned_locator,
-			request_body: TEST_BLOCK,
+			method:      "PUT",
+			uri:         unsignedLocator,
+			requestBody: TestBlock,
 		})
 
 	ExpectStatusCode(t,
 		"Unauthenticated request, no server key", http.StatusOK, response)
 	ExpectBody(t,
 		"Unauthenticated request, no server key",
-		TEST_HASH_PUT_RESPONSE, response)
+		TestHashPutResp, response)
 
 	// ------------------
 	// With a server key.
 
-	PermissionSecret = []byte(known_key)
-	permission_ttl = time.Duration(300) * time.Second
+	PermissionSecret = []byte(knownKey)
+	blobSignatureTTL = 300 * time.Second
 
 	// When a permission key is available, the locator returned
 	// from an authenticated PUT request will be signed.
@@ -185,29 +185,29 @@ func TestPutHandler(t *testing.T) {
 	// => OK (signed response)
 	response = IssueRequest(
 		&RequestTester{
-			method:       "PUT",
-			uri:          unsigned_locator,
-			request_body: TEST_BLOCK,
-			api_token:    known_token,
+			method:      "PUT",
+			uri:         unsignedLocator,
+			requestBody: TestBlock,
+			apiToken:    knownToken,
 		})
 
 	ExpectStatusCode(t,
 		"Authenticated PUT, signed locator, with server key",
 		http.StatusOK, response)
-	response_locator := strings.TrimSpace(response.Body.String())
-	if !VerifySignature(response_locator, known_token) {
+	responseLocator := strings.TrimSpace(response.Body.String())
+	if VerifySignature(responseLocator, knownToken) != nil {
 		t.Errorf("Authenticated PUT, signed locator, with server key:\n"+
 			"response '%s' does not contain a valid signature",
-			response_locator)
+			responseLocator)
 	}
 
 	// Unauthenticated PUT, unsigned locator
 	// => OK
 	response = IssueRequest(
 		&RequestTester{
-			method:       "PUT",
-			uri:          unsigned_locator,
-			request_body: TEST_BLOCK,
+			method:      "PUT",
+			uri:         unsignedLocator,
+			requestBody: TestBlock,
 		})
 
 	ExpectStatusCode(t,
@@ -215,7 +215,54 @@ func TestPutHandler(t *testing.T) {
 		http.StatusOK, response)
 	ExpectBody(t,
 		"Unauthenticated PUT, unsigned locator, with server key",
-		TEST_HASH_PUT_RESPONSE, response)
+		TestHashPutResp, response)
+}
+
+func TestPutAndDeleteSkipReadonlyVolumes(t *testing.T) {
+	defer teardown()
+	dataManagerToken = "fake-data-manager-token"
+	vols := []*MockVolume{CreateMockVolume(), CreateMockVolume()}
+	vols[0].Readonly = true
+	KeepVM = MakeRRVolumeManager([]Volume{vols[0], vols[1]})
+	defer KeepVM.Close()
+	IssueRequest(
+		&RequestTester{
+			method:      "PUT",
+			uri:         "/" + TestHash,
+			requestBody: TestBlock,
+		})
+	defer func(orig bool) {
+		neverDelete = orig
+	}(neverDelete)
+	neverDelete = false
+	IssueRequest(
+		&RequestTester{
+			method:      "DELETE",
+			uri:         "/" + TestHash,
+			requestBody: TestBlock,
+			apiToken:    dataManagerToken,
+		})
+	type expect struct {
+		volnum    int
+		method    string
+		callcount int
+	}
+	for _, e := range []expect{
+		{0, "Get", 0},
+		{0, "Compare", 0},
+		{0, "Touch", 0},
+		{0, "Put", 0},
+		{0, "Delete", 0},
+		{1, "Get", 0},
+		{1, "Compare", 1},
+		{1, "Touch", 1},
+		{1, "Put", 1},
+		{1, "Delete", 1},
+	} {
+		if calls := vols[e.volnum].CallCount(e.method); calls != e.callcount {
+			t.Errorf("Got %d %s() on vol %d, expect %d", calls, e.method, e.volnum, e.callcount)
+		}
+	}
 }
 
 // Test /index requests:
@@ -227,7 +274,7 @@ func TestPutHandler(t *testing.T) {
 //   - authenticated   /index/prefix request | superuser
 //
 // The only /index requests that should succeed are those issued by the
-// superuser. They should pass regardless of the value of enforce_permissions.
+// superuser. They should pass regardless of the value of enforcePermissions.
 //
 func TestIndexHandler(t *testing.T) {
 	defer teardown()
@@ -236,64 +283,74 @@ func TestIndexHandler(t *testing.T) {
 	// Include multiple blocks on different volumes, and
 	// some metadata files (which should be omitted from index listings)
 	KeepVM = MakeTestVolumeManager(2)
-	defer KeepVM.Quit()
+	defer KeepVM.Close()
 
-	vols := KeepVM.Volumes()
-	vols[0].Put(TEST_HASH, TEST_BLOCK)
-	vols[1].Put(TEST_HASH_2, TEST_BLOCK_2)
-	vols[0].Put(TEST_HASH+".meta", []byte("metadata"))
-	vols[1].Put(TEST_HASH_2+".meta", []byte("metadata"))
+	vols := KeepVM.AllWritable()
+	vols[0].Put(TestHash, TestBlock)
+	vols[1].Put(TestHash2, TestBlock2)
+	vols[0].Put(TestHash+".meta", []byte("metadata"))
+	vols[1].Put(TestHash2+".meta", []byte("metadata"))
 
-	data_manager_token = "DATA MANAGER TOKEN"
+	dataManagerToken = "DATA MANAGER TOKEN"
 
-	unauthenticated_req := &RequestTester{
+	unauthenticatedReq := &RequestTester{
 		method: "GET",
 		uri:    "/index",
 	}
-	authenticated_req := &RequestTester{
-		method:    "GET",
-		uri:       "/index",
-		api_token: known_token,
+	authenticatedReq := &RequestTester{
+		method:   "GET",
+		uri:      "/index",
+		apiToken: knownToken,
 	}
-	superuser_req := &RequestTester{
-		method:    "GET",
-		uri:       "/index",
-		api_token: data_manager_token,
+	superuserReq := &RequestTester{
+		method:   "GET",
+		uri:      "/index",
+		apiToken: dataManagerToken,
 	}
-	unauth_prefix_req := &RequestTester{
+	unauthPrefixReq := &RequestTester{
 		method: "GET",
-		uri:    "/index/" + TEST_HASH[0:3],
+		uri:    "/index/" + TestHash[0:3],
 	}
-	auth_prefix_req := &RequestTester{
-		method:    "GET",
-		uri:       "/index/" + TEST_HASH[0:3],
-		api_token: known_token,
+	authPrefixReq := &RequestTester{
+		method:   "GET",
+		uri:      "/index/" + TestHash[0:3],
+		apiToken: knownToken,
 	}
-	superuser_prefix_req := &RequestTester{
-		method:    "GET",
-		uri:       "/index/" + TEST_HASH[0:3],
-		api_token: data_manager_token,
+	superuserPrefixReq := &RequestTester{
+		method:   "GET",
+		uri:      "/index/" + TestHash[0:3],
+		apiToken: dataManagerToken,
+	}
+	superuserNoSuchPrefixReq := &RequestTester{
+		method:   "GET",
+		uri:      "/index/abcd",
+		apiToken: dataManagerToken,
+	}
+	superuserInvalidPrefixReq := &RequestTester{
+		method:   "GET",
+		uri:      "/index/xyz",
+		apiToken: dataManagerToken,
 	}
 
 	// -------------------------------------------------------------
 	// Only the superuser should be allowed to issue /index requests.
 
 	// ---------------------------
-	// enforce_permissions enabled
+	// enforcePermissions enabled
 	// This setting should not affect tests passing.
-	enforce_permissions = true
+	enforcePermissions = true
 
 	// unauthenticated /index request
 	// => UnauthorizedError
-	response := IssueRequest(unauthenticated_req)
+	response := IssueRequest(unauthenticatedReq)
 	ExpectStatusCode(t,
-		"enforce_permissions on, unauthenticated request",
+		"enforcePermissions on, unauthenticated request",
 		UnauthorizedError.HTTPCode,
 		response)
 
 	// unauthenticated /index/prefix request
 	// => UnauthorizedError
-	response = IssueRequest(unauth_prefix_req)
+	response = IssueRequest(unauthPrefixReq)
 	ExpectStatusCode(t,
 		"permissions on, unauthenticated /index/prefix request",
 		UnauthorizedError.HTTPCode,
@@ -301,7 +358,7 @@ func TestIndexHandler(t *testing.T) {
 
 	// authenticated /index request, non-superuser
 	// => UnauthorizedError
-	response = IssueRequest(authenticated_req)
+	response = IssueRequest(authenticatedReq)
 	ExpectStatusCode(t,
 		"permissions on, authenticated request, non-superuser",
 		UnauthorizedError.HTTPCode,
@@ -309,7 +366,7 @@ func TestIndexHandler(t *testing.T) {
 
 	// authenticated /index/prefix request, non-superuser
 	// => UnauthorizedError
-	response = IssueRequest(auth_prefix_req)
+	response = IssueRequest(authPrefixReq)
 	ExpectStatusCode(t,
 		"permissions on, authenticated /index/prefix request, non-superuser",
 		UnauthorizedError.HTTPCode,
@@ -317,27 +374,27 @@ func TestIndexHandler(t *testing.T) {
 
 	// superuser /index request
 	// => OK
-	response = IssueRequest(superuser_req)
+	response = IssueRequest(superuserReq)
 	ExpectStatusCode(t,
 		"permissions on, superuser request",
 		http.StatusOK,
 		response)
 
 	// ----------------------------
-	// enforce_permissions disabled
+	// enforcePermissions disabled
 	// Valid Request should still pass.
-	enforce_permissions = false
+	enforcePermissions = false
 
 	// superuser /index request
 	// => OK
-	response = IssueRequest(superuser_req)
+	response = IssueRequest(superuserReq)
 	ExpectStatusCode(t,
 		"permissions on, superuser request",
 		http.StatusOK,
 		response)
 
-	expected := `^` + TEST_HASH + `\+\d+ \d+\n` +
-		TEST_HASH_2 + `\+\d+ \d+\n$`
+	expected := `^` + TestHash + `\+\d+ \d+\n` +
+		TestHash2 + `\+\d+ \d+\n\n$`
 	match, _ := regexp.MatchString(expected, response.Body.String())
 	if !match {
 		t.Errorf(
@@ -347,19 +404,39 @@ func TestIndexHandler(t *testing.T) {
 
 	// superuser /index/prefix request
 	// => OK
-	response = IssueRequest(superuser_prefix_req)
+	response = IssueRequest(superuserPrefixReq)
 	ExpectStatusCode(t,
 		"permissions on, superuser request",
 		http.StatusOK,
 		response)
 
-	expected = `^` + TEST_HASH + `\+\d+ \d+\n$`
+	expected = `^` + TestHash + `\+\d+ \d+\n\n$`
 	match, _ = regexp.MatchString(expected, response.Body.String())
 	if !match {
 		t.Errorf(
 			"permissions on, superuser /index/prefix request: expected %s, got:\n%s",
 			expected, response.Body.String())
 	}
+
+	// superuser /index/{no-such-prefix} request
+	// => OK
+	response = IssueRequest(superuserNoSuchPrefixReq)
+	ExpectStatusCode(t,
+		"permissions on, superuser request",
+		http.StatusOK,
+		response)
+
+	if "\n" != response.Body.String() {
+		t.Errorf("Expected empty response for %s. Found %s", superuserNoSuchPrefixReq.uri, response.Body.String())
+	}
+
+	// superuser /index/{invalid-prefix} request
+	// => StatusBadRequest
+	response = IssueRequest(superuserInvalidPrefixReq)
+	ExpectStatusCode(t,
+		"permissions on, superuser request",
+		http.StatusBadRequest,
+		response)
 }
 
 // TestDeleteHandler
@@ -395,52 +472,54 @@ func TestDeleteHandler(t *testing.T) {
 	// Include multiple blocks on different volumes, and
 	// some metadata files (which should be omitted from index listings)
 	KeepVM = MakeTestVolumeManager(2)
-	defer KeepVM.Quit()
+	defer KeepVM.Close()
 
-	vols := KeepVM.Volumes()
-	vols[0].Put(TEST_HASH, TEST_BLOCK)
+	vols := KeepVM.AllWritable()
+	vols[0].Put(TestHash, TestBlock)
 
-	// Explicitly set the permission_ttl to 0 for these
+	// Explicitly set the blobSignatureTTL to 0 for these
 	// tests, to ensure the MockVolume deletes the blocks
 	// even though they have just been created.
-	permission_ttl = time.Duration(0)
+	blobSignatureTTL = time.Duration(0)
+
+	var userToken = "NOT DATA MANAGER TOKEN"
+	dataManagerToken = "DATA MANAGER TOKEN"
 
-	var user_token = "NOT DATA MANAGER TOKEN"
-	data_manager_token = "DATA MANAGER TOKEN"
+	neverDelete = false
 
-	unauth_req := &RequestTester{
+	unauthReq := &RequestTester{
 		method: "DELETE",
-		uri:    "/" + TEST_HASH,
+		uri:    "/" + TestHash,
 	}
 
-	user_req := &RequestTester{
-		method:    "DELETE",
-		uri:       "/" + TEST_HASH,
-		api_token: user_token,
+	userReq := &RequestTester{
+		method:   "DELETE",
+		uri:      "/" + TestHash,
+		apiToken: userToken,
 	}
 
-	superuser_existing_block_req := &RequestTester{
-		method:    "DELETE",
-		uri:       "/" + TEST_HASH,
-		api_token: data_manager_token,
+	superuserExistingBlockReq := &RequestTester{
+		method:   "DELETE",
+		uri:      "/" + TestHash,
+		apiToken: dataManagerToken,
 	}
 
-	superuser_nonexistent_block_req := &RequestTester{
-		method:    "DELETE",
-		uri:       "/" + TEST_HASH_2,
-		api_token: data_manager_token,
+	superuserNonexistentBlockReq := &RequestTester{
+		method:   "DELETE",
+		uri:      "/" + TestHash2,
+		apiToken: dataManagerToken,
 	}
 
 	// Unauthenticated request returns PermissionError.
 	var response *httptest.ResponseRecorder
-	response = IssueRequest(unauth_req)
+	response = IssueRequest(unauthReq)
 	ExpectStatusCode(t,
 		"unauthenticated request",
 		PermissionError.HTTPCode,
 		response)
 
 	// Authenticated non-admin request returns PermissionError.
-	response = IssueRequest(user_req)
+	response = IssueRequest(userReq)
 	ExpectStatusCode(t,
 		"authenticated non-admin request",
 		PermissionError.HTTPCode,
@@ -451,62 +530,62 @@ func TestDeleteHandler(t *testing.T) {
 		Deleted int `json:"copies_deleted"`
 		Failed  int `json:"copies_failed"`
 	}
-	var response_dc, expected_dc deletecounter
+	var responseDc, expectedDc deletecounter
 
-	response = IssueRequest(superuser_nonexistent_block_req)
+	response = IssueRequest(superuserNonexistentBlockReq)
 	ExpectStatusCode(t,
 		"data manager request, nonexistent block",
 		http.StatusNotFound,
 		response)
 
-	// Authenticated admin request for existing block while never_delete is set.
-	never_delete = true
-	response = IssueRequest(superuser_existing_block_req)
+	// Authenticated admin request for existing block while neverDelete is set.
+	neverDelete = true
+	response = IssueRequest(superuserExistingBlockReq)
 	ExpectStatusCode(t,
 		"authenticated request, existing block, method disabled",
 		MethodDisabledError.HTTPCode,
 		response)
-	never_delete = false
+	neverDelete = false
 
 	// Authenticated admin request for existing block.
-	response = IssueRequest(superuser_existing_block_req)
+	response = IssueRequest(superuserExistingBlockReq)
 	ExpectStatusCode(t,
 		"data manager request, existing block",
 		http.StatusOK,
 		response)
 	// Expect response {"copies_deleted":1,"copies_failed":0}
-	expected_dc = deletecounter{1, 0}
-	json.NewDecoder(response.Body).Decode(&response_dc)
-	if response_dc != expected_dc {
-		t.Errorf("superuser_existing_block_req\nexpected: %+v\nreceived: %+v",
-			expected_dc, response_dc)
+	expectedDc = deletecounter{1, 0}
+	json.NewDecoder(response.Body).Decode(&responseDc)
+	if responseDc != expectedDc {
+		t.Errorf("superuserExistingBlockReq\nexpected: %+v\nreceived: %+v",
+			expectedDc, responseDc)
 	}
 	// Confirm the block has been deleted
-	_, err := vols[0].Get(TEST_HASH)
-	var block_deleted = os.IsNotExist(err)
-	if !block_deleted {
-		t.Error("superuser_existing_block_req: block not deleted")
+	_, err := vols[0].Get(TestHash)
+	var blockDeleted = os.IsNotExist(err)
+	if !blockDeleted {
+		t.Error("superuserExistingBlockReq: block not deleted")
 	}
 
-	// A DELETE request on a block newer than permission_ttl should return
-	// success but leave the block on the volume.
-	vols[0].Put(TEST_HASH, TEST_BLOCK)
-	permission_ttl = time.Duration(1) * time.Hour
+	// A DELETE request on a block newer than blobSignatureTTL
+	// should return success but leave the block on the volume.
+	vols[0].Put(TestHash, TestBlock)
+	blobSignatureTTL = time.Hour
 
-	response = IssueRequest(superuser_existing_block_req)
+	response = IssueRequest(superuserExistingBlockReq)
 	ExpectStatusCode(t,
 		"data manager request, existing block",
 		http.StatusOK,
 		response)
 	// Expect response {"copies_deleted":1,"copies_failed":0}
-	expected_dc = deletecounter{1, 0}
-	json.NewDecoder(response.Body).Decode(&response_dc)
-	if response_dc != expected_dc {
-		t.Errorf("superuser_existing_block_req\nexpected: %+v\nreceived: %+v",
-			expected_dc, response_dc)
+	expectedDc = deletecounter{1, 0}
+	json.NewDecoder(response.Body).Decode(&responseDc)
+	if responseDc != expectedDc {
+		t.Errorf("superuserExistingBlockReq\nexpected: %+v\nreceived: %+v",
+			expectedDc, responseDc)
 	}
 	// Confirm the block has NOT been deleted.
-	_, err = vols[0].Get(TEST_HASH)
+	_, err = vols[0].Get(TestHash)
 	if err != nil {
 		t.Errorf("testing delete on new block: %s\n", err)
 	}
@@ -542,10 +621,12 @@ func TestDeleteHandler(t *testing.T) {
 func TestPullHandler(t *testing.T) {
 	defer teardown()
 
-	var user_token = "USER TOKEN"
-	data_manager_token = "DATA MANAGER TOKEN"
+	var userToken = "USER TOKEN"
+	dataManagerToken = "DATA MANAGER TOKEN"
+
+	pullq = NewWorkQueue()
 
-	good_json := []byte(`[
+	goodJSON := []byte(`[
 		{
 			"locator":"locator_with_two_servers",
 			"servers":[
@@ -563,45 +644,45 @@ func TestPullHandler(t *testing.T) {
 		}
 	]`)
 
-	bad_json := []byte(`{ "key":"I'm a little teapot" }`)
+	badJSON := []byte(`{ "key":"I'm a little teapot" }`)
 
 	type pullTest struct {
-		name          string
-		req           RequestTester
-		response_code int
-		response_body string
+		name         string
+		req          RequestTester
+		responseCode int
+		responseBody string
 	}
 	var testcases = []pullTest{
 		{
 			"Valid pull list from an ordinary user",
-			RequestTester{"/pull", user_token, "PUT", good_json},
+			RequestTester{"/pull", userToken, "PUT", goodJSON},
 			http.StatusUnauthorized,
 			"Unauthorized\n",
 		},
 		{
 			"Invalid pull request from an ordinary user",
-			RequestTester{"/pull", user_token, "PUT", bad_json},
+			RequestTester{"/pull", userToken, "PUT", badJSON},
 			http.StatusUnauthorized,
 			"Unauthorized\n",
 		},
 		{
 			"Valid pull request from the data manager",
-			RequestTester{"/pull", data_manager_token, "PUT", good_json},
+			RequestTester{"/pull", dataManagerToken, "PUT", goodJSON},
 			http.StatusOK,
 			"Received 3 pull requests\n",
 		},
 		{
 			"Invalid pull request from the data manager",
-			RequestTester{"/pull", data_manager_token, "PUT", bad_json},
+			RequestTester{"/pull", dataManagerToken, "PUT", badJSON},
 			http.StatusBadRequest,
-			"Bad Request\n",
+			"",
 		},
 	}
 
 	for _, tst := range testcases {
 		response := IssueRequest(&tst.req)
-		ExpectStatusCode(t, tst.name, tst.response_code, response)
-		ExpectBody(t, tst.name, tst.response_body, response)
+		ExpectStatusCode(t, tst.name, tst.responseCode, response)
+		ExpectBody(t, tst.name, tst.responseBody, response)
 	}
 
 	// The Keep pull manager should have received one good list with 3
@@ -646,10 +727,12 @@ func TestPullHandler(t *testing.T) {
 func TestTrashHandler(t *testing.T) {
 	defer teardown()
 
-	var user_token = "USER TOKEN"
-	data_manager_token = "DATA MANAGER TOKEN"
+	var userToken = "USER TOKEN"
+	dataManagerToken = "DATA MANAGER TOKEN"
 
-	good_json := []byte(`[
+	trashq = NewWorkQueue()
+
+	goodJSON := []byte(`[
 		{
 			"locator":"block1",
 			"block_mtime":1409082153
@@ -664,46 +747,46 @@ func TestTrashHandler(t *testing.T) {
 		}
 	]`)
 
-	bad_json := []byte(`I am not a valid JSON string`)
+	badJSON := []byte(`I am not a valid JSON string`)
 
 	type trashTest struct {
-		name          string
-		req           RequestTester
-		response_code int
-		response_body string
+		name         string
+		req          RequestTester
+		responseCode int
+		responseBody string
 	}
 
 	var testcases = []trashTest{
 		{
 			"Valid trash list from an ordinary user",
-			RequestTester{"/trash", user_token, "PUT", good_json},
+			RequestTester{"/trash", userToken, "PUT", goodJSON},
 			http.StatusUnauthorized,
 			"Unauthorized\n",
 		},
 		{
 			"Invalid trash list from an ordinary user",
-			RequestTester{"/trash", user_token, "PUT", bad_json},
+			RequestTester{"/trash", userToken, "PUT", badJSON},
 			http.StatusUnauthorized,
 			"Unauthorized\n",
 		},
 		{
 			"Valid trash list from the data manager",
-			RequestTester{"/trash", data_manager_token, "PUT", good_json},
+			RequestTester{"/trash", dataManagerToken, "PUT", goodJSON},
 			http.StatusOK,
 			"Received 3 trash requests\n",
 		},
 		{
 			"Invalid trash list from the data manager",
-			RequestTester{"/trash", data_manager_token, "PUT", bad_json},
+			RequestTester{"/trash", dataManagerToken, "PUT", badJSON},
 			http.StatusBadRequest,
-			"Bad Request\n",
+			"",
 		},
 	}
 
 	for _, tst := range testcases {
 		response := IssueRequest(&tst.req)
-		ExpectStatusCode(t, tst.name, tst.response_code, response)
-		ExpectBody(t, tst.name, tst.response_body, response)
+		ExpectStatusCode(t, tst.name, tst.responseCode, response)
+		ExpectBody(t, tst.name, tst.responseBody, response)
 	}
 
 	// The trash collector should have received one good list with 3
@@ -726,12 +809,12 @@ func TestTrashHandler(t *testing.T) {
 // REST router.  It returns the HTTP response to the request.
 func IssueRequest(rt *RequestTester) *httptest.ResponseRecorder {
 	response := httptest.NewRecorder()
-	body := bytes.NewReader(rt.request_body)
+	body := bytes.NewReader(rt.requestBody)
 	req, _ := http.NewRequest(rt.method, rt.uri, body)
-	if rt.api_token != "" {
-		req.Header.Set("Authorization", "OAuth2 "+rt.api_token)
+	if rt.apiToken != "" {
+		req.Header.Set("Authorization", "OAuth2 "+rt.apiToken)
 	}
-	loggingRouter := MakeLoggingRESTRouter()
+	loggingRouter := MakeRESTRouter()
 	loggingRouter.ServeHTTP(response, req)
 	return response
 }
@@ -741,21 +824,252 @@ func IssueRequest(rt *RequestTester) *httptest.ResponseRecorder {
 func ExpectStatusCode(
 	t *testing.T,
 	testname string,
-	expected_status int,
+	expectedStatus int,
 	response *httptest.ResponseRecorder) {
-	if response.Code != expected_status {
-		t.Errorf("%s: expected status %s, got %+v",
-			testname, expected_status, response)
+	if response.Code != expectedStatus {
+		t.Errorf("%s: expected status %d, got %+v",
+			testname, expectedStatus, response)
 	}
 }
 
 func ExpectBody(
 	t *testing.T,
 	testname string,
-	expected_body string,
+	expectedBody string,
 	response *httptest.ResponseRecorder) {
-	if response.Body.String() != expected_body {
+	if expectedBody != "" && response.Body.String() != expectedBody {
 		t.Errorf("%s: expected response body '%s', got %+v",
-			testname, expected_body, response)
+			testname, expectedBody, response)
+	}
+}
+
+// See #7121
+func TestPutNeedsOnlyOneBuffer(t *testing.T) {
+	defer teardown()
+	KeepVM = MakeTestVolumeManager(1)
+	defer KeepVM.Close()
+
+	defer func(orig *bufferPool) {
+		bufs = orig
+	}(bufs)
+	bufs = newBufferPool(1, BlockSize)
+
+	ok := make(chan struct{})
+	go func() {
+		for i := 0; i < 2; i++ {
+			response := IssueRequest(
+				&RequestTester{
+					method:      "PUT",
+					uri:         "/" + TestHash,
+					requestBody: TestBlock,
+				})
+			ExpectStatusCode(t,
+				"TestPutNeedsOnlyOneBuffer", http.StatusOK, response)
+		}
+		ok <- struct{}{}
+	}()
+
+	select {
+	case <-ok:
+	case <-time.After(time.Second):
+		t.Fatal("PUT deadlocks with maxBuffers==1")
+	}
+}
+
+// Invoke the PutBlockHandler a bunch of times to test for bufferpool resource
+// leak.
+func TestPutHandlerNoBufferleak(t *testing.T) {
+	defer teardown()
+
+	// Prepare two test Keep volumes.
+	KeepVM = MakeTestVolumeManager(2)
+	defer KeepVM.Close()
+
+	ok := make(chan bool)
+	go func() {
+		for i := 0; i < maxBuffers+1; i++ {
+			// Unauthenticated request, no server key
+			// => OK (unsigned response)
+			unsignedLocator := "/" + TestHash
+			response := IssueRequest(
+				&RequestTester{
+					method:      "PUT",
+					uri:         unsignedLocator,
+					requestBody: TestBlock,
+				})
+			ExpectStatusCode(t,
+				"TestPutHandlerBufferleak", http.StatusOK, response)
+			ExpectBody(t,
+				"TestPutHandlerBufferleak",
+				TestHashPutResp, response)
+		}
+		ok <- true
+	}()
+	select {
+	case <-time.After(20 * time.Second):
+		// If the buffer pool leaks, the test goroutine hangs.
+		t.Fatal("test did not finish, assuming pool leaked")
+	case <-ok:
+	}
+}
+
+// Invoke the GetBlockHandler a bunch of times to test for bufferpool resource
+// leak.
+func TestGetHandlerNoBufferleak(t *testing.T) {
+	defer teardown()
+
+	// Prepare two test Keep volumes. Our block is stored on the second volume.
+	KeepVM = MakeTestVolumeManager(2)
+	defer KeepVM.Close()
+
+	vols := KeepVM.AllWritable()
+	if err := vols[0].Put(TestHash, TestBlock); err != nil {
+		t.Error(err)
+	}
+
+	ok := make(chan bool)
+	go func() {
+		for i := 0; i < maxBuffers+1; i++ {
+			// Unauthenticated request, unsigned locator
+			// => OK
+			unsignedLocator := "/" + TestHash
+			response := IssueRequest(
+				&RequestTester{
+					method: "GET",
+					uri:    unsignedLocator,
+				})
+			ExpectStatusCode(t,
+				"Unauthenticated request, unsigned locator", http.StatusOK, response)
+			ExpectBody(t,
+				"Unauthenticated request, unsigned locator",
+				string(TestBlock),
+				response)
+		}
+		ok <- true
+	}()
+	select {
+	case <-time.After(20 * time.Second):
+		// If the buffer pool leaks, the test goroutine hangs.
+		t.Fatal("test did not finish, assuming pool leaked")
+	case <-ok:
+	}
+}
+
+func TestPutReplicationHeader(t *testing.T) {
+	defer teardown()
+
+	KeepVM = MakeTestVolumeManager(2)
+	defer KeepVM.Close()
+
+	resp := IssueRequest(&RequestTester{
+		method:      "PUT",
+		uri:         "/" + TestHash,
+		requestBody: TestBlock,
+	})
+	if r := resp.Header().Get("X-Keep-Replicas-Stored"); r != "1" {
+		t.Errorf("Got X-Keep-Replicas-Stored: %q, expected %q", r, "1")
+	}
+}
+
+func TestUntrashHandler(t *testing.T) {
+	defer teardown()
+
+	// Set up Keep volumes
+	KeepVM = MakeTestVolumeManager(2)
+	defer KeepVM.Close()
+	vols := KeepVM.AllWritable()
+	vols[0].Put(TestHash, TestBlock)
+
+	dataManagerToken = "DATA MANAGER TOKEN"
+
+	// unauthenticatedReq => UnauthorizedError
+	unauthenticatedReq := &RequestTester{
+		method: "PUT",
+		uri:    "/untrash/" + TestHash,
+	}
+	response := IssueRequest(unauthenticatedReq)
+	ExpectStatusCode(t,
+		"Unauthenticated request",
+		UnauthorizedError.HTTPCode,
+		response)
+
+	// notDataManagerReq => UnauthorizedError
+	notDataManagerReq := &RequestTester{
+		method:   "PUT",
+		uri:      "/untrash/" + TestHash,
+		apiToken: knownToken,
+	}
+
+	response = IssueRequest(notDataManagerReq)
+	ExpectStatusCode(t,
+		"Non-datamanager token",
+		UnauthorizedError.HTTPCode,
+		response)
+
+	// datamanagerWithBadHashReq => StatusBadRequest
+	datamanagerWithBadHashReq := &RequestTester{
+		method:   "PUT",
+		uri:      "/untrash/thisisnotalocator",
+		apiToken: dataManagerToken,
+	}
+	response = IssueRequest(datamanagerWithBadHashReq)
+	ExpectStatusCode(t,
+		"Bad locator in untrash request",
+		http.StatusBadRequest,
+		response)
+
+	// datamanagerWrongMethodReq => StatusBadRequest
+	datamanagerWrongMethodReq := &RequestTester{
+		method:   "GET",
+		uri:      "/untrash/" + TestHash,
+		apiToken: dataManagerToken,
+	}
+	response = IssueRequest(datamanagerWrongMethodReq)
+	ExpectStatusCode(t,
+		"Only PUT method is supported for untrash",
+		http.StatusBadRequest,
+		response)
+
+	// datamanagerReq => StatusOK
+	datamanagerReq := &RequestTester{
+		method:   "PUT",
+		uri:      "/untrash/" + TestHash,
+		apiToken: dataManagerToken,
+	}
+	response = IssueRequest(datamanagerReq)
+	ExpectStatusCode(t,
+		"",
+		http.StatusOK,
+		response)
+	expected := "Successfully untrashed on: [MockVolume],[MockVolume]"
+	if response.Body.String() != expected {
+		t.Errorf(
+			"Untrash response mismatched: expected %s, got:\n%s",
+			expected, response.Body.String())
 	}
 }
+
+func TestUntrashHandlerWithNoWritableVolumes(t *testing.T) {
+	defer teardown()
+
+	// Set up readonly Keep volumes
+	vols := []*MockVolume{CreateMockVolume(), CreateMockVolume()}
+	vols[0].Readonly = true
+	vols[1].Readonly = true
+	KeepVM = MakeRRVolumeManager([]Volume{vols[0], vols[1]})
+	defer KeepVM.Close()
+
+	dataManagerToken = "DATA MANAGER TOKEN"
+
+	// datamanagerReq => StatusOK
+	datamanagerReq := &RequestTester{
+		method:   "PUT",
+		uri:      "/untrash/" + TestHash,
+		apiToken: dataManagerToken,
+	}
+	response := IssueRequest(datamanagerReq)
+	ExpectStatusCode(t,
+		"No writable volumes",
+		http.StatusNotFound,
+		response)
+}