feat: add config update hook to cdk integration tests (#50849)

airbyte-cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/load/check/CheckIntegrationTest.kt

@@ -6,6 +6,8 @@ package io.airbyte.cdk.load.check
 
 import io.airbyte.cdk.command.ConfigurationSpecification
 import io.airbyte.cdk.command.FeatureFlag
+import io.airbyte.cdk.load.test.util.ConfigurationUpdater
+import io.airbyte.cdk.load.test.util.FakeConfigurationUpdater
 import io.airbyte.cdk.load.test.util.FakeDataDumper
 import io.airbyte.cdk.load.test.util.IntegrationTest
 import io.airbyte.cdk.load.test.util.NoopDestinationCleaner
@@ -27,16 +29,18 @@ data class CheckTestConfig(val configPath: Path, val featureFlags: Set<FeatureFl
 open class CheckIntegrationTest<T : ConfigurationSpecification>(
     val successConfigFilenames: List<CheckTestConfig>,
     val failConfigFilenamesAndFailureReasons: Map<CheckTestConfig, Pattern>,
+    configUpdater: ConfigurationUpdater = FakeConfigurationUpdater,
 ) :
     IntegrationTest(
-        FakeDataDumper,
-        NoopDestinationCleaner,
-        NoopExpectedRecordMapper,
+        dataDumper = FakeDataDumper,
+        destinationCleaner = NoopDestinationCleaner,
+        recordMangler = NoopExpectedRecordMapper,
+        configUpdater = configUpdater,
     ) {
     @Test
     open fun testSuccessConfigs() {
         for ((path, featureFlags) in successConfigFilenames) {
-            val config = Files.readString(path, StandardCharsets.UTF_8)
+            val config = updateConfig(Files.readString(path, StandardCharsets.UTF_8))
             val process =
                 destinationProcessFactory.createDestinationProcess(
                     "check",
@@ -64,7 +68,7 @@ open class CheckIntegrationTest<T : ConfigurationSpecification>(
     open fun testFailConfigs() {
         for ((checkTestConfig, failurePattern) in failConfigFilenamesAndFailureReasons) {
             val (path, featureFlags) = checkTestConfig
-            val config = Files.readString(path)
+            val config = updateConfig(Files.readString(path))
             val process =
                 destinationProcessFactory.createDestinationProcess(
                     "check",

airbyte-cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/load/spec/SpecTest.kt

@@ -36,9 +36,9 @@ import org.junit.jupiter.api.assertAll
  */
 abstract class SpecTest :
     IntegrationTest(
-        FakeDataDumper,
-        NoopDestinationCleaner,
-        NoopExpectedRecordMapper,
+        dataDumper = FakeDataDumper,
+        destinationCleaner = NoopDestinationCleaner,
+        recordMangler = NoopExpectedRecordMapper,
     ) {
     private val testResourcesPath = Path.of("src/test-integration/resources")
 

airbyte-cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/load/test/util/ConfigurationUpdater.kt

@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.load.test.util
+
+/** Utility that may update/modify a connector configuration for test purposes. */
+interface ConfigurationUpdater {
+
+    /**
+     * May modify one or more entry in the provided configuration.
+     * @param config The connector configuration.
+     * @return The potentially modified configuration.
+     */
+    fun update(config: String): String
+}
+
+/**
+ * Basic implementation of the [ConfigurationUpdater] interface that does not modify the
+ * configuration.
+ */
+object FakeConfigurationUpdater : ConfigurationUpdater {
+    override fun update(config: String): String = config
+}

airbyte-cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/load/test/util/IntegrationTest.kt

@@ -55,6 +55,7 @@ abstract class IntegrationTest(
     val nameMapper: NameMapper = NoopNameMapper,
     /** See [RecordDiffer.nullEqualsUnset]. */
     val nullEqualsUnset: Boolean = false,
+    val configUpdater: ConfigurationUpdater = FakeConfigurationUpdater,
 ) {
     // Intentionally don't inject the actual destination process - we need a full factory
     // because some tests want to run multiple syncs, so we need to run the destination
@@ -268,6 +269,8 @@ abstract class IntegrationTest(
         }
     }
 
+    fun updateConfig(config: String): String = configUpdater.update(config)
+
     companion object {
         val randomizedNamespaceRegex = Regex("test(\\d{8})[A-Za-z]{4}")
         val randomizedNamespaceDateFormatter: DateTimeFormatter =

airbyte-cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/load/write/BasicFunctionalityIntegrationTest.kt

@@ -41,9 +41,11 @@ import io.airbyte.cdk.load.message.InputRecord
 import io.airbyte.cdk.load.message.InputStreamCheckpoint
 import io.airbyte.cdk.load.message.Meta.Change
 import io.airbyte.cdk.load.message.StreamCheckpoint
+import io.airbyte.cdk.load.test.util.ConfigurationUpdater
 import io.airbyte.cdk.load.test.util.DestinationCleaner
 import io.airbyte.cdk.load.test.util.DestinationDataDumper
 import io.airbyte.cdk.load.test.util.ExpectedRecordMapper
+import io.airbyte.cdk.load.test.util.FakeConfigurationUpdater
 import io.airbyte.cdk.load.test.util.IntegrationTest
 import io.airbyte.cdk.load.test.util.NameMapper
 import io.airbyte.cdk.load.test.util.NoopExpectedRecordMapper
@@ -130,8 +132,18 @@ abstract class BasicFunctionalityIntegrationTest(
     val allTypesBehavior: AllTypesBehavior,
     val failOnUnknownTypes: Boolean = false,
     nullEqualsUnset: Boolean = false,
-) : IntegrationTest(dataDumper, destinationCleaner, recordMangler, nameMapper, nullEqualsUnset) {
-    val parsedConfig = ValidatedJsonUtils.parseOne(configSpecClass, configContents)
+    configUpdater: ConfigurationUpdater = FakeConfigurationUpdater,
+) :
+    IntegrationTest(
+        dataDumper = dataDumper,
+        destinationCleaner = destinationCleaner,
+        recordMangler = recordMangler,
+        nameMapper = nameMapper,
+        nullEqualsUnset = nullEqualsUnset,
+        configUpdater = configUpdater
+    ) {
+    val parsedConfig =
+        ValidatedJsonUtils.parseOne(configSpecClass, configUpdater.update(configContents))
 
     @Test
     open fun testBasicWrite() {