diff --git a/src/main/kotlin/com/redhat/devtools/gateway/openshift/DevWorkspacePods.kt b/src/main/kotlin/com/redhat/devtools/gateway/openshift/DevWorkspacePods.kt index e6536e67..c1e83a8a 100644 --- a/src/main/kotlin/com/redhat/devtools/gateway/openshift/DevWorkspacePods.kt +++ b/src/main/kotlin/com/redhat/devtools/gateway/openshift/DevWorkspacePods.kt @@ -13,9 +13,7 @@ package com.redhat.devtools.gateway.openshift import com.intellij.openapi.diagnostic.logger import com.redhat.devtools.gateway.util.isCancellationException -import com.redhat.devtools.gateway.openshift.apiclient.ApiClientUtils import io.kubernetes.client.PortForward -import io.kubernetes.client.custom.IOTrio import io.kubernetes.client.openapi.ApiClient import io.kubernetes.client.openapi.ApiException import io.kubernetes.client.openapi.apis.CoreV1Api @@ -27,8 +25,6 @@ import java.io.IOException import java.io.InputStream import java.io.OutputStream import java.net.* -import kotlin.coroutines.resume -import kotlin.coroutines.resumeWithException class DevWorkspacePods(private val client: ApiClient) { @@ -60,141 +56,19 @@ class DevWorkspacePods(private val client: ApiClient) { container: String, timeout: Long = 60, checkCancelled: (() -> Unit)? = null - ): String = suspendCancellableCoroutine { cont -> - val metadata = pod.metadata - ?: throw IllegalArgumentException("Pod metadata is missing") - val namespace = metadata.namespace - ?: throw IllegalArgumentException("Pod namespace is missing") - val podName = metadata.name - ?: throw IllegalArgumentException("Pod name is missing") - - val closed = CompletableDeferred() - val stdout = StringBuilder() - val stderr = StringBuilder() - - val scope = CoroutineScope(Dispatchers.IO + SupervisorJob()) - - val execClientApi = createIsolatedExecClient(client) - var stdoutJob: Job? = null - var stderrJob: Job? = null - lateinit var stdoutStream: InputStream - lateinit var stderrStream: InputStream - - try { - val execHandle = ContainerAwareExec(execClientApi).containerAwareExec( - namespace = namespace, - pod = podName, - container = container, - command = command, - onOpen = { io -> - launchCheckCancelled(checkCancelled, scope, io) - - stdoutJob = scope.launch { readStream(io.stdout, stdout, checkCancelled) } - stderrJob = scope.launch { readStream(io.stderr, stderr, checkCancelled) } - launchJoinStdOutStdErr(scope, stdoutJob, stderrJob, checkCancelled, closed, cont, stdout, execClientApi) - }, - onClosed = { _, _ -> - closed.complete(Unit) - }, - onError = { err, _ -> - closed.complete(Unit) - shutdownExecClient(execClientApi) - cont.resumeWithException(err) - }, - timeoutMs = timeout * 1000, - tty = false - ) - - cont.invokeOnCancellation { cause -> - try { stdoutStream.close() } catch (_: Throwable) {} - try { stderrStream.close() } catch (_: Throwable) {} - try { - execHandle.job.cancel(CancellationException("Pods.exec cancellation")) - execHandle.future.cancel(true) - } catch (_: Throwable) {} - scope.cancel() - shutdownExecClient(execClientApi) - } - } catch (e: Exception) { - shutdownExecClient(execClientApi) - if (cont.isActive) cont.resumeWithException(e) - } + ): String { + val metadata = pod.metadata ?: throw IOException("Pod metadata is missing") + return PodExecSession( + client = client, + namespace = metadata.namespace, + podName = metadata.name, + container = container, + command = command, + timeout = timeout, + checkCancelled = checkCancelled + ).execute() } - private fun launchJoinStdOutStdErr( - scope: CoroutineScope, - stdoutJob: Job, - stderrJob: Job, - checkCancelled: (() -> Unit)?, - closed: CompletableDeferred, - cont: CancellableContinuation, - stdout: StringBuilder, - execClientApi: ApiClient - ) { - scope.launch { - try { - listOfNotNull(stdoutJob, stderrJob).joinAll() - checkCancelled?.invoke() - closed.await() - - checkCancelled?.invoke() - if (cont.isActive) cont.resume(stdout.toString()) - } catch (e: Throwable) { - if (e.isCancellationException()) cont.cancel(e) - else if (cont.isActive) cont.resumeWithException(e) - } finally { - scope.cancel() - shutdownExecClient(execClientApi) - } - } - } - - private fun launchCheckCancelled( - checkCancelled: (() -> Unit)?, - scope: CoroutineScope, - io: IOTrio - ) { - if (checkCancelled == null) { - return - } - scope.launch { - try { - while (isActive) { - checkCancelled.invoke() - delay(200) - } - } catch (_: Throwable) { - runCatching { io.stdout.close() } - runCatching { io.stderr.close() } - } - } - } - - private fun shutdownExecClient(client: ApiClient) { - runCatching { client.httpClient.dispatcher.executorService.shutdownNow() } - runCatching { client.httpClient.connectionPool.evictAll() } - } - - private fun readStream( - input: InputStream, - output: StringBuilder, - checkCancelled: (() -> Unit)? - ) { - try { - while (true) { - checkCancelled?.invoke() - val b = input.read() - if (b == -1) break - output.append(b.toChar()) - } - } catch (_: IOException) { - // Stream was closed (possibly due to cancellation) - } - } - - private fun createIsolatedExecClient(base: ApiClient): ApiClient = - ApiClientUtils.cloneForExec(base) - @Throws(IOException::class) fun forward(pod: V1Pod, localPort: Int, remotePort: Int): Closeable { val serverSocket = ServerSocket(localPort, 50, InetAddress.getLoopbackAddress()) @@ -204,7 +78,7 @@ class DevWorkspacePods(private val client: ApiClient) { ) scope.acceptConnections(serverSocket, pod, localPort, remotePort) return Closeable { - runCatching { serverSocket.close() } + closeQuietly(serverSocket) scope.cancel() } } @@ -282,7 +156,7 @@ class DevWorkspacePods(private val client: ApiClient) { "Could not port forward to pod ${pod.metadata?.name} using port $localPort -> $remotePort", e) } finally { - runCatching { clientSocket.close() } + closeQuietly(clientSocket) } } @@ -314,8 +188,13 @@ class DevWorkspacePods(private val client: ApiClient) { } private fun closeStreams(port: Int, forwardResult: PortForward.PortForwardResult?) { - runCatching { forwardResult?.getInputStream(port)?.close() } - runCatching { forwardResult?.getOutboundStream(port)?.close() } + // getInputStream/getOutboundStream can throw; isolate so one failure does not skip the other close + runCatching { forwardResult?.getInputStream(port) } + .onSuccess { closeQuietly(it) } + .onFailure { logger.debug("Could not get input stream for port $port while closing port-forward", it) } + runCatching { forwardResult?.getOutboundStream(port) } + .onSuccess { closeQuietly(it) } + .onFailure { logger.debug("Could not get outbound stream for port $port while closing port-forward", it) } } private fun InputStream.copyToAndFlush(destination: OutputStream) { diff --git a/src/main/kotlin/com/redhat/devtools/gateway/openshift/PodExecSession.kt b/src/main/kotlin/com/redhat/devtools/gateway/openshift/PodExecSession.kt new file mode 100644 index 00000000..3d2173cb --- /dev/null +++ b/src/main/kotlin/com/redhat/devtools/gateway/openshift/PodExecSession.kt @@ -0,0 +1,208 @@ +/* + * Copyright (c) 2024-2025 Red Hat, Inc. + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + * + * Contributors: + * Red Hat, Inc. - initial API and implementation + */ +package com.redhat.devtools.gateway.openshift + +import com.redhat.devtools.gateway.openshift.apiclient.ApiClientUtils +import com.redhat.devtools.gateway.util.isCancellationException +import io.kubernetes.client.custom.IOTrio +import io.kubernetes.client.openapi.ApiClient +import kotlinx.coroutines.* +import java.io.IOException +import java.io.InputStream +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicReference +import kotlin.coroutines.resume +import kotlin.coroutines.resumeWithException + +internal class PodExecSession( + private val client: ApiClient, + namespace: String?, + podName: String?, + private val container: String, + private val command: Array, + private val timeout: Long, + private val checkCancelled: (() -> Unit)? +) { + private val namespace: String = namespace + ?: throw IOException("Pod namespace is missing") + private val podName: String = podName + ?: throw IOException("Pod name is missing") + + suspend fun execute(): String = suspendCancellableCoroutine { cont -> + val ctx = ExecContext() + val joinerJob = launchJoiner(ctx, cont) + + var execHandle: ContainerAwareExec.ExecHandle? = null + cont.invokeOnCancellation { + ctx.streamsReady.complete(Unit) + closeQuietly(ctx.io?.stdin) + closeQuietly(ctx.io?.stdout) + closeQuietly(ctx.io?.stderr) + runCatching { joinerJob.cancel("Pods.exec cancellation") } + execHandle?.let { handle -> + runCatching { + handle.job.cancel(CancellationException("Pods.exec cancellation")) + handle.future.cancel(true) + } + } + ctx.scope.cancel() + } + + try { + execHandle = runExec(ctx) + } catch (e: Exception) { + ctx.streamsReady.complete(Unit) + ctx.exitCode.completeExceptionally(e) + } + } + + private fun launchJoiner(ctx: ExecContext, cont: CancellableContinuation): Job = + ctx.scope.launch { + try { + ctx.streamsReady.await() + listOfNotNull(ctx.stdoutJobRef.get(), ctx.stderrJobRef.get()).joinAll() + checkCancelled?.invoke() + val code = ctx.exitCode.await() + + checkCancelled?.invoke() + val stderrMsg = ctx.stderr.toString().takeIf { it.isNotBlank() } + ?.let { "; stderr: ${it.take(2000)}" }.orEmpty() + when { + code == Int.MAX_VALUE -> { + if (cont.isActive) cont.resumeWithException(IOException("Pod exec timed out after ${timeout}s$stderrMsg")) + } + code != 0 -> { + if (cont.isActive) cont.resumeWithException(IOException("Pod exec failed with exit code $code$stderrMsg")) + } + else -> { + val readError = ctx.streamReadError.get() + if (readError != null) { + if (cont.isActive) { + cont.resumeWithException( + IOException( + "Pod exec stream closed before output was fully read$stderrMsg", + readError + ) + ) + } + } else if (cont.isActive) { + cont.resume(ctx.stdout.toString()) + } + } + } + } catch (e: Throwable) { + if (e.isCancellationException()) cont.cancel(e) + else if (cont.isActive) cont.resumeWithException(e) + } finally { + ctx.scope.cancel() + shutdownExecClient(ctx.execClient) + } + } + + private fun runExec(ctx: ExecContext): ContainerAwareExec.ExecHandle = + ContainerAwareExec(ctx.execClient).containerAwareExec( + namespace = namespace, + pod = podName, + container = container, + command = command, + onOpen = { i -> + ctx.io = i + launchCheckCancelled(checkCancelled, ctx.scope, i) + ctx.stdoutJobRef.set( + ctx.scope.launch { readStream(i.stdout, ctx.stdout, checkCancelled, ctx.streamReadError) } + ) + ctx.stderrJobRef.set( + ctx.scope.launch { readStream(i.stderr, ctx.stderr, checkCancelled, ctx.streamReadError) } + ) + ctx.streamsReady.complete(Unit) + }, + onClosed = { code, _ -> + ctx.streamsReady.complete(Unit) + ctx.exitCode.complete(code) + }, + onError = { err, _ -> + ctx.exitCode.completeExceptionally(err) + closeQuietly(ctx.io?.stdout) + closeQuietly(ctx.io?.stderr) + ctx.streamsReady.complete(Unit) + }, + timeoutMs = timeout * 1000, + tty = false + ) + + private fun launchCheckCancelled( + checkCancelled: (() -> Unit)?, + scope: CoroutineScope, + io: IOTrio + ) { + if (checkCancelled == null) return + scope.launch { + try { + while (isActive) { + checkCancelled.invoke() + @Suppress("ConvertLongToDuration") + delay(200) + } + } catch (_: Throwable) { + closeQuietly(io.stdout) + closeQuietly(io.stderr) + } + } + } + + private fun shutdownExecClient(client: ApiClient) { + runCatching { + val executor = client.httpClient.dispatcher.executorService + executor.shutdownNow() + executor.awaitTermination(500, TimeUnit.MILLISECONDS) + } + runCatching { client.httpClient.connectionPool.evictAll() } + } + + private fun readStream( + input: InputStream, + output: StringBuilder, + checkCancelled: (() -> Unit)?, + streamReadError: AtomicReference + ) { + try { + val reader = input.reader(Charsets.UTF_8) + val buffer = CharArray(4096) + while (true) { + checkCancelled?.invoke() + val read = reader.read(buffer) + if (read == -1) break + output.appendRange(buffer, 0, read) + } + } catch (e: IOException) { + // Closed during cancel/timeout/onError — recorded so exit 0 cannot return a partial buffer + streamReadError.compareAndSet(null, e) + } + } + + private fun createIsolatedExecClient(base: ApiClient): ApiClient = + ApiClientUtils.cloneForExec(base) + + private inner class ExecContext( + val scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()), + val exitCode: CompletableDeferred = CompletableDeferred(), + val streamsReady: CompletableDeferred = CompletableDeferred(), + val stdout: StringBuilder = StringBuilder(), + val stderr: StringBuilder = StringBuilder(), + val stdoutJobRef: AtomicReference = AtomicReference(null), + val stderrJobRef: AtomicReference = AtomicReference(null), + val streamReadError: AtomicReference = AtomicReference(null), + val execClient: ApiClient = createIsolatedExecClient(client), + var io: IOTrio? = null + ) + +} diff --git a/src/main/kotlin/com/redhat/devtools/gateway/openshift/Utils.kt b/src/main/kotlin/com/redhat/devtools/gateway/openshift/Utils.kt index 88e06141..ac7c07e6 100644 --- a/src/main/kotlin/com/redhat/devtools/gateway/openshift/Utils.kt +++ b/src/main/kotlin/com/redhat/devtools/gateway/openshift/Utils.kt @@ -11,6 +11,8 @@ */ package com.redhat.devtools.gateway.openshift +import java.io.Closeable + object Utils { @JvmStatic fun getValue(obj: Any?, path: Array): Any? { @@ -77,3 +79,5 @@ fun mapOfNotNull(vararg pairs: Pair): Map { }.toMap() } +fun closeQuietly(closeable: Closeable?) = runCatching { closeable?.close() } + diff --git a/src/test/kotlin/com/redhat/devtools/gateway/openshift/DevWorkspacePodsTest.kt b/src/test/kotlin/com/redhat/devtools/gateway/openshift/DevWorkspacePodsTest.kt index e13a7c3c..b24a31c7 100644 --- a/src/test/kotlin/com/redhat/devtools/gateway/openshift/DevWorkspacePodsTest.kt +++ b/src/test/kotlin/com/redhat/devtools/gateway/openshift/DevWorkspacePodsTest.kt @@ -30,6 +30,8 @@ import io.mockk.slot import io.mockk.unmockkAll import io.mockk.unmockkConstructor import io.mockk.verify +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.async import kotlinx.coroutines.CancellationException import kotlinx.coroutines.delay import kotlinx.coroutines.launch @@ -551,6 +553,192 @@ class DevWorkspacePodsTest { unmockkConstructor(CoreV1Api::class) } + @Test + fun `#exec onError after onOpen completes with one failure and shuts down client once`() = runBlocking { + // given — ContainerAwareExec calls onOpen then onError (mid-stream error) + mockkConstructor(ContainerAwareExec::class) + val onErrorCalled = java.util.concurrent.atomic.AtomicBoolean() + val fakeHandle = ContainerAwareExec.ExecHandle( + future = java.util.concurrent.CompletableFuture.completedFuture(0), + job = mockk(relaxed = true) + ) + every { + anyConstructed().containerAwareExec( + any(), any(), any(), any(), any(), any(), any(), any(), any() + ) + } answers { + @Suppress("UNCHECKED_CAST") + val onOpen = it.invocation.args[4] as java.util.function.Consumer + val onError = it.invocation.args[6] as java.util.function.BiConsumer + val io = IOTrio() + io.stdout = ByteArrayInputStream(ByteArray(0)) + io.stderr = ByteArrayInputStream(ByteArray(0)) + io.stdin = mockk(relaxed = true) + onOpen.accept(io) + // Simulate mid-stream error after onOpen has fired + onError.accept(IOException("mid-stream connection reset"), io) + onErrorCalled.set(true) + fakeHandle + } + + mockkObject(ApiClientUtils) + val execClient = mockk(relaxed = true) + every { ApiClientUtils.cloneForExec(any()) } returns execClient + + val testPod = V1Pod().apply { + metadata = V1ObjectMeta().apply { + name = "test-pod" + namespace = "test-ns" + } + } + + // when / then + assertThatThrownBy { + runBlocking { + pods.exec( + pod = testPod, + command = arrayOf("echo"), + container = "test-container" + ) + } + }.isInstanceOf(IOException::class.java) + + // exactly one failure (not double-completed) + assertThat(onErrorCalled).isTrue() + + // client shut down at least once (joiner finally block handles it) + verify(atLeast = 1) { + execClient.httpClient.dispatcher.executorService.shutdownNow() + } + verify(atLeast = 1) { + execClient.httpClient.connectionPool.evictAll() + } + } + + private class CloseTrackingInputStream(private val delegate: java.io.InputStream) : java.io.InputStream() { + var closed = false + private set + override fun read(): Int = delegate.read() + override fun close() { closed = true; delegate.close() } + } + + @Test + fun `#exec closes stdout and stderr on cancellation`() = runBlocking { + // given — ContainerAwareExec returns a fake process with close-tracking streams + val stdout = CloseTrackingInputStream(ByteArrayInputStream("data".toByteArray())) + val stderr = CloseTrackingInputStream(ByteArrayInputStream(ByteArray(0))) + val fakeProcess = mockk(relaxed = true).also { + every { it.inputStream } returns stdout + every { it.errorStream } returns stderr + every { it.outputStream } returns mockk(relaxed = true) + every { it.isAlive } returns true + } + + mockkConstructor(ContainerAwareExec::class) + val fakeHandle = ContainerAwareExec.ExecHandle( + future = java.util.concurrent.CompletableFuture.completedFuture(0), + job = mockk(relaxed = true) + ) + val onOpenReady = CompletableDeferred() + every { + anyConstructed().containerAwareExec( + any(), any(), any(), any(), any(), any(), any(), any(), any() + ) + } answers { + @Suppress("UNCHECKED_CAST") + val onOpen = it.invocation.args[4] as java.util.function.Consumer + val io = IOTrio() + io.stdout = fakeProcess.inputStream + io.stderr = fakeProcess.errorStream + io.stdin = fakeProcess.outputStream + onOpen.accept(io) + onOpenReady.complete(Unit) + fakeHandle + } + + mockkObject(ApiClientUtils) + val execClient = mockk(relaxed = true) + every { ApiClientUtils.cloneForExec(any()) } returns execClient + + val scope = kotlinx.coroutines.CoroutineScope(kotlinx.coroutines.SupervisorJob()) + val testPod = V1Pod().apply { + metadata = V1ObjectMeta().apply { + name = "test-pod" + namespace = "test-ns" + } + } + + // when — launch exec, wait for onOpen, then cancel + val job = scope.launch { + pods.exec( + pod = testPod, + command = arrayOf("echo"), + container = "test-container", + timeout = 60 + ) + } + onOpenReady.await() + job.cancel() + job.join() + + // then — stdout and stderr should be closed + assertThat(stdout.closed).isTrue() + assertThat(stderr.closed).isTrue() + } + + @Test + fun `#exec fails when error occurs before onOpen`() = runBlocking { + // given — ContainerAwareExec calls onError without onOpen (connection failed before ready) + + mockkConstructor(ContainerAwareExec::class) + val fakeHandle = ContainerAwareExec.ExecHandle( + future = java.util.concurrent.CompletableFuture.completedFuture(0), + job = mockk(relaxed = true) + ) + every { + anyConstructed().containerAwareExec( + any(), any(), any(), any(), any(), any(), any(), any(), any() + ) + } answers { + @Suppress("UNCHECKED_CAST") + val onError = it.invocation.args[6] as java.util.function.BiConsumer + val io = IOTrio() + io.stdout = ByteArrayInputStream(ByteArray(0)) + io.stderr = ByteArrayInputStream(ByteArray(0)) + io.stdin = mockk(relaxed = true) + // Simulate error callback fired without onOpen (connection failed before ready) + onError.accept(IOException("connection refused before onOpen"), io) + fakeHandle + } + + mockkObject(ApiClientUtils) + val execClient = mockk(relaxed = true) + every { ApiClientUtils.cloneForExec(any()) } returns execClient + + val testPod = V1Pod().apply { + metadata = V1ObjectMeta().apply { + name = "test-pod" + namespace = "test-ns" + } + } + + // when / then + assertThatThrownBy { + runBlocking { + pods.exec( + pod = testPod, + command = arrayOf("echo"), + container = "test-container" + ) + } + }.isInstanceOf(IOException::class.java) + + verify { + execClient.httpClient.dispatcher.executorService.shutdownNow() + execClient.httpClient.connectionPool.evictAll() + } + } + @Test fun `#exec cancels cleanly when checkCancelled throws`() = runBlocking { // given — ContainerAwareExec returns a fake process that produces data @@ -713,4 +901,268 @@ class DevWorkspacePodsTest { execClient.httpClient.connectionPool.evictAll() } } + + @Test + fun `#exec throws IOException when process exits non-zero`() = runBlocking { + // given + mockkConstructor(ContainerAwareExec::class) + val fakeHandle = ContainerAwareExec.ExecHandle( + future = java.util.concurrent.CompletableFuture.completedFuture(0), + job = mockk(relaxed = true) + ) + every { + anyConstructed().containerAwareExec( + any(), any(), any(), any(), any(), any(), any(), any(), any() + ) + } answers { + @Suppress("UNCHECKED_CAST") + val onClosed = it.invocation.args[5] as java.util.function.BiConsumer + val onOpen = it.invocation.args[4] as java.util.function.Consumer + val io = IOTrio() + io.stdout = ByteArrayInputStream("stdout".toByteArray()) + io.stderr = ByteArrayInputStream(ByteArray(0)) + io.stdin = mockk(relaxed = true) + onOpen.accept(io) + onClosed.accept(1, io) + fakeHandle + } + + mockkObject(ApiClientUtils) + val execClient = mockk(relaxed = true) + every { ApiClientUtils.cloneForExec(any()) } returns execClient + + val testPod = V1Pod().apply { + metadata = V1ObjectMeta().apply { + name = "test-pod" + namespace = "test-ns" + } + } + + // when / then + assertThatThrownBy { + runBlocking { + pods.exec( + pod = testPod, + command = arrayOf("echo"), + container = "test-container", + timeout = 60 + ) + } + }.isInstanceOf(IOException::class.java) + .hasMessageContaining("exit code 1") + } + + @Test + fun `#exec throws IOException when exec times out`() = runBlocking { + // given + mockkConstructor(ContainerAwareExec::class) + val fakeHandle = ContainerAwareExec.ExecHandle( + future = java.util.concurrent.CompletableFuture.completedFuture(0), + job = mockk(relaxed = true) + ) + every { + anyConstructed().containerAwareExec( + any(), any(), any(), any(), any(), any(), any(), any(), any() + ) + } answers { + @Suppress("UNCHECKED_CAST") + val onClosed = it.invocation.args[5] as java.util.function.BiConsumer + val onOpen = it.invocation.args[4] as java.util.function.Consumer + val io = IOTrio() + io.stdout = ByteArrayInputStream("stdout".toByteArray()) + io.stderr = ByteArrayInputStream(ByteArray(0)) + io.stdin = mockk(relaxed = true) + onOpen.accept(io) + onClosed.accept(Int.MAX_VALUE, io) + fakeHandle + } + + mockkObject(ApiClientUtils) + val execClient = mockk(relaxed = true) + every { ApiClientUtils.cloneForExec(any()) } returns execClient + + val testPod = V1Pod().apply { + metadata = V1ObjectMeta().apply { + name = "test-pod" + namespace = "test-ns" + } + } + + // when / then + assertThatThrownBy { + runBlocking { + pods.exec( + pod = testPod, + command = arrayOf("echo"), + container = "test-container", + timeout = 60 + ) + } + }.isInstanceOf(IOException::class.java) + .hasMessageContaining("timed out") + } + + @Test + fun `#exec includes stderr in IOException when exit code is non-zero`() = runBlocking { + // given + mockkConstructor(ContainerAwareExec::class) + val fakeHandle = ContainerAwareExec.ExecHandle( + future = java.util.concurrent.CompletableFuture.completedFuture(0), + job = mockk(relaxed = true) + ) + every { + anyConstructed().containerAwareExec( + any(), any(), any(), any(), any(), any(), any(), any(), any() + ) + } answers { + @Suppress("UNCHECKED_CAST") + val onClosed = it.invocation.args[5] as java.util.function.BiConsumer + val onOpen = it.invocation.args[4] as java.util.function.Consumer + val io = IOTrio() + io.stdout = ByteArrayInputStream("stdout".toByteArray()) + io.stderr = ByteArrayInputStream("boom".toByteArray()) + io.stdin = mockk(relaxed = true) + onOpen.accept(io) + onClosed.accept(1, io) + fakeHandle + } + + mockkObject(ApiClientUtils) + val execClient = mockk(relaxed = true) + every { ApiClientUtils.cloneForExec(any()) } returns execClient + + val testPod = V1Pod().apply { + metadata = V1ObjectMeta().apply { + name = "test-pod" + namespace = "test-ns" + } + } + + // when / then + assertThatThrownBy { + runBlocking { + pods.exec( + pod = testPod, + command = arrayOf("echo"), + container = "test-container", + timeout = 60 + ) + } + }.isInstanceOf(IOException::class.java) + .hasMessageContaining("exit code 1") + .hasMessageContaining("boom") + } + + @Test + fun `#exec returns stdout when exit code is 0`() = runBlocking { + // given + mockkConstructor(ContainerAwareExec::class) + val fakeHandle = ContainerAwareExec.ExecHandle( + future = java.util.concurrent.CompletableFuture.completedFuture(0), + job = mockk(relaxed = true) + ) + every { + anyConstructed().containerAwareExec( + any(), any(), any(), any(), any(), any(), any(), any(), any() + ) + } answers { + @Suppress("UNCHECKED_CAST") + val onClosed = it.invocation.args[5] as java.util.function.BiConsumer + val onOpen = it.invocation.args[4] as java.util.function.Consumer + val io = IOTrio() + io.stdout = ByteArrayInputStream("hello world".toByteArray()) + io.stderr = ByteArrayInputStream(ByteArray(0)) + io.stdin = mockk(relaxed = true) + onOpen.accept(io) + onClosed.accept(0, io) + fakeHandle + } + + mockkObject(ApiClientUtils) + val execClient = mockk(relaxed = true) + every { ApiClientUtils.cloneForExec(any()) } returns execClient + + val testPod = V1Pod().apply { + metadata = V1ObjectMeta().apply { + name = "test-pod" + namespace = "test-ns" + } + } + + // when / then + assertThat(pods.exec( + pod = testPod, + command = arrayOf("echo"), + container = "test-container", + timeout = 60 + )).isEqualTo("hello world") + } + + @Test + fun `#exec fails when stdout stream errors even if exit code is 0`() = runBlocking { + // given — stdout delivers partial data then throws (no clean EOF) + mockkConstructor(ContainerAwareExec::class) + val fakeHandle = ContainerAwareExec.ExecHandle( + future = java.util.concurrent.CompletableFuture.completedFuture(0), + job = mockk(relaxed = true) + ) + every { + anyConstructed().containerAwareExec( + any(), any(), any(), any(), any(), any(), any(), any(), any() + ) + } answers { + @Suppress("UNCHECKED_CAST") + val onClosed = it.invocation.args[5] as java.util.function.BiConsumer + val onOpen = it.invocation.args[4] as java.util.function.Consumer + val io = IOTrio() + io.stdout = ThrowingAfterDataInputStream("partial".toByteArray()) + io.stderr = ByteArrayInputStream(ByteArray(0)) + io.stdin = mockk(relaxed = true) + onOpen.accept(io) + onClosed.accept(0, io) + fakeHandle + } + + mockkObject(ApiClientUtils) + val execClient = mockk(relaxed = true) + every { ApiClientUtils.cloneForExec(any()) } returns execClient + + val testPod = V1Pod().apply { + metadata = V1ObjectMeta().apply { + name = "test-pod" + namespace = "test-ns" + } + } + + // when / then — must not return the partial buffer as success + assertThatThrownBy { + runBlocking { + pods.exec( + pod = testPod, + command = arrayOf("echo"), + container = "test-container", + timeout = 60 + ) + } + }.isInstanceOf(IOException::class.java) + .hasMessageContaining("stream closed before output was fully read") + } + + /** Returns all bytes once, then throws instead of EOF (-1). */ + private class ThrowingAfterDataInputStream(private val data: ByteArray) : java.io.InputStream() { + private var index = 0 + override fun read(): Int { + if (index < data.size) return data[index++].toInt() and 0xFF + throw IOException("connection reset") + } + + override fun read(b: ByteArray, off: Int, len: Int): Int { + if (len == 0) return 0 + if (index >= data.size) throw IOException("connection reset") + val n = minOf(len, data.size - index) + System.arraycopy(data, index, b, off, n) + index += n + return n + } + } } \ No newline at end of file