Bulk CDK: Enable setting deployment mode in integration tests; update…

… destinations check test appropriately (#45929)
airbytehq · Oct 1, 2024 · 7213920 · 7213920
1 parent 4043fe0
commit 7213920
Show file tree

Hide file tree

Showing 8 changed files with 79 additions and 19 deletions.
diff --git a/airbyte-cdk/bulk/core/base/src/main/kotlin/io/airbyte/cdk/AirbyteConnectorRunner.kt b/airbyte-cdk/bulk/core/base/src/main/kotlin/io/airbyte/cdk/AirbyteConnectorRunner.kt
@@ -8,6 +8,7 @@ import io.micronaut.context.ApplicationContext
 import io.micronaut.context.RuntimeBeanDefinition
 import io.micronaut.context.env.CommandLinePropertySource
 import io.micronaut.context.env.Environment
+import io.micronaut.context.env.MapPropertySource
 import io.micronaut.core.cli.CommandLine as MicronautCommandLine
 import java.nio.file.Path
 import kotlin.system.exitProcess
@@ -35,9 +36,10 @@ class AirbyteSourceRunner(
 class AirbyteDestinationRunner(
     /** CLI args. */
     args: Array<out String>,
+    testEnvironments: Map<String, String> = emptyMap(),
     /** Micronaut bean definition overrides, used only for tests. */
     vararg testBeanDefinitions: RuntimeBeanDefinition<*>,
-) : AirbyteConnectorRunner("destination", args, testBeanDefinitions) {
+) : AirbyteConnectorRunner("destination", args, testBeanDefinitions, testEnvironments) {
     companion object {
         @JvmStatic
         fun run(vararg args: String) {
@@ -54,6 +56,7 @@ sealed class AirbyteConnectorRunner(
     val connectorType: String,
     val args: Array<out String>,
     val testBeanDefinitions: Array<out RuntimeBeanDefinition<*>>,
+    val testProperties: Map<String, String> = emptyMap(),
 ) {
     val envs: Array<String> = arrayOf(Environment.CLI, connectorType)
 
@@ -69,9 +72,13 @@ sealed class AirbyteConnectorRunner(
         val ctx: ApplicationContext =
             ApplicationContext.builder(R::class.java, *envs)
                 .propertySources(
-                    airbytePropertySource,
-                    commandLinePropertySource,
-                    MetadataYamlPropertySource(),
+                    *listOfNotNull(
+                            MapPropertySource("additional_properties", testProperties),
+                            airbytePropertySource,
+                            commandLinePropertySource,
+                            MetadataYamlPropertySource(),
+                        )
+                        .toTypedArray(),
                 )
                 .beanDefinitions(*testBeanDefinitions)
                 .start()

diff --git a/airbyte-cdk/bulk/core/base/src/testFixtures/kotlin/io/airbyte/cdk/command/CliRunner.kt b/airbyte-cdk/bulk/core/base/src/testFixtures/kotlin/io/airbyte/cdk/command/CliRunner.kt
@@ -52,6 +52,7 @@ data object CliRunner {
         catalog: ConfiguredAirbyteCatalog? = null,
         state: List<AirbyteStateMessage>? = null,
         inputStream: InputStream,
+        testProperties: Map<String, String> = emptyMap(),
     ): CliRunnable {
         val inputBeanDefinition: RuntimeBeanDefinition<InputStream> =
             RuntimeBeanDefinition.builder(InputStream::class.java) { inputStream }
@@ -60,7 +61,12 @@ data object CliRunner {
         val out = CliRunnerOutputStream()
         val runnable: Runnable =
             makeRunnable(op, config, catalog, state) { args: Array<String> ->
-                AirbyteDestinationRunner(args, inputBeanDefinition, out.beanDefinition)
+                AirbyteDestinationRunner(
+                    args,
+                    testProperties,
+                    inputBeanDefinition,
+                    out.beanDefinition
+                )
             }
         return CliRunnable(runnable, out.results)
     }

diff --git a/.../bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/test/check/CheckIntegrationTest.kt b/.../bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/test/check/CheckIntegrationTest.kt
@@ -10,6 +10,7 @@ import io.airbyte.cdk.test.util.FakeDataDumper
 import io.airbyte.cdk.test.util.IntegrationTest
 import io.airbyte.cdk.test.util.NoopDestinationCleaner
 import io.airbyte.cdk.test.util.NoopExpectedRecordMapper
+import io.airbyte.cdk.test.util.TestDeploymentMode
 import io.airbyte.protocol.models.v0.AirbyteConnectionStatus
 import io.airbyte.protocol.models.v0.AirbyteMessage
 import java.nio.charset.StandardCharsets
@@ -21,10 +22,12 @@ import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertAll
 
+data class CheckTestConfig(val configPath: String, val deploymentMode: TestDeploymentMode)
+
 open class CheckIntegrationTest<T : ConfigurationSpecification>(
     val configurationClass: Class<T>,
-    val successConfigFilenames: List<String>,
-    val failConfigFilenamesAndFailureReasons: Map<String, Pattern>,
+    val successConfigFilenames: List<CheckTestConfig>,
+    val failConfigFilenamesAndFailureReasons: Map<CheckTestConfig, Pattern>,
 ) :
     IntegrationTest(
         FakeDataDumper,
@@ -33,11 +36,15 @@ open class CheckIntegrationTest<T : ConfigurationSpecification>(
     ) {
     @Test
     open fun testSuccessConfigs() {
-        for (path in successConfigFilenames) {
+        for ((path, deploymentMode) in successConfigFilenames) {
             val fileContents = Files.readString(Path.of(path), StandardCharsets.UTF_8)
             val config = ValidatedJsonUtils.parseOne(configurationClass, fileContents)
             val process =
-                destinationProcessFactory.createDestinationProcess("check", config = config)
+                destinationProcessFactory.createDestinationProcess(
+                    "check",
+                    config = config,
+                    deploymentMode = deploymentMode,
+                )
             process.run()
             val messages = process.readMessages()
             val checkMessages = messages.filter { it.type == AirbyteMessage.Type.CONNECTION_STATUS }
@@ -49,18 +56,24 @@ open class CheckIntegrationTest<T : ConfigurationSpecification>(
             )
             assertEquals(
                 AirbyteConnectionStatus.Status.SUCCEEDED,
-                checkMessages.first().connectionStatus.status
+                checkMessages.first().connectionStatus.status,
+                "Expected check to be successful, but message was ${checkMessages.first().connectionStatus}"
             )
         }
     }
 
     @Test
     open fun testFailConfigs() {
-        for ((path, failurePattern) in failConfigFilenamesAndFailureReasons) {
+        for ((checkTestConfig, failurePattern) in failConfigFilenamesAndFailureReasons) {
+            val (path, deploymentMode) = checkTestConfig
             val fileContents = Files.readString(Path.of(path))
             val config = ValidatedJsonUtils.parseOne(configurationClass, fileContents)
             val process =
-                destinationProcessFactory.createDestinationProcess("check", config = config)
+                destinationProcessFactory.createDestinationProcess(
+                    "check",
+                    config = config,
+                    deploymentMode = deploymentMode,
+                )
             process.run()
             val messages = process.readMessages()
             val checkMessages = messages.filter { it.type == AirbyteMessage.Type.CONNECTION_STATUS }
@@ -76,7 +89,7 @@ open class CheckIntegrationTest<T : ConfigurationSpecification>(
                 { assertEquals(AirbyteConnectionStatus.Status.FAILED, connectionStatus.status) },
                 {
                     assertTrue(
-                        failurePattern.matcher(connectionStatus.message).matches(),
+                        failurePattern.matcher(connectionStatus.message).find(),
                         "Expected to match ${failurePattern.pattern()}, but got ${connectionStatus.message}"
                     )
                 }

diff --git a/...cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/test/util/DestinationProcess.kt b/...cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/test/util/DestinationProcess.kt
@@ -49,18 +49,25 @@ interface DestinationProcess {
     fun shutdown()
 }
 
+enum class TestDeploymentMode {
+    CLOUD,
+    OSS
+}
+
 interface DestinationProcessFactory {
     fun createDestinationProcess(
         command: String,
         config: ConfigurationSpecification? = null,
         catalog: ConfiguredAirbyteCatalog? = null,
+        deploymentMode: TestDeploymentMode = TestDeploymentMode.OSS,
     ): DestinationProcess
 }
 
 class NonDockerizedDestination(
     command: String,
     config: ConfigurationSpecification?,
     catalog: ConfiguredAirbyteCatalog?,
+    testDeploymentMode: TestDeploymentMode,
 ) : DestinationProcess {
     private val destinationStdinPipe: PrintWriter
     private val destination: CliRunnable
@@ -75,11 +82,19 @@ class NonDockerizedDestination(
             // from PrintWriter(outputStream) ).
             // Thanks, spotbugs.
             PrintWriter(PipedOutputStream(destinationStdin), false, Charsets.UTF_8)
+        val testEnvironments =
+            when (testDeploymentMode) {
+                // the env var is DEPLOYMENT_MODE, which micronaut parses to
+                // a property called deployment.mode.
+                TestDeploymentMode.CLOUD -> mapOf("deployment.mode" to "CLOUD")
+                TestDeploymentMode.OSS -> mapOf("deployment.mode" to "OSS")
+            }
         destination =
             CliRunner.destination(
                 command,
                 config = config,
                 catalog = catalog,
+                testProperties = testEnvironments,
                 inputStream = destinationStdin,
             )
     }
@@ -108,9 +123,10 @@ class NonDockerizedDestinationFactory : DestinationProcessFactory {
     override fun createDestinationProcess(
         command: String,
         config: ConfigurationSpecification?,
-        catalog: ConfiguredAirbyteCatalog?
+        catalog: ConfiguredAirbyteCatalog?,
+        deploymentMode: TestDeploymentMode,
     ): DestinationProcess {
-        return NonDockerizedDestination(command, config, catalog)
+        return NonDockerizedDestination(command, config, catalog, deploymentMode)
     }
 }
 

diff --git a/airbyte-integrations/connectors/destination-dev-null/metadata.yaml b/airbyte-integrations/connectors/destination-dev-null/metadata.yaml
@@ -2,7 +2,7 @@ data:
   connectorSubtype: file
   connectorType: destination
   definitionId: a7bcc9d8-13b3-4e49-b80d-d020b90045e3
-  dockerImageTag: 0.7.1
+  dockerImageTag: 0.7.2
   dockerRepository: airbyte/destination-dev-null
   githubIssueLabel: destination-dev-null
   icon: airbyte.svg

diff --git a/...n-dev-null/src/main/kotlin/io/airbyte/integrations/destination/dev_null/DevNullChecker.kt b/...n-dev-null/src/main/kotlin/io/airbyte/integrations/destination/dev_null/DevNullChecker.kt
@@ -8,7 +8,7 @@ import io.airbyte.cdk.check.DestinationChecker
 import jakarta.inject.Singleton
 
 @Singleton
-class DevNullChecker() : DestinationChecker<DevNullConfiguration> {
+class DevNullChecker : DestinationChecker<DevNullConfiguration> {
     override fun check(config: DevNullConfiguration) {
         // Do nothing
     }

diff --git a/...ration/kotlin/io/airbyte/integrations/destination/dev_null/DevNullCheckIntegrationTest.kt b/...ration/kotlin/io/airbyte/integrations/destination/dev_null/DevNullCheckIntegrationTest.kt
@@ -5,17 +5,34 @@
 package io.airbyte.integrations.destination.dev_null
 
 import io.airbyte.cdk.test.check.CheckIntegrationTest
+import io.airbyte.cdk.test.check.CheckTestConfig
+import io.airbyte.cdk.test.util.TestDeploymentMode
+import java.util.regex.Pattern
 import org.junit.jupiter.api.Test
 
 class DevNullCheckIntegrationTest :
     CheckIntegrationTest<DevNullSpecificationOss>(
         DevNullSpecificationOss::class.java,
-        successConfigFilenames = listOf(DevNullTestUtils.LOGGING_CONFIG_PATH),
-        failConfigFilenamesAndFailureReasons = mapOf(),
+        successConfigFilenames =
+            listOf(
+                CheckTestConfig(DevNullTestUtils.LOGGING_CONFIG_PATH, TestDeploymentMode.OSS),
+            ),
+        failConfigFilenamesAndFailureReasons =
+            mapOf(
+                // cloud doesn't support logging mode, so this should fail
+                // when trying to parse the config
+                CheckTestConfig(DevNullTestUtils.LOGGING_CONFIG_PATH, TestDeploymentMode.CLOUD) to
+                    Pattern.compile("Value 'LOGGING' is not defined in the schema")
+            ),
     ) {
 
     @Test
     override fun testSuccessConfigs() {
         super.testSuccessConfigs()
     }
+
+    @Test
+    override fun testFailConfigs() {
+        super.testFailConfigs()
+    }
 }
diff --git a/docs/integrations/destinations/dev-null.md b/docs/integrations/destinations/dev-null.md
@@ -49,6 +49,7 @@ The OSS and Cloud variants have the same version number starting from version `0
 
 | Version | Date       | Pull Request                                             | Subject                                                    |
 |:--------|:-----------|:---------------------------------------------------------|:-----------------------------------------------------------|
+| 0.7.2   | 2024-10-01 | [45929](https://github.com/airbytehq/airbyte/pull/45929) | Internal code changes                                      |
 | 0.7.1   | 2024-09-30 | [46276](https://github.com/airbytehq/airbyte/pull/46276) | Upgrade to latest bulk CDK                                 |
 | 0.7.0   | 2024-09-20 | [45704](https://github.com/airbytehq/airbyte/pull/45704) |                                                            |
 | 0.6.1   | 2024-09-20 | [45715](https://github.com/airbytehq/airbyte/pull/45715) | add destination to cloud registry                          |