+
+// 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 + " "
+ }
+ manifest += "0:1:dummyfile.txt\n"
+ }
+
+ collection := make(Dict)
+ err := arv.Create("collections",
+ arvadosclient.Dict{"collection": arvadosclient.Dict{"manifest_text": manifest}},
+ &collection)
+
+ if err != nil {
+ t.Fatalf("Error creating collection %v", err)
+ }
+
+ var locs []string
+ for k, _ := range locators {
+ locs = append(locs, k)
+ }
+
+ return collection["uuid"].(string), locs
+}
+
+// Create collection with multiple streams and blocks; backdate the blocks and but do not delete the collection.
+// Also, create stray block and backdate it.
+// After datamanager run: expect blocks from the collection, but not the stray block.
+func TestManifestWithMultipleStreamsAndBlocks(t *testing.T) {
+ defer TearDownDataManagerTest(t)
+ SetupDataManagerTest(t)
+
+ // create collection whose blocks will be backdated
+ collectionWithOldBlocks, oldBlocks := createMultiStreamBlockCollection(t, "old block", 100, 10)
+ if collectionWithOldBlocks == "" {
+ t.Fatalf("Failed to create collection with 1000 blocks")
+ }
+ if len(oldBlocks) != 1000 {
+ t.Fatalf("Not all blocks are created: expected %v, found %v", 1000, len(oldBlocks))
+ }
+
+ // create a stray block that will be backdated
+ strayOldBlock := putBlock(t, "this stray block is old")
+
+ expected := []string{strayOldBlock}
+ expected = append(expected, oldBlocks...)
+ verifyBlocks(t, nil, expected, 2)
+
+ // Backdate old blocks; but the collection still references these blocks
+ backdateBlocks(t, oldBlocks)
+
+ // also backdate the stray old block
+ backdateBlocks(t, []string{strayOldBlock})
+
+ // run datamanager
+ dataManagerSingleRun(t)
+
+ // verify that strayOldBlock is not to be found, but the collections blocks are still there
+ verifyBlocks(t, []string{strayOldBlock}, oldBlocks, 2)
+}