Fix a bug in attach where stdin wasn't copied. Add tests.

Author: brendandburns

node-client/src/attach.ts

@@ -2,18 +2,22 @@ import querystring = require('querystring');
 import stream = require('stream');
 
 import { KubeConfig } from './config';
-import { WebSocketHandler } from './web-socket-handler';
+import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
 
 export class Attach {
-    public 'handler': WebSocketHandler;
+    public 'handler': WebSocketInterface;
 
-    public constructor(config: KubeConfig) {
-        this.handler = new WebSocketHandler(config);
+    public constructor(config: KubeConfig, websocketInterface?: WebSocketInterface) {
+        if (websocketInterface) {
+            this.handler = websocketInterface;
+        } else {
+            this.handler = new WebSocketHandler(config);
+        }
     }
 
-    public attach(namespace: string, podName: string, containerName: string,
-                  stdout: stream.Writable | any, stderr: stream.Writable | any, stdin: stream.Readable | any,
-                  tty: boolean): Promise<void> {
+    public async attach(namespace: string, podName: string, containerName: string,
+                        stdout: stream.Writable | any, stderr: stream.Writable | any, stdin: stream.Readable | any,
+                        tty: boolean): Promise<void> {
         const query = {
             container: containerName,
             stderr: stderr != null,
@@ -23,13 +27,12 @@ export class Attach {
         };
         const queryStr = querystring.stringify(query);
         const path = `/api/v1/namespaces/${namespace}/pods/${podName}/attach?${queryStr}`;
-        const promise = this.handler.connect(path, null, (streamNum: number, buff: Buffer): boolean => {
+        const conn = await this.handler.connect(path, null, (streamNum: number, buff: Buffer): boolean => {
             WebSocketHandler.handleStandardStreams(streamNum, buff, stdout, stderr);
             return true;
         });
-        const result = new Promise<void>((resolvePromise, reject) => {
-            promise.then(() => resolvePromise(), (err) => reject(err));
-        });
-        return result;
+        if (stdin != null) {
+            WebSocketHandler.handleStandardInput(conn, stdin);
+        }
     }
 }

node-client/src/attach_test.ts

@@ -0,0 +1,107 @@
+import { expect } from 'chai';
+import { ReadableStreamBuffer, WritableStreamBuffer } from 'stream-buffers';
+import { anyFunction, anything, capture, instance, mock, verify, when } from 'ts-mockito';
+import ws = require('websocket');
+
+import { Attach } from './attach';
+import { KubeConfig } from './config';
+import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
+
+describe('Attach', () => {
+    describe('basic', () => {
+        it('should correctly attach to a url', async () => {
+            const kc = new KubeConfig();
+            const fakeWebSocket: WebSocketInterface = mock(WebSocketHandler);
+            const attach = new Attach(kc, instance(fakeWebSocket));
+            const osStream = new WritableStreamBuffer();
+            const errStream = new WritableStreamBuffer();
+            const isStream = new ReadableStreamBuffer();
+
+            const namespace = 'somenamespace';
+            const pod = 'somepod';
+            const container = 'somecontainer';
+
+            await attach.attach(
+                namespace, pod, container, osStream, errStream, isStream, false);
+
+            const path = `/api/v1/namespaces/${namespace}/pods/${pod}/attach`;
+            let args = `container=${container}&stderr=true&stdin=true&stdout=true&tty=false`;
+            verify(fakeWebSocket.connect(`${path}?${args}`, null, anyFunction())).called();
+
+            await attach.attach(
+                namespace, pod, container, null, null, null, false);
+            args = `container=${container}&stderr=false&stdin=false&stdout=false&tty=false`;
+            verify(fakeWebSocket.connect(`${path}?${args}`, null, anyFunction())).called();
+
+            await attach.attach(
+                namespace, pod, container, osStream, null, null, false);
+            args = `container=${container}&stderr=false&stdin=false&stdout=true&tty=false`;
+            verify(fakeWebSocket.connect(`${path}?${args}`, null, anyFunction())).called();
+
+            await attach.attach(
+                namespace, pod, container, osStream, errStream, null, false);
+            args = `container=${container}&stderr=true&stdin=false&stdout=true&tty=false`;
+            verify(fakeWebSocket.connect(`${path}?${args}`, null, anyFunction())).called();
+
+            await attach.attach(
+                namespace, pod, container, osStream, errStream, null, true);
+            args = `container=${container}&stderr=true&stdin=false&stdout=true&tty=true`;
+            verify(fakeWebSocket.connect(`${path}?${args}`, null, anyFunction())).called();
+        });
+
+        it('should correctly attach to streams', async () => {
+            const kc = new KubeConfig();
+            const fakeWebSocket: WebSocketInterface = mock(WebSocketHandler);
+            const attach = new Attach(kc, instance(fakeWebSocket));
+            const osStream = new WritableStreamBuffer();
+            const errStream = new WritableStreamBuffer();
+            const isStream = new ReadableStreamBuffer();
+
+            const namespace = 'somenamespace';
+            const pod = 'somepod';
+            const container = 'somecontainer';
+
+            const path = `/api/v1/namespaces/${namespace}/pods/${pod}/attach`;
+            const args = `container=${container}&stderr=true&stdin=true&stdout=true&tty=false`;
+
+            const fakeConn: ws.connection = mock(ws.connection);
+            when(fakeWebSocket.connect(`${path}?${args}`, null, anyFunction())).thenResolve(fakeConn);
+
+            await attach.attach(
+                namespace, pod, container, osStream, errStream, isStream, false);
+            const [, , outputFn] = capture(fakeWebSocket.connect).last();
+
+            /* tslint:disable:no-unused-expression */
+            expect(outputFn).to.not.be.null;
+
+            // this is redundant but needed for the compiler, sigh...
+            if (!outputFn) {
+                return;
+            }
+
+            let buffer = Buffer.alloc(1024, 10);
+
+            outputFn(WebSocketHandler.StdoutStream, buffer);
+            expect(osStream.size()).to.equal(1024);
+            let buff = osStream.getContents() as Buffer;
+            for (let i = 0; i < 1024; i++) {
+                expect(buff[i]).to.equal(10);
+            }
+
+            buffer = Buffer.alloc(1024, 20);
+            outputFn(WebSocketHandler.StderrStream, buffer);
+            expect(errStream.size()).to.equal(1024);
+            buff = errStream.getContents() as Buffer;
+            for (let i = 0; i < 1024; i++) {
+                expect(buff[i]).to.equal(20);
+            }
+
+            const msg = 'This is test data';
+            isStream.put(msg);
+            verify(fakeConn.send(msg));
+
+            isStream.stop();
+            verify(fakeConn.close());
+        });
+    });
+});

node-client/src/web-socket-handler.ts

@@ -43,7 +43,7 @@ export class WebSocketHandler implements WebSocketInterface {
             }
             return JSON.parse(buff.toString('utf8')) as V1Status;
         } else {
-            throw new Error('Unknown stream: ' + stream);
+            throw new Error('Unknown stream: ' + streamNum);
         }
         return null;
     }