Method UpdateCSVs

backend/helpers/e2ehelper/data_flow_tester.go:259–305

UpdateCSVs scans the supplied directory for .csv files and applies the `updater` function to each. It saves back the result. This is useful when you need to bulk-change a lot of entries across many CSV files in a systematic way.

(csvDir string, updater func(fileName string, record *pluginhelper.CsvRecord))

// UpdateCSVs scans the supplied directory for .csv files and applies the `updater` function to each. It saves back the result.
// This is useful when you need to bulk-change a lot of entries across many CSV files in a systematic way.
func (t *DataFlowTester) UpdateCSVs(csvDir string, updater func(fileName string, record *pluginhelper.CsvRecord)) errors.Error {
	var csvFiles []string
	err := filepath.WalkDir(csvDir, func(file string, dir fs.DirEntry, e error) error {
		if e != nil {
			return e
		}
		if filepath.Ext(file) == ".csv" {
			csvFiles = append(csvFiles, file)
		}
		return nil
	})
	if err != nil {
		return errors.Convert(err)
	}
	for _, csvFile := range csvFiles {
		csvIter, err := pluginhelper.NewCsvFileIterator(csvFile)
		if err != nil {
			return errors.Convert(err)
		}
		columns := csvIter.GetColumns()
		var entries [][]string
		func() {
			defer csvIter.Close()
			for csvIter.HasNext() {
				records := csvIter.FetchRecords()
				var csvEntry []string
				for _, record := range records {
					updater(filepath.Base(csvFile), record)
					csvEntry = append(csvEntry, cast.ToString(record.Value))
				}
				entries = append(entries, csvEntry)
			}
		}()
		if len(entries) == 0 {
			continue
		}
		func() {
			csvWriter, _ := pluginhelper.NewCsvFileWriter(csvFile, columns)
			defer csvWriter.Close()
			for _, entry := range entries {
				csvWriter.Write(entry)
			}
			csvWriter.Flush()
		}()
	}
	return nil
}

// CreateSnapshot reads rows from database and write them into .csv file.
func (t *DataFlowTester) CreateSnapshot(dst schema.Tabler, opts TableOptions) {