Merge pull request #180 from brendandburns/test

Fix sending protocols for web sockets.

Author: k8s-ci-robot

examples/typescript/exec/exec-example.ts

@@ -1,4 +1,5 @@
-import k8s = require('@kubernetes/client-node');
+import * as k8s from '@kubernetes/client-node';
+import * as stream from 'stream';
 
 const command = process.argv[2];
 
@@ -7,4 +8,11 @@ kc.loadFromDefault();
 
 const exec = new k8s.Exec(kc);
 exec.exec('default', 'nginx-4217019353-9gl4s', 'nginx', command,
-    process.stdout, process.stderr, process.stdin, true /* tty */);
+    process.stdout as stream.Writable, process.stderr as stream.Writable, process.stdin as stream.Readable,
+    true /* tty */,
+    (status: k8s.V1Status) => {
+        // tslint:disable-next-line:no-console
+        console.log('Exited with status:');
+        // tslint:disable-next-line:no-console
+        console.log(JSON.stringify(status, null, 2));
+    });

src/exec.ts

@@ -2,6 +2,7 @@ import WebSocket = require('isomorphic-ws');
 import querystring = require('querystring');
 import stream = require('stream');
 
+import { V1Status } from './api';
 import { KubeConfig } from './config';
 import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
 
@@ -17,9 +18,23 @@ export class Exec {
     }
 
     // TODO: make command an array and support multiple args
+    /**
+     * @param {string}  namespace - The namespace of the pod to exec the command inside.
+     * @param {string} podName - The name of the pod to exec the command inside.
+     * @param {string} containerName - The name of the container in the pod to exec the command inside.
+     * @param {string} command - The command to execute.
+     * @param {stream.Writable} stdout - The stream to write stdout data from the command.
+     * @param {stream.Writable} stderr - The stream to write stderr data from the command.
+     * @param {stream.Readable} stdin - The strream to write stdin data into the command.
+     * @param {boolean} tty - Should the command execute in a TTY enabled session.
+     * @param {(V1Status) => void} statusCallback -
+     *       A callback to received the status (e.g. exit code) from the command, optional.
+     * @return {string} This is the result
+     */
     public async exec(namespace: string, podName: string, containerName: string, command: string,
-                      stdout: stream.Writable | any, stderr: stream.Writable | any, stdin: stream.Readable | any,
-                      tty: boolean): Promise<WebSocket> {
+                      stdout: stream.Writable | null, stderr: stream.Writable | null, stdin: stream.Readable | null,
+                      tty: boolean,
+                      statusCallback?: (status: V1Status) => void): Promise<WebSocket> {
         const query = {
             stdout: stdout != null,
             stderr: stderr != null,
@@ -31,7 +46,13 @@ export class Exec {
         const queryStr = querystring.stringify(query);
         const path = `/api/v1/namespaces/${namespace}/pods/${podName}/exec?${queryStr}`;
         const conn = await this.handler.connect(path, null, (streamNum: number, buff: Buffer): boolean => {
-            WebSocketHandler.handleStandardStreams(streamNum, buff, stdout, stderr);
+            const status = WebSocketHandler.handleStandardStreams(streamNum, buff, stdout, stderr);
+            if (status != null) {
+                if (statusCallback) {
+                    statusCallback(status);
+                }
+                return false;
+            }
             return true;
         });
         if (stdin != null) {

src/exec_test.ts

@@ -3,6 +3,7 @@ import WebSocket = require('isomorphic-ws');
 import { ReadableStreamBuffer, WritableStreamBuffer } from 'stream-buffers';
 import { anyFunction, capture, instance, mock, verify, when } from 'ts-mockito';
 
+import { V1Status } from './api';
 import { KubeConfig } from './config';
 import { Exec } from './exec';
 import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
@@ -65,11 +66,15 @@ describe('Exec', () => {
             const path = `/api/v1/namespaces/${namespace}/pods/${pod}/exec`;
             const args = `stdout=true&stderr=true&stdin=true&tty=false&command=${cmd}&container=${container}`;
 
+            let statusOut = {} as V1Status;
+
             const fakeConn: WebSocket = mock(WebSocket);
             when(fakeWebSocket.connect(`${path}?${args}`, null, anyFunction())).thenResolve(fakeConn);
 
             await exec.exec(
-                namespace, pod, container, cmd, osStream, errStream, isStream, false);
+                namespace, pod, container, cmd, osStream, errStream, isStream, false, (status: V1Status) => {
+                    statusOut = status;
+                });
 
             const [, , outputFn] = capture(fakeWebSocket.connect).last();
 
@@ -102,6 +107,14 @@ describe('Exec', () => {
             isStream.put(msg);
             verify(fakeConn.send(msg));
 
+            const statusIn = {
+                code: 100,
+                message: 'this is a test',
+            } as V1Status;
+
+            outputFn(WebSocketHandler.StatusStream, Buffer.from(JSON.stringify(statusIn)));
+            expect(statusOut).to.deep.equal(statusIn);
+
             isStream.stop();
             verify(fakeConn.close());
         });

src/web-socket-handler.ts

@@ -27,21 +27,21 @@ export class WebSocketHandler implements WebSocketInterface {
 
     public static handleStandardStreams(
         streamNum: number, buff: Buffer,
-        stdout: stream.Writable, stderr: stream.Writable,
+        stdout: stream.Writable | null, stderr: stream.Writable | null,
     ): V1Status | null {
         if (buff.length < 1) {
             return null;
         }
-        if (streamNum === WebSocketHandler.StdoutStream) {
+        if (stdout && streamNum === WebSocketHandler.StdoutStream) {
             stdout.write(buff);
-        } else if (streamNum === WebSocketHandler.StderrStream) {
+        } else if (stderr && streamNum === WebSocketHandler.StderrStream) {
             stderr.write(buff);
         } else if (streamNum === WebSocketHandler.StatusStream) {
             // stream closing.
-            if (stdout) {
+            if (stdout && stdout !== process.stdout) {
                 stdout.end();
             }
-            if (stderr) {
+            if (stderr && stderr !== process.stderr) {
                 stderr.end();
             }
             return JSON.parse(buff.toString('utf8')) as V1Status;
@@ -109,7 +109,7 @@ export class WebSocketHandler implements WebSocketInterface {
         this.config.applytoHTTPSOptions(opts);
 
         return new Promise((resolve, reject) => {
-            const client = (this.socketFactory ? this.socketFactory(uri, opts) : new WebSocket(uri, opts));
+            const client = (this.socketFactory ? this.socketFactory(uri, opts) : new WebSocket(uri, protocols, opts));
             let resolved = false;
 
             client.onopen = () => {