Use path struct while streaming data to kopia (#840)

* Helper function to append elements to a path

Kopia wrapper will need to create the complete path of an item by
joining the collection path and the item name. This allows it to do so
without having to drop to a path Builder (not service/category safe) and
go back to a resource path. Right now it does not handle escaping.

* Use path struct while streaming entries

Use new path struct while streaming entries to kopia. Preparation for
FullPath returning a path struct.

* Update tests to use valid path structures

src/internal/kopia/wrapper.go

@@ -152,6 +152,19 @@ func getStreamItemFunc(
 			return nil
 		}
 
+		itemPath, err := path.FromDataLayerPath(
+			stdpath.Join(streamedEnts.FullPath()...),
+			false,
+		)
+		if err != nil {
+			err = errors.Wrap(err, "parsing collection path")
+			errs = multierror.Append(errs, err)
+
+			logger.Ctx(ctx).Error(err)
+
+			return errs.ErrorOrNil()
+		}
+
 		items := streamedEnts.Items()
 
 		for {
@@ -164,7 +177,16 @@ func getStreamItemFunc(
 					return errs.ErrorOrNil()
 				}
 
-				itemPath := stdpath.Join(append(streamedEnts.FullPath(), e.UUID())...)
+				// For now assuming that item IDs don't need escaping.
+				itemPath, err := itemPath.Append(e.UUID(), true)
+				if err != nil {
+					err = errors.Wrap(err, "getting full item path")
+					errs = multierror.Append(errs, err)
+
+					logger.Ctx(ctx).Error(err)
+
+					continue
+				}
 
 				ei, ok := e.(data.StreamInfo)
 				if !ok {
@@ -180,8 +202,8 @@ func getStreamItemFunc(
 				// Relative path given to us in the callback is missing the root
 				// element. Add to pending set before calling the callback to avoid race
 				// conditions when the item is completed.
-				p := stdpath.Join(append(streamedEnts.FullPath()[1:], e.UUID())...)
+				p := itemPath.PopFront().String()
-				d := &itemDetails{info: ei.Info(), repoRef: itemPath}
+				d := &itemDetails{info: ei.Info(), repoRef: itemPath.String()}
 
 				progress.put(p, d)
 

src/internal/kopia/wrapper_test.go

@@ -38,19 +38,20 @@ const (
 )
 
 var (
-	testEmailDir = path.EmailCategory.String()
+	service  = path.ExchangeService.String()
-	testPath     = []string{
+	category = path.EmailCategory.String()
+	testPath = []string{
 		testTenant,
-		path.ExchangeService.String(),
+		service,
 		testUser,
-		path.EmailCategory.String(),
+		category,
 		testInboxDir,
 	}
 	testPath2 = []string{
 		testTenant,
-		path.ExchangeService.String(),
+		service,
 		testUser,
-		path.EmailCategory.String(),
+		category,
 		testArchiveDir,
 	}
 	testFileData  = []byte("abcdefghijklmnopqrstuvwxyz")
@@ -63,16 +64,6 @@ var (
 	testFileData6 = testFileData
 )
 
-func entriesToNames(entries []fs.Entry) []string {
-	res := make([]string, 0, len(entries))
-
-	for _, e := range entries {
-		res = append(res, e.Name())
-	}
-
-	return res
-}
-
 func testForFiles(
 	t *testing.T,
 	expected map[string][]byte,
@@ -99,6 +90,42 @@ func testForFiles(
 	assert.Equal(t, len(expected), count)
 }
 
+func expectDirs(
+	t *testing.T,
+	entries []fs.Entry,
+	dirs []string,
+	exactly bool,
+) {
+	t.Helper()
+
+	if exactly {
+		require.Len(t, entries, len(dirs))
+	}
+
+	names := make([]string, 0, len(entries))
+	for _, e := range entries {
+		names = append(names, e.Name())
+	}
+
+	assert.Subset(t, names, dirs)
+}
+
+//revive:disable:context-as-argument
+func getDirEntriesForEntry(
+	t *testing.T,
+	ctx context.Context,
+	entry fs.Entry,
+) []fs.Entry {
+	//revive:enable:context-as-argument
+	d, ok := entry.(fs.Directory)
+	require.True(t, ok, "returned entry is not a directory")
+
+	entries, err := fs.GetAllEntries(ctx, d)
+	require.NoError(t, err)
+
+	return entries
+}
+
 // ---------------
 // unit tests
 // ---------------
@@ -203,11 +230,11 @@ func (suite *KopiaUnitSuite) TestCloseWithoutInitDoesNotPanic() {
 func (suite *KopiaUnitSuite) TestBuildDirectoryTree() {
 	tester.LogTimeOfTest(suite.T())
 
+	t := suite.T()
 	ctx := context.Background()
 	tenant := "a-tenant"
-	user1 := "user1"
+	user1 := testUser
 	user2 := "user2"
-	emails := "emails"
 
 	expectedFileCount := map[string]int{
 		user1: 5,
@@ -218,48 +245,49 @@ func (suite *KopiaUnitSuite) TestBuildDirectoryTree() {
 
 	collections := []data.Collection{
 		mockconnector.NewMockExchangeCollection(
-			[]string{tenant, user1, emails},
+			[]string{tenant, service, user1, category, testInboxDir},
 			expectedFileCount[user1],
 		),
 		mockconnector.NewMockExchangeCollection(
-			[]string{tenant, user2, emails},
+			[]string{tenant, service, user2, category, testInboxDir},
 			expectedFileCount[user2],
 		),
 	}
 
 	// Returned directory structure should look like:
 	// - a-tenant
-	//   - user1
+	//   - exchange
-	//     - emails
+	//     - user1
-	//       - 5 separate files
+	//       - emails
-	//   - user2
+	//         - Inbox
-	//     - emails
+	//           - 5 separate files
-	//       - 42 separate files
+	//     - user2
+	//       - emails
+	//         - Inbox
+	//           - 42 separate files
 	dirTree, err := inflateDirTree(ctx, collections, progress)
-	require.NoError(suite.T(), err)
+	require.NoError(t, err)
-	assert.Equal(suite.T(), dirTree.Name(), tenant)
+	assert.Equal(t, testTenant, dirTree.Name())
 
 	entries, err := fs.GetAllEntries(ctx, dirTree)
-	require.NoError(suite.T(), err)
+	require.NoError(t, err)
 
-	names := entriesToNames(entries)
+	expectDirs(t, entries, []string{service}, true)
-	assert.Len(suite.T(), names, 2)
+
-	assert.Contains(suite.T(), names, user1)
+	entries = getDirEntriesForEntry(t, ctx, entries[0])
-	assert.Contains(suite.T(), names, user2)
+	expectDirs(t, entries, []string{user1, user2}, true)
 
 	for _, entry := range entries {
-		dir, ok := entry.(fs.Directory)
+		userName := entry.Name()
-		require.True(suite.T(), ok)
 
-		subEntries, err := fs.GetAllEntries(ctx, dir)
+		entries = getDirEntriesForEntry(t, ctx, entry)
-		require.NoError(suite.T(), err)
+		expectDirs(t, entries, []string{category}, true)
-		require.Len(suite.T(), subEntries, 1)
-		assert.Contains(suite.T(), subEntries[0].Name(), emails)
 
-		subDir := subEntries[0].(fs.Directory)
+		entries = getDirEntriesForEntry(t, ctx, entries[0])
-		emailFiles, err := fs.GetAllEntries(ctx, subDir)
+		expectDirs(t, entries, []string{testInboxDir}, true)
-		require.NoError(suite.T(), err)
+
-		assert.Len(suite.T(), emailFiles, expectedFileCount[entry.Name()])
+		entries = getDirEntriesForEntry(t, ctx, entries[0])
+		assert.Len(t, entries, expectedFileCount[userName])
 	}
 
 	totalFileCount := 0
@@ -267,45 +295,22 @@ func (suite *KopiaUnitSuite) TestBuildDirectoryTree() {
 		totalFileCount += c
 	}
 
-	assert.Len(suite.T(), progress.pending, totalFileCount)
+	assert.Len(t, progress.pending, totalFileCount)
-}
-
-func (suite *KopiaUnitSuite) TestBuildDirectoryTree_NoAncestorDirs() {
-	tester.LogTimeOfTest(suite.T())
-
-	ctx := context.Background()
-	emails := "emails"
-	expectedFileCount := 42
-
-	progress := &corsoProgress{pending: map[string]*itemDetails{}}
-	collections := []data.Collection{
-		mockconnector.NewMockExchangeCollection(
-			[]string{emails},
-			expectedFileCount,
-		),
-	}
-
-	// Returned directory structure should look like:
-	// - emails
-	//   - 42 separate files
-	dirTree, err := inflateDirTree(ctx, collections, progress)
-	require.NoError(suite.T(), err)
-	assert.Equal(suite.T(), dirTree.Name(), emails)
-
-	entries, err := fs.GetAllEntries(ctx, dirTree)
-	require.NoError(suite.T(), err)
-	assert.Len(suite.T(), entries, expectedFileCount)
 }
 
 func (suite *KopiaUnitSuite) TestBuildDirectoryTree_MixedDirectory() {
 	ctx := context.Background()
+	subdir := "subfolder"
 	// Test multiple orders of items because right now order can matter. Both
 	// orders result in a directory structure like:
 	// - a-tenant
-	//   - user1
+	//   - exchange
-	//     - emails
+	//     - user1
-	//       - 5 separate files
+	//       - emails
-	//     - 42 separate files
+	//         - Inbox
+	//           - subfolder
+	//             - 5 separate files
+	//           - 42 separate files
 	table := []struct {
 		name   string
 		layout []data.Collection
@@ -314,11 +319,11 @@ func (suite *KopiaUnitSuite) TestBuildDirectoryTree_MixedDirectory() {
 			name: "SubdirFirst",
 			layout: []data.Collection{
 				mockconnector.NewMockExchangeCollection(
-					[]string{testTenant, testUser, testEmailDir},
+					[]string{testTenant, service, testUser, category, testInboxDir, subdir},
 					5,
 				),
 				mockconnector.NewMockExchangeCollection(
-					[]string{testTenant, testUser},
+					[]string{testTenant, service, testUser, category, testInboxDir},
 					42,
 				),
 			},
@@ -327,11 +332,11 @@ func (suite *KopiaUnitSuite) TestBuildDirectoryTree_MixedDirectory() {
 			name: "SubdirLast",
 			layout: []data.Collection{
 				mockconnector.NewMockExchangeCollection(
-					[]string{testTenant, testUser},
+					[]string{testTenant, service, testUser, category, testInboxDir},
 					42,
 				),
 				mockconnector.NewMockExchangeCollection(
-					[]string{testTenant, testUser, testEmailDir},
+					[]string{testTenant, service, testUser, category, testInboxDir, subdir},
 					5,
 				),
 			},
@@ -348,14 +353,19 @@ func (suite *KopiaUnitSuite) TestBuildDirectoryTree_MixedDirectory() {
 
 			entries, err := fs.GetAllEntries(ctx, dirTree)
 			require.NoError(t, err)
-			require.Len(t, entries, 1)
-			assert.Equal(t, testUser, entries[0].Name())
 
-			d, ok := entries[0].(fs.Directory)
+			expectDirs(t, entries, []string{service}, true)
-			require.True(t, ok, "returned entry is not a directory")
 
-			entries, err = fs.GetAllEntries(ctx, d)
+			entries = getDirEntriesForEntry(t, ctx, entries[0])
-			require.NoError(t, err)
+			expectDirs(t, entries, []string{testUser}, true)
+
+			entries = getDirEntriesForEntry(t, ctx, entries[0])
+			expectDirs(t, entries, []string{category}, true)
+
+			entries = getDirEntriesForEntry(t, ctx, entries[0])
+			expectDirs(t, entries, []string{testInboxDir}, true)
+
+			entries = getDirEntriesForEntry(t, ctx, entries[0])
 			// 42 files and 1 subdirectory.
 			assert.Len(t, entries, 43)
 
@@ -368,13 +378,12 @@ func (suite *KopiaUnitSuite) TestBuildDirectoryTree_MixedDirectory() {
 				}
 
 				subDirs = append(subDirs, d)
-				assert.Equal(t, testEmailDir, e.Name())
+				assert.Equal(t, "subfolder", d.Name())
 			}
 
 			require.Len(t, subDirs, 1)
 
-			entries, err = fs.GetAllEntries(ctx, subDirs[0])
+			entries = getDirEntriesForEntry(t, ctx, entries[0])
-			assert.NoError(t, err)
 			assert.Len(t, entries, 5)
 		})
 	}
@@ -485,11 +494,11 @@ func (suite *KopiaIntegrationSuite) TestBackupCollections() {
 
 	collections := []data.Collection{
 		mockconnector.NewMockExchangeCollection(
-			[]string{"a-tenant", "user1", "emails"},
+			[]string{"a-tenant", service, "user1", category, testInboxDir},
 			5,
 		),
 		mockconnector.NewMockExchangeCollection(
-			[]string{"a-tenant", "user2", "emails"},
+			[]string{"a-tenant", service, "user2", category, testInboxDir},
 			42,
 		),
 	}
@@ -497,7 +506,7 @@ func (suite *KopiaIntegrationSuite) TestBackupCollections() {
 	stats, rp, err := suite.w.BackupCollections(suite.ctx, collections)
 	assert.NoError(t, err)
 	assert.Equal(t, stats.TotalFileCount, 47)
-	assert.Equal(t, stats.TotalDirectoryCount, 5)
+	assert.Equal(t, stats.TotalDirectoryCount, 8)
 	assert.Equal(t, stats.IgnoredErrorCount, 0)
 	assert.Equal(t, stats.ErrorCount, 0)
 	assert.False(t, stats.Incomplete)
@@ -518,16 +527,16 @@ func (suite *KopiaIntegrationSuite) TestRestoreAfterCompressionChange() {
 	tid := uuid.NewString()
 	p1 := []string{
 		tid,
-		path.ExchangeService.String(),
+		service,
 		"uid",
-		path.EmailCategory.String(),
+		category,
 		"fid",
 	}
 	p2 := []string{
 		tid,
-		path.ExchangeService.String(),
+		service,
 		"uid2",
-		path.EmailCategory.String(),
+		category,
 		"fid",
 	}
 	dc1 := mockconnector.NewMockExchangeCollection(p1, 1)
@@ -833,16 +842,16 @@ func (suite *KopiaSimpleRepoIntegrationSuite) TestRestoreMultipleItems() {
 	tid := uuid.NewString()
 	p1 := []string{
 		tid,
-		path.ExchangeService.String(),
+		service,
 		"uid",
-		path.EmailCategory.String(),
+		category,
 		"fid",
 	}
 	p2 := []string{
 		tid,
-		path.ExchangeService.String(),
+		service,
 		"uid2",
-		path.EmailCategory.String(),
+		category,
 		"fid",
 	}
 	dc1 := mockconnector.NewMockExchangeCollection(p1, 1)
@@ -914,13 +923,13 @@ func (suite *KopiaIntegrationSuite) TestDeleteSnapshot() {
 	t := suite.T()
 
 	dc1 := mockconnector.NewMockExchangeCollection(
-		[]string{"a-tenant", "user1", "emails"},
+		[]string{"a-tenant", service, "user1", category, testInboxDir},
 		5,
 	)
 	collections := []data.Collection{
 		dc1,
 		mockconnector.NewMockExchangeCollection(
-			[]string{"a-tenant", "user2", "emails"},
+			[]string{"a-tenant", service, "user2", category, testInboxDir},
 			42,
 		),
 	}

src/internal/path/path.go

@@ -74,6 +74,10 @@ type Path interface {
 	// If removing the right-most element would discard one of the required prefix
 	// elements then an error is returned.
 	Dir() (Path, error)
+	// Append returns a new Path object with the given element added to the end of
+	// the old Path if possible. If the old Path is an item Path then Append
+	// returns an error.
+	Append(element string, isItem bool) (Path, error)
 }
 
 // Builder is a simple path representation that only tracks path elements. It

src/internal/path/resource_path.go

@@ -177,3 +177,19 @@ func (rp dataLayerResourcePath) Dir() (Path, error) {
 		hasItem:  false,
 	}, nil
 }
+
+func (rp dataLayerResourcePath) Append(
+	element string,
+	isItem bool,
+) (Path, error) {
+	if rp.hasItem {
+		return nil, errors.New("appending to an item path")
+	}
+
+	return &dataLayerResourcePath{
+		Builder:  *rp.Builder.Append(element),
+		service:  rp.service,
+		category: rp.category,
+		hasItem:  isItem,
+	}, nil
+}

src/internal/path/resource_path_test.go

@@ -348,3 +348,49 @@ func (suite *PopulatedDataLayerResourcePath) TestItem() {
 		})
 	}
 }
+
+func (suite *PopulatedDataLayerResourcePath) TestAppend() {
+	newElement := "someElement"
+	isItem := []struct {
+		name    string
+		hasItem bool
+		// Used if the starting path is a folder.
+		expectedFolder string
+		expectedItem   string
+	}{
+		{
+			name:           "Item",
+			hasItem:        true,
+			expectedFolder: strings.Join(rest, "/"),
+			expectedItem:   newElement,
+		},
+		{
+			name:    "Directory",
+			hasItem: false,
+			expectedFolder: strings.Join(
+				append(append([]string{}, rest...), newElement),
+				"/",
+			),
+			expectedItem: "",
+		},
+	}
+
+	for _, m := range modes {
+		suite.T().Run(m.name, func(t1 *testing.T) {
+			for _, test := range isItem {
+				t1.Run(test.name, func(t *testing.T) {
+					newPath, err := suite.paths[m.isItem].Append(newElement, test.hasItem)
+
+					// Items don't allow appending.
+					if m.isItem {
+						assert.Error(t, err)
+						return
+					}
+
+					assert.Equal(t, test.expectedFolder, newPath.Folder())
+					assert.Equal(t, test.expectedItem, newPath.Item())
+				})
+			}
+		})
+	}
+}