test: users context traits and populate users fields (#5135)

.github/workflows/tests.yaml

@@ -168,6 +168,11 @@ jobs:
           go-version-file: 'go.mod'
       - run: go version
       - run: go mod download
+      - name: Login to DockerHub
+        uses: docker/login-action@v3.3.0
+        with:
+          username: rudderlabs
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
       - name: Package Unit [ ${{ matrix.package }} ]
         env:
           TEST_KAFKA_CONFLUENT_CLOUD_HOST: ${{ secrets.TEST_KAFKA_CONFLUENT_CLOUD_HOST }}

warehouse/integrations/azure-synapse/azure_synapse_test.go

@@ -153,6 +153,8 @@ func TestIntegration(t *testing.T) {
 					WithConfigOption("endPoint", minioEndpoint).
 					WithConfigOption("useRudderStorage", false).
 					WithConfigOption("syncFrequency", "30").
+					WithConfigOption("allowUsersContextTraits", true).
+					WithConfigOption("underscoreDivideNumbers", true).
 					Build()
 
 				workspaceConfig := backendconfigtest.NewConfigBuilder().

warehouse/integrations/clickhouse/clickhouse_test.go

@@ -222,7 +222,9 @@ func TestIntegration(t *testing.T) {
 					WithConfigOption("secure", false).
 					WithConfigOption("endPoint", minioEndpoint).
 					WithConfigOption("useRudderStorage", false).
-					WithConfigOption("syncFrequency", "30")
+					WithConfigOption("syncFrequency", "30").
+					WithConfigOption("allowUsersContextTraits", true).
+					WithConfigOption("underscoreDivideNumbers", true)
 				for k, v := range tc.configOverride {
 					destinationBuilder = destinationBuilder.WithConfigOption(k, v)
 				}

warehouse/integrations/datalake/datalake_test.go

@@ -333,7 +333,9 @@ func TestIntegration(t *testing.T) {
 					WithID(destinationID).
 					WithRevisionID(destinationID).
 					WithConfigOption("namespace", namespace).
-					WithConfigOption("syncFrequency", "30")
+					WithConfigOption("syncFrequency", "30").
+					WithConfigOption("allowUsersContextTraits", true).
+					WithConfigOption("underscoreDivideNumbers", true)
 				for k, v := range tc.configOverride {
 					destinationBuilder = destinationBuilder.WithConfigOption(k, v)
 				}
@@ -570,6 +572,8 @@ func TestIntegration(t *testing.T) {
 			WithConfigOption("s3ForcePathStyle", true).
 			WithConfigOption("disableSSL", true).
 			WithConfigOption("prefix", "some-prefix").
+			WithConfigOption("allowUsersContextTraits", true).
+			WithConfigOption("underscoreDivideNumbers", true).
 			Build()
 		workspaceConfig := backendconfigtest.NewConfigBuilder().
 			WithSource(
@@ -853,6 +857,8 @@ func TestIntegration(t *testing.T) {
 			WithConfigOption("s3ForcePathStyle", true).
 			WithConfigOption("disableSSL", true).
 			WithConfigOption("prefix", "some-prefix").
+			WithConfigOption("allowUsersContextTraits", true).
+			WithConfigOption("underscoreDivideNumbers", true).
 			Build()
 		workspaceConfig := backendconfigtest.NewConfigBuilder().
 			WithSource(

warehouse/integrations/deltalake/deltalake_test.go

@@ -281,7 +281,9 @@ func TestIntegration(t *testing.T) {
 					WithConfigOption("enableSSE", false).
 					WithConfigOption("accountName", credentials.AccountName).
 					WithConfigOption("accountKey", credentials.AccountKey).
-					WithConfigOption("syncFrequency", "30")
+					WithConfigOption("syncFrequency", "30").
+					WithConfigOption("allowUsersContextTraits", true).
+					WithConfigOption("underscoreDivideNumbers", true)
 				for k, v := range tc.configOverride {
 					destinationBuilder = destinationBuilder.WithConfigOption(k, v)
 				}

warehouse/integrations/mssql/mssql_test.go

@@ -183,6 +183,8 @@ func TestIntegration(t *testing.T) {
 					WithConfigOption("endPoint", minioEndpoint).
 					WithConfigOption("useRudderStorage", false).
 					WithConfigOption("syncFrequency", "30").
+					WithConfigOption("allowUsersContextTraits", true).
+					WithConfigOption("underscoreDivideNumbers", true).
 					Build()
 
 				workspaceConfig := backendconfigtest.NewConfigBuilder().

warehouse/integrations/postgres/postgres_test.go

@@ -309,7 +309,9 @@ func TestIntegration(t *testing.T) {
 					WithConfigOption("useSSL", false).
 					WithConfigOption("endPoint", minioEndpoint).
 					WithConfigOption("useRudderStorage", false).
-					WithConfigOption("syncFrequency", "30")
+					WithConfigOption("syncFrequency", "30").
+					WithConfigOption("allowUsersContextTraits", true).
+					WithConfigOption("underscoreDivideNumbers", true)
 				for k, v := range tc.configOverride {
 					destinationBuilder = destinationBuilder.WithConfigOption(k, v)
 				}
@@ -530,6 +532,8 @@ func TestIntegration(t *testing.T) {
 					WithConfigOption("sshHost", tunnelledSSHHost).
 					WithConfigOption("sshPort", strconv.Itoa(sshPort)).
 					WithConfigOption("sshPrivateKey", strings.ReplaceAll(tunnelledPrivateKey, "\\n", "\n")).
+					WithConfigOption("allowUsersContextTraits", true).
+					WithConfigOption("underscoreDivideNumbers", true).
 					Build()
 
 				workspaceConfig := backendconfigtest.NewConfigBuilder().

warehouse/integrations/redshift/redshift_test.go

@@ -443,7 +443,9 @@ func TestIntegration(t *testing.T) {
 					WithConfigOption("namespace", namespace).
 					WithConfigOption("enableSSE", false).
 					WithConfigOption("useRudderStorage", false).
-					WithConfigOption("syncFrequency", "30")
+					WithConfigOption("syncFrequency", "30").
+					WithConfigOption("allowUsersContextTraits", true).
+					WithConfigOption("underscoreDivideNumbers", true)
 				for k, v := range tc.configOverride {
 					destinationBuilder = destinationBuilder.WithConfigOption(k, v)
 				}

warehouse/integrations/snowflake/snowflake_test.go

@@ -486,7 +486,9 @@ func TestIntegration(t *testing.T) {
 					WithConfigOption("namespace", namespace).
 					WithConfigOption("enableSSE", false).
 					WithConfigOption("useRudderStorage", false).
-					WithConfigOption("syncFrequency", "30")
+					WithConfigOption("syncFrequency", "30").
+					WithConfigOption("allowUsersContextTraits", true).
+					WithConfigOption("underscoreDivideNumbers", true)
 				for k, v := range tc.configOverride {
 					destinationBuilder = destinationBuilder.WithConfigOption(k, v)
 				}

warehouse/integrations/testhelper/records.go

@@ -207,7 +207,6 @@ func UploadJobUsersRecords(userIDFormat, sourceID, destID, destType string) [][]
 // UploadJobUsersRecordsForDatalake returns a set of records for testing upload job users scenarios.
 // It uses upload-job.events-1.json, upload-job.events-2.json as the source of data.
 // sent_at, timestamp, original_timestamp will not be present in the records.
-// TODO: Once we start populating it in the records, we need to update the expected records.
 func UploadJobUsersRecordsForDatalake(userIDFormat, sourceID, destID, destType string) [][]string {
 	uuidTS := timeutil.Now().Format("2006-01-02")
 	return [][]string{
@@ -235,12 +234,11 @@ func UploadJobUsersAppendRecords(userIDFormat, sourceID, destID, destType string
 // UploadJobUsersAppendRecordsUsingUsersLoadFiles returns a set of records for testing upload job users scenarios.
 // It uses twice upload-job.events-1.json as the source of data.
 // sent_at, timestamp, original_timestamp will not be present in the records.
-// TODO: Once we start populating it in the records, we need to update the expected records.
 func UploadJobUsersAppendRecordsUsingUsersLoadFiles(userIDFormat, sourceID, destID, destType string) [][]string {
 	uuidTS := timeutil.Now().Format("2006-01-02")
 	return [][]string{
-		{sourceID, destType, "[::1]", "Richard Hendricks", "non escaped column", "non escaped column", "2", "", "2", "[::1]", "non escaped column", "rhedricks+4@example.com", "", destID, "rhedricks+4@example.com", "non escaped column", "HTTP", userIDFormat, uuidTS, "2023-05-12T04:26:48Z", "Richard Hendricks", ""},
-		{sourceID, destType, "[::1]", "Richard Hendricks", "non escaped column", "non escaped column", "2", "", "2", "[::1]", "non escaped column", "rhedricks+4@example.com", "", destID, "rhedricks+4@example.com", "non escaped column", "HTTP", userIDFormat, uuidTS, "2023-05-12T04:26:48Z", "Richard Hendricks", ""},
+		{sourceID, destType, "[::1]", "Richard Hendricks", "non escaped column", "non escaped column", "2", "2023-05-12T04:26:48Z", "2", "[::1]", "non escaped column", "rhedricks+4@example.com", "2023-05-12T04:26:48Z", destID, "rhedricks+4@example.com", "non escaped column", "HTTP", userIDFormat, uuidTS, "2023-05-12T04:26:48Z", "Richard Hendricks", "2023-05-12T04:26:48Z"},
+		{sourceID, destType, "[::1]", "Richard Hendricks", "non escaped column", "non escaped column", "2", "2023-05-12T04:26:48Z", "2", "[::1]", "non escaped column", "rhedricks+4@example.com", "2023-05-12T04:26:48Z", destID, "rhedricks+4@example.com", "non escaped column", "HTTP", userIDFormat, uuidTS, "2023-05-12T04:26:48Z", "Richard Hendricks", "2023-05-12T04:26:48Z"},
 	}
 }
 
@@ -258,12 +256,11 @@ func UploadJobUsersMergeRecord(userIDFormat, sourceID, destID, destType string)
 // sent_at, timestamp, original_timestamp will not be present in the records.
 // For AggregatingMergeTree ClickHouse replaces all rows with the same primary key (or more accurately, with the same sorting key) with a single row (within a one data part) that stores a combination of states of aggregate functions.
 // So received_at will be record for the first record only.
-// TODO: Once we start populating it in the records, we need to update the expected records.
 func UploadJobUsersRecordsUsingUsersLoadFilesForClickhouse(userIDFormat, sourceID, destID, destType string) [][]string {
 	uuidTS := timeutil.Now().Format("2006-01-02")
 	return [][]string{
-		{sourceID, destType, "[::1]", "Richard Hendricks", "non escaped column", "non escaped column", "2", "", "2", "[::1]", "non escaped column", "rhedricks+4@example.com", "", destID, "rhedricks+4@example.com", "non escaped column", "HTTP", userIDFormat, uuidTS, "2023-05-12T04:08:48Z", "Richard Hendricks", ""},
-		{sourceID, destType, "[::1]", "Richard Hendricks", "non escaped column", "non escaped column", "2", "", "2", "[::1]", "non escaped column", "rhedricks+8@example.com", "", destID, "rhedricks+8@example.com", "non escaped column", "HTTP", userIDFormat, uuidTS, "2023-05-12T04:32:48Z", "Richard Hendricks", ""},
+		{sourceID, destType, "[::1]", "Richard Hendricks", "non escaped column", "non escaped column", "2", "2023-05-12T04:26:48Z", "2", "[::1]", "non escaped column", "rhedricks+4@example.com", "2023-05-12T04:26:48Z", destID, "rhedricks+4@example.com", "non escaped column", "HTTP", userIDFormat, uuidTS, "2023-05-12T04:08:48Z", "Richard Hendricks", "2023-05-12T04:26:48Z"},
+		{sourceID, destType, "[::1]", "Richard Hendricks", "non escaped column", "non escaped column", "2", "2023-05-12T04:50:48Z", "2", "[::1]", "non escaped column", "rhedricks+8@example.com", "2023-05-12T04:50:48Z", destID, "rhedricks+8@example.com", "non escaped column", "HTTP", userIDFormat, uuidTS, "2023-05-12T04:32:48Z", "Richard Hendricks", "2023-05-12T04:50:48Z"},
 	}
 }