X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/8ba8aece3e72247d5dbf885b5845a1663956032d..143a5f355100b837daa428231df0370b525a1f9f:/services/keepstore/pull_worker_test.go diff --git a/services/keepstore/pull_worker_test.go b/services/keepstore/pull_worker_test.go index 84e951c454..37d83b3280 100644 --- a/services/keepstore/pull_worker_test.go +++ b/services/keepstore/pull_worker_test.go @@ -8,58 +8,60 @@ import ( . "gopkg.in/check.v1" "io" "net/http" - "strings" "testing" "time" ) -var testPullLists map[string]string -var processedPullLists map[string]string - type PullWorkerTestSuite struct{} // Gocheck boilerplate -func Test(t *testing.T) { +func TestPullWorker(t *testing.T) { TestingT(t) } // 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. - arv, err := arvadosclient.MakeArvadosClient() - c.Assert(err, Equals, nil) - keepClient, err := keepclient.MakeKeepClient(&arv) - c.Assert(err, Equals, nil) - go RunPullWorker(pullq.NextItem, keepClient) +var testPullLists map[string]string +var readContent string +var readError error +var putContent []byte +var putError error +var currentTestData PullWorkerTestData + +func (s *PullWorkerTestSuite) SetUpTest(c *C) { + readContent = "" + readError = nil + putContent = []byte("") + putError = nil // When a new pull request arrives, the old one will be overwritten. - // This behavior is simulated with delay tests below. + // This behavior is verified using these two maps in the + // "TestPullWorker_pull_list_with_two_items_latest_replacing_old" testPullLists = make(map[string]string) - processedPullLists = make(map[string]string) } -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) +// Since keepstore does not come into picture in tests, +// we need to explicitly start the goroutine in tests. +func RunTestPullWorker(c *C) { + arv, err := arvadosclient.MakeArvadosClient() + c.Assert(err, Equals, nil) + keepClient, err := keepclient.MakeKeepClient(&arv) + c.Assert(err, Equals, nil) - c.Assert(len(processedPullLists), Not(Equals), len(testPullLists)) + pullq = NewWorkQueue() + go RunPullWorker(pullq, keepClient) } var first_pull_list = []byte(`[ { - "locator":"locator1", + "locator":"acbd18db4cc2f85cedef654fccc4a4d8+3", "servers":[ "server_1", "server_2" ] - }, - { - "locator":"locator2", + },{ + "locator":"37b51d194a7513e45b56f6524f2d51f2+3", "servers":[ "server_3" ] @@ -68,10 +70,10 @@ var first_pull_list = []byte(`[ var second_pull_list = []byte(`[ { - "locator":"locator3", + "locator":"73feffa4b7f6bb68e44cf984c85f6e88+3", "servers":[ "server_1", - "server_2" + "server_2" ] } ]`) @@ -91,13 +93,13 @@ func (s *PullWorkerTestSuite) TestPullWorker_pull_list_with_two_locators(c *C) { 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, + name: "TestPullWorker_pull_list_with_two_locators", + req: RequestTester{"/pull", data_manager_token, "PUT", first_pull_list}, + response_code: http.StatusOK, + response_body: "Received 2 pull requests\n", + read_content: "hello", + read_error: false, + put_error: false, } performTest(testData, c) @@ -108,159 +110,219 @@ func (s *PullWorkerTestSuite) TestPullWorker_pull_list_with_one_locator(c *C) { 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, + name: "TestPullWorker_pull_list_with_one_locator", + req: RequestTester{"/pull", data_manager_token, "PUT", second_pull_list}, + response_code: http.StatusOK, + response_body: "Received 1 pull requests\n", + read_content: "hola", + read_error: false, + put_error: false, } 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) { +func (s *PullWorkerTestSuite) TestPullWorker_error_on_get_one_locator(c *C) { defer teardown() data_manager_token = "DATA MANAGER TOKEN" 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: "TestPullWorker_error_on_get_one_locator", + req: RequestTester{"/pull", data_manager_token, "PUT", second_pull_list}, + response_code: http.StatusOK, + response_body: "Received 1 pull requests\n", + read_content: "unused", + read_error: true, + put_error: false, } performTest(testData, c) } -func (s *PullWorkerTestSuite) TestPullWorker_pull_list_with_one_locator_with_delay_2(c *C) { +func (s *PullWorkerTestSuite) TestPullWorker_error_on_get_two_locators(c *C) { defer teardown() data_manager_token = "DATA MANAGER TOKEN" 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: "TestPullWorker_error_on_get_two_locators", + req: RequestTester{"/pull", data_manager_token, "PUT", first_pull_list}, + response_code: http.StatusOK, + response_body: "Received 2 pull requests\n", + read_content: "unused", + read_error: true, + put_error: false, } performTest(testData, c) } -func (s *PullWorkerTestSuite) TestPullWorker_error_on_get_one_locator(c *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_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_put_one_locator", + req: RequestTester{"/pull", data_manager_token, "PUT", second_pull_list}, + response_code: http.StatusOK, + response_body: "Received 1 pull requests\n", + read_content: "hello hello", + read_error: false, + put_error: true, } performTest(testData, c) } -func (s *PullWorkerTestSuite) TestPullWorker_error_on_get_two_locators(c *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_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_put_two_locators", + req: RequestTester{"/pull", data_manager_token, "PUT", first_pull_list}, + response_code: http.StatusOK, + response_body: "Received 2 pull requests\n", + read_content: "hello again", + read_error: false, + put_error: true, } performTest(testData, c) } -func (s *PullWorkerTestSuite) TestPullWorker_error_on_put_one_locator(c *C) { +// When a new pull request arrives, the old one is replaced. This test +// is used to check that behavior by first putting an item on the queue, +// and then performing the test. Thus the "testPullLists" has two entries; +// however, processedPullLists will see only the newest item in the list. +func (s *PullWorkerTestSuite) TestPullWorker_pull_list_with_two_items_latest_replacing_old(c *C) { defer teardown() + var firstInput = []int{1} + pullq = NewWorkQueue() + pullq.ReplaceQueue(makeTestWorkList(firstInput)) + testPullLists["Added_before_actual_test_item"] = string(1) + 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_pull_list_with_two_items_latest_replacing_old", + req: RequestTester{"/pull", data_manager_token, "PUT", second_pull_list}, + response_code: http.StatusOK, + response_body: "Received 1 pull requests\n", + read_content: "hola de nuevo", + read_error: false, + put_error: false, } performTest(testData, c) } -func (s *PullWorkerTestSuite) TestPullWorker_error_on_put_two_locators(c *C) { +// In this case, the item will not be placed on pullq +func (s *PullWorkerTestSuite) TestPullWorker_invalid_data_manager_token(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_pull_list_with_two_locators", + req: RequestTester{"/pull", "invalid_data_manager_token", "PUT", first_pull_list}, + response_code: http.StatusUnauthorized, + response_body: "Unauthorized\n", + read_content: "hello", + read_error: false, + put_error: false, } performTest(testData, c) } func performTest(testData PullWorkerTestData, c *C) { - testPullLists[testData.name] = testData.response_body + KeepVM = MakeTestVolumeManager(2) + defer KeepVM.Close() - // 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) + RunTestPullWorker(c) + defer pullq.Close() - // 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) - } + currentTestData = testData + testPullLists[testData.name] = testData.response_body + + 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.response_body if testData.read_error { - return nil, 0, "", errors.New("Error getting data") + err = errors.New("Error getting data") + readError = err + return nil, 0, "", err } else { - cb := &ClosingBuffer{bytes.NewBufferString("Hi!")} + readContent = testData.read_content + cb := &ClosingBuffer{bytes.NewBufferString(testData.read_content)} var rc io.ReadCloser rc = cb - return rc, 3, "", nil + return rc, int64(len(testData.read_content)), "", nil } } // Override PutContent to mock PutBlock functionality + defer func(orig func([]byte, string) error) { PutContent = orig }(PutContent) PutContent = func(content []byte, locator string) (err error) { if testData.put_error { - return errors.New("Error putting data") + err = errors.New("Error putting data") + putError = err + return err } else { + putContent = content return nil } } + c.Assert(getStatusItem("PullQueue", "InProgress"), Equals, float64(0)) + c.Assert(getStatusItem("PullQueue", "Queued"), Equals, float64(0)) + response := IssueRequest(&testData.req) - c.Assert(testData.response_code, Equals, response.Code) - c.Assert(testData.response_body, Equals, response.Body.String()) + c.Assert(response.Code, Equals, testData.response_code) + c.Assert(response.Body.String(), Equals, testData.response_body) + + expectEqualWithin(c, time.Second, 0, func() interface{} { + st := pullq.Status() + return st.InProgress + st.Queued + }) + + if testData.name == "TestPullWorker_pull_list_with_two_items_latest_replacing_old" { + c.Assert(len(testPullLists), Equals, 2) + c.Assert(len(processedPullLists), Equals, 1) + c.Assert(testPullLists["Added_before_actual_test_item"], NotNil) + c.Assert(testPullLists["TestPullWorker_pull_list_with_two_items_latest_replacing_old"], NotNil) + c.Assert(processedPullLists["TestPullWorker_pull_list_with_two_items_latest_replacing_old"], NotNil) + } else { + if testData.response_code == http.StatusOK { + c.Assert(len(testPullLists), Equals, 1) + c.Assert(len(processedPullLists), Equals, 1) + c.Assert(testPullLists[testData.name], NotNil) + } else { + c.Assert(len(testPullLists), Equals, 1) + c.Assert(len(processedPullLists), Equals, 0) + } + } + + if testData.read_error { + c.Assert(readError, NotNil) + } else if testData.response_code == http.StatusOK { + c.Assert(readError, IsNil) + c.Assert(readContent, Equals, testData.read_content) + if testData.put_error { + c.Assert(putError, NotNil) + } else { + c.Assert(putError, IsNil) + c.Assert(string(putContent), Equals, testData.read_content) + } + } + + expectChannelEmpty(c, pullq.NextItem) } type ClosingBuffer struct { @@ -270,11 +332,3 @@ type ClosingBuffer struct { 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: - } -}