Fix transaction handling logic and implement tests

Reviewers: teon.banek

Reviewed By: teon.banek

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D1492

src/communication/bolt/v1/decoder/decoded_value.hpp

@@ -150,6 +150,7 @@ class DecodedValue {
   DecodedValue(const std::string &value) : type_(Type::String) {
     new (&string_v) std::string(value);
   }
+  DecodedValue(const char *value) : DecodedValue(std::string(value)) {}
   DecodedValue(const std::vector<DecodedValue> &value) : type_(Type::List) {
     new (&list_v) std::vector<DecodedValue>(value);
   }

src/communication/bolt/v1/session.hpp

@@ -47,9 +47,6 @@ class Session {
 
   virtual ~Session() {}
 
-  /** Return `true` if we are no longer running and accepting queries */
-  virtual bool IsShuttingDown() = 0;
-
   /**
    * Put results in the `result_stream` by processing the given `query` with
    * `params`.

src/communication/bolt/v1/states/executing.hpp

@@ -62,15 +62,6 @@ State HandleRun(TSession &session, State state, Marker marker) {
 
   DLOG(INFO) << fmt::format("[Run] '{}'", query.ValueString());
 
-  // TODO: Possible (but very unlikely) race condition, where we have alive
-  // session during shutdown, but IsAcceptingTransactions isn't yet false.
-  // We should probably create transactions under some locking mechanism.
-  if (session.IsShuttingDown()) {
-    // Db is shutting down and doesn't accept new transactions so we should
-    // close this session.
-    return State::Close;
-  }
-
   try {
     // PullAll can throw.
     session.PullAll(query.ValueString(), params.ValueMap());

src/memgraph_bolt.cpp

@@ -18,6 +18,7 @@
 #include "glue/conversion.hpp"
 #include "query/exceptions.hpp"
 #include "query/interpreter.hpp"
+#include "query/transaction_engine.hpp"
 #include "stats/stats.hpp"
 #include "telemetry/telemetry.hpp"
 #include "utils/flag_validation.hpp"
@@ -73,16 +74,7 @@ class BoltSession final
       : communication::bolt::Session<communication::InputStream,
                                      communication::OutputStream>(
             input_stream, output_stream),
-        db_(data.db),
+        transaction_engine_(data.db, data.interpreter) {}
-        interpreter_(data.interpreter) {}
-
-  ~BoltSession() {
-    if (db_accessor_) {
-      Abort();
-    }
-  }
-
-  bool IsShuttingDown() override { return !db_.is_accepting_transactions(); }
 
   using communication::bolt::Session<
       communication::InputStream, communication::OutputStream>::ResultStreamT;
@@ -91,78 +83,20 @@ class BoltSession final
       const std::string &query,
       const std::map<std::string, communication::bolt::DecodedValue> &params,
       ResultStreamT *result_stream) override {
-    bool in_explicit_transaction = !!db_accessor_;
-    if (!db_accessor_)
-      db_accessor_ = std::make_unique<database::GraphDbAccessor>(db_);
-    // TODO: Queries below should probably move to interpreter, but there is
-    // only one interpreter in GraphDb. We probably need some kind of
-    // per-session access for the interpreter.
-    // TODO: Also, write tests for these queries
-    if (expect_rollback_ && query != "ROLLBACK") {
-      // Client could potentially recover if we move to error state, but we
-      // don't implement rollback of single command in transaction, only
-      // rollback of whole transaction so we can't continue in this transaction
-      // if we receive new query.
-      throw communication::bolt::ClientError(
-          "Expected ROLLBACK, because previous query contained an error");
-    }
-    if (query == "ROLLBACK") {
-      if (!in_explicit_transaction)
-        throw communication::bolt::ClientError(
-            "ROLLBACK can only be used after BEGIN");
-      Abort();
-      result_stream->Header({});
-      result_stream->Summary({});
-      return;
-    } else if (query == "BEGIN") {
-      if (in_explicit_transaction)
-        throw communication::bolt::ClientError("BEGIN already called");
-      // We accept BEGIN, so send the empty results.
-      result_stream->Header({});
-      result_stream->Summary({});
-      return;
-    } else if (query == "COMMIT") {
-      if (!in_explicit_transaction)
-        throw communication::bolt::ClientError(
-            "COMMIT can only be used after BEGIN");
-      Commit();
-      result_stream->Header({});
-      result_stream->Summary({});
-      return;
-    }
-    // Any other query in BEGIN block advances the command.
-    if (in_explicit_transaction) AdvanceCommand();
-    // Handle regular Cypher queries below
     std::map<std::string, query::TypedValue> params_tv;
     for (const auto &kv : params)
       params_tv.emplace(kv.first, glue::ToTypedValue(kv.second));
-    auto abort_tx = [this, in_explicit_transaction]() {
-      if (in_explicit_transaction)
-        expect_rollback_ = true;
-      else
-        this->Abort();
-    };
     try {
       TypedValueResultStream stream(result_stream);
-      interpreter_(query, *db_accessor_, params_tv, in_explicit_transaction)
+      transaction_engine_.PullAll(query, params_tv, &stream);
-          .PullAll(stream);
-      if (!in_explicit_transaction) Commit();
     } catch (const query::QueryException &e) {
-      abort_tx();
       // Wrap QueryException into ClientError, because we want to allow the
       // client to fix their query.
       throw communication::bolt::ClientError(e.what());
-    } catch (const utils::BasicException &) {
-      abort_tx();
-      throw;
     }
   }
 
-  void Abort() override {
+  void Abort() override { transaction_engine_.Abort(); }
-    if (!db_accessor_) return;
-    db_accessor_->Abort();
-    db_accessor_ = nullptr;
-  }
 
  private:
   // Wrapper around ResultStreamT which converts TypedValue to DecodedValue
@@ -197,28 +131,7 @@ class BoltSession final
     ResultStreamT *result_stream_;
   };
 
-  database::MasterBase &db_;
+  query::TransactionEngine transaction_engine_;
-  query::Interpreter &interpreter_;
-  // GraphDbAccessor of active transaction in the session, can be null if
-  // there is no associated transaction.
-  std::unique_ptr<database::GraphDbAccessor> db_accessor_;
-  bool expect_rollback_{false};
-
-  void Commit() {
-    DCHECK(db_accessor_) << "Commit called and there is no transaction";
-    db_accessor_->Commit();
-    db_accessor_ = nullptr;
-  }
-
-  void AdvanceCommand() {
-    db_accessor_->AdvanceCommand();
-    if (db_.type() == database::GraphDb::Type::DISTRIBUTED_MASTER) {
-      auto tx_id = db_accessor_->transaction_id();
-      auto futures =
-          db_.pull_clients().NotifyAllTransactionCommandAdvanced(tx_id);
-      for (auto &future : futures) future.wait();
-    }
-  }
 };
 
 using ServerT = communication::Server<BoltSession, SessionData>;

src/query/transaction_engine.hpp

@@ -0,0 +1,118 @@
+#pragma once
+
+#include "database/graph_db.hpp"
+#include "database/graph_db_accessor.hpp"
+#include "query/exceptions.hpp"
+#include "query/interpreter.hpp"
+#include "utils/string.hpp"
+
+namespace query {
+
+class TransactionEngine final {
+ public:
+  TransactionEngine(database::MasterBase &db, Interpreter &interpreter)
+      : db_(db), interpreter_(interpreter) {}
+
+  ~TransactionEngine() { Abort(); }
+
+  template <typename TStream>
+  void PullAll(const std::string &query,
+               const std::map<std::string, TypedValue> &params,
+               TStream *result_stream) {
+    auto query_upper = utils::Trim(utils::ToUpperCase(query));
+    if (query_upper == "BEGIN") {
+      if (in_explicit_transaction_) {
+        throw QueryException("Nested transactions are not supported.");
+      }
+      result_stream->Header({});
+      result_stream->Summary({});
+      in_explicit_transaction_ = true;
+      expect_rollback_ = false;
+      return;
+    } else if (query_upper == "COMMIT") {
+      if (!in_explicit_transaction_) {
+        throw QueryException("No current transaction to commit.");
+      }
+      if (expect_rollback_) {
+        throw QueryException(
+            "Transaction can't be committed because there was a previous "
+            "error. Please invoke a rollback instead.");
+      }
+      Commit();
+      result_stream->Header({});
+      result_stream->Summary({});
+      expect_rollback_ = false;
+      in_explicit_transaction_ = false;
+      return;
+    } else if (query_upper == "ROLLBACK") {
+      if (!in_explicit_transaction_) {
+        throw QueryException("No current transaction to rollback.");
+      }
+      Abort();
+      result_stream->Header({});
+      result_stream->Summary({});
+      expect_rollback_ = false;
+      in_explicit_transaction_ = false;
+      return;
+    }
+
+    // Any other query in an explicit transaction block advances the command.
+    if (in_explicit_transaction_ && db_accessor_) AdvanceCommand();
+
+    // Create a DB accessor if we don't yet have one.
+    if (!db_accessor_)
+      db_accessor_ = std::make_unique<database::GraphDbAccessor>(db_);
+
+    // Interpret the query and stream the results
+    try {
+      interpreter_(query, *db_accessor_, params, in_explicit_transaction_)
+          .PullAll(*result_stream);
+      if (!in_explicit_transaction_) Commit();
+    } catch (const utils::BasicException &) {
+      AbortCommand();
+      throw;
+    }
+  }
+
+  void Abort() {
+    if (!db_accessor_) return;
+    db_accessor_->Abort();
+    db_accessor_ = nullptr;
+  }
+
+ private:
+  database::MasterBase &db_;
+  Interpreter &interpreter_;
+  // GraphDbAccessor of active transaction in the session, can be null if
+  // there is no associated transaction.
+  std::unique_ptr<database::GraphDbAccessor> db_accessor_;
+  bool in_explicit_transaction_{false};
+  bool expect_rollback_{false};
+
+  void Commit() {
+    if (!db_accessor_) return;
+    db_accessor_->Commit();
+    db_accessor_ = nullptr;
+  }
+
+  void AdvanceCommand() {
+    if (!db_accessor_) return;
+    db_accessor_->AdvanceCommand();
+    // TODO: this logic shouldn't be here!
+    if (db_.type() == database::GraphDb::Type::DISTRIBUTED_MASTER) {
+      auto tx_id = db_accessor_->transaction_id();
+      auto futures =
+          db_.pull_clients().NotifyAllTransactionCommandAdvanced(tx_id);
+      for (auto &future : futures) future.wait();
+    }
+  }
+
+  void AbortCommand() {
+    if (in_explicit_transaction_) {
+      expect_rollback_ = true;
+    } else {
+      Abort();
+    }
+  }
+};
+}  // namespace query

tests/integration/CMakeLists.txt

@@ -3,3 +3,6 @@ add_subdirectory(telemetry)
 
 # ssl test binaries
 add_subdirectory(ssl)
+
+# transactions test binaries
+add_subdirectory(transactions)

tests/integration/apollo_runs.yaml

@@ -14,3 +14,11 @@
     - runner.sh # runner script
     - ../../../build_debug/tests/integration/ssl/tester # tester binary
   enable_network: true
+
+- name: integration__transactions
+  cd: transactions
+  commands: ./runner.sh
+  infiles:
+    - runner.sh # runner script
+    - ../../../build_debug/memgraph # memgraph binary
+    - ../../../build_debug/tests/integration/transactions/tester # tester binary

tests/integration/transactions/CMakeLists.txt

@@ -0,0 +1,6 @@
+set(target_name memgraph__integration__transactions)
+set(tester_target_name ${target_name}__tester)
+
+add_executable(${tester_target_name} tester.cpp)
+set_target_properties(${tester_target_name} PROPERTIES OUTPUT_NAME tester)
+target_link_libraries(${tester_target_name} mg-communication gtest)

tests/integration/transactions/runner.sh

@@ -0,0 +1,36 @@
+#!/bin/bash
+
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+cd "$DIR"
+
+binary_dir="$DIR/../../../build"
+if [ ! -d $binary_dir ]; then
+    binary_dir="$DIR/../../../build_debug"
+fi
+
+# Start the memgraph process.
+$binary_dir/memgraph &
+pid=$!
+
+# Wait for the database to start up.
+while ! nc -z -w 1 127.0.0.1 7687; do
+    sleep 0.5
+done
+
+# Start the test.
+$binary_dir/tests/integration/transactions/tester
+code=$?
+
+# Shutdown the memgraph process.
+kill $pid
+wait -n
+code_mg=$?
+
+# Check memgraph exit code.
+if [ $code_mg -ne 0 ]; then
+    echo "The memgraph process didn't terminate properly!"
+    exit $code_mg
+fi
+
+# Exit with the exitcode of the test.
+exit $code

tests/integration/transactions/tester.cpp

@@ -0,0 +1,486 @@
+#include <iostream>
+
+#include <gflags/gflags.h>
+#include <glog/logging.h>
+#include <gtest/gtest.h>
+
+#include "communication/bolt/client.hpp"
+#include "io/network/endpoint.hpp"
+#include "io/network/utils.hpp"
+#include "utils/exceptions.hpp"
+
+DEFINE_string(address, "127.0.0.1", "Server address");
+DEFINE_int32(port, 7687, "Server port");
+DEFINE_string(username, "", "Username for the database");
+DEFINE_string(password, "", "Password for the database");
+DEFINE_bool(use_ssl, false, "Set to true to connect with SSL to the server.");
+
+using namespace communication::bolt;
+
+class BoltClient : public ::testing::Test {
+ protected:
+  virtual void SetUp() {
+    if (!client_.Connect(endpoint_, FLAGS_username, FLAGS_password)) {
+      throw utils::BasicException("Couldn't connect to database!");
+    }
+  }
+
+  virtual void TearDown() {}
+
+  bool Execute(const std::string &query, const std::string &message = "") {
+    try {
+      auto ret = client_.Execute(query, {});
+    } catch (const ClientQueryException &e) {
+      if (message != "") {
+        EXPECT_EQ(e.what(), message);
+      }
+      throw;
+    }
+    return true;
+  }
+
+  int64_t GetCount() {
+    auto ret = client_.Execute("match (n) return count(n)", {});
+    EXPECT_EQ(ret.records.size(), 1);
+    EXPECT_EQ(ret.records[0].size(), 1);
+    EXPECT_TRUE(ret.records[0][0].IsInt());
+    return ret.records[0][0].ValueInt();
+  }
+
+  bool TransactionActive() {
+    try {
+      client_.Execute("begin", {});
+    } catch (const ClientQueryException &e) {
+      return true;
+    }
+    return false;
+  }
+
+  io::network::Endpoint endpoint_{io::network::ResolveHostname(FLAGS_address),
+                                  static_cast<uint16_t>(FLAGS_port)};
+  communication::ClientContext context_{FLAGS_use_ssl};
+  Client client_{&context_};
+};
+
+const std::string kNoCurrentTransactionToCommit =
+    "No current transaction to commit.";
+const std::string kNoCurrentTransactionToRollback =
+    "No current transaction to rollback.";
+const std::string kNestedTransactions =
+    "Nested transactions are not supported.";
+const std::string kCommitInvalid =
+    "Transaction can't be committed because there was a previous error. Please "
+    "invoke a rollback instead.";
+
+TEST_F(BoltClient, CommitWithoutTransaction) {
+  EXPECT_THROW(Execute("commit", kNoCurrentTransactionToCommit),
+               ClientQueryException);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, RollbackWithoutTransaction) {
+  EXPECT_THROW(Execute("rollback", kNoCurrentTransactionToRollback),
+               ClientQueryException);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, DoubleCommitWithoutTransaction) {
+  EXPECT_THROW(Execute("commit", kNoCurrentTransactionToCommit),
+               ClientQueryException);
+  EXPECT_THROW(Execute("commit", kNoCurrentTransactionToCommit),
+               ClientQueryException);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, DoubleRollbackWithoutTransaction) {
+  EXPECT_THROW(Execute("rollback", kNoCurrentTransactionToRollback),
+               ClientQueryException);
+  EXPECT_THROW(Execute("rollback", kNoCurrentTransactionToRollback),
+               ClientQueryException);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, DoubleBegin) {
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, DoubleBeginAndCommit) {
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(Execute("commit"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, DoubleBeginAndRollback) {
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, BeginAndDoubleCommit) {
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_TRUE(Execute("commit"));
+  EXPECT_THROW(Execute("commit", kNoCurrentTransactionToCommit),
+               ClientQueryException);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, BeginAndDoubleRollback) {
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_THROW(Execute("rollback", kNoCurrentTransactionToRollback),
+               ClientQueryException);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, BeginAndCorrectQueriesAndCommit) {
+  EXPECT_TRUE(Execute("begin"));
+  auto count = GetCount();
+  EXPECT_TRUE(Execute("create (n)"));
+  ASSERT_EQ(GetCount(), count + 1);
+  EXPECT_TRUE(Execute("commit"));
+  EXPECT_EQ(GetCount(), count + 1);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, BeginAndCorrectQueriesAndRollback) {
+  EXPECT_TRUE(Execute("begin"));
+  auto count = GetCount();
+  EXPECT_TRUE(Execute("create (n)"));
+  ASSERT_EQ(GetCount(), count + 1);
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_EQ(GetCount(), count);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, BeginAndCorrectQueriesAndBegin) {
+  EXPECT_TRUE(Execute("begin"));
+  auto count = GetCount();
+  EXPECT_TRUE(Execute("create (n)"));
+  ASSERT_EQ(GetCount(), count + 1);
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_EQ(GetCount(), count + 1);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, BeginAndWrongQueryAndRollback) {
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, BeginAndWrongQueryAndCommit) {
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  EXPECT_THROW(Execute("commit", kCommitInvalid), ClientQueryException);
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, BeginAndWrongQueryAndBegin) {
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  EXPECT_THROW(Execute("commit", kCommitInvalid), ClientQueryException);
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, CommitAndCorrectQueryAndCommit) {
+  EXPECT_THROW(Execute("commit", kNoCurrentTransactionToCommit),
+               ClientQueryException);
+  EXPECT_TRUE(Execute("create (n)"));
+  EXPECT_THROW(Execute("commit", kNoCurrentTransactionToCommit),
+               ClientQueryException);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, CommitAndWrongQueryAndCommit) {
+  EXPECT_THROW(Execute("commit", kNoCurrentTransactionToCommit),
+               ClientQueryException);
+  EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  EXPECT_THROW(Execute("commit", kNoCurrentTransactionToCommit),
+               ClientQueryException);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, RollbackAndCorrectQueryAndRollback) {
+  EXPECT_THROW(Execute("rollback", kNoCurrentTransactionToRollback),
+               ClientQueryException);
+  EXPECT_TRUE(Execute("create (n)"));
+  EXPECT_THROW(Execute("rollback", kNoCurrentTransactionToRollback),
+               ClientQueryException);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, RollbackAndWrongQueryAndRollback) {
+  EXPECT_THROW(Execute("rollback", kNoCurrentTransactionToRollback),
+               ClientQueryException);
+  EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  EXPECT_THROW(Execute("rollback", kNoCurrentTransactionToRollback),
+               ClientQueryException);
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueryAndBeginAndCommit) {
+  EXPECT_TRUE(Execute("match (n) return count(n)"));
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_TRUE(Execute("commit"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueryAndBeginAndRollback) {
+  EXPECT_TRUE(Execute("match (n) return count(n)"));
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueryAndBeginAndBegin) {
+  EXPECT_TRUE(Execute("match (n) return count(n)"));
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueryAndBeginAndCommit) {
+  EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_TRUE(Execute("commit"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueryAndBeginAndRollback) {
+  EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueryAndBeginAndBegin) {
+  EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueriesAndBeginAndCommit) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_TRUE(Execute("commit"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueriesAndBeginAndRollback) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueriesAndBeginAndBegin) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueriesAndBeginAndCommit) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_TRUE(Execute("commit"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueriesAndBeginAndRollback) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueriesAndBeginAndBegin) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("begin"));
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueriesAndBeginAndCorrectQueriesAndCommit) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("commit"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueriesAndBeginAndCorrectQueriesAndRollback) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueriesAndBeginAndCorrectQueriesAndBegin) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueriesAndBeginAndCorrectQueriesAndCommit) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("commit"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueriesAndBeginAndCorrectQueriesAndRollback) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueriesAndBeginAndCorrectQueriesAndBegin) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueriesAndBeginAndWrongQueriesAndCommit) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_THROW(Execute("commit", kCommitInvalid), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueriesAndBeginAndWrongQueriesAndRollback) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, CorrectQueriesAndBeginAndWrongQueriesAndBegin) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_TRUE(Execute("match (n) return count(n)"));
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueriesAndBeginAndWrongQueriesAndCommit) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_THROW(Execute("commit", kCommitInvalid), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueriesAndBeginAndWrongQueriesAndRollback) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("rollback"));
+  EXPECT_FALSE(TransactionActive());
+}
+
+TEST_F(BoltClient, WrongQueriesAndBeginAndWrongQueriesAndBegin) {
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_TRUE(Execute("begin"));
+  for (int i = 0; i < 3; ++i) {
+    EXPECT_THROW(Execute("asdasd"), ClientQueryException);
+  }
+  EXPECT_THROW(Execute("begin", kNestedTransactions), ClientQueryException);
+  EXPECT_TRUE(TransactionActive());
+}
+
+TEST_F(BoltClient, MixedCaseAndWhitespace) {
+  EXPECT_TRUE(Execute("   bEgiN   \n\n"));
+  auto count = GetCount();
+  EXPECT_TRUE(Execute("   cReATe     (   n   )   \n\n"));
+  ASSERT_EQ(GetCount(), count + 1);
+  EXPECT_TRUE(Execute("    COMmit  "));
+  EXPECT_EQ(GetCount(), count + 1);
+  EXPECT_FALSE(TransactionActive());
+}
+
+int main(int argc, char **argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  gflags::ParseCommandLineFlags(&argc, &argv, true);
+  google::InitGoogleLogging(argv[0]);
+
+  communication::Init();
+
+  return RUN_ALL_TESTS();
+}

tests/unit/bolt_session.cpp

@@ -25,8 +25,6 @@ class TestSession : public Session<TestInputStream, TestOutputStream> {
       : Session<TestInputStream, TestOutputStream>(input_stream,
                                                    output_stream) {}
 
-  bool IsShuttingDown() override { return false; }
-
   void PullAll(const std::string &query,
                const std::map<std::string, DecodedValue> &params,
                ResultStreamT *result_stream) override {