Adding Watch conformance tests (#176)

* Adding Watch conformance tests

* Adding newline

Author: schmidt-sebastian

conformance/runner.js

@@ -22,6 +22,7 @@ const is = require('is');
 const through = require('through2');
 const googleProtoFiles = require('google-proto-files');
 const protobufjs = require('protobufjs');
+const duplexify = require('duplexify');
 const grpc = require('google-gax').grpc().grpc;
 
 const Firestore = require('../');
@@ -52,6 +53,10 @@ const collRef = function(path) {
   return firestore.collection(relativePath);
 };
 
+const watchQuery = collRef(
+  'projects/projectID/databases/(default)/documents/C'
+).orderBy('a');
+
 /** Converts JSON test data into JavaScript types suitable for the Node API. */
 const convertInput = {
   argument: json => {
@@ -128,6 +133,40 @@ const convertInput = {
     }
     return args;
   },
+  snapshot: snapshot => {
+    const docs = [];
+    const changes = [];
+    const readTime = DocumentSnapshot.toISOTime(snapshot.readTime);
+
+    for (const doc of snapshot.docs) {
+      const deepCopy = JSON.parse(JSON.stringify(doc));
+      deepCopy.fields = convert.documentFromJson(deepCopy.fields);
+      docs.push(firestore.snapshot_(deepCopy, readTime, 'json'));
+    }
+
+    for (const change of snapshot.changes) {
+      const deepCopy = JSON.parse(JSON.stringify(change.doc));
+      deepCopy.fields = convert.documentFromJson(deepCopy.fields);
+      const doc = firestore.snapshot_(deepCopy, readTime, 'json');
+      const type = ['unspecified', 'added', 'removed', 'modified'][change.kind];
+      changes.push(
+        new Firestore.DocumentChange(
+          type,
+          doc,
+          change.oldIndex,
+          change.newIndex
+        )
+      );
+    }
+
+    return new Firestore.QuerySnapshot(
+      watchQuery,
+      readTime,
+      docs.length,
+      () => docs,
+      () => changes
+    );
+  },
 };
 
 /** Converts Firestore Protobuf types in Proto3 JSON format to Protobuf JS. */
@@ -173,6 +212,23 @@ const convertProto = {
     }
     return deepCopy;
   },
+  listenRequest: listenRequest => {
+    const deepCopy = JSON.parse(JSON.stringify(listenRequest));
+
+    if (deepCopy.targetChange) {
+      if (deepCopy.targetChange.targetChangeType === undefined) {
+        deepCopy.targetChange.targetChangeType = 'NO_CHANGE';
+      }
+    }
+
+    if (deepCopy.documentChange) {
+      deepCopy.documentChange.document.fields = convert.documentFromJson(
+        deepCopy.documentChange.document.fields
+      );
+    }
+
+    return deepCopy;
+  },
 };
 
 /** Request handler for _commit. */
@@ -366,6 +422,38 @@ function runTest(spec) {
     });
   };
 
+  const watchTest = function(spec) {
+    let expectedSnapshots = spec.snapshots;
+
+    const writeStream = through.obj();
+
+    firestore.api.Firestore._listen = () => {
+      return duplexify.obj(through.obj(), writeStream);
+    };
+
+    return new Promise((resolve, reject) => {
+      const unlisten = watchQuery.onSnapshot(acutalSnap => {
+        const expectedSnapshot = expectedSnapshots.shift();
+        if (expectedSnapshot) {
+          if (!acutalSnap.isEqual(convertInput.snapshot(expectedSnapshot))) {
+            reject(new Error('Expected and actual snapshots do not match.'));
+          }
+
+          if (expectedSnapshots.length === 0) {
+            unlisten();
+            resolve();
+          }
+        } else {
+          reject(new Error('Received unexpected snapshot'));
+        }
+      });
+
+      for (const response of spec.responses) {
+        writeStream.write(convertProto.listenRequest(response));
+      }
+    });
+  };
+
   let testSpec;
   let testPromise;
 
@@ -375,6 +463,7 @@ function runTest(spec) {
   const updateSpec = spec.update || spec.updatePaths;
   const deleteSpec = spec.delete;
   const querySpec = spec.query;
+  const listenSpec = spec.listen;
 
   if (getSpec) {
     testSpec = getSpec;
@@ -394,6 +483,9 @@ function runTest(spec) {
   } else if (querySpec) {
     testSpec = querySpec;
     testPromise = queryTest(querySpec);
+  } else if (listenSpec) {
+    testSpec = listenSpec;
+    testPromise = watchTest(listenSpec);
   } else {
     return Promise.reject(new Error(`Unhandled Spec: ${JSON.stringify(spec)}`));
   }

conformance/test-definition.proto

@@ -4,12 +4,15 @@ syntax = "proto3";
 
 package tests;
 
+option php_namespace = "Google\\Cloud\\Firestore\\Tests\\Conformance";
 option csharp_namespace = "Google.Cloud.Firestore.Tests.Proto";
 option java_package = "com.google.cloud.firestore.conformance";
 
-import "google/firestore/v1beta1/firestore.proto";
 import "google/firestore/v1beta1/common.proto";
+import "google/firestore/v1beta1/document.proto";
+import "google/firestore/v1beta1/firestore.proto";
 import "google/firestore/v1beta1/query.proto";
+import "google/protobuf/timestamp.proto";
 
 // A collection of tests.
 message TestSuite {
@@ -28,6 +31,7 @@ message Test {
     UpdatePathsTest update_paths = 6;
     DeleteTest      delete = 7;
     QueryTest       query = 8;
+    ListenTest      listen = 9;
   }
 }
 
@@ -153,3 +157,40 @@ message DocSnapshot {
 message FieldPath {
   repeated string field = 1;
 }
+
+// A test of the Listen streaming RPC (a.k.a. FireStore watch).
+// If the sequence of responses is provided to the implementation,
+// it should produce the sequence of snapshots.
+// If is_error is true, an error should occur after the snapshots.
+//
+// The tests assume that the query is
+// Collection("projects/projectID/databases/(default)/documents/C").OrderBy("a", Ascending)
+//
+// The watch target ID used in these tests is 1. Test interpreters
+// should either change their client's ID for testing,
+// or change the ID in the tests before running them.
+message ListenTest {
+  repeated google.firestore.v1beta1.ListenResponse responses = 1;
+  repeated Snapshot snapshots = 2;
+  bool is_error = 3;
+}
+
+message Snapshot {
+  repeated google.firestore.v1beta1.Document docs = 1;
+  repeated DocChange changes = 2;
+  google.protobuf.Timestamp read_time = 3;
+}
+
+message DocChange {
+  enum Kind {
+    KIND_UNSPECIFIED = 0;
+    ADDED = 1;
+    REMOVED = 2;
+    MODIFIED = 3;
+  }
+
+  Kind kind = 1;
+  google.firestore.v1beta1.Document doc = 2;
+  int32 old_index = 3;
+  int32 new_index = 4;
+}

conformance/test-suite.binproto

@@ -1257,7 +1257,7 @@ module.exports.QuerySnapshot = reference.QuerySnapshot;
  * @see DocumentChange
  * @type DocumentChange
  */
-module.exports.DocumentChange = document.DocumentChange;
+module.exports.DocumentChange = reference.DocumentChange;
 
 /**
  * {@link Query} class.

src/index.js

@@ -67,7 +67,7 @@ let ResourcePath = require('./path').ResourcePath;
  * @private
  * @type {number}
  */
-const WATCH_TARGET_ID = 0xf0;
+const WATCH_TARGET_ID = 0x1;
 
 /*!
  * The change type for document change events.

src/watch.js

@@ -598,7 +598,7 @@ describe('Query watch', function() {
 
     firestore = createInstance();
 
-    targetId = 0xf0;
+    targetId = 0x1;
 
     streamHelper = new StreamHelper(firestore);
     watchHelper = new WatchHelper(
@@ -2125,7 +2125,7 @@ describe('DocumentReference watch', function() {
     Backoff.setTimeoutHandler(setImmediate);
 
     firestore = createInstance();
-    targetId = 0xf0;
+    targetId = 0x1;
     doc = firestore.doc('col/doc');
     streamHelper = new StreamHelper(firestore);
     watchHelper = new WatchHelper(streamHelper, doc, targetId);