+ time.Sleep(100 * time.Millisecond)
+ }
+ }
+}
+
+func TestRunDatamanagerWithBogusServer(t *testing.T) {
+ defer TearDownDataManagerTest(t)
+ SetupDataManagerTest(t)
+
+ arv.ApiServer = "bogus-server"
+
+ err := singlerun(arv)
+ if err == nil {
+ t.Fatalf("Expected error during singlerun with bogus server")
+ }
+}
+
+func TestRunDatamanagerAsNonAdminUser(t *testing.T) {
+ defer TearDownDataManagerTest(t)
+ SetupDataManagerTest(t)
+
+ arv.ApiToken = arvadostest.ActiveToken
+
+ err := singlerun(arv)
+ if err == nil {
+ t.Fatalf("Expected error during singlerun as non-admin user")
+ }
+}
+
+func TestPutAndGetBlocks_NoErrorDuringSingleRun(t *testing.T) {
+ testOldBlocksNotDeletedOnDataManagerError(t, "", "", false, false)
+}
+
+func createBadPath(t *testing.T) (badpath string) {
+ tempdir, err := ioutil.TempDir("", "bad")
+ if err != nil {
+ t.Fatalf("Could not create temporary directory for bad path: %v", err)
+ }
+ badpath = path.Join(tempdir, "bad")
+ return
+}
+
+func destroyBadPath(t *testing.T, badpath string) () {
+ tempdir := path.Join(badpath, "..")
+ err := os.Remove(tempdir)
+ if err != nil {
+ t.Fatalf("Could not remove bad path temporary directory %v: %v", tempdir, err)
+ return
+ }
+ return
+}
+
+func TestPutAndGetBlocks_ErrorDuringGetCollectionsBadWriteTo(t *testing.T) {
+ badpath := createBadPath(t)
+ defer destroyBadPath(t, badpath)
+ testOldBlocksNotDeletedOnDataManagerError(t, path.Join(badpath, "writetofile"), "", true, true)
+}
+
+func TestPutAndGetBlocks_ErrorDuringGetCollectionsBadHeapProfileFilename(t *testing.T) {
+ badpath := createBadPath(t)
+ defer destroyBadPath(t, badpath)
+ testOldBlocksNotDeletedOnDataManagerError(t, "", path.Join(badpath, "heapprofilefile"), true, true)
+}
+
+// Create some blocks and backdate some of them.
+// Run datamanager while producing an error condition.
+// Verify that the blocks are hence not deleted.
+func testOldBlocksNotDeletedOnDataManagerError(t *testing.T, writeDataTo string, heapProfileFile string, expectError bool, expectOldBlocks bool) {
+ defer TearDownDataManagerTest(t)
+ SetupDataManagerTest(t)
+
+ // Put some blocks and backdate them.
+ var oldUnusedBlockLocators []string
+ oldUnusedBlockData := "this block will have older mtime"
+ for i := 0; i < 5; i++ {
+ oldUnusedBlockLocators = append(oldUnusedBlockLocators, putBlock(t, fmt.Sprintf("%s%d", oldUnusedBlockData, i)))
+ }
+ backdateBlocks(t, oldUnusedBlockLocators)
+
+ // Run data manager
+ summary.WriteDataTo = writeDataTo
+ collection.HeapProfileFilename = heapProfileFile
+
+ err := singlerun(arv)
+ if !expectError {
+ if err != nil {
+ t.Fatalf("Got an error during datamanager singlerun: %v", err)
+ }
+ } else {
+ if err == nil {
+ t.Fatalf("Expected error during datamanager singlerun")
+ }
+ }
+ waitUntilQueuesFinishWork(t)
+
+ // Get block indexes and verify that all backdated blocks are not/deleted as expected
+ if expectOldBlocks {
+ verifyBlocks(t, nil, oldUnusedBlockLocators, 2)
+ } else {
+ verifyBlocks(t, oldUnusedBlockLocators, nil, 2)
+ }
+}
+
+// Create a collection with multiple streams and blocks
+func createMultiStreamBlockCollection(t *testing.T, data string, numStreams, numBlocks int) (string, []string) {
+ defer switchToken(arvadostest.AdminToken)()
+
+ manifest := ""
+ locators := make(map[string]bool)
+ for s := 0; s < numStreams; s++ {
+ manifest += fmt.Sprintf("./stream%d ", s)
+ for b := 0; b < numBlocks; b++ {
+ locator, _, err := keepClient.PutB([]byte(fmt.Sprintf("%s in stream %d and block %d", data, s, b)))
+ if err != nil {
+ t.Fatalf("Error creating block %d in stream %d: %v", b, s, err)
+ }
+ locators[strings.Split(locator, "+A")[0]] = true
+ manifest += locator + " "