path: root/src/mongo/db/repl/base_cloner_test_fixture.cpp

/**
 *    Copyright 2015 MongoDB Inc.
 *
 *    This program is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the GNU Affero General Public License in all respects for
 *    all of the code used other than as permitted herein. If you modify file(s)
 *    with this exception, you may extend this exception to your version of the
 *    file(s), but you are not obligated to do so. If you do not wish to do so,
 *    delete this exception statement from your version. If you delete this
 *    exception statement from all source files in the program, then also delete
 *    it in the license file.
 */

#include "mongo/platform/basic.h"

#include "mongo/db/repl/base_cloner_test_fixture.h"

#include <memory>

#include "mongo/db/jsobj.h"
#include "mongo/stdx/thread.h"
#include "mongo/util/mongoutils/str.h"

namespace mongo {
namespace repl {
using executor::RemoteCommandRequest;
using executor::RemoteCommandResponse;
using namespace unittest;

const HostAndPort BaseClonerTest::target("localhost", -1);
const NamespaceString BaseClonerTest::nss("db.coll");
const BSONObj BaseClonerTest::idIndexSpec = BSON("v" << 1 << "key" << BSON("_id" << 1) << "name"
                                                     << "_id_"
                                                     << "ns"
                                                     << nss.ns());

// static
BSONObj BaseClonerTest::createCountResponse(int documentCount) {
    return BSON("n" << documentCount << "ok" << 1);
}

// static
BSONObj BaseClonerTest::createCursorResponse(CursorId cursorId,
                                             const std::string& ns,
                                             const BSONArray& docs,
                                             const char* batchFieldName) {
    return BSON("cursor" << BSON("id" << cursorId << "ns" << ns << batchFieldName << docs) << "ok"
                         << 1);
}

// static
BSONObj BaseClonerTest::createCursorResponse(CursorId cursorId,
                                             const BSONArray& docs,
                                             const char* batchFieldName) {
    return createCursorResponse(cursorId, nss.toString(), docs, batchFieldName);
}

// static
BSONObj BaseClonerTest::createCursorResponse(CursorId cursorId, const BSONArray& docs) {
    return createCursorResponse(cursorId, docs, "firstBatch");
}

// static
BSONObj BaseClonerTest::createFinalCursorResponse(const BSONArray& docs) {
    return createCursorResponse(0, docs, "nextBatch");
}

// static
BSONObj BaseClonerTest::createListCollectionsResponse(CursorId cursorId,
                                                      const BSONArray& colls,
                                                      const char* fieldName) {
    return createCursorResponse(cursorId, "test.$cmd.listCollections.coll", colls, fieldName);
}

// static
BSONObj BaseClonerTest::createListCollectionsResponse(CursorId cursorId, const BSONArray& colls) {
    return createListCollectionsResponse(cursorId, colls, "firstBatch");
}

// static
BSONObj BaseClonerTest::createListIndexesResponse(CursorId cursorId,
                                                  const BSONArray& specs,
                                                  const char* batchFieldName) {
    return createCursorResponse(cursorId, "test.$cmd.listIndexes.coll", specs, batchFieldName);
}

// static
BSONObj BaseClonerTest::createListIndexesResponse(CursorId cursorId, const BSONArray& specs) {
    return createListIndexesResponse(cursorId, specs, "firstBatch");
}

namespace {
struct EnsureClientHasBeenInitialized : public executor::ThreadPoolMock::Options {
    EnsureClientHasBeenInitialized() : executor::ThreadPoolMock::Options() {
        onCreateThread = []() { Client::initThread("CollectionClonerTestThread"); };
    }
};
}  // namespace

BaseClonerTest::BaseClonerTest()
    : ThreadPoolExecutorTest(EnsureClientHasBeenInitialized()),
      _mutex(),
      _setStatusCondition(),
      _status(getDetectableErrorStatus()) {}

void BaseClonerTest::setUp() {
    executor::ThreadPoolExecutorTest::setUp();
    clear();
    launchExecutorThread();

    Client::initThread("CollectionClonerTest");
    ThreadPool::Options options;
    options.minThreads = 1U;
    options.maxThreads = 1U;
    options.onCreateThread = [](StringData threadName) { Client::initThread(threadName); };
    dbWorkThreadPool = stdx::make_unique<ThreadPool>(options);
    dbWorkThreadPool->startup();

    storageInterface.reset(new StorageInterfaceMock());
}

void BaseClonerTest::tearDown() {
    getExecutor().shutdown();
    getExecutor().join();

    storageInterface.reset();

    dbWorkThreadPool.reset();
    Client::releaseCurrent();
}

void BaseClonerTest::clear() {
    _status = getDetectableErrorStatus();
}

void BaseClonerTest::setStatus(const Status& status) {
    stdx::unique_lock<stdx::mutex> lk(_mutex);
    _status = status;
    _setStatusCondition.notify_all();
}

const Status& BaseClonerTest::getStatus() const {
    stdx::unique_lock<stdx::mutex> lk(_mutex);
    return _status;
}

void BaseClonerTest::scheduleNetworkResponse(NetworkOperationIterator noi, const BSONObj& obj) {
    auto net = getNet();
    Milliseconds millis(0);
    RemoteCommandResponse response(obj, BSONObj(), millis);
    log() << "Scheduling response to request:" << noi->getDiagnosticString() << " -- resp:" << obj;
    net->scheduleResponse(noi, net->now(), response);
}

void BaseClonerTest::scheduleNetworkResponse(NetworkOperationIterator noi,
                                             ErrorCodes::Error code,
                                             const std::string& reason) {
    auto net = getNet();
    RemoteCommandResponse responseStatus(code, reason);
    log() << "Scheduling error response to request:" << noi->getDiagnosticString()
          << " -- status:" << responseStatus.status.toString();
    net->scheduleResponse(noi, net->now(), responseStatus);
}

void BaseClonerTest::scheduleNetworkResponse(const BSONObj& obj) {
    if (!getNet()->hasReadyRequests()) {
        BSONObjBuilder b;
        getExecutor().appendDiagnosticBSON(&b);
        log() << "Expected network request for resp: " << obj;
        log() << "      replExec: " << b.done();
        log() << "      net:" << getNet()->getDiagnosticString();
    }
    if (getStatus() != getDetectableErrorStatus()) {
        log() << "Status has changed during network response playback to: " << getStatus();
        return;
    }
    ASSERT_TRUE(getNet()->hasReadyRequests());
    scheduleNetworkResponse(getNet()->getNextReadyRequest(), obj);
}

void BaseClonerTest::scheduleNetworkResponse(ErrorCodes::Error code, const std::string& reason) {
    ASSERT_TRUE(getNet()->hasReadyRequests());
    scheduleNetworkResponse(getNet()->getNextReadyRequest(), code, reason);
}

void BaseClonerTest::processNetworkResponse(const BSONObj& obj) {
    scheduleNetworkResponse(obj);
    finishProcessingNetworkResponse();
}

void BaseClonerTest::processNetworkResponse(ErrorCodes::Error code, const std::string& reason) {
    scheduleNetworkResponse(code, reason);
    finishProcessingNetworkResponse();
}

void BaseClonerTest::finishProcessingNetworkResponse() {
    clear();
    getNet()->runReadyNetworkOperations();
}

void BaseClonerTest::testLifeCycle() {
    // IsActiveAfterStart
    ASSERT_FALSE(getCloner()->isActive());
    ASSERT_OK(getCloner()->startup());
    ASSERT_TRUE(getCloner()->isActive());
    tearDown();

    // StartWhenActive
    setUp();
    ASSERT_OK(getCloner()->startup());
    ASSERT_TRUE(getCloner()->isActive());
    ASSERT_NOT_OK(getCloner()->startup());
    ASSERT_TRUE(getCloner()->isActive());
    tearDown();

    // CancelWithoutStart
    setUp();
    ASSERT_FALSE(getCloner()->isActive());
    getCloner()->shutdown();
    ASSERT_FALSE(getCloner()->isActive());
    tearDown();

    // WaitWithoutStart
    setUp();
    ASSERT_FALSE(getCloner()->isActive());
    getCloner()->join();
    ASSERT_FALSE(getCloner()->isActive());
    tearDown();

    // ShutdownBeforeStart
    setUp();
    getExecutor().shutdown();
    ASSERT_NOT_OK(getCloner()->startup());
    ASSERT_FALSE(getCloner()->isActive());
    tearDown();

    // StartAndCancel
    setUp();
    ASSERT_OK(getCloner()->startup());
    getCloner()->shutdown();
    {
        executor::NetworkInterfaceMock::InNetworkGuard guard(getNet());
        finishProcessingNetworkResponse();
    }
    ASSERT_EQUALS(ErrorCodes::CallbackCanceled, getStatus().code());
    ASSERT_FALSE(getCloner()->isActive());
    tearDown();

    // StartButShutdown
    setUp();
    ASSERT_OK(getCloner()->startup());
    {
        executor::NetworkInterfaceMock::InNetworkGuard guard(getNet());
        scheduleNetworkResponse(BSON("ok" << 1));
        // Network interface should not deliver mock response to callback.
        getExecutor().shutdown();
        finishProcessingNetworkResponse();
    }
    ASSERT_EQUALS(ErrorCodes::CallbackCanceled, getStatus().code());
    ASSERT_FALSE(getCloner()->isActive());
}

}  // namespace repl
}  // namespace mongo