Function CheckImportFileTracking

integration/metadb_test.go:25–86

(t *testing.T, importer *i.Importer)

)

func CheckImportFileTracking(t *testing.T, importer *i.Importer) { // uses valid dataset
	t.Helper()
	var result struct {
		Count uint64 `ch:"count"`
	}

	// set context with importID and database parameters
	ctx := clickhouse.Context(importer.Database.GetContext(), clickhouse.WithParameters(clickhouse.Parameters{
		"importID": importer.ImportID.Hex(),
		"database": importer.Database.GetSelectedDB(),
	}))

	// make sure total (post duplicate filtering) file count to be imported matches file count in metadb
	err := importer.Database.Conn.QueryRow(ctx, `
			SELECT count() AS count FROM metadatabase.files
			WHERE import_id = unhex({importID:String}) AND database = {database:String}
		`).ScanStruct(&result)
	require.NoError(t, err)
	require.EqualValues(t, importer.TotalFileCount, result.Count, "total file count matches imported file count")

	var allFiles []string
	allFiles = append(allFiles, importer.FileMap[c.ConnPrefix]...)
	allFiles = append(allFiles, importer.FileMap[c.OpenConnPrefix]...)
	allFiles = append(allFiles, importer.FileMap[c.HTTPPrefix]...)
	allFiles = append(allFiles, importer.FileMap[c.OpenHTTPPrefix]...)
	allFiles = append(allFiles, importer.FileMap[c.SSLPrefix]...)
	allFiles = append(allFiles, importer.FileMap[c.OpenSSLPrefix]...)
	allFiles = append(allFiles, importer.FileMap[c.DNSPrefix]...)

	var filesResult struct {
		Files []string `ch:"files"`
	}

	// set context with importID parameter
	ctx = clickhouse.Context(importer.Database.GetContext(), clickhouse.WithParameters(clickhouse.Parameters{
		"importID": importer.ImportID.Hex(),
	}))

	// make sure all files in created filemap were imported and saved in metadb
	err = importer.Database.Conn.QueryRow(ctx, `
			SELECT groupArray(path) as files FROM metadatabase.files
			WHERE import_id = unhex({importID:String})
		`).ScanStruct(&filesResult)
	require.NoError(t, err)
	require.ElementsMatch(t, allFiles, filesResult.Files, "files in filemap created for import match list of imported files in metadb")

	// Make sure duplicate files in same import don't get imported (conn.log & conn.log.gz)
	for _, file := range allFiles {
		if strings.HasSuffix(file, ".gz") {
			hasUncompressedDuplicate := slices.Contains(allFiles, strings.TrimSuffix(file, ".gz"))
			require.False(t, hasUncompressedDuplicate, "ucompressed duplicate of .gz file was not imported")
		} else {
			hasCompressedDuplicate := slices.Contains(allFiles, (file + ".gz"))
			require.False(t, hasCompressedDuplicate, "compressed duplicate of uncompressed file was not imported")
		}
	}

	// importing again should fail because all files are imported