Add unit tests for watch.

Author: brendandburns

node-client/src/config.ts

@@ -67,11 +67,18 @@ export class KubeConfig {
     }
 
     public getContextObject(name: string) {
+        if (!this.contexts) {
+            return null;
+        }
         return KubeConfig.findObject(this.contexts, name, 'context');
     }
 
-    public getCurrentCluster() {
-        return this.getCluster(this.getCurrentContextObject().cluster);
+    public getCurrentCluster(): Cluster | null {
+        const context = this.getCurrentContextObject();
+        if (!context) {
+            return null;
+        }
+        return this.getCluster(context.cluster);
     }
 
     public getCluster(name: string): Cluster {
@@ -106,7 +113,7 @@ export class KubeConfig {
 
         this.applyOptions(opts);
 
-        if (cluster.skipTLSVerify) {
+        if (cluster && cluster.skipTLSVerify) {
             opts.strictSSL = false;
         }
 
@@ -210,7 +217,11 @@ export class KubeConfig {
     }
 
     public makeApiClient<T extends ApiType>(apiClientType: ApiConstructor<T>) {
-        const apiClient = new apiClientType(this.getCurrentCluster().server);
+        const cluster = this.getCurrentCluster();
+        if (!cluster) {
+            throw new Error('No active cluster!');
+        }
+        const apiClient = new apiClientType(cluster.server);
         apiClient.setDefaultAuthentication(this);
 
         return apiClient;
@@ -234,7 +245,7 @@ export class KubeConfig {
         const cluster = this.getCurrentCluster();
         const user = this.getCurrentUser();
 
-        const ca = this.bufferFromFileOrString(cluster.caFile, cluster.caData);
+        const ca = cluster != null ? this.bufferFromFileOrString(cluster.caFile, cluster.caData) : null;
         if (ca) {
             opts.ca = ca;
         }
@@ -345,7 +356,12 @@ export class Config {
         const kc = new KubeConfig();
         kc.loadFromCluster();
 
-        const k8sApi = new apiClientType(kc.getCurrentCluster().server);
+        const cluster = kc.getCurrentCluster();
+        if (!cluster) {
+            throw new Error('No active cluster!');
+        }
+
+        const k8sApi = new apiClientType(cluster.server);
         k8sApi.setDefaultAuthentication(kc);
 
         return k8sApi;

node-client/src/config_test.ts

@@ -611,8 +611,13 @@ describe('KubeConfig', () => {
             delete process.env.KUBERNETES_SERVICE_HOST;
             delete process.env.KUBERNETES_SERVICE_PORT;
 
-            expect(kc.getCurrentCluster().caFile).to.equal('/var/run/secrets/kubernetes.io/serviceaccount/ca.crt');
-            expect(kc.getCurrentCluster().server).to.equal('https://kubernetes:443');
+            const cluster = kc.getCurrentCluster();
+            expect(cluster).to.not.be.null;
+            if (!cluster) {
+                return;
+            }
+            expect(cluster.caFile).to.equal('/var/run/secrets/kubernetes.io/serviceaccount/ca.crt');
+            expect(cluster.server).to.equal('https://kubernetes:443');
             expect(kc.getCurrentUser().token).to.equal(token);
         });
 
@@ -634,7 +639,12 @@ describe('KubeConfig', () => {
             delete process.env.KUBERNETES_SERVICE_HOST;
             delete process.env.KUBERNETES_SERVICE_PORT;
 
-            expect(kc.getCurrentCluster().server).to.equal('http://kubernetes:80');
+            const cluster = kc.getCurrentCluster();
+            expect(cluster).to.not.be.null;
+            if (!cluster) {
+                return;
+            }
+            expect(cluster.server).to.equal('http://kubernetes:80');
         });
 
         it('should default to localhost', () => {
@@ -644,8 +654,13 @@ describe('KubeConfig', () => {
             kc.loadFromDefault();
             process.env.HOME = currentHome;
 
+            const cluster = kc.getCurrentCluster();
+            expect(cluster).to.not.be.null;
+            if (!cluster) {
+                return;
+            }
             expect(kc.getCurrentUser().name).to.equal('user');
-            expect(kc.getCurrentCluster().server).to.equal('http://localhost:8080');
+            expect(cluster.server).to.equal('http://localhost:8080');
         });
     });
 

node-client/src/watch.ts

@@ -7,17 +7,37 @@ export interface WatchUpdate {
     object: object;
 }
 
+export interface RequestInterface {
+    webRequest(opts: request.Options, callback: (err, response, body) => void): any;
+}
+
+export class DefaultRequest implements RequestInterface {
+    public webRequest(opts: request.Options, callback: (err, response, body) => void): any {
+        return request(opts, callback);
+    }
+}
+
 export class Watch {
-    public 'config': KubeConfig;
+    public config: KubeConfig;
+    private readonly requestImpl: RequestInterface;
 
-    public constructor(config: KubeConfig) {
+    public constructor(config: KubeConfig, requestImpl?: RequestInterface) {
         this.config = config;
+        if (requestImpl) {
+            this.requestImpl = requestImpl;
+        } else {
+            this.requestImpl = new DefaultRequest();
+        }
     }
 
     public watch(path: string, queryParams: any,
                  callback: (phase: string, obj: any) => void,
                  done: (err: any) => void): any {
-        const url = this.config.getCurrentCluster().server + path;
+        const cluster = this.config.getCurrentCluster();
+        if (!cluster) {
+            throw new Error('No currently active cluster');
+        }
+        const url = cluster.server + path;
 
         queryParams.watch = true;
         const headerParams: any = {};
@@ -47,11 +67,12 @@ export class Watch {
             }
         });
 
-        const req = request(requestOptions, (error, response, body) => {
+        const req = this.requestImpl.webRequest(requestOptions, (error, response, body) => {
             if (error) {
                 done(error);
+            } else {
+                done(null);
             }
-            done(null);
         });
         req.pipe(stream);
         return req;

node-client/src/watch_test.ts

@@ -0,0 +1,100 @@
+import { expect } from 'chai';
+import request = require('request');
+import { ReadableStreamBuffer, WritableStreamBuffer } from 'stream-buffers';
+import { anyFunction, anything, capture, instance, mock, reset, verify, when  } from 'ts-mockito';
+
+import { KubeConfig } from './config';
+import { Cluster, Context, User } from './config_types';
+import { DefaultRequest, Watch } from './watch';
+
+describe('Watch', () => {
+    it('should construct correctly', () => {
+        const kc = new KubeConfig();
+        const watch = new Watch(kc);
+    });
+
+    it('should watch correctly', () => {
+        const kc = new KubeConfig();
+        const server = 'foo.company.com';
+        kc.clusters = [
+            {
+                name: 'cluster',
+                server,
+            } as Cluster,
+        ] as Cluster[];
+        kc.contexts = [
+            {
+                cluster: 'cluster',
+                user: 'user',
+            } as Context,
+        ] as Context[];
+        kc.users = [
+            {
+                name: 'user',
+            } as User,
+        ];
+        const fakeRequestor = mock(DefaultRequest);
+        const watch = new Watch(kc, instance(fakeRequestor));
+
+        const obj1 = {
+            type: 'ADDED',
+            object: {
+                foo: 'bar',
+            },
+        };
+
+        const obj2 = {
+            type: 'MODIFIED',
+            object: {
+                baz: 'blah',
+            },
+        };
+
+        const fakeRequest = {
+            pipe: (stream) => {
+                stream.write(JSON.stringify(obj1) + '\n');
+                stream.write(JSON.stringify(obj2) + '\n');
+            },
+        };
+
+        when(fakeRequestor.webRequest(anything(), anyFunction())).thenReturn(fakeRequest);
+
+        const path = '/some/path/to/object';
+
+        const receivedTypes: string[] = [];
+        const receivedObjects: string[] = [];
+        let doneCalled = false;
+        let doneErr: any;
+
+        watch.watch(path, {}, (phase: string, obj: string) => {
+            receivedTypes.push(phase);
+            receivedObjects.push(obj);
+        }, (err: any) => {
+            doneCalled = true;
+            doneErr = err;
+        });
+
+        verify(fakeRequestor.webRequest(anything(), anyFunction()));
+
+        const [opts, doneCallback] = capture(fakeRequestor.webRequest).last();
+        const reqOpts: request.OptionsWithUri = opts as request.OptionsWithUri;
+
+        expect(reqOpts.uri).to.equal(`${server}${path}`);
+        expect(reqOpts.method).to.equal('GET');
+        expect(reqOpts.json).to.equal(true);
+
+        expect(receivedTypes).to.deep.equal([obj1.type, obj2.type]);
+        expect(receivedObjects).to.deep.equal([obj1.object, obj2.object]);
+
+        expect(doneCalled).to.equal(false);
+
+        doneCallback(null, null, null);
+
+        expect(doneCalled).to.equal(true);
+        expect(doneErr).to.equal(null);
+
+        const errIn = {error: 'err'};
+        doneCallback(errIn, null, null);
+        expect(doneErr).to.deep.equal(errIn);
+    });
+});

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

@@ -86,7 +86,11 @@ export class WebSocketHandler implements WebSocketInterface {
                    textHandler: ((text: string) => boolean) | null,
                    binaryHandler: ((stream: number, buff: Buffer) => boolean) | null): Promise<ws.connection> {
 
-        const server = this.config.getCurrentCluster().server;
+        const cluster = this.config.getCurrentCluster();
+        if (!cluster) {
+            throw new Error('No cluster is defined.');
+        }
+        const server = cluster.server;
         const ssl = server.startsWith('https://');
         const target = ssl ? server.substr(8) : server.substr(7);
         const proto = ssl ? 'wss' : 'ws';