9066: Add keepstore -max-requests argument.
[arvados.git] / services / keepstore / handler_test.go
index 05b410c97dd28c74cdb5e4639c15572497975a89..33d585ae1e69f18868f7eb57278637c7f855761b 100644 (file)
@@ -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)
+}