Merge pull request #109 from Unity-Technologies/dev/improve-test-cove…

…rage

Improve test coverage

lib/cache/cache_base.js

@@ -10,6 +10,7 @@ const { promisify } = require('util');
 const loki = require('lokijs');
 const ReliabilityManager = require('./reliability_manager');
 const _ = require('lodash');
+const { Writable } = require('stream');
 
 const kDbName = 'cache.db';
 
@@ -176,7 +177,7 @@ class CacheBase extends EventEmitter {
      * @returns {Promise<any>}
      */
     createPutTransaction(guid, hash) {
-        return Promise.reject(new Error("Not implemented"));
+        return Promise.resolve(new PutTransaction(guid, hash));
     }
 
     /**
@@ -300,7 +301,9 @@ class PutTransaction extends EventEmitter {
      * @returns {Promise<any>}
      */
     getWriteStream(type, size) {
-        return Promise.reject(new Error("Not implemented"));
+        return Promise.resolve(new Writable({
+            write(chunk, encoding, cb){ setImmediate(cb); }
+        }));
     }
 
     async invalidate() {
@@ -313,7 +316,7 @@ class PutTransaction extends EventEmitter {
      * @returns {Promise<any>}
      */
     async writeFilesToPath(targetPath) {
-        return Promise.reject(new Error("Not implemented"));
+        return Promise.resolve();
     }
 }
 

lib/server/server.js

@@ -57,6 +57,14 @@ class CacheServer {
         return this._server;
     }
 
+    /**
+     *
+     * @returns {Array|*}
+     */
+    get mirrors() {
+        return this._mirrors;
+    }
+
     /**
      *
      * @param {Function} cb
@@ -82,10 +90,7 @@ class CacheServer {
             const mirrors = this._mirrors;
             if(mirrors.length > 0) {
                 cmdProc.on('onTransactionEnd', (trx) => {
-                    mirrors.forEach(m => {
-                        if(m.address !== socket.remoteAddress)
-                            m.queueTransaction(trx);
-                    });
+                    mirrors.forEach(m => m.queueTransaction(trx));
                 });
             }
 
@@ -103,9 +108,7 @@ class CacheServer {
                 .pipe(socket);     // Connect back to socket to send files
 
             socket['commandProcessor'] = cmdProc;
-        });
-
-        this._server.on('error', err => {
+        }).on('error', err => {
             if (err.code === 'EADDRINUSE') {
                 helpers.log(consts.LOG_ERR, `Port ${this.port} is already in use...`);
             }

lib/server/transaction_mirror.js

@@ -37,6 +37,10 @@ class TransactionMirror {
         return this._connectOptions.host;
     }
 
+    get port() {
+        return this._connectOptions.port;
+    }
+
     _connect() {
         helpers.log(consts.LOG_INFO, `[TransactionMirror] Connecting to ${this._connectOptions.host}:${this._connectOptions.port}`);
         return this._client.connect();

test/cache_base.js

@@ -8,6 +8,7 @@ const path = require('path');
 const randomBuffer = require('./test_utils').randomBuffer;
 const consts = require('../lib/constants');
 const sinon = require('sinon');
+const { Writable } = require('stream');
 
 describe("Cache: Base Class", () => {
     let cache;
@@ -124,9 +125,9 @@ describe("Cache: Base Class", () => {
     });
 
     describe("createPutTransaction", () => {
-        it("should require override implementation in subclasses by returning an error", () => {
-            return cache.createPutTransaction()
-                .then(() => { throw new Error("Expected error!"); }, () => {});
+        it("should return an instance of a PutTransaction", async () => {
+            const t = await cache.createPutTransaction();
+            assert.ok(t instanceof PutTransaction);
         });
     });
 
@@ -264,16 +265,16 @@ describe("PutTransaction: Base Class", () => {
     });
 
     describe("getWriteStream", () => {
-        it("should require override implementation in subclasses by returning an error", () => {
-            return trx.getWriteStream(consts.FILE_TYPE.INFO, 0)
-                .then(() => { throw new Error("Expected error!"); }, () => {});
+        it("should return a Writable stream", async() => {
+            const s = await trx.getWriteStream(consts.FILE_TYPE.INFO, 0);
+            assert.ok(s instanceof Writable);
         });
     });
 
     describe("writeFilesToPath", () => {
-        it("should require override implementation in subclasses by returning an error", () => {
-            return trx.writeFilesToPath()
-                .then(() => { throw new Error("Expected error!"); }, () => {});
+        it("should return a promise", () => {
+            const p = trx.writeFilesToPath();
+            assert.ok(p instanceof Promise);
         });
     });
 });

test/protocol.js

@@ -323,6 +323,7 @@ describe("Protocol", () => {
                 });
 
                 const tests = [
+                    {cmd: cmd.getAsset, blob: self.data.bin, type: 'bin', packetSize: 1},
                     {cmd: cmd.getAsset, blob: self.data.bin, type: 'bin', packetSize: SMALL_PACKET_SIZE},
                     {cmd: cmd.getInfo, blob: self.data.info, type: 'info', packetSize: MED_PACKET_SIZE},
                     {cmd: cmd.getResource, blob: self.data.resource, type: 'resource', packetSize: LARGE_PACKET_SIZE}

test/server.js

@@ -6,15 +6,61 @@ const os = require('os');
 const helpers = require('../lib/helpers');
 const consts = require('../lib/constants');
 const CacheServer = require('../lib/server/server');
-const Cache = require('../lib/cache/cache_base').CacheBase;
-const sleep = require('./test_utils').sleep;
-const cmd = require('./test_utils').cmd;
+const CacheBase = require('../lib/cache/cache_base').CacheBase;
+const TransactionMirror = require('../lib/server/transaction_mirror');
+const { generateCommandData, encodeCommand, clientWrite, sleep, cmd } = require('./test_utils');
+const sinon = require('sinon');
 
-const cache = new Cache();
-const server = new CacheServer(cache, {port: 0});
+const cache = new CacheBase();
 let client;
 
+describe("Server constructor", function() {
+    it("should use the default port if no port is specified in options", () => {
+        const s = new CacheServer(cache, { mirror:[] });
+        assert.strictEqual(s.port, consts.DEFAULT_PORT);
+    });
+});
+
+describe("Server mirroring", function() {
+    const server = new CacheServer(cache, {
+        port: 0,
+        mirror: [{host: "127.0.0.1", port: 8126}, {host: "1.2.3.4", port: 8126}, {host: "4.3.2.1", port: 8126}]
+    });
+
+    before(function () {
+        return server.start(err => { return Promise.reject(err); });
+    });
+
+    after(function() {
+        server.stop();
+    });
+
+    beforeEach(function (done) {
+        client = net.connect({port: server.port}, done);
+    });
+
+    afterEach(() => client.end());
+
+    it("should mirror transactions to the configured list of mirrors", async () => {
+        const spies = server.mirrors.map(m => {
+            return sinon.spy(m, "queueTransaction");
+        });
+
+        const testData = generateCommandData();
+
+        const buf = Buffer.from(helpers.encodeInt32(consts.PROTOCOL_VERSION) +
+            encodeCommand(cmd.transactionStart, testData.guid, testData.hash) +
+            encodeCommand(cmd.putAsset, null, null, testData.bin) +
+            encodeCommand(cmd.transactionEnd), 'ascii');
+
+        await clientWrite(client, buf);
+
+        spies.forEach(s => assert(s.calledOnce));
+    });
+});
+
 describe("Server common", function() {
+    const server = new CacheServer(cache, {port: 0});
 
     before(function () {
         this._defaultErrCallback = err => assert(!err, `Cache Server reported error! ${err}`);
@@ -71,7 +117,7 @@ describe("Server common", function() {
     describe("Ipv6", function() {
         const ipv6Server = new CacheServer(cache, {port: 0, allowIpv6: true});
 
-        before(function () {  
+        before(function () {
             const interfaces = os.networkInterfaces();
             let ipv6Available = false;
             Object.keys(interfaces).forEach(function (interfaceName){
@@ -85,15 +131,15 @@ describe("Server common", function() {
             if(!ipv6Available){
                 console.log("Skipping IPv6 tests because IPv6 is not available on this machine");
                 this.skip();
-            }   
+            }
 
             return ipv6Server.start(err => assert(!err, `Cache Server reported error! ${err}`));
         });
-    
+
         after(function() {
             ipv6Server.stop();
         });
-    
+
         it("should bind to ipv6 when allowed", function(done) {
             const serverAddress = ipv6Server._server.address();
             assert.strictEqual(serverAddress.family, "IPv6");
@@ -108,7 +154,7 @@ describe("Server common", function() {
         before(function () {
             return ipv4Server.start(err => assert(!err, `Cache Server reported error! ${err}`));
         });
-    
+
         after(function() {
             ipv4Server.stop();
         });

test/unity_cache_server.js

@@ -208,6 +208,9 @@ describe("Unity Cache Server bootstrap", () => {
                             cachePath: tmpPath
                         }
                     }
+                },
+                Server: {
+                    port: 0
                 }
             });