dart/test/integration/syncbase_database_test.dart

// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

library syncbase_database_test;

import 'dart:async';
import 'dart:convert' show UTF8;

import 'package:test/test.dart';

import 'package:ether/src/testing_instrumentation.dart' as testing;
import 'package:ether/syncbase_client.dart'
    show SyncbaseClient, WatchChangeTypes, WatchChange, WatchGlobStreamImpl;

import './utils.dart' as utils;

runDatabaseTests(SyncbaseClient c) {
  test('getting a handle to a database', () {
    var app = c.app(utils.uniqueName('app'));
    var dbName = utils.uniqueName('db');
    var db = app.noSqlDatabase(dbName);
    expect(db.name, equals(dbName));
    expect(db.fullName, equals(app.fullName + '/' + dbName));
  });

  test('creating and destroying a database', () async {
    var app = c.app(utils.uniqueName('app'));
    await app.create(utils.emptyPerms());

    var db = app.noSqlDatabase(utils.uniqueName('db'));

    expect(await db.exists(), equals(false));
    await db.create(utils.emptyPerms());
    expect(await db.exists(), equals(true));
    await db.destroy();
    expect(await db.exists(), equals(false));
  });

  test('basic watch', () async {
    var app = c.app(utils.uniqueName('app'));
    await app.create(utils.emptyPerms());
    var db = app.noSqlDatabase(utils.uniqueName('db'));
    await db.create(utils.emptyPerms());
    var table = db.table(utils.uniqueName('table'));
    await table.create(utils.emptyPerms());

    // Perform some operations that we won't be watching.
    await table.put('row1', UTF8.encode('value1'));
    await table.delete('row1');

    // Start watching everything from now.
    var resumeMarker = await db.getResumeMarker();
    var prefix = '';
    var watchStream = db.watch(table.name, prefix, resumeMarker);

    // Perform some operations while are watching.
    var expectedChanges = new List<WatchChange>();

    await table.put('row2', UTF8.encode('value2'));
    resumeMarker = await db.getResumeMarker();
    var expectedChange = SyncbaseClient.watchChange(
        tableName: table.name,
        rowKey: 'row2',
        changeType: WatchChangeTypes.put,
        valueBytes: UTF8.encode('value2'),
        resumeMarker: resumeMarker);
    expectedChanges.add(expectedChange);

    await table.delete('row2');
    resumeMarker = await db.getResumeMarker();
    expectedChange = SyncbaseClient.watchChange(
        tableName: table.name,
        rowKey: 'row2',
        changeType: WatchChangeTypes.delete,
        resumeMarker: resumeMarker);
    expectedChanges.add(expectedChange);

    // Ensure we see all the expected changes in order in the watch stream.
    var changeNum = 0;
    await for (var change in watchStream) {
      // Classes generated by mojom Dart compiler do not override == and hashCode
      // but they do override toString to print all properties. So we use toString
      // to assert equality.
      expect(change.toString(), equals(expectedChanges[changeNum].toString()));
      changeNum++;
      // We need to break out of awaiting for watch stream values when we get everything we expected.
      // because watch stream does not end until canceled by design and we don't have canceling mechanism yet.
      if (changeNum == expectedChanges.length) {
        break;
      }
    }
  });

  test('watch flow control', () async {
    var app = c.app(utils.uniqueName('app'));
    await app.create(utils.emptyPerms());
    var db = app.noSqlDatabase(utils.uniqueName('db'));
    await db.create(utils.emptyPerms());
    var table = db.table(utils.uniqueName('table'));
    await table.create(utils.emptyPerms());

    var resumeMarker = await db.getResumeMarker();
    var aFewMoments = new Duration(seconds: 1);
    const int numOperations = 10;
    var allOperations = [];

    // Do several put operations in parallel and wait until they are all done.
    for (var i = 0; i < numOperations; i++) {
      allOperations.add(table.put('row $i', UTF8.encode('value$i')));
    }
    await Future.wait(allOperations);

    // Reset testing instrumentations.
    testing.DatabaseWatch.onChangeCounter.reset();

    // Create a watch stream.
    var watchStream = db.watch(table.name, '', resumeMarker);

    // Listen for the data on the stream.
    var allExpectedChangesReceived = new Completer();
    onData(_) {
      if (testing.DatabaseWatch.onChangeCounter.count == numOperations) {
        allExpectedChangesReceived.complete();
      }
    }
    var streamListener = watchStream.listen(onData);

    // Pause the stream.
    streamListener.pause();

    // Wait a few moments.
    await new Future.delayed(aFewMoments);

    // Assert that we did not got any* events from server when paused.
    // testing.DatabaseWatch.onChangeCounter instrumentation is used to ensure
    // that client did not receive any updates from mojo server, guaranteeing
    // flow control propagated properly all the way to the other end of the pipe
    // *Note: We always get 1 change before we can tell the server to block by
    // not acking that single change.
    expect(testing.DatabaseWatch.onChangeCounter.count, equals(1));

    // Resume the stream.
    streamListener.resume();

    // Wait until we get all expected changes.
    await allExpectedChangesReceived.future;

    // Assert we've got all the expected changes after resuming.
    expect(testing.DatabaseWatch.onChangeCounter.count, equals(numOperations));
  });

  // TODO(nlacasse): Test database.get/setPermissions.
}