fix(public-api): Return all job types when jobType filter is not specified (#18592)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Benoit Moriceau <benoit@airbyte.io>

Author: devin-ai-integration[bot]

airbyte-api/server-api/src/main/openapi/api_documentation_jobs.yaml

@@ -932,7 +932,7 @@ components:
       x-speakeasy-component: true
     JobTypeEnum:
       description: "Enum that describes the different types of jobs that the platform\
-        \ runs."
+        \ runs. When not specified, all job types are returned."
       enum:
       - "sync"
       - "reset"

airbyte-api/server-api/src/main/openapi/config.yaml

@@ -20501,7 +20501,7 @@ components:
       type: string
       x-sdk-component: true
     JobTypeEnum:
-      description: Enum that describes the different types of jobs that the platform runs.
+      description: Enum that describes the different types of jobs that the platform runs. When not specified, all job types are returned.
       enum:
         - sync
         - reset

airbyte-server/src/main/kotlin/io/airbyte/server/apis/publicapi/services/JobService.kt

@@ -12,7 +12,6 @@ import io.airbyte.api.model.generated.JobListForWorkspacesRequestBody.OrderByFie
 import io.airbyte.api.model.generated.JobListForWorkspacesRequestBody.OrderByMethodEnum
 import io.airbyte.api.model.generated.JobListRequestBody
 import io.airbyte.api.model.generated.Pagination
-import io.airbyte.api.problems.throwable.generated.UnprocessableEntityProblem
 import io.airbyte.commons.server.handlers.JobHistoryHandler
 import io.airbyte.commons.server.handlers.SchedulerHandler
 import io.airbyte.commons.server.support.CurrentUserService
@@ -220,14 +219,13 @@ class JobServiceImpl(
   private fun getJobConfigTypes(jobType: JobTypeEnum?): List<JobConfigType> {
     val configTypes: MutableList<JobConfigType> = ArrayList()
     if (jobType == null) {
-      configTypes.addAll(listOf(JobConfigType.SYNC, JobConfigType.RESET_CONNECTION))
+      configTypes.addAll(listOf(JobConfigType.SYNC, JobConfigType.RESET_CONNECTION, JobConfigType.CLEAR, JobConfigType.REFRESH))
     } else {
       when (jobType) {
         JobTypeEnum.SYNC -> configTypes.add(JobConfigType.SYNC)
         JobTypeEnum.RESET -> configTypes.add(JobConfigType.RESET_CONNECTION)
         JobTypeEnum.CLEAR -> configTypes.add(JobConfigType.CLEAR)
         JobTypeEnum.REFRESH -> configTypes.add(JobConfigType.REFRESH)
-        else -> throw UnprocessableEntityProblem()
       }
     }
     return configTypes

airbyte-server/src/test/kotlin/io/airbyte/server/apis/publicapi/services/JobServiceTest.kt

@@ -4,22 +4,36 @@
 
 package io.airbyte.server.apis.publicapi.services
 
+import io.airbyte.api.model.generated.JobConfigType
+import io.airbyte.api.model.generated.JobReadList
 import io.airbyte.api.problems.throwable.generated.StateConflictProblem
 import io.airbyte.api.problems.throwable.generated.TryAgainLaterConflictProblem
 import io.airbyte.commons.server.errors.ValueConflictKnownException
+import io.airbyte.commons.server.handlers.JobHistoryHandler
 import io.airbyte.commons.server.handlers.SchedulerHandler
 import io.airbyte.micronaut.runtime.AirbyteApiConfig
+import io.airbyte.publicApi.server.generated.models.JobTypeEnum
 import io.airbyte.server.apis.publicapi.errorHandlers.JOB_NOT_RUNNING_MESSAGE
+import io.airbyte.server.apis.publicapi.filters.JobsFilter
 import io.mockk.every
 import io.mockk.mockk
+import io.mockk.slot
+import io.mockk.verify
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.Arguments
+import org.junit.jupiter.params.provider.MethodSource
 import java.util.UUID
+import java.util.stream.Stream
 
 class JobServiceTest {
   private lateinit var jobService: JobServiceImpl
   private val schedulerHandler: SchedulerHandler = mockk()
+  private val jobHistoryHandler: JobHistoryHandler = mockk()
 
   private val connectionId = UUID.randomUUID()
 
@@ -28,7 +42,7 @@ class JobServiceTest {
     jobService =
       JobServiceImpl(
         schedulerHandler = schedulerHandler,
-        jobHistoryHandler = mockk(),
+        jobHistoryHandler = jobHistoryHandler,
         currentUserService = mockk(),
         userService = mockk(),
         airbyteApiConfig = AirbyteApiConfig(),
@@ -66,4 +80,66 @@ class JobServiceTest {
     every { schedulerHandler.syncConnection(any()) } throws IllegalStateException(failureReason)
     assertThrows<StateConflictProblem>(JOB_NOT_RUNNING_MESSAGE) { jobService.sync(connectionId) }
   }
+
+  @Test
+  fun `test getJobList with null jobType returns all job types`() {
+    val requestBodySlot = slot<io.airbyte.api.model.generated.JobListRequestBody>()
+    every { jobHistoryHandler.listJobsForLight(capture(requestBodySlot)) } returns JobReadList().jobs(emptyList()).totalJobCount(0L)
+
+    val jobsFilter =
+      JobsFilter(
+        createdAtStart = null,
+        createdAtEnd = null,
+        updatedAtStart = null,
+        updatedAtEnd = null,
+        limit = 10,
+        offset = 0,
+        jobType = null,
+        status = null,
+      )
+    jobService.getJobList(connectionId, jobsFilter)
+
+    verify { jobHistoryHandler.listJobsForLight(any()) }
+    val capturedConfigTypes = requestBodySlot.captured.configTypes
+    assertEquals(4, capturedConfigTypes.size)
+    assertTrue(capturedConfigTypes.contains(JobConfigType.SYNC))
+    assertTrue(capturedConfigTypes.contains(JobConfigType.RESET_CONNECTION))
+    assertTrue(capturedConfigTypes.contains(JobConfigType.CLEAR))
+    assertTrue(capturedConfigTypes.contains(JobConfigType.REFRESH))
+  }
+
+  @ParameterizedTest(name = "jobType {0} should return config type {1}")
+  @MethodSource("jobTypeToConfigTypeProvider")
+  fun `test getJobList with specific jobType returns only that type`(
+    jobType: JobTypeEnum,
+    expectedConfigType: JobConfigType,
+  ) {
+    val requestBodySlot = slot<io.airbyte.api.model.generated.JobListRequestBody>()
+    every { jobHistoryHandler.listJobsForLight(capture(requestBodySlot)) } returns JobReadList().jobs(emptyList()).totalJobCount(0L)
+
+    val filter =
+      JobsFilter(
+        createdAtStart = null,
+        createdAtEnd = null,
+        updatedAtStart = null,
+        updatedAtEnd = null,
+        limit = 10,
+        offset = 0,
+        jobType = jobType,
+        status = null,
+      )
+    jobService.getJobList(connectionId, filter)
+    assertEquals(listOf(expectedConfigType), requestBodySlot.captured.configTypes)
+  }
+
+  companion object {
+    @JvmStatic
+    fun jobTypeToConfigTypeProvider(): Stream<Arguments> =
+      Stream.of(
+        Arguments.of(JobTypeEnum.SYNC, JobConfigType.SYNC),
+        Arguments.of(JobTypeEnum.RESET, JobConfigType.RESET_CONNECTION),
+        Arguments.of(JobTypeEnum.CLEAR, JobConfigType.CLEAR),
+        Arguments.of(JobTypeEnum.REFRESH, JobConfigType.REFRESH),
+      )
+  }
 }