<commit_before>// Copyright 2018-present MongoDB Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include <iostream>

#include <bsoncxx/builder/basic/document.hpp>
#include <bsoncxx/document/value.hpp>
#include <bsoncxx/private/libbson.hh>
#include <bsoncxx/test_util/catch.hh>
#include <mongocxx/client.hpp>
#include <mongocxx/collection.hpp>
#include <mongocxx/instance.hpp>
#include <mongocxx/options/insert.hpp>
#include <mongocxx/pipeline.hpp>
#include <mongocxx/private/libbson.hh>
#include <mongocxx/test_util/client_helpers.hh>
#include <mongocxx/write_concern.hpp>
#include <third_party/catch/include/helpers.hpp>

namespace {

using bsoncxx::builder::basic::document;
using bsoncxx::builder::basic::kvp;
using bsoncxx::builder::basic::make_array;
using bsoncxx::builder::basic::make_document;

using namespace mongocxx;

// Create a single-item document.
// E.g. doc("foo", 123) creates {"foo":123}.
template <typename T>
bsoncxx::document::value doc(std::string key, T val) {
    bsoncxx::builder::basic::document out{};
    out.append(kvp(key, val));
    return out.extract();
}

// Generates lambda/interpose for change_stream_next
//
// Phrased as a lambda instead of function because c++11 doesn't have decltype(auto) and the
// return-type is haunting.
const auto gen_next = [](bool has_next) {
    static mongocxx::libbson::scoped_bson_t next_bson{make_document(kvp("some", "doc"))};
    return [=](mongoc_change_stream_t*, const bson_t** bson) mutable -> bool {
        if (has_next) {
            *bson = next_bson.bson();
        }
        return has_next;
    };
};

bson_t err_doc;

// Generates lambda/interpose for change_stream_error_document.
const auto gen_error = [](bool has_error) {
    bson_init(&err_doc);  // Will fit on stack.
    bson_append_int32(&err_doc, "ok", -1, 0);
    return [=](const mongoc_change_stream_t*, bson_error_t* err, const bson_t** bson) -> bool {
        if (has_error) {
            bson_set_error(err,
                           MONGOC_ERROR_CURSOR,
                           MONGOC_ERROR_CHANGE_STREAM_NO_RESUME_TOKEN,
                           "expected error");
            *bson = &err_doc;
        } else {
            *bson = NULL;
        }
        return has_error;
    };
};

const auto watch_interpose = [](const mongoc_collection_t*,
                                const bson_t*,
                                const bson_t*) -> mongoc_change_stream_t* { return nullptr; };

const auto destroy_interpose = [](mongoc_change_stream_t*) -> void {};

TEST_CASE("Change stream options") {
    instance::current();
    client mongodb_client{uri{}, test_util::add_test_server_api()};

    if (!test_util::is_replica_set(mongodb_client)) {
        WARN("skip: change streams require replica set");
        return;
    }

    SECTION("Error if both resumeAfter and startAfter are set") {
        bsoncxx::document::value resume_after = make_document(kvp("resume", "token"));
        bsoncxx::document::value start_after = make_document(kvp("start", "token"));

        // Invalid to set both resumeAfter and startAfter.
        options::change_stream cs_opts;
        cs_opts.resume_after(resume_after.view());
        cs_opts.start_after(start_after.view());

        auto cs = mongodb_client.watch(cs_opts);
        REQUIRE_THROWS(cs.begin());
    }
}

TEST_CASE("Spec Prose Tests") {
    instance::current();
    client client{uri{}, test_util::add_test_server_api()};

    if (!test_util::is_replica_set(client)) {
        WARN("skip: change streams require replica set");
        return;
    }

    auto db = client["db"];
    auto coll = db["coll"];
    coll.drop();

    write_concern wc_majority;
    wc_majority.majority(std::chrono::milliseconds(0));

    // As a sanity check, we implement the first prose test. The behavior tested
    // by the prose tests is implemented and tested by the C driver, so we won't
    // replicate the full suite of prose tests here.

    SECTION("1. ChangeStream must continuously track the last seen resumeToken") {
        // Set the batch size to 1 so we read 1 doc at a time.
        options::change_stream opts;
        opts.batch_size(1);
        auto cs = client.watch(std::move(opts));

        // With WC majority, insert some documents to listen for.
        auto doc1 = make_document(kvp("_id", 1));
        auto doc2 = make_document(kvp("_id", 2));
        auto doc3 = make_document(kvp("_id", 3));

        options::insert insert_opts{};
        insert_opts.write_concern(wc_majority);

        coll.insert_one(doc1.view(), insert_opts);
        coll.insert_one(doc2.view(), insert_opts);
        coll.insert_one(doc3.view(), insert_opts);

        // For each read, check the resume token is updated. We should
        // be reading the postBatchResumeToken on each read, since our
        // batch size is 1.
        auto it = cs.begin();
        REQUIRE(cs.get_resume_token());
        auto token1 = bsoncxx::document::value(*cs.get_resume_token());

        it++;
        REQUIRE(cs.get_resume_token());
        auto token2 = bsoncxx::document::value(*cs.get_resume_token());
        REQUIRE(token1 != token2);

        it++;
        REQUIRE(cs.get_resume_token());
        auto token3 = bsoncxx::document::value(*cs.get_resume_token());
        REQUIRE(token2 != token3);
        REQUIRE(token1 != token3);

        // When out of docs, check that the resume token is the same as the last doc.
        it++;
        REQUIRE(it == cs.end());
        REQUIRE(*cs.get_resume_token() == token3);
    }
}

TEST_CASE("Mock streams and error-handling") {
    MOCK_CHANGE_STREAM

    instance::current();
    client mongodb_client{uri{}, test_util::add_test_server_api()};
    options::change_stream options{};
    database db = mongodb_client["streams"];
    collection events = db["events"];
<commit_msg>CXX-2337 Raise wtimeout of resumeToken prose test to 30 seconds (#886)<commit_after>// Copyright 2018-present MongoDB Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include <iostream>

#include <bsoncxx/builder/basic/document.hpp>
#include <bsoncxx/document/value.hpp>
#include <bsoncxx/private/libbson.hh>
#include <bsoncxx/test_util/catch.hh>
#include <mongocxx/client.hpp>
#include <mongocxx/collection.hpp>
#include <mongocxx/instance.hpp>
#include <mongocxx/options/insert.hpp>
#include <mongocxx/pipeline.hpp>
#include <mongocxx/private/libbson.hh>
#include <mongocxx/test_util/client_helpers.hh>
#include <mongocxx/write_concern.hpp>
#include <third_party/catch/include/helpers.hpp>

namespace {

using bsoncxx::builder::basic::document;
using bsoncxx::builder::basic::kvp;
using bsoncxx::builder::basic::make_array;
using bsoncxx::builder::basic::make_document;

using namespace mongocxx;

// Create a single-item document.
// E.g. doc("foo", 123) creates {"foo":123}.
template <typename T>
bsoncxx::document::value doc(std::string key, T val) {
    bsoncxx::builder::basic::document out{};
    out.append(kvp(key, val));
    return out.extract();
}

// Generates lambda/interpose for change_stream_next
//
// Phrased as a lambda instead of function because c++11 doesn't have decltype(auto) and the
// return-type is haunting.
const auto gen_next = [](bool has_next) {
    static mongocxx::libbson::scoped_bson_t next_bson{make_document(kvp("some", "doc"))};
    return [=](mongoc_change_stream_t*, const bson_t** bson) mutable -> bool {
        if (has_next) {
            *bson = next_bson.bson();
        }
        return has_next;
    };
};

bson_t err_doc;

// Generates lambda/interpose for change_stream_error_document.
const auto gen_error = [](bool has_error) {
    bson_init(&err_doc);  // Will fit on stack.
    bson_append_int32(&err_doc, "ok", -1, 0);
    return [=](const mongoc_change_stream_t*, bson_error_t* err, const bson_t** bson) -> bool {
        if (has_error) {
            bson_set_error(err,
                           MONGOC_ERROR_CURSOR,
                           MONGOC_ERROR_CHANGE_STREAM_NO_RESUME_TOKEN,
                           "expected error");
            *bson = &err_doc;
        } else {
            *bson = NULL;
        }
        return has_error;
    };
};

const auto watch_interpose = [](const mongoc_collection_t*,
                                const bson_t*,
                                const bson_t*) -> mongoc_change_stream_t* { return nullptr; };

const auto destroy_interpose = [](mongoc_change_stream_t*) -> void {};

TEST_CASE("Change stream options") {
    instance::current();
    client mongodb_client{uri{}, test_util::add_test_server_api()};

    if (!test_util::is_replica_set(mongodb_client)) {
        WARN("skip: change streams require replica set");
        return;
    }

    SECTION("Error if both resumeAfter and startAfter are set") {
        bsoncxx::document::value resume_after = make_document(kvp("resume", "token"));
        bsoncxx::document::value start_after = make_document(kvp("start", "token"));

        // Invalid to set both resumeAfter and startAfter.
        options::change_stream cs_opts;
        cs_opts.resume_after(resume_after.view());
        cs_opts.start_after(start_after.view());

        auto cs = mongodb_client.watch(cs_opts);
        REQUIRE_THROWS(cs.begin());
    }
}

TEST_CASE("Spec Prose Tests") {
    instance::current();
    client client{uri{}, test_util::add_test_server_api()};

    if (!test_util::is_replica_set(client)) {
        WARN("skip: change streams require replica set");
        return;
    }

    auto db = client["db"];
    auto coll = db["coll"];
    coll.drop();

    write_concern wc_majority;
    wc_majority.majority(std::chrono::seconds(30));

    // As a sanity check, we implement the first prose test. The behavior tested
    // by the prose tests is implemented and tested by the C driver, so we won't
    // replicate the full suite of prose tests here.

    SECTION("1. ChangeStream must continuously track the last seen resumeToken") {
        // Set the batch size to 1 so we read 1 doc at a time.
        options::change_stream opts;
        opts.batch_size(1);
        auto cs = client.watch(std::move(opts));

        // With WC majority, insert some documents to listen for.
        auto doc1 = make_document(kvp("a", 1));
        auto doc2 = make_document(kvp("b", 2));
        auto doc3 = make_document(kvp("c", 3));

        options::insert insert_opts{};
        insert_opts.write_concern(wc_majority);

        {
            auto res = coll.insert_one(doc1.view(), insert_opts);
            REQUIRE(res);
            REQUIRE(res->result().inserted_count() == 1);
        }
        {
            auto res = coll.insert_one(doc2.view(), insert_opts);
            REQUIRE(res);
            REQUIRE(res->result().inserted_count() == 1);
        }
        {
            auto res = coll.insert_one(doc3.view(), insert_opts);
            REQUIRE(res);
            REQUIRE(res->result().inserted_count() == 1);
        }

        // For each read, check the resume token is updated. We should
        // be reading the postBatchResumeToken on each read, since our
        // batch size is 1.
        auto it = cs.begin();
        REQUIRE(it != cs.end());
        REQUIRE(cs.get_resume_token());
        auto token1 = bsoncxx::document::value(*cs.get_resume_token());

        it++;
        REQUIRE(it != cs.end());
        REQUIRE(cs.get_resume_token());
        auto token2 = bsoncxx::document::value(*cs.get_resume_token());
        REQUIRE(token1 != token2);

        it++;
        REQUIRE(it != cs.end());
        REQUIRE(cs.get_resume_token());
        auto token3 = bsoncxx::document::value(*cs.get_resume_token());
        REQUIRE(token2 != token3);
        REQUIRE(token1 != token3);

        // When out of docs, check that the resume token is the same as the last doc.
        it++;
        REQUIRE(it == cs.end());
        REQUIRE(cs.get_resume_token());
        REQUIRE(*cs.get_resume_token() == token3);
    }
}

TEST_CASE("Mock streams and error-handling") {
    MOCK_CHANGE_STREAM

    instance::current();
    client mongodb_client{uri{}, test_util::add_test_server_api()};
    options::change_stream options{};
    database db = mongodb_client["streams"];
    collection events = db["events"];

{'fraction_non_alphanumeric': 0.12727960650775633, 'fraction_numerical': 0.007794173287930382, 'mean_word_length': 3.0012110202845896, 'pattern_counts': {'":': 2, '<': 33, '<?xml version=': 0, '>': 44, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5433483', 'n_tokens_mistral': 4247, 'n_tokens_neox': 4119, 'n_words': 1036}