X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/ba032ff921ffb7be57d9f831e2a3dce8b38e266e..0eb72b526bf8bbb011551ecf019f604e17a534f1:/services/keepstore/pull_worker_test.go diff --git a/services/keepstore/pull_worker_test.go b/services/keepstore/pull_worker_test.go index de9ec4142f..9e547f30d0 100644 --- a/services/keepstore/pull_worker_test.go +++ b/services/keepstore/pull_worker_test.go @@ -1,274 +1,315 @@ +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + package main import ( "bytes" "errors" - . "gopkg.in/check.v1" "io" + "io/ioutil" "net/http" - "strings" - "testing" "time" -) -var testPullLists map[string]string -var processedPullLists map[string]string + "git.curoverse.com/arvados.git/sdk/go/arvadosclient" + "git.curoverse.com/arvados.git/sdk/go/keepclient" + . "gopkg.in/check.v1" +) -type PullWorkerTestSuite struct{} +var _ = Suite(&PullWorkerTestSuite{}) -// Gocheck boilerplate -func Test(t *testing.T) { - TestingT(t) +type PullWorkerTestSuite struct { + testPullLists map[string]string + readContent string + readError error + putContent []byte + putError error } -// Gocheck boilerplate -var _ = Suite(&PullWorkerTestSuite{}) - -func (s *PullWorkerTestSuite) SetUpSuite(c *C) { - // Since keepstore does not come into picture in tests, - // we need to explicitly start the goroutine in tests. - go RunPullWorker(pullq.NextItem) +func (s *PullWorkerTestSuite) SetUpTest(c *C) { + theConfig.systemAuthToken = "arbitrary data manager token" + s.readContent = "" + s.readError = nil + s.putContent = []byte{} + s.putError = nil // When a new pull request arrives, the old one will be overwritten. - // This behavior is simulated with delay tests below. - testPullLists = make(map[string]string) - processedPullLists = make(map[string]string) + // This behavior is verified using these two maps in the + // "TestPullWorkerPullList_with_two_items_latest_replacing_old" + s.testPullLists = make(map[string]string) + + KeepVM = MakeTestVolumeManager(2) + + // Normally the pull queue and workers are started by main() + // -- tests need to set up their own. + arv, err := arvadosclient.MakeArvadosClient() + c.Assert(err, IsNil) + keepClient, err := keepclient.MakeKeepClient(arv) + c.Assert(err, IsNil) + pullq = NewWorkQueue() + go RunPullWorker(pullq, keepClient) } -func (s *PullWorkerTestSuite) TearDownSuite(c *C) { - // give the channel enough time to read and process all pull list entries - time.Sleep(1000 * time.Millisecond) - - expectWorkerChannelEmpty(c, pullq.NextItem) - - c.Assert(len(processedPullLists), Not(Equals), len(testPullLists)) +func (s *PullWorkerTestSuite) TearDownTest(c *C) { + KeepVM.Close() + KeepVM = nil + pullq.Close() + pullq = nil + teardown() + theConfig = DefaultConfig() + theConfig.Start() } -var first_pull_list = []byte(`[ +var firstPullList = []byte(`[ { - "locator":"locator1", + "locator":"acbd18db4cc2f85cedef654fccc4a4d8+3", "servers":[ "server_1", "server_2" ] - }, - { - "locator":"locator2", + },{ + "locator":"37b51d194a7513e45b56f6524f2d51f2+3", "servers":[ "server_3" ] } ]`) -var second_pull_list = []byte(`[ +var secondPullList = []byte(`[ { - "locator":"locator3", + "locator":"73feffa4b7f6bb68e44cf984c85f6e88+3", "servers":[ "server_1", - "server_2" + "server_2" ] } ]`) type PullWorkerTestData struct { - name string - req RequestTester - response_code int - response_body string - read_content string - read_error bool - put_error bool + name string + req RequestTester + responseCode int + responseBody string + readContent string + readError bool + putError bool } -func (s *PullWorkerTestSuite) TestPullWorker_pull_list_with_two_locators(c *C) { - defer teardown() - - data_manager_token = "DATA MANAGER TOKEN" - testData := PullWorkerTestData{ - "TestPullWorker_pull_list_with_two_locators", - RequestTester{"/pull", data_manager_token, "PUT", first_pull_list}, - http.StatusOK, - "Received 2 pull requests\n", - "hello", - false, - false, - } - - performTest(testData, c) -} - -func (s *PullWorkerTestSuite) TestPullWorker_pull_list_with_one_locator(c *C) { - defer teardown() +// Ensure MountUUID in a pull list is correctly translated to a Volume +// argument passed to writePulledBlock(). +func (s *PullWorkerTestSuite) TestSpecifyMountUUID(c *C) { + defer func(f func(Volume, []byte, string)) { + writePulledBlock = f + }(writePulledBlock) + + for _, spec := range []struct { + sendUUID string + expectVolume Volume + }{ + { + sendUUID: "", + expectVolume: nil, + }, + { + sendUUID: KeepVM.Mounts()[0].UUID, + expectVolume: KeepVM.Mounts()[0].volume, + }, + } { + writePulledBlock = func(v Volume, _ []byte, _ string) { + c.Check(v, Equals, spec.expectVolume) + } - data_manager_token = "DATA MANAGER TOKEN" - testData := PullWorkerTestData{ - "TestPullWorker_pull_list_with_one_locator", - RequestTester{"/pull", data_manager_token, "PUT", second_pull_list}, - http.StatusOK, - "Received 1 pull requests\n", - "hola", - false, - false, + resp := IssueRequest(&RequestTester{ + uri: "/pull", + apiToken: theConfig.systemAuthToken, + method: "PUT", + requestBody: []byte(`[{ + "locator":"acbd18db4cc2f85cedef654fccc4a4d8+3", + "servers":["server_1","server_2"], + "mountuuid":"` + spec.sendUUID + `"}]`), + }) + c.Assert(resp.Code, Equals, http.StatusOK) + expectEqualWithin(c, time.Second, 0, func() interface{} { + st := pullq.Status() + return st.InProgress + st.Queued + }) } - - performTest(testData, c) } -// When a new pull request arrives, the old one will be overwritten. -// Simulate this behavior by inducing delay in GetContent for the delay test(s). -// To ensure this delay test is not the last one executed and -// hence we cannot verify this behavior, let's run the delay test twice. -func (s *PullWorkerTestSuite) TestPullWorker_pull_list_with_one_locator_with_delay_1(c *C) { - defer teardown() - - data_manager_token = "DATA MANAGER TOKEN" +func (s *PullWorkerTestSuite) TestPullWorkerPullList_with_two_locators(c *C) { testData := PullWorkerTestData{ - "TestPullWorker_pull_list_with_one_locator_with_delay_1", - RequestTester{"/pull", data_manager_token, "PUT", second_pull_list}, - http.StatusOK, - "Received 1 pull requests\n", - "hola", - false, - false, + name: "TestPullWorkerPullList_with_two_locators", + req: RequestTester{"/pull", theConfig.systemAuthToken, "PUT", firstPullList}, + responseCode: http.StatusOK, + responseBody: "Received 2 pull requests\n", + readContent: "hello", + readError: false, + putError: false, } - performTest(testData, c) + s.performTest(testData, c) } -func (s *PullWorkerTestSuite) TestPullWorker_pull_list_with_one_locator_with_delay_2(c *C) { - defer teardown() - - data_manager_token = "DATA MANAGER TOKEN" +func (s *PullWorkerTestSuite) TestPullWorkerPullList_with_one_locator(c *C) { testData := PullWorkerTestData{ - "TestPullWorker_pull_list_with_one_locator_with_delay_2", - RequestTester{"/pull", data_manager_token, "PUT", second_pull_list}, - http.StatusOK, - "Received 1 pull requests\n", - "hola", - false, - false, + name: "TestPullWorkerPullList_with_one_locator", + req: RequestTester{"/pull", theConfig.systemAuthToken, "PUT", secondPullList}, + responseCode: http.StatusOK, + responseBody: "Received 1 pull requests\n", + readContent: "hola", + readError: false, + putError: false, } - performTest(testData, c) + s.performTest(testData, c) } func (s *PullWorkerTestSuite) TestPullWorker_error_on_get_one_locator(c *C) { - defer teardown() - - data_manager_token = "DATA MANAGER TOKEN" testData := PullWorkerTestData{ - "TestPullWorker_error_on_get_one_locator", - RequestTester{"/pull", data_manager_token, "PUT", second_pull_list}, - http.StatusOK, - "Received 1 pull requests\n", - "unused", - true, - false, + name: "TestPullWorker_error_on_get_one_locator", + req: RequestTester{"/pull", theConfig.systemAuthToken, "PUT", secondPullList}, + responseCode: http.StatusOK, + responseBody: "Received 1 pull requests\n", + readContent: "unused", + readError: true, + putError: false, } - performTest(testData, c) + s.performTest(testData, c) } func (s *PullWorkerTestSuite) TestPullWorker_error_on_get_two_locators(c *C) { - defer teardown() - - data_manager_token = "DATA MANAGER TOKEN" testData := PullWorkerTestData{ - "TestPullWorker_error_on_get_two_locators", - RequestTester{"/pull", data_manager_token, "PUT", first_pull_list}, - http.StatusOK, - "Received 2 pull requests\n", - "unused", - true, - false, + name: "TestPullWorker_error_on_get_two_locators", + req: RequestTester{"/pull", theConfig.systemAuthToken, "PUT", firstPullList}, + responseCode: http.StatusOK, + responseBody: "Received 2 pull requests\n", + readContent: "unused", + readError: true, + putError: false, } - performTest(testData, c) + s.performTest(testData, c) } func (s *PullWorkerTestSuite) TestPullWorker_error_on_put_one_locator(c *C) { - defer teardown() - - data_manager_token = "DATA MANAGER TOKEN" testData := PullWorkerTestData{ - "TestPullWorker_error_on_put_one_locator", - RequestTester{"/pull", data_manager_token, "PUT", second_pull_list}, - http.StatusOK, - "Received 1 pull requests\n", - "unused", - false, - true, + name: "TestPullWorker_error_on_put_one_locator", + req: RequestTester{"/pull", theConfig.systemAuthToken, "PUT", secondPullList}, + responseCode: http.StatusOK, + responseBody: "Received 1 pull requests\n", + readContent: "hello hello", + readError: false, + putError: true, } - performTest(testData, c) + s.performTest(testData, c) } func (s *PullWorkerTestSuite) TestPullWorker_error_on_put_two_locators(c *C) { - defer teardown() - - data_manager_token = "DATA MANAGER TOKEN" testData := PullWorkerTestData{ - "TestPullWorker_error_on_put_two_locators", - RequestTester{"/pull", data_manager_token, "PUT", first_pull_list}, - http.StatusOK, - "Received 2 pull requests\n", - "unused", - false, - true, + name: "TestPullWorker_error_on_put_two_locators", + req: RequestTester{"/pull", theConfig.systemAuthToken, "PUT", firstPullList}, + responseCode: http.StatusOK, + responseBody: "Received 2 pull requests\n", + readContent: "hello again", + readError: false, + putError: true, } - performTest(testData, c) + s.performTest(testData, c) } -func performTest(testData PullWorkerTestData, c *C) { - testPullLists[testData.name] = testData.response_body +// In this case, the item will not be placed on pullq +func (s *PullWorkerTestSuite) TestPullWorker_invalidToken(c *C) { + testData := PullWorkerTestData{ + name: "TestPullWorkerPullList_with_two_locators", + req: RequestTester{"/pull", "invalidToken", "PUT", firstPullList}, + responseCode: http.StatusUnauthorized, + responseBody: "Unauthorized\n", + readContent: "hello", + readError: false, + putError: false, + } - // We need to make sure the tests have a slight delay so that we can verify the pull list channel overwrites. - time.Sleep(25 * time.Millisecond) + s.performTest(testData, c) +} - // Override GetContent to mock keepclient functionality - GetContent = func(locator string, signedLocator string) (reader io.ReadCloser, contentLength int64, url string, err error) { - if strings.HasPrefix(testData.name, "TestPullWorker_pull_list_with_one_locator_with_delay") { - time.Sleep(100 * time.Millisecond) +func (s *PullWorkerTestSuite) performTest(testData PullWorkerTestData, c *C) { + s.testPullLists[testData.name] = testData.responseBody + + processedPullLists := make(map[string]string) + + // Override GetContent to mock keepclient Get functionality + defer func(orig func(string, *keepclient.KeepClient) (io.ReadCloser, int64, string, error)) { + GetContent = orig + }(GetContent) + GetContent = func(signedLocator string, keepClient *keepclient.KeepClient) (reader io.ReadCloser, contentLength int64, url string, err error) { + c.Assert(getStatusItem("PullQueue", "InProgress"), Equals, float64(1)) + processedPullLists[testData.name] = testData.responseBody + if testData.readError { + err = errors.New("Error getting data") + s.readError = err + return } + s.readContent = testData.readContent + reader = ioutil.NopCloser(bytes.NewBufferString(testData.readContent)) + contentLength = int64(len(testData.readContent)) + return + } - processedPullLists[testData.name] = testData.response_body - if testData.read_error { - return nil, 0, "", errors.New("Error getting data") - } else { - cb := &ClosingBuffer{bytes.NewBufferString("Hi!")} - var rc io.ReadCloser - rc = cb - return rc, 3, "", nil + // Override writePulledBlock to mock PutBlock functionality + defer func(orig func(Volume, []byte, string)) { writePulledBlock = orig }(writePulledBlock) + writePulledBlock = func(v Volume, content []byte, locator string) { + if testData.putError { + s.putError = errors.New("Error putting data") + return } + s.putContent = content } - // Override PutContent to mock PutBlock functionality - PutContent = func(content []byte, locator string) (err error) { - if testData.put_error { - return errors.New("Error putting data") + c.Check(getStatusItem("PullQueue", "InProgress"), Equals, float64(0)) + c.Check(getStatusItem("PullQueue", "Queued"), Equals, float64(0)) + + response := IssueRequest(&testData.req) + c.Assert(response.Code, Equals, testData.responseCode) + c.Assert(response.Body.String(), Equals, testData.responseBody) + + expectEqualWithin(c, time.Second, 0, func() interface{} { + st := pullq.Status() + return st.InProgress + st.Queued + }) + + if testData.name == "TestPullWorkerPullList_with_two_items_latest_replacing_old" { + c.Assert(len(s.testPullLists), Equals, 2) + c.Assert(len(processedPullLists), Equals, 1) + c.Assert(s.testPullLists["Added_before_actual_test_item"], NotNil) + c.Assert(s.testPullLists["TestPullWorkerPullList_with_two_items_latest_replacing_old"], NotNil) + c.Assert(processedPullLists["TestPullWorkerPullList_with_two_items_latest_replacing_old"], NotNil) + } else { + if testData.responseCode == http.StatusOK { + c.Assert(len(s.testPullLists), Equals, 1) + c.Assert(len(processedPullLists), Equals, 1) + c.Assert(s.testPullLists[testData.name], NotNil) } else { - return nil + c.Assert(len(s.testPullLists), Equals, 1) + c.Assert(len(processedPullLists), Equals, 0) } } - response := IssueRequest(&testData.req) - c.Assert(testData.response_code, Equals, response.Code) - c.Assert(testData.response_body, Equals, response.Body.String()) -} - -type ClosingBuffer struct { - *bytes.Buffer -} - -func (cb *ClosingBuffer) Close() (err error) { - return -} - -func expectWorkerChannelEmpty(c *C, workerChannel <-chan interface{}) { - select { - case item := <-workerChannel: - c.Fatalf("Received value (%v) from channel that was expected to be empty", item) - default: + if testData.readError { + c.Assert(s.readError, NotNil) + } else if testData.responseCode == http.StatusOK { + c.Assert(s.readError, IsNil) + c.Assert(s.readContent, Equals, testData.readContent) + if testData.putError { + c.Assert(s.putError, NotNil) + } else { + c.Assert(s.putError, IsNil) + c.Assert(string(s.putContent), Equals, testData.readContent) + } } + + expectChannelEmpty(c, pullq.NextItem) }