jorgen.edelbo_402

Committed 21 Aug 2024 11:10AM UTC coverage: 91.054% (-0.03%) from 91.085%

Build # jorgen.edelbo_402

Build Type

Pull #7803

Evergreen

Committed by

jedelbo

Commit Message

Small fix to Table::typed_write

When writing the realm to a new file from a write transaction,
the Table may be COW so that the top ref is changed. So don't
use the ref that is present in the group when the operation starts.

Pull Request Pull Request #7803: Feature/string compression

Run Details

103494 of 181580 branches covered (57.0%)

1929 of 1999 new or added lines in 46 files covered. (96.5%)

695 existing lines in 51 files now uncovered.

220142 of 241772 relevant lines covered (91.05%)

7344461.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.0

/test/object-store/sync/app.cpp

////////////////////////////////////////////////////////////////////////////
//
// Copyright 2016 Realm 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 "collection_fixtures.hpp"
#include "util/sync/baas_admin_api.hpp"
#include "util/sync/sync_test_utils.hpp"
#include "util/test_path.hpp"
#include "util/unit_test_transport.hpp"
#include "util/test_path.hpp"

#include <realm/object-store/impl/object_accessor_impl.hpp>
#include <realm/object-store/sync/app_credentials.hpp>
#include <realm/object-store/sync/app_utils.hpp>
#include <realm/object-store/sync/async_open_task.hpp>
#include <realm/object-store/sync/generic_network_transport.hpp>
#include <realm/object-store/sync/mongo_client.hpp>
#include <realm/object-store/sync/mongo_collection.hpp>
#include <realm/object-store/sync/mongo_database.hpp>
#include <realm/object-store/sync/sync_session.hpp>
#include <realm/object-store/sync/sync_user.hpp>
#include <realm/object-store/thread_safe_reference.hpp>
#include <realm/object-store/util/uuid.hpp>
#include <realm/sync/network/default_socket.hpp>
#include <realm/sync/network/websocket.hpp>
#include <realm/sync/noinst/server/access_token.hpp>
#include <realm/util/base64.hpp>
#include <realm/util/overload.hpp>
#include <realm/util/platform_info.hpp>
#include <realm/util/future.hpp>
#include <realm/util/uri.hpp>

#include <catch2/catch_all.hpp>
#include <external/json/json.hpp>
#include <external/mpark/variant.hpp>

#include <condition_variable>
#include <iostream>
#include <list>
#include <mutex>

using namespace realm;
using namespace realm::app;
using util::any_cast;
using util::Optional;

using namespace std::string_view_literals;
using namespace std::literals::string_literals;
using namespace std::chrono_literals;
using namespace Catch::Matchers;


namespace {
std::shared_ptr<User> log_in(std::shared_ptr<App> app, AppCredentials credentials = AppCredentials::anonymous())
{
    if (auto transport = dynamic_cast<UnitTestTransport*>(app->config().transport.get())) {
        transport->set_provider_type(credentials.provider_as_string());
    }
    std::shared_ptr<User> user;
    app->log_in_with_credentials(credentials, [&](std::shared_ptr<User> user_arg, Optional<AppError> error) {
        REQUIRE_FALSE(error);
        REQUIRE(user_arg);
        user = std::move(user_arg);
    });
    REQUIRE(user);
    return user;
}

AppError failed_log_in(std::shared_ptr<App> app, AppCredentials credentials = AppCredentials::anonymous())
{
    Optional<AppError> err;
    app->log_in_with_credentials(credentials, [&](std::shared_ptr<User> user, Optional<AppError> error) {
        REQUIRE(error);
        REQUIRE_FALSE(user);
        err = error;
    });
    REQUIRE(err);
    return *err;
}

} // namespace

namespace realm {
class TestHelper {
public:
    static DBRef get_db(Realm& realm)
    {
        return Realm::Internal::get_db(realm);
    }
};
} // namespace realm

static const std::string profile_0_name = "Ursus americanus Ursus boeckhi";
static const std::string profile_0_first_name = "Ursus americanus";
static const std::string profile_0_last_name = "Ursus boeckhi";
static const std::string profile_0_email = "Ursus ursinus";
static const std::string profile_0_picture_url = "Ursus malayanus";
static const std::string profile_0_gender = "Ursus thibetanus";
static const std::string profile_0_birthday = "Ursus americanus";
static const std::string profile_0_min_age = "Ursus maritimus";
static const std::string profile_0_max_age = "Ursus arctos";

static const nlohmann::json profile_0 = {
    {"name", profile_0_name},         {"first_name", profile_0_first_name},   {"last_name", profile_0_last_name},
    {"email", profile_0_email},       {"picture_url", profile_0_picture_url}, {"gender", profile_0_gender},
    {"birthday", profile_0_birthday}, {"min_age", profile_0_min_age},         {"max_age", profile_0_max_age}};

static nlohmann::json user_json(std::string access_token, std::string user_id = random_string(15))
{
    return {{"access_token", access_token},
            {"refresh_token", access_token},
            {"user_id", user_id},
            {"device_id", "Panda Bear"}};
}

static nlohmann::json user_profile_json(std::string user_id = random_string(15),
                                        std::string identity_0_id = "Ursus arctos isabellinus",
                                        std::string identity_1_id = "Ursus arctos horribilis",
                                        std::string provider_type = "anon-user")
{
    return {{"user_id", user_id},
            {"identities",
             {{{"id", identity_0_id}, {"provider_type", provider_type}},
              {{"id", identity_1_id}, {"provider_type", "lol_wut"}}}},
            {"data", profile_0}};
}

static const std::string good_access_token =
    "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
    "eyJleHAiOjE1ODE1MDc3OTYsImlhdCI6MTU4MTUwNTk5NiwiaXNzIjoiNWU0M2RkY2M2MzZlZTEwNmVhYTEyYmRjIiwic3RpdGNoX2RldklkIjoi"
    "MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwIiwic3RpdGNoX2RvbWFpbklkIjoiNWUxNDk5MTNjOTBiNGFmMGViZTkzNTI3Iiwic3ViIjoiNWU0M2Rk"
    "Y2M2MzZlZTEwNmVhYTEyYmRhIiwidHlwIjoiYWNjZXNzIn0.0q3y9KpFxEnbmRwahvjWU1v9y1T1s3r2eozu93vMc3s";

static const std::string good_access_token2 =
    "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
    "eyJleHAiOjE1ODkzMDE3MjAsImlhdCI6MTU4NDExODcyMCwiaXNzIjoiNWU2YmJiYzBhNmI3ZGZkM2UyNTA0OGI3Iiwic3RpdGNoX2RldklkIjoi"
    "MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwIiwic3RpdGNoX2RvbWFpbklkIjoiNWUxNDk5MTNjOTBiNGFmMGViZTkzNTI3Iiwic3ViIjoiNWU2YmJi"
    "YzBhNmI3ZGZkM2UyNTA0OGIzIiwidHlwIjoiYWNjZXNzIn0.eSX4QMjIOLbdOYOPzQrD_racwLUk1HGFgxtx2a34k80";

#if REALM_ENABLE_AUTH_TESTS

#include <realm/util/sha_crypto.hpp>

static std::string create_jwt(const std::string& appId)
{
    nlohmann::json header = {{"alg", "HS256"}, {"typ", "JWT"}};
    nlohmann::json payload = {{"aud", appId}, {"sub", "someUserId"}, {"exp", 1961896476}};

    payload["user_data"]["name"] = "Foo Bar";
    payload["user_data"]["occupation"] = "firefighter";

    payload["my_metadata"]["name"] = "Bar Foo";
    payload["my_metadata"]["occupation"] = "stock analyst";

    std::string header_str = header.dump();
    std::string payload_str = payload.dump();

    std::string encoded_header;
    encoded_header.resize(util::base64_encoded_size(header_str.length()));
    util::base64_encode(header_str, encoded_header);

    std::string encoded_payload;
    encoded_payload.resize(util::base64_encoded_size(payload_str.length()));
    util::base64_encode(payload_str, encoded_payload);

    // Remove padding characters.
    while (encoded_header.back() == '=')
        encoded_header.pop_back();
    while (encoded_payload.back() == '=')
        encoded_payload.pop_back();

    std::string jwtPayload = encoded_header + "." + encoded_payload;

    std::array<char, 32> hmac;
    unsigned char key[] = "My_very_confidential_secretttttt";
    util::hmac_sha256(util::unsafe_span_cast<uint8_t>(jwtPayload), util::unsafe_span_cast<uint8_t>(hmac),
                      util::Span<uint8_t, 32>(key, 32));

    std::string signature;
    signature.resize(util::base64_encoded_size(hmac.size()));
    util::base64_encode(hmac, signature);
    while (signature.back() == '=')
        signature.pop_back();
    std::replace(signature.begin(), signature.end(), '+', '-');
    std::replace(signature.begin(), signature.end(), '/', '_');

    return jwtPayload + "." + signature;
}

// MARK: - Verify AppError with all error codes
TEST_CASE("app: verify app error codes", "[sync][app][local]") {
    auto error_codes = ErrorCodes::get_error_list();
    std::vector<std::pair<int, std::string>> http_status_codes = {
        {0, ""},
        {100, "http error code considered fatal: some http error. Informational: 100"},
        {200, ""},
        {300, "http error code considered fatal: some http error. Redirection: 300"},
        {400, "http error code considered fatal: some http error. Client Error: 400"},
        {500, "http error code considered fatal: some http error. Server Error: 500"},
        {600, "http error code considered fatal: some http error. Unknown HTTP Error: 600"}};

    auto make_http_error = [](std::optional<std::string_view> error_code, int http_status = 500,
                              std::optional<std::string_view> error = "some error",
                              std::optional<std::string_view> link = "http://dummy-link/") -> app::Response {
        nlohmann::json body;
        if (error_code) {
            body["error_code"] = *error_code;
        }
        if (error) {
            body["error"] = *error;
        }
        if (link) {
            body["link"] = *link;
        }

        return {
            http_status,
            0,
            {{"Content-Type", "application/json"}},
            body.empty() ? "{}" : body.dump(),
        };
    };

    auto validate_json_body = [](std::string body, std::optional<std::string_view> error_code,
                                 std::optional<std::string_view> error = "some error",
                                 std::optional<std::string_view> logs_link = "http://dummy-link/") -> bool {
        if (body.empty()) {
            return false;
        }
        try {
            auto json_body = nlohmann::json::parse(body);
            // If provided, check the error_code value against the 'error_code' value in the json body
            auto code = json_body.find("error_code");
            if (error_code && !error_code->empty()) {
                if (code == json_body.end() || code->get<std::string>() != *error_code) {
                    return false;
                }
            }
            // If not provided, it's an error if the value is included in the json body
            else if (code != json_body.end()) {
                return false;
            }
            // If provided, check the message value against the 'error' value in the json body
            auto message = json_body.find("error");
            if (error && !error->empty()) {
                if (message == json_body.end() || message->get<std::string>() != *error) {
                    return false;
                }
            }
            // If not provided, it's an error if the value is included in the json body
            else if (message != json_body.end()) {
                return false;
            }
            // If provided, check the logs_link value against the 'link' value in the json body
            auto link = json_body.find("link");
            if (logs_link && !logs_link->empty()) {
                if (link == json_body.end() || link->get<std::string>() != *logs_link) {
                    return false;
                }
            }
            // If not provided, it's an error if the value is included in the json body
            else if (link != json_body.end()) {
                return false;
            }
        }
        catch (const nlohmann::json::exception&) {
            // It's also a failure if parsing the json body throws an exception
            return false;
        }
        return true;
    };

    // Success responses
    app::Response response = {200, 0, {}, ""};
    auto app_error = AppUtils::check_for_errors(response);
    REQUIRE(!app_error);

    response = {0, 0, {}, ""};
    app_error = AppUtils::check_for_errors(response);
    REQUIRE(!app_error);

    // Empty error code
    response = make_http_error("");
    app_error = AppUtils::check_for_errors(response);
    REQUIRE(app_error);
    REQUIRE(app_error->code() == ErrorCodes::AppUnknownError);
    REQUIRE(app_error->code_string() == "AppUnknownError");
    REQUIRE(app_error->server_error.empty());
    REQUIRE(app_error->reason() == "some error");
    REQUIRE(app_error->link_to_server_logs == "http://dummy-link/");
    REQUIRE(*app_error->additional_status_code == 500);

    // Re-compose back into a Response
    auto err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 500);
    REQUIRE(!err_response.body.empty());
    REQUIRE(validate_json_body(err_response.body, ""));
    REQUIRE(!err_response.client_error_code);
    REQUIRE(err_response.custom_status_code == 0);
    auto ct = AppUtils::find_header("content-type", err_response.headers);
    REQUIRE(ct);
    REQUIRE(ct->second == "application/json");

    // Missing error code
    response = make_http_error(std::nullopt);
    app_error = AppUtils::check_for_errors(response);
    REQUIRE(app_error);
    REQUIRE(app_error->code() == ErrorCodes::AppUnknownError);
    REQUIRE(app_error->code_string() == "AppUnknownError");
    REQUIRE(app_error->server_error.empty());
    REQUIRE(app_error->reason() == "some error");
    REQUIRE(app_error->link_to_server_logs == "http://dummy-link/");
    REQUIRE(*app_error->additional_status_code == 500);

    // Re-compose back into a Response
    err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 500);
    REQUIRE(!err_response.body.empty());
    REQUIRE(validate_json_body(err_response.body, std::nullopt));
    REQUIRE(!err_response.client_error_code);
    REQUIRE(err_response.custom_status_code == 0);
    ct = AppUtils::find_header("content-type", err_response.headers);
    REQUIRE(ct);
    REQUIRE(ct->second == "application/json");

    // Missing error message
    response = make_http_error("InvalidParameter", 404, std::nullopt);
    app_error = AppUtils::check_for_errors(response);
    REQUIRE(app_error);
    REQUIRE(app_error->code() == ErrorCodes::InvalidParameter);
    REQUIRE(app_error->code_string() == "InvalidParameter");
    REQUIRE(app_error->server_error == "InvalidParameter");
    REQUIRE(app_error->reason() == "no error message");
    REQUIRE(app_error->link_to_server_logs == "http://dummy-link/");
    REQUIRE(*app_error->additional_status_code == 404);

    // Re-compose back into a Response
    err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 404);
    REQUIRE(!err_response.body.empty());
    REQUIRE(validate_json_body(err_response.body, "InvalidParameter", "no error message"));
    REQUIRE(!err_response.client_error_code);
    REQUIRE(err_response.custom_status_code == 0);
    ct = AppUtils::find_header("content-type", err_response.headers);
    REQUIRE(ct);
    REQUIRE(ct->second == "application/json");

    // Missing logs link
    response = make_http_error("InvalidParameter", 403, "some error occurred", std::nullopt);
    app_error = AppUtils::check_for_errors(response);
    REQUIRE(app_error);
    REQUIRE(app_error->code() == ErrorCodes::InvalidParameter);
    REQUIRE(app_error->code_string() == "InvalidParameter");
    REQUIRE(app_error->server_error == "InvalidParameter");
    REQUIRE(app_error->reason() == "some error occurred");
    REQUIRE(app_error->link_to_server_logs == "");
    REQUIRE(*app_error->additional_status_code == 403);

    // Re-compose back into a Response
    err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 403);
    REQUIRE(!err_response.body.empty());
    REQUIRE(validate_json_body(err_response.body, "InvalidParameter", "some error occurred", std::nullopt));
    REQUIRE(!err_response.client_error_code);
    REQUIRE(err_response.custom_status_code == 0);
    ct = AppUtils::find_header("content-type", err_response.headers);
    REQUIRE(ct);
    REQUIRE(ct->second == "application/json");

    // Missing error code and error message with success http status
    response = make_http_error(std::nullopt, 200, std::nullopt);
    app_error = AppUtils::check_for_errors(response);
    REQUIRE(!app_error);

    for (auto [name, error] : error_codes) {
        // All error codes should not cause an exception
        if (error != ErrorCodes::HTTPError && error != ErrorCodes::OK) {
            response = make_http_error(name);
            app_error = AppUtils::check_for_errors(response);
            REQUIRE(app_error);
            if (ErrorCodes::error_categories(error).test(ErrorCategory::app_error)) {
                REQUIRE(app_error->code() == error);
                REQUIRE(app_error->code_string() == name);
            }
            else {
                REQUIRE(app_error->code() == ErrorCodes::AppServerError);
                REQUIRE(app_error->code_string() == "AppServerError");
            }
            REQUIRE(app_error->server_error == name);
            REQUIRE(app_error->reason() == "some error");
            REQUIRE(app_error->link_to_server_logs == "http://dummy-link/");
            REQUIRE(app_error->additional_status_code);
            REQUIRE(*app_error->additional_status_code == 500);

            // Re-compose back into a Response
            err_response = AppUtils::make_apperror_response(*app_error);
            REQUIRE(err_response.http_status_code == 500);
            REQUIRE(!err_response.body.empty());
            REQUIRE(validate_json_body(err_response.body, name));
            REQUIRE(!err_response.client_error_code);
            REQUIRE(err_response.custom_status_code == 0);
            ct = AppUtils::find_header("content-type", err_response.headers);
            REQUIRE(ct);
            REQUIRE(ct->second == "application/json");
        }
    }

    response = make_http_error("AppErrorMissing", 404);
    app_error = AppUtils::check_for_errors(response);
    REQUIRE(app_error);
    REQUIRE(app_error->code() == ErrorCodes::AppServerError);
    REQUIRE(app_error->code_string() == "AppServerError");
    REQUIRE(app_error->server_error == "AppErrorMissing");
    REQUIRE(app_error->reason() == "some error");
    REQUIRE(app_error->link_to_server_logs == "http://dummy-link/");
    REQUIRE(app_error->additional_status_code);
    REQUIRE(*app_error->additional_status_code == 404);

    // Re-compose back into a Response
    err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 404);
    REQUIRE(!err_response.body.empty());
    REQUIRE(validate_json_body(err_response.body, "AppErrorMissing"));
    REQUIRE(!err_response.client_error_code);
    REQUIRE(err_response.custom_status_code == 0);
    ct = AppUtils::find_header("content-type", err_response.headers);
    REQUIRE(ct);
    REQUIRE(ct->second == "application/json");

    // HTTPError with different status values
    for (auto [status, message] : http_status_codes) {
        response = {
            status,
            0,
            {},
            "some http error",
        };
        app_error = AppUtils::check_for_errors(response);
        if (message.empty()) {
            REQUIRE(!app_error);
            continue;
        }
        REQUIRE(app_error);
        REQUIRE(app_error->code() == ErrorCodes::HTTPError);
        REQUIRE(app_error->code_string() == "HTTPError");
        REQUIRE(app_error->server_error.empty());
        REQUIRE(app_error->reason() == message);
        REQUIRE(app_error->link_to_server_logs.empty());
        REQUIRE(app_error->additional_status_code);
        REQUIRE(*app_error->additional_status_code == status);

        // Recompose back into a Response
        err_response = AppUtils::make_apperror_response(*app_error);
        REQUIRE(err_response.http_status_code == status);
        REQUIRE(err_response.body == "some http error");
        REQUIRE(!err_response.client_error_code);
        REQUIRE(err_response.custom_status_code == 0);
        REQUIRE(err_response.headers.empty());
    }

    // Missing error code and error message with fatal http status
    response = {
        501,
        0,
        {},
        "",
    };
    app_error = AppUtils::check_for_errors(response);
    REQUIRE(app_error);
    REQUIRE(app_error->code() == ErrorCodes::HTTPError);
    REQUIRE(app_error->code_string() == "HTTPError");
    REQUIRE(app_error->server_error.empty());
    REQUIRE(app_error->reason() == "http error code considered fatal. Server Error: 501");
    REQUIRE(app_error->link_to_server_logs.empty());
    REQUIRE(app_error->additional_status_code);
    REQUIRE(*app_error->additional_status_code == 501);

    // Re-compose back into a Response
    err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 501);
    REQUIRE(err_response.body.empty());
    REQUIRE(!err_response.client_error_code);
    REQUIRE(err_response.custom_status_code == 0);
    REQUIRE(err_response.headers.empty());

    // Missing error code and error message contains period with redirect http status
    response = {
        308,
        0,
        {},
        "some http error. ocurred",
    };
    app_error = AppUtils::check_for_errors(response);
    REQUIRE(app_error);
    REQUIRE(app_error->code() == ErrorCodes::HTTPError);
    REQUIRE(app_error->code_string() == "HTTPError");
    REQUIRE(app_error->server_error.empty());
    REQUIRE(app_error->reason() == "http error code considered fatal: some http error. ocurred. Redirection: 308");
    REQUIRE(app_error->link_to_server_logs.empty());
    REQUIRE(app_error->additional_status_code);
    REQUIRE(*app_error->additional_status_code == 308);

    // Re-compose back into a Response
    err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 308);
    REQUIRE(err_response.body == "some http error. ocurred");
    REQUIRE(!err_response.client_error_code);
    REQUIRE(err_response.custom_status_code == 0);
    REQUIRE(err_response.headers.empty());

    // Valid client error code, with body, but no json
    app::Response client_response = {
        501,
        0,
        {},
        "Some error occurred",
        ErrorCodes::BadBsonParse, // client_error_code
    };
    app_error = AppUtils::check_for_errors(client_response);
    REQUIRE(app_error);
    REQUIRE(app_error->code() == ErrorCodes::BadBsonParse);
    REQUIRE(app_error->code_string() == "BadBsonParse");
    REQUIRE(app_error->server_error.empty());
    REQUIRE(app_error->reason() == "Some error occurred");
    REQUIRE(app_error->link_to_server_logs.empty());
    REQUIRE(app_error->additional_status_code);
    REQUIRE(*app_error->additional_status_code == 501);

    // Re-compose back into a Response
    err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 501);
    REQUIRE(err_response.body == "Some error occurred");
    REQUIRE(err_response.client_error_code == ErrorCodes::BadBsonParse);
    REQUIRE(err_response.custom_status_code == 0);
    REQUIRE(err_response.headers.empty());

    // Same response with client error code, but no body
    client_response.body = "";
    app_error = AppUtils::check_for_errors(client_response);
    REQUIRE(app_error);
    REQUIRE(app_error->reason() == "client error code value considered fatal");

    // Re-compose back into a Response
    err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 501);
    REQUIRE(err_response.body == "client error code value considered fatal");
    REQUIRE(err_response.client_error_code == ErrorCodes::BadBsonParse);
    REQUIRE(err_response.custom_status_code == 0);
    REQUIRE(err_response.headers.empty());

    // Valid custom status code, with body, but no json
    app::Response custom_response = {501,
                                     4999, // custom_status_code
                                     {},
                                     "Some custom error occurred"};
    app_error = AppUtils::check_for_errors(custom_response);
    REQUIRE(app_error);
    REQUIRE(app_error->code() == ErrorCodes::CustomError);
    REQUIRE(app_error->code_string() == "CustomError");
    REQUIRE(app_error->server_error.empty());
    REQUIRE(app_error->reason() == "Some custom error occurred");
    REQUIRE(app_error->link_to_server_logs.empty());
    REQUIRE(app_error->additional_status_code);
    REQUIRE(*app_error->additional_status_code == 4999);

    // Re-compose back into a Response
    err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 0);
    REQUIRE(err_response.body == "Some custom error occurred");
    REQUIRE(!err_response.client_error_code);
    REQUIRE(err_response.custom_status_code == 4999);
    REQUIRE(err_response.headers.empty());

    // Same response with custom status code, but no body
    custom_response.body = "";
    app_error = AppUtils::check_for_errors(custom_response);
    REQUIRE(app_error);
    REQUIRE(app_error->reason() == "non-zero custom status code considered fatal");

    // Re-compose back into a Response
    err_response = AppUtils::make_apperror_response(*app_error);
    REQUIRE(err_response.http_status_code == 0);
    REQUIRE(err_response.body == "non-zero custom status code considered fatal");
    REQUIRE(!err_response.client_error_code);
    REQUIRE(err_response.custom_status_code == 4999);
    REQUIRE(err_response.headers.empty());
}

// MARK: - Verify generic app utils helper functions
TEST_CASE("app: verify app utils helpers", "[sync][app][local]") {
    SECTION("find_header") {
        std::map<std::string, std::string> headers1 = {{"header1", "header1-value"},
                                                       {"HEADER2", "header2-value"},
                                                       {"HeAdEr3", "header3-value"},
                                                       {"header@4", "header4-value"}};

        std::map<std::string, std::string> headers2 = {
            {"", "no-key-value"},
            {"header1", "header1-value"},
        };

        CHECK(AppUtils::find_header("", headers1) == nullptr);
        CHECK(AppUtils::find_header("header", headers1) == nullptr);
        CHECK(AppUtils::find_header("header*4", headers1) == nullptr);
        CHECK(AppUtils::find_header("header5", headers1) == nullptr);
        auto value = AppUtils::find_header("header1", headers1);
        CHECK(value != nullptr);
        CHECK(value->first == "header1");
        CHECK(value->second == "header1-value");
        value = AppUtils::find_header("HEADER1", headers1);
        CHECK(value != nullptr);
        CHECK(value->first == "header1");
        CHECK(value->second == "header1-value");
        value = AppUtils::find_header("header2", headers1);
        CHECK(value != nullptr);
        CHECK(value->first == "HEADER2");
        CHECK(value->second == "header2-value");
        value = AppUtils::find_header("hEaDeR2", headers1);
        CHECK(value != nullptr);
        CHECK(value->first == "HEADER2");
        CHECK(value->second == "header2-value");
        value = AppUtils::find_header("HEADER3", headers1);
        CHECK(value != nullptr);
        CHECK(value->first == "HeAdEr3");
        CHECK(value->second == "header3-value");
        value = AppUtils::find_header("header3", headers1);
        CHECK(value != nullptr);
        CHECK(value->first == "HeAdEr3");
        CHECK(value->second == "header3-value");
        value = AppUtils::find_header("HEADER@4", headers1);
        CHECK(value != nullptr);
        CHECK(value->first == "header@4");
        CHECK(value->second == "header4-value");
        value = AppUtils::find_header("", headers2);
        CHECK(value != nullptr);
        CHECK(value->first == "");
        CHECK(value->second == "no-key-value");
        value = AppUtils::find_header("HeAdEr1", headers2);
        CHECK(value != nullptr);
        CHECK(value->first == "header1");
        CHECK(value->second == "header1-value");
    }

    SECTION("is_success_status_code") {
        CHECK(AppUtils::is_success_status_code(0));
        for (int code = 200; code < 300; code++) {
            CHECK(AppUtils::is_success_status_code(code));
        }
        CHECK(!AppUtils::is_success_status_code(1));
        CHECK(!AppUtils::is_success_status_code(199));
        CHECK(!AppUtils::is_success_status_code(300));
        CHECK(!AppUtils::is_success_status_code(99999));
    }

    SECTION("is_redirect_status_code") {
        // Only MovedPermanently(301) and PermanentRedirect(308) return true
        CHECK(AppUtils::is_redirect_status_code(301));
        CHECK(AppUtils::is_redirect_status_code(308));
        CHECK(!AppUtils::is_redirect_status_code(0));
        CHECK(!AppUtils::is_redirect_status_code(200));
        CHECK(!AppUtils::is_redirect_status_code(300));
        CHECK(!AppUtils::is_redirect_status_code(403));
        CHECK(!AppUtils::is_redirect_status_code(99999));
    }

    SECTION("extract_redir_location") {
        auto comp = AppUtils::extract_redir_location(
            {{"Content-Type", "application/json"}, {"Location", "http://redirect.host"}});
        CHECK(comp == "http://redirect.host");
        comp = AppUtils::extract_redir_location({{"location", "http://redirect.host"}});
        CHECK(comp == "http://redirect.host");
        comp = AppUtils::extract_redir_location({{"LoCaTiOn", "http://redirect.host/"}});
        CHECK(comp == "http://redirect.host/");
        comp = AppUtils::extract_redir_location({{"LOCATION", "http://redirect.host/includes/path"}});
        CHECK(comp == "http://redirect.host/includes/path");
        comp = AppUtils::extract_redir_location({{"Content-Type", "application/json"}});
        CHECK(!comp);
        comp = AppUtils::extract_redir_location({{"some-location", "http://redirect.host"}});
        CHECK(!comp);
        comp = AppUtils::extract_redir_location({{"location", ""}});
        CHECK(!comp);
        comp = AppUtils::extract_redir_location({});
        CHECK(!comp);
        comp = AppUtils::extract_redir_location({{"location", "bad-server-url"}});
        CHECK(!comp);
    }
}

// MARK: - Login with Credentials Tests

TEST_CASE("app: login_with_credentials integration", "[sync][app][user][baas]") {
    SECTION("login") {
        TestAppSession session;
        auto app = session.app();
        app->log_out([](auto) {});

        int subscribe_processed = 0;

        auto token = app->subscribe([&subscribe_processed](auto&) {
            subscribe_processed++;
        });

        REQUIRE_FALSE(app->current_user());
        auto user = log_in(app);
        CHECK(!user->device_id().empty());
        CHECK(user->has_device_id());
        REQUIRE(app->current_user());
        CHECK(subscribe_processed == 1);

        bool processed = false;
        app->log_out([&](auto error) {
            REQUIRE_FALSE(error);
            processed = true;
        });
        REQUIRE_FALSE(app->current_user());
        CHECK(processed);
        CHECK(subscribe_processed == 2);

        app->unsubscribe(token);
    }
}

// MARK: - UsernamePasswordProviderClient Tests

TEST_CASE("app: UsernamePasswordProviderClient integration", "[sync][app][user][baas]") {
    const std::string base_url = get_real_base_url();
    AutoVerifiedEmailCredentials creds;
    auto email = creds.email;
    auto password = creds.password;

    TestAppSession session;
    auto app = session.app();
    auto client = app->provider_client<App::UsernamePasswordProviderClient>();

    bool processed = false;

    client.register_email(email, password, [&](Optional<AppError> error) {
        CAPTURE(email);
        CAPTURE(password);
        REQUIRE_FALSE(error); // first registration success
    });

    SECTION("double registration should fail") {
        client.register_email(email, password, [&](Optional<AppError> error) {
            // Error returned states the account has already been created
            REQUIRE(error);
            CHECK(error->reason() == "name already in use");
            CHECK(error->code() == ErrorCodes::AccountNameInUse);
            CHECK(!error->link_to_server_logs.empty());
            CHECK_THAT(error->link_to_server_logs, ContainsSubstring(base_url));
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("double registration should fail") {
        // the server registration function will reject emails that do not contain "realm_tests_do_autoverify"
        std::string email_to_reject = util::format("%1@%2.com", random_string(10), random_string(10));
        client.register_email(email_to_reject, password, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == util::format("failed to confirm user \"%1\"", email_to_reject));
            CHECK(error->code() == ErrorCodes::BadRequest);
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("can login with registered account") {
        auto user = log_in(app, creds);
        CHECK(user->user_profile().email() == email);
    }

    SECTION("cannot login with wrong password") {
        app->log_in_with_credentials(AppCredentials::username_password(email, "boogeyman"),
                                     [&](std::shared_ptr<User> user, Optional<AppError> error) {
                                         CHECK(!user);
                                         REQUIRE(error);
                                         REQUIRE(error->code() == ErrorCodes::InvalidPassword);
                                         processed = true;
                                     });
        CHECK(processed);
    }

    SECTION("confirm user") {
        client.confirm_user("a_token", "a_token_id", [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == "invalid token data");
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("resend confirmation email") {
        client.resend_confirmation_email(email, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == "already confirmed");
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("reset password invalid tokens") {
        client.reset_password(password, "token_sample", "token_id_sample", [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == "invalid token data");
            CHECK(!error->link_to_server_logs.empty());
            CHECK_THAT(error->link_to_server_logs, ContainsSubstring(base_url));
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("reset password function success") {
        // the imported test app will accept password reset if the password contains "realm_tests_do_reset" via a
        // function
        std::string accepted_new_password = util::format("realm_tests_do_reset%1", random_string(10));
        client.call_reset_password_function(email, accepted_new_password, {}, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("reset password function failure") {
        std::string rejected_password = util::format("%1", random_string(10));
        client.call_reset_password_function(email, rejected_password, {"foo", "bar"}, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == util::format("failed to reset password for user \"%1\"", email));
            CHECK(error->is_service_error());
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("reset password function for invalid user fails") {
        client.call_reset_password_function(util::format("%1@%2.com", random_string(5), random_string(5)), password,
                                            {"foo", "bar"}, [&](Optional<AppError> error) {
                                                REQUIRE(error);
                                                CHECK(error->reason() == "user not found");
                                                CHECK(error->is_service_error());
                                                CHECK(error->code() == ErrorCodes::UserNotFound);
                                                processed = true;
                                            });
        CHECK(processed);
    }

    SECTION("retry custom confirmation") {
        client.retry_custom_confirmation(email, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == "already confirmed");
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("retry custom confirmation for invalid user fails") {
        client.retry_custom_confirmation(util::format("%1@%2.com", random_string(5), random_string(5)),
                                         [&](Optional<AppError> error) {
                                             REQUIRE(error);
                                             CHECK(error->reason() == "user not found");
                                             CHECK(error->is_service_error());
                                             CHECK(error->code() == ErrorCodes::UserNotFound);
                                             processed = true;
                                         });
        CHECK(processed);
    }

    SECTION("log in, remove, log in") {
        app->remove_user(app->current_user(), [](auto) {});
        CHECK(app->all_users().size() == 0);
        CHECK(app->current_user() == nullptr);

        auto user = log_in(app, AppCredentials::username_password(email, password));
        CHECK(user->user_profile().email() == email);
        CHECK(user->state() == SyncUser::State::LoggedIn);

        app->remove_user(user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });
        CHECK(user->state() == SyncUser::State::Removed);

        log_in(app, AppCredentials::username_password(email, password));
        CHECK(user->state() == SyncUser::State::Removed);
        CHECK(app->current_user() != user);
        user = app->current_user();
        CHECK(user->user_profile().email() == email);
        CHECK(user->state() == SyncUser::State::LoggedIn);

        app->remove_user(user, [&](Optional<AppError> error) {
            REQUIRE(!error);
            CHECK(app->all_users().size() == 0);
            processed = true;
        });

        CHECK(user->state() == SyncUser::State::Removed);
        CHECK(processed);
        CHECK(app->all_users().size() == 0);
    }
}

// MARK: - UserAPIKeyProviderClient Tests

TEST_CASE("app: UserAPIKeyProviderClient integration", "[sync][app][api key][baas]") {
    TestAppSession session;
    auto app = session.app();
    auto client = app->provider_client<App::UserAPIKeyProviderClient>();

    bool processed = false;
    App::UserAPIKey api_key;

    SECTION("api-key") {
        std::shared_ptr<User> logged_in_user = app->current_user();
        auto api_key_name = util::format("%1", random_string(15));
        client.create_api_key(api_key_name, logged_in_user,
                              [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
                                  REQUIRE_FALSE(error);
                                  CHECK(user_api_key.name == api_key_name);
                                  api_key = user_api_key;
                              });

        client.fetch_api_key(api_key.id, logged_in_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(user_api_key.name == api_key_name);
            CHECK(user_api_key.id == api_key.id);
        });

        client.fetch_api_keys(logged_in_user, [&](std::vector<App::UserAPIKey> api_keys, Optional<AppError> error) {
            CHECK(api_keys.size() == 1);
            for (auto key : api_keys) {
                CHECK(key.id.to_string() == api_key.id.to_string());
                CHECK(api_key.name == api_key_name);
                CHECK(key.id == api_key.id);
            }
            REQUIRE_FALSE(error);
        });

        client.enable_api_key(api_key.id, logged_in_user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        client.fetch_api_key(api_key.id, logged_in_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(user_api_key.disabled == false);
            CHECK(user_api_key.name == api_key_name);
            CHECK(user_api_key.id == api_key.id);
        });

        client.disable_api_key(api_key.id, logged_in_user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        client.fetch_api_key(api_key.id, logged_in_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(user_api_key.disabled == true);
            CHECK(user_api_key.name == api_key_name);
        });

        client.delete_api_key(api_key.id, logged_in_user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        client.fetch_api_key(api_key.id, logged_in_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            CHECK(user_api_key.name == "");
            CHECK(error);
            processed = true;
        });

        CHECK(processed);
    }

    SECTION("api-key without a user") {
        std::shared_ptr<User> no_user = nullptr;
        auto api_key_name = util::format("%1", random_string(15));
        client.create_api_key(api_key_name, no_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->is_service_error());
            CHECK(error->reason() == "must authenticate first");
            CHECK(user_api_key.name == "");
        });

        client.fetch_api_key(api_key.id, no_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->is_service_error());
            CHECK(error->reason() == "must authenticate first");
            CHECK(user_api_key.name == "");
        });

        client.fetch_api_keys(no_user, [&](std::vector<App::UserAPIKey> api_keys, Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->is_service_error());
            CHECK(error->reason() == "must authenticate first");
            CHECK(api_keys.size() == 0);
        });

        client.enable_api_key(api_key.id, no_user, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->is_service_error());
            CHECK(error->reason() == "must authenticate first");
        });

        client.fetch_api_key(api_key.id, no_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->is_service_error());
            CHECK(error->reason() == "must authenticate first");
            CHECK(user_api_key.name == "");
        });

        client.disable_api_key(api_key.id, no_user, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->is_service_error());
            CHECK(error->reason() == "must authenticate first");
        });

        client.fetch_api_key(api_key.id, no_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->is_service_error());
            CHECK(error->reason() == "must authenticate first");
            CHECK(user_api_key.name == "");
        });

        client.delete_api_key(api_key.id, no_user, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->is_service_error());
            CHECK(error->reason() == "must authenticate first");
        });

        client.fetch_api_key(api_key.id, no_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            CHECK(user_api_key.name == "");
            REQUIRE(error);
            CHECK(error->is_service_error());
            CHECK(error->reason() == "must authenticate first");
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("api-key against the wrong user") {
        std::shared_ptr<User> first_user = app->current_user();
        create_user_and_log_in(app);
        std::shared_ptr<User> second_user = app->current_user();
        REQUIRE(first_user != second_user);
        auto api_key_name = util::format("%1", random_string(15));
        App::UserAPIKey api_key;
        App::UserAPIKeyProviderClient provider = app->provider_client<App::UserAPIKeyProviderClient>();

        provider.create_api_key(api_key_name, first_user,
                                [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
                                    REQUIRE_FALSE(error);
                                    CHECK(user_api_key.name == api_key_name);
                                    api_key = user_api_key;
                                });

        provider.fetch_api_key(api_key.id, first_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(user_api_key.name == api_key_name);
            CHECK(user_api_key.id.to_string() == user_api_key.id.to_string());
        });

        provider.fetch_api_key(api_key.id, second_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == "API key not found");
            CHECK(error->is_service_error());
            CHECK(error->code() == ErrorCodes::APIKeyNotFound);
            CHECK(user_api_key.name == "");
        });

        provider.fetch_api_keys(first_user, [&](std::vector<App::UserAPIKey> api_keys, Optional<AppError> error) {
            CHECK(api_keys.size() == 1);
            for (auto api_key : api_keys) {
                CHECK(api_key.name == api_key_name);
            }
            REQUIRE_FALSE(error);
        });

        provider.fetch_api_keys(second_user, [&](std::vector<App::UserAPIKey> api_keys, Optional<AppError> error) {
            CHECK(api_keys.size() == 0);
            REQUIRE_FALSE(error);
        });

        provider.enable_api_key(api_key.id, first_user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        provider.enable_api_key(api_key.id, second_user, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == "API key not found");
            CHECK(error->is_service_error());
            CHECK(error->code() == ErrorCodes::APIKeyNotFound);
        });

        provider.fetch_api_key(api_key.id, first_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(user_api_key.disabled == false);
            CHECK(user_api_key.name == api_key_name);
        });

        provider.fetch_api_key(api_key.id, second_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE(error);
            CHECK(user_api_key.name == "");
            CHECK(error->reason() == "API key not found");
            CHECK(error->is_service_error());
            CHECK(error->code() == ErrorCodes::APIKeyNotFound);
        });

        provider.disable_api_key(api_key.id, first_user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        provider.disable_api_key(api_key.id, second_user, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == "API key not found");
            CHECK(error->is_service_error());
            CHECK(error->code() == ErrorCodes::APIKeyNotFound);
        });

        provider.fetch_api_key(api_key.id, first_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(user_api_key.disabled == true);
            CHECK(user_api_key.name == api_key_name);
        });

        provider.fetch_api_key(api_key.id, second_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE(error);
            CHECK(user_api_key.name == "");
            CHECK(error->reason() == "API key not found");
            CHECK(error->is_service_error());
            CHECK(error->code() == ErrorCodes::APIKeyNotFound);
        });

        provider.delete_api_key(api_key.id, second_user, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == "API key not found");
            CHECK(error->is_service_error());
            CHECK(error->code() == ErrorCodes::APIKeyNotFound);
        });

        provider.delete_api_key(api_key.id, first_user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        provider.fetch_api_key(api_key.id, first_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            CHECK(user_api_key.name == "");
            REQUIRE(error);
            CHECK(error->reason() == "API key not found");
            CHECK(error->is_service_error());
            CHECK(error->code() == ErrorCodes::APIKeyNotFound);
            processed = true;
        });

        provider.fetch_api_key(api_key.id, second_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            CHECK(user_api_key.name == "");
            REQUIRE(error);
            CHECK(error->reason() == "API key not found");
            CHECK(error->is_service_error());
            CHECK(error->code() == ErrorCodes::APIKeyNotFound);
            processed = true;
        });

        CHECK(processed);
    }
}

// MARK: - Auth Providers Function Tests

TEST_CASE("app: auth providers function integration", "[sync][app][user][baas]") {
    TestAppSession session;
    auto app = session.app();

    SECTION("auth providers function integration") {
        bson::BsonDocument function_params{{"realmCustomAuthFuncUserId", "123456"}};
        auto credentials = AppCredentials::function(function_params);
        auto user = log_in(app, credentials);
        REQUIRE(user->identities()[0].provider_type == IdentityProviderFunction);
    }
}

// MARK: - Link User Tests

TEST_CASE("app: Linking user identities", "[sync][app][user][baas]") {
    TestAppSession session;
    auto app = session.app();
    auto user = log_in(app);

    AutoVerifiedEmailCredentials creds;
    app->provider_client<App::UsernamePasswordProviderClient>().register_email(creds.email, creds.password,
                                                                               [&](Optional<AppError> error) {
                                                                                   REQUIRE_FALSE(error);
                                                                               });

    SECTION("anonymous users are reused before they are linked to an identity") {
        REQUIRE(user == log_in(app));
    }

    SECTION("linking a user adds that identity to the user") {
        REQUIRE(user->identities().size() == 1);
        CHECK(user->identities()[0].provider_type == IdentityProviderAnonymous);

        app->link_user(user, creds, [&](std::shared_ptr<User> user2, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            REQUIRE(user == user2);
            REQUIRE(user->identities().size() == 2);
            CHECK(user->identities()[0].provider_type == IdentityProviderAnonymous);
            CHECK(user->identities()[1].provider_type == IdentityProviderUsernamePassword);
        });
    }

    SECTION("linking an identity makes the user no longer returned by anonymous logins") {
        app->link_user(user, creds, [&](std::shared_ptr<User>, Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });
        auto user2 = log_in(app);
        REQUIRE(user != user2);
    }

    SECTION("existing users are reused when logging in via linked identities") {
        app->link_user(user, creds, [](std::shared_ptr<User>, Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });
        app->log_out([](auto error) {
            REQUIRE_FALSE(error);
        });
        REQUIRE(user->state() == SyncUser::State::LoggedOut);
        // Should give us the same user instance despite logging in with a
        // different identity
        REQUIRE(user == log_in(app, creds));
        REQUIRE(user->state() == SyncUser::State::LoggedIn);
    }
}

// MARK: - Delete User Tests

TEST_CASE("app: delete anonymous user integration", "[sync][app][user][baas]") {
    TestAppSession session;
    auto app = session.app();

    SECTION("delete user expect success") {
        CHECK(app->all_users().size() == 1);

        // Log in user 1
        auto user_a = app->current_user();
        CHECK(user_a->state() == SyncUser::State::LoggedIn);
        app->delete_user(user_a, [&](Optional<app::AppError> error) {
            REQUIRE_FALSE(error);
            // a logged out anon user will be marked as Removed, not LoggedOut
            CHECK(user_a->state() == SyncUser::State::Removed);
        });
        CHECK(app->all_users().empty());
        CHECK(app->current_user() == nullptr);

        app->delete_user(user_a, [&](Optional<app::AppError> error) {
            CHECK(error->reason() == "User must be logged in to be deleted.");
            CHECK(app->all_users().size() == 0);
        });

        // Log in user 2
        auto user_b = log_in(app);
        CHECK(app->current_user() == user_b);
        CHECK(user_b->state() == SyncUser::State::LoggedIn);
        CHECK(app->all_users().size() == 1);

        app->delete_user(user_b, [&](Optional<app::AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(app->all_users().size() == 0);
        });

        CHECK(app->current_user() == nullptr);

        // check both handles are no longer valid
        CHECK(user_a->state() == SyncUser::State::Removed);
        CHECK(user_b->state() == SyncUser::State::Removed);
    }
}

TEST_CASE("app: delete user with credentials integration", "[sync][app][user][baas]") {
    TestAppSession session;
    auto app = session.app();
    app->remove_user(app->current_user(), [](auto) {});

    SECTION("log in and delete") {
        CHECK(app->all_users().size() == 0);
        CHECK(app->current_user() == nullptr);

        auto credentials = create_user_and_log_in(app);
        auto user = app->current_user();

        CHECK(app->current_user() == user);
        CHECK(user->state() == SyncUser::State::LoggedIn);
        app->delete_user(user, [&](Optional<app::AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(app->all_users().size() == 0);
        });
        CHECK(user->state() == SyncUser::State::Removed);
        CHECK(app->current_user() == nullptr);

        app->log_in_with_credentials(credentials, [](std::shared_ptr<User> user, util::Optional<AppError> error) {
            CHECK(!user);
            REQUIRE(error);
            REQUIRE(error->code() == ErrorCodes::InvalidPassword);
        });
        CHECK(app->current_user() == nullptr);

        CHECK(app->all_users().size() == 0);
        app->delete_user(user, [](Optional<app::AppError> err) {
            CHECK(err->code() > 0);
        });

        CHECK(app->current_user() == nullptr);
        CHECK(app->all_users().size() == 0);
        CHECK(user->state() == SyncUser::State::Removed);
    }
}

// MARK: - Call Function Tests

TEST_CASE("app: call function", "[sync][app][function][baas]") {
    TestAppSession session;
    auto app = session.app();

    bson::BsonArray toSum(5);
    std::iota(toSum.begin(), toSum.end(), static_cast<int64_t>(1));
    const auto checkFn = [](Optional<int64_t>&& sum, Optional<AppError>&& error) {
        REQUIRE(!error);
        CHECK(*sum == 15);
    };
    app->call_function<int64_t>("sumFunc", toSum, checkFn);
    app->call_function<int64_t>(app->current_user(), "sumFunc", toSum, checkFn);
}

// MARK: - Remote Mongo Client Tests

TEST_CASE("app: remote mongo client", "[sync][app][mongo][baas]") {
    TestAppSession session;
    auto app = session.app();

    auto remote_client = app->current_user()->mongo_client("BackingDB");
    auto app_session = get_runtime_app_session();
    auto db = remote_client.db(app_session.config.mongo_dbname);
    auto dog_collection = db["Dog"];
    auto cat_collection = db["Cat"];
    auto person_collection = db["Person"];

    bson::BsonDocument dog_document{{"name", "fido"}, {"breed", "king charles"}};

    bson::BsonDocument dog_document2{{"name", "bob"}, {"breed", "french bulldog"}};

    auto dog3_object_id = ObjectId::gen();
    bson::BsonDocument dog_document3{
        {"_id", dog3_object_id},
        {"name", "petunia"},
        {"breed", "french bulldog"},
    };

    auto cat_id_string = random_string(10);
    bson::BsonDocument cat_document{
        {"_id", cat_id_string},
        {"name", "luna"},
        {"breed", "scottish fold"},
    };

    bson::BsonDocument person_document{
        {"firstName", "John"},
        {"lastName", "Johnson"},
        {"age", 30},
    };

    bson::BsonDocument person_document2{
        {"firstName", "Bob"},
        {"lastName", "Johnson"},
        {"age", 30},
    };

    bson::BsonDocument bad_document{{"bad", "value"}};

    dog_collection.delete_many(dog_document, [&](uint64_t, Optional<AppError> error) {
        REQUIRE_FALSE(error);
    });

    dog_collection.delete_many(dog_document2, [&](uint64_t, Optional<AppError> error) {
        REQUIRE_FALSE(error);
    });

    dog_collection.delete_many({}, [&](uint64_t, Optional<AppError> error) {
        REQUIRE_FALSE(error);
    });

    dog_collection.delete_many(person_document, [&](uint64_t, Optional<AppError> error) {
        REQUIRE_FALSE(error);
    });

    dog_collection.delete_many(person_document2, [&](uint64_t, Optional<AppError> error) {
        REQUIRE_FALSE(error);
    });

    SECTION("insert") {
        bool processed = false;
        ObjectId dog_object_id;
        ObjectId dog2_object_id;

        dog_collection.insert_one_bson(bad_document, [&](Optional<bson::Bson> bson, Optional<AppError> error) {
            CHECK(error);
            CHECK(!bson);
        });

        dog_collection.insert_one_bson(dog_document3, [&](Optional<bson::Bson> value, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            auto bson = static_cast<bson::BsonDocument>(*value);
            CHECK(static_cast<ObjectId>(bson["insertedId"]) == dog3_object_id);
        });

        cat_collection.insert_one_bson(cat_document, [&](Optional<bson::Bson> value, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            auto bson = static_cast<bson::BsonDocument>(*value);
            CHECK(static_cast<std::string>(bson["insertedId"]) == cat_id_string);
        });

        dog_collection.delete_many({}, [&](uint64_t, Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        cat_collection.delete_one(cat_document, [&](uint64_t, Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        dog_collection.insert_one(bad_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            CHECK(error);
            CHECK(!object_id);
        });

        dog_collection.insert_one(dog_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
            dog_object_id = static_cast<ObjectId>(*object_id);
        });

        dog_collection.insert_one(dog_document2, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
            dog2_object_id = static_cast<ObjectId>(*object_id);
        });

        dog_collection.insert_one(dog_document3, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(object_id->type() == bson::Bson::Type::ObjectId);
            CHECK(static_cast<ObjectId>(*object_id) == dog3_object_id);
        });

        cat_collection.insert_one(cat_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(object_id->type() == bson::Bson::Type::String);
            CHECK(static_cast<std::string>(*object_id) == cat_id_string);
        });

        person_document["dogs"] = bson::BsonArray({dog_object_id, dog2_object_id, dog3_object_id});
        person_collection.insert_one(person_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
        });

        dog_collection.delete_many({}, [&](uint64_t, Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        cat_collection.delete_one(cat_document, [&](uint64_t, Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        bson::BsonArray documents{
            dog_document,
            dog_document2,
            dog_document3,
        };

        dog_collection.insert_many_bson(documents, [&](Optional<bson::Bson> value, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            auto bson = static_cast<bson::BsonDocument>(*value);
            auto insertedIds = static_cast<bson::BsonArray>(bson["insertedIds"]);
        });

        dog_collection.delete_many({}, [&](uint64_t, Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        dog_collection.insert_many(documents, [&](bson::BsonArray inserted_docs, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(inserted_docs.size() == 3);
            CHECK(inserted_docs[0].type() == bson::Bson::Type::ObjectId);
            CHECK(inserted_docs[1].type() == bson::Bson::Type::ObjectId);
            CHECK(inserted_docs[2].type() == bson::Bson::Type::ObjectId);
            CHECK(static_cast<ObjectId>(inserted_docs[2]) == dog3_object_id);
            processed = true;
        });

        CHECK(processed);
    }

    SECTION("find") {
        bool processed = false;

        dog_collection.find(dog_document, [&](Optional<bson::BsonArray> document_array, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*document_array).size() == 0);
        });

        dog_collection.find_bson(dog_document, {}, [&](Optional<bson::Bson> bson, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(static_cast<bson::BsonArray>(*bson).size() == 0);
        });

        dog_collection.find_one(dog_document, [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(!document);
        });

        dog_collection.find_one_bson(dog_document, {}, [&](Optional<bson::Bson> bson, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((!bson || bson::holds_alternative<util::None>(*bson)));
        });

        ObjectId dog_object_id;
        ObjectId dog2_object_id;

        dog_collection.insert_one(dog_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
            dog_object_id = static_cast<ObjectId>(*object_id);
        });

        dog_collection.insert_one(dog_document2, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
            dog2_object_id = static_cast<ObjectId>(*object_id);
        });

        person_document["dogs"] = bson::BsonArray({dog_object_id, dog2_object_id});
        person_collection.insert_one(person_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
        });

        dog_collection.find(dog_document, [&](Optional<bson::BsonArray> documents, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*documents).size() == 1);
        });

        dog_collection.find_bson(dog_document, {}, [&](Optional<bson::Bson> bson, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(static_cast<bson::BsonArray>(*bson).size() == 1);
        });

        person_collection.find(person_document, [&](Optional<bson::BsonArray> documents, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*documents).size() == 1);
        });

        MongoCollection::FindOptions options{
            2,                                                         // document limit
            Optional<bson::BsonDocument>({{"name", 1}, {"breed", 1}}), // project
            Optional<bson::BsonDocument>({{"breed", 1}})               // sort
        };

        dog_collection.find(dog_document, options,
                            [&](Optional<bson::BsonArray> document_array, Optional<AppError> error) {
                                REQUIRE_FALSE(error);
                                CHECK((*document_array).size() == 1);
                            });

        dog_collection.find({{"name", "fido"}}, options,
                            [&](Optional<bson::BsonArray> document_array, Optional<AppError> error) {
                                REQUIRE_FALSE(error);
                                CHECK((*document_array).size() == 1);
                                auto king_charles = static_cast<bson::BsonDocument>((*document_array)[0]);
                                CHECK(king_charles["breed"] == "king charles");
                            });

        dog_collection.find_one(dog_document, [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            auto name = (*document)["name"];
            CHECK(name == "fido");
        });

        dog_collection.find_one(dog_document, options,
                                [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                    REQUIRE_FALSE(error);
                                    auto name = (*document)["name"];
                                    CHECK(name == "fido");
                                });

        dog_collection.find_one_bson(dog_document, options, [&](Optional<bson::Bson> bson, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            auto name = (static_cast<bson::BsonDocument>(*bson))["name"];
            CHECK(name == "fido");
        });

        dog_collection.find(dog_document, [&](Optional<bson::BsonArray> documents, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*documents).size() == 1);
        });

        dog_collection.find_one_and_delete(dog_document,
                                           [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                               REQUIRE_FALSE(error);
                                               REQUIRE(document);
                                           });

        dog_collection.find_one_and_delete({{}},
                                           [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                               REQUIRE_FALSE(error);
                                               REQUIRE(document);
                                           });

        dog_collection.find_one_and_delete({{"invalid", "key"}},
                                           [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                               REQUIRE_FALSE(error);
                                               CHECK(!document);
                                           });

        dog_collection.find_one_and_delete_bson({{"invalid", "key"}}, {},
                                                [&](Optional<bson::Bson> bson, Optional<AppError> error) {
                                                    REQUIRE_FALSE(error);
                                                    CHECK((!bson || bson::holds_alternative<util::None>(*bson)));
                                                });

        dog_collection.find(dog_document, [&](Optional<bson::BsonArray> documents, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*documents).size() == 0);
            processed = true;
        });

        CHECK(processed);
    }

    SECTION("count and aggregate") {
        bool processed = false;

        ObjectId dog_object_id;
        ObjectId dog2_object_id;

        dog_collection.insert_one(dog_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
        });

        dog_collection.insert_one(dog_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
            dog_object_id = static_cast<ObjectId>(*object_id);
        });

        dog_collection.insert_one(dog_document2, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
            dog2_object_id = static_cast<ObjectId>(*object_id);
        });

        person_document["dogs"] = bson::BsonArray({dog_object_id, dog2_object_id});
        person_collection.insert_one(person_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
        });

        bson::BsonDocument match{{"$match", bson::BsonDocument({{"name", "fido"}})}};

        bson::BsonDocument group{{"$group", bson::BsonDocument({{"_id", "$name"}})}};

        bson::BsonArray pipeline{match, group};

        dog_collection.aggregate(pipeline, [&](Optional<bson::BsonArray> documents, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*documents).size() == 1);
        });

        dog_collection.aggregate_bson(pipeline, [&](Optional<bson::Bson> bson, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(static_cast<bson::BsonArray>(*bson).size() == 1);
        });

        dog_collection.count({{"breed", "king charles"}}, [&](uint64_t count, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(count == 2);
        });

        dog_collection.count_bson({{"breed", "king charles"}}, 0,
                                  [&](Optional<bson::Bson> bson, Optional<AppError> error) {
                                      REQUIRE_FALSE(error);
                                      CHECK(static_cast<int64_t>(*bson) == 2);
                                  });

        dog_collection.count({{"breed", "french bulldog"}}, [&](uint64_t count, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(count == 1);
        });

        dog_collection.count({{"breed", "king charles"}}, 1, [&](uint64_t count, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(count == 1);
        });

        person_collection.count(
            {{"firstName", "John"}, {"lastName", "Johnson"}, {"age", bson::BsonDocument({{"$gt", 25}})}}, 1,
            [&](uint64_t count, Optional<AppError> error) {
                REQUIRE_FALSE(error);
                CHECK(count == 1);
                processed = true;
            });

        CHECK(processed);
    }

    SECTION("find and update") {
        bool processed = false;

        MongoCollection::FindOneAndModifyOptions find_and_modify_options{
            Optional<bson::BsonDocument>({{"name", 1}, {"breed", 1}}), // project
            Optional<bson::BsonDocument>({{"name", 1}}),               // sort,
            true,                                                      // upsert
            true                                                       // return new doc
        };

        dog_collection.find_one_and_update(dog_document, dog_document2,
                                           [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                               REQUIRE_FALSE(error);
                                               CHECK(!document);
                                           });

        dog_collection.insert_one(dog_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
        });

        dog_collection.find_one_and_update(dog_document, dog_document2, find_and_modify_options,
                                           [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                               REQUIRE_FALSE(error);
                                               auto breed = static_cast<std::string>((*document)["breed"]);
                                               CHECK(breed == "french bulldog");
                                           });

        dog_collection.find_one_and_update(dog_document2, dog_document, find_and_modify_options,
                                           [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                               REQUIRE_FALSE(error);
                                               auto breed = static_cast<std::string>((*document)["breed"]);
                                               CHECK(breed == "king charles");
                                           });

        dog_collection.find_one_and_update_bson(dog_document, dog_document2, find_and_modify_options,
                                                [&](Optional<bson::Bson> bson, Optional<AppError> error) {
                                                    REQUIRE_FALSE(error);
                                                    auto breed = static_cast<std::string>(
                                                        static_cast<bson::BsonDocument>(*bson)["breed"]);
                                                    CHECK(breed == "french bulldog");
                                                });

        dog_collection.find_one_and_update_bson(dog_document2, dog_document, find_and_modify_options,
                                                [&](Optional<bson::Bson> bson, Optional<AppError> error) {
                                                    REQUIRE_FALSE(error);
                                                    auto breed = static_cast<std::string>(
                                                        static_cast<bson::BsonDocument>(*bson)["breed"]);
                                                    CHECK(breed == "king charles");
                                                });

        dog_collection.find_one_and_update({{"name", "invalid name"}}, {{"name", "some name"}},
                                           [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                               REQUIRE_FALSE(error);
                                               CHECK(!document);
                                               processed = true;
                                           });
        CHECK(processed);
        processed = false;

        dog_collection.find_one_and_update({{"name", "invalid name"}}, {{}}, find_and_modify_options,
                                           [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                               REQUIRE(error);
                                               CHECK(error->reason() == "insert not permitted");
                                               CHECK(!document);
                                               processed = true;
                                           });
        CHECK(processed);
    }

    SECTION("update") {
        bool processed = false;
        ObjectId dog_object_id;

        dog_collection.update_one(dog_document, dog_document2, true,
                                  [&](MongoCollection::UpdateResult result, Optional<AppError> error) {
                                      REQUIRE_FALSE(error);
                                      CHECK((*result.upserted_id).to_string() != "");
                                  });

        dog_collection.update_one(dog_document2, dog_document,
                                  [&](MongoCollection::UpdateResult result, Optional<AppError> error) {
                                      REQUIRE_FALSE(error);
                                      CHECK(!result.upserted_id);
                                  });

        cat_collection.update_one({}, cat_document, true,
                                  [&](MongoCollection::UpdateResult result, Optional<AppError> error) {
                                      REQUIRE_FALSE(error);
                                      CHECK(result.upserted_id->type() == bson::Bson::Type::String);
                                      CHECK(result.upserted_id == cat_id_string);
                                  });

        dog_collection.delete_many({}, [&](uint64_t, Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        cat_collection.delete_many({}, [&](uint64_t, Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        dog_collection.update_one_bson(dog_document, dog_document2, true,
                                       [&](Optional<bson::Bson> bson, Optional<AppError> error) {
                                           REQUIRE_FALSE(error);
                                           auto upserted_id = static_cast<bson::BsonDocument>(*bson)["upsertedId"];

                                           REQUIRE(upserted_id.type() == bson::Bson::Type::ObjectId);
                                       });

        dog_collection.update_one_bson(dog_document2, dog_document, true,
                                       [&](Optional<bson::Bson> bson, Optional<AppError> error) {
                                           REQUIRE_FALSE(error);
                                           auto document = static_cast<bson::BsonDocument>(*bson);
                                           auto foundUpsertedId = document.find("upsertedId");
                                           REQUIRE(!foundUpsertedId);
                                       });

        cat_collection.update_one_bson({}, cat_document, true,
                                       [&](Optional<bson::Bson> bson, Optional<AppError> error) {
                                           REQUIRE_FALSE(error);
                                           auto upserted_id = static_cast<bson::BsonDocument>(*bson)["upsertedId"];
                                           REQUIRE(upserted_id.type() == bson::Bson::Type::String);
                                           REQUIRE(upserted_id == cat_id_string);
                                       });

        person_document["dogs"] = bson::BsonArray();
        bson::BsonDocument person_document_copy = bson::BsonDocument(person_document);
        person_document_copy["dogs"] = bson::BsonArray({dog_object_id});
        person_collection.update_one(person_document, person_document, true,
                                     [&](MongoCollection::UpdateResult, Optional<AppError> error) {
                                         REQUIRE_FALSE(error);
                                         processed = true;
                                     });

        CHECK(processed);
    }

    SECTION("update many") {
        bool processed = false;

        dog_collection.insert_one(dog_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
        });

        dog_collection.update_many(dog_document2, dog_document, true,
                                   [&](MongoCollection::UpdateResult result, Optional<AppError> error) {
                                       REQUIRE_FALSE(error);
                                       CHECK((*result.upserted_id).to_string() != "");
                                   });

        dog_collection.update_many(dog_document2, dog_document,
                                   [&](MongoCollection::UpdateResult result, Optional<AppError> error) {
                                       REQUIRE_FALSE(error);
                                       CHECK(!result.upserted_id);
                                       processed = true;
                                   });

        CHECK(processed);
    }

    SECTION("find and replace") {
        bool processed = false;
        ObjectId dog_object_id;
        ObjectId person_object_id;

        MongoCollection::FindOneAndModifyOptions find_and_modify_options{
            Optional<bson::BsonDocument>({{"name", "fido"}}), // project
            Optional<bson::BsonDocument>({{"name", 1}}),      // sort,
            true,                                             // upsert
            true                                              // return new doc
        };

        dog_collection.find_one_and_replace(dog_document, dog_document2,
                                            [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                                REQUIRE_FALSE(error);
                                                CHECK(!document);
                                            });

        dog_collection.insert_one(dog_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
            dog_object_id = static_cast<ObjectId>(*object_id);
        });

        dog_collection.find_one_and_replace(dog_document, dog_document2,
                                            [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                                REQUIRE_FALSE(error);
                                                auto name = static_cast<std::string>((*document)["name"]);
                                                CHECK(name == "fido");
                                            });

        dog_collection.find_one_and_replace(dog_document2, dog_document, find_and_modify_options,
                                            [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                                REQUIRE_FALSE(error);
                                                auto name = static_cast<std::string>((*document)["name"]);
                                                CHECK(static_cast<std::string>(name) == "fido");
                                            });

        person_document["dogs"] = bson::BsonArray({dog_object_id});
        person_document2["dogs"] = bson::BsonArray({dog_object_id});
        person_collection.insert_one(person_document, [&](Optional<bson::Bson> object_id, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK((*object_id).to_string() != "");
            person_object_id = static_cast<ObjectId>(*object_id);
        });

        MongoCollection::FindOneAndModifyOptions person_find_and_modify_options{
            Optional<bson::BsonDocument>({{"firstName", 1}}), // project
            Optional<bson::BsonDocument>({{"firstName", 1}}), // sort,
            false,                                            // upsert
            true                                              // return new doc
        };

        person_collection.find_one_and_replace(person_document, person_document2,
                                               [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                                   REQUIRE_FALSE(error);
                                                   auto name = static_cast<std::string>((*document)["firstName"]);
                                                   // Should return the old document
                                                   CHECK(name == "John");
                                                   processed = true;
                                               });

        person_collection.find_one_and_replace(person_document2, person_document, person_find_and_modify_options,
                                               [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                                   REQUIRE_FALSE(error);
                                                   auto name = static_cast<std::string>((*document)["firstName"]);
                                                   // Should return new document, Bob -> John
                                                   CHECK(name == "John");
                                               });

        person_collection.find_one_and_replace({{"invalid", "item"}}, {{}},
                                               [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                                   // If a document is not found then null will be returned for the
                                                   // document and no error will be returned
                                                   REQUIRE_FALSE(error);
                                                   CHECK(!document);
                                               });

        person_collection.find_one_and_replace({{"invalid", "item"}}, {{}}, person_find_and_modify_options,
                                               [&](Optional<bson::BsonDocument> document, Optional<AppError> error) {
                                                   REQUIRE_FALSE(error);
                                                   CHECK(!document);
                                                   processed = true;
                                               });

        CHECK(processed);
    }

    SECTION("delete") {

        bool processed = false;

        bson::BsonArray documents;
        documents.push_back(dog_document);
        documents.push_back(dog_document);
        documents.push_back(dog_document);

        dog_collection.insert_many(documents, [&](bson::BsonArray inserted_docs, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(inserted_docs.size() == 3);
        });

        MongoCollection::FindOneAndModifyOptions find_and_modify_options{
            Optional<bson::BsonDocument>({{"name", "fido"}}), // project
            Optional<bson::BsonDocument>({{"name", 1}}),      // sort,
            true,                                             // upsert
            true                                              // return new doc
        };

        dog_collection.delete_one(dog_document, [&](uint64_t deleted_count, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(deleted_count >= 1);
        });

        dog_collection.delete_many(dog_document, [&](uint64_t deleted_count, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(deleted_count >= 1);
            processed = true;
        });

        person_collection.delete_many_bson(person_document, [&](Optional<bson::Bson> bson, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(static_cast<int32_t>(static_cast<bson::BsonDocument>(*bson)["deletedCount"]) >= 1);
            processed = true;
        });

        CHECK(processed);
    }
}

// MARK: - Push Notifications Tests

TEST_CASE("app: push notifications", "[sync][app][notifications][baas]") {
    TestAppSession session;
    auto app = session.app();
    std::shared_ptr<User> sync_user = app->current_user();

    SECTION("register") {
        bool processed;

        app->push_notification_client("gcm").register_device("hello", sync_user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
            processed = true;
        });

        CHECK(processed);
    }
    /*
        // FIXME: It seems this test fails when the two register_device calls are invoked too quickly,
        // The error returned will be 'Device not found' on the second register_device call.
        SECTION("register twice") {
            // registering the same device twice should not result in an error
            bool processed;

            app->push_notification_client("gcm").register_device("hello",
                                                                 sync_user,
                                                                 [&](Optional<AppError> error) {
                REQUIRE_FALSE(error);
            });

            app->push_notification_client("gcm").register_device("hello",
                                                                 sync_user,
                                                                 [&](Optional<AppError> error) {
                REQUIRE_FALSE(error);
                processed = true;
            });

            CHECK(processed);
        }
    */
    SECTION("deregister") {
        bool processed;

        app->push_notification_client("gcm").deregister_device(sync_user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("register with unavailable service") {
        bool processed;

        app->push_notification_client("gcm_blah").register_device("hello", sync_user, [&](Optional<AppError> error) {
            REQUIRE(error);
            CHECK(error->reason() == "service not found: 'gcm_blah'");
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("register with logged out user") {
        bool processed;

        app->log_out([=](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        app->push_notification_client("gcm").register_device("hello", sync_user, [&](Optional<AppError> error) {
            REQUIRE(error);
            processed = true;
        });

        app->push_notification_client("gcm").register_device("hello", nullptr, [&](Optional<AppError> error) {
            REQUIRE(error);
            processed = true;
        });

        CHECK(processed);
    }
}

// MARK: - Token refresh

TEST_CASE("app: token refresh", "[sync][app][token][baas]") {
    TestAppSession session;
    auto app = session.app();
    std::shared_ptr<User> sync_user = app->current_user();
    sync_user->update_data_for_testing([](UserData& data) {
        data.access_token = RealmJWT(ENCODE_FAKE_JWT("fake_access_token"));
    });

    auto remote_client = app->current_user()->mongo_client("BackingDB");
    auto app_session = get_runtime_app_session();
    auto db = remote_client.db(app_session.config.mongo_dbname);
    auto dog_collection = db["Dog"];
    bson::BsonDocument dog_document{{"name", "fido"}, {"breed", "king charles"}};

    SECTION("access token should refresh") {
        /*
         Expected sequence of events:
         - `find_one` tries to hit the server with a bad access token
         - Server returns an error because of the bad token, error should be something like:
            {\"error\":\"json: cannot unmarshal array into Go value of type map[string]interface
         {}\",\"link\":\"http://localhost:9090/groups/5f84167e776aa0f9dc27081a/apps/5f841686776aa0f9dc270876/logs?co_id=5f844c8c776aa0f9dc273db6\"}
            http_status_code = 401
            custom_status_code = 0
         - App::handle_auth_failure is then called and an attempt to refresh the access token will be peformed.
         - If the token refresh was successful, the original request will retry and we should expect no error in the
         callback of `find_one`
         */
        dog_collection.find_one(dog_document, [&](Optional<bson::BsonDocument>, Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });
    }
}

// MARK: - Sync Tests

TEST_CASE("app: mixed lists with object links", "[sync][pbs][app][links][baas]") {
    const std::string valid_pk_name = "_id";

    Schema schema{
        {"TopLevel",
         {
             {valid_pk_name, PropertyType::ObjectId, Property::IsPrimary{true}},
             {"mixed_array", PropertyType::Mixed | PropertyType::Array | PropertyType::Nullable},
         }},
        {"Target",
         {
             {valid_pk_name, PropertyType::ObjectId, Property::IsPrimary{true}},
             {"value", PropertyType::Int},
         }},
    };

    auto server_app_config = minimal_app_config("set_new_embedded_object", schema);
    auto app_session = create_app(server_app_config);
    auto partition = random_string(100);

    auto obj_id = ObjectId::gen();
    auto target_id = ObjectId::gen();
    auto mixed_list_values = AnyVector{
        Mixed{int64_t(1234)},
        Mixed{},
        Mixed{target_id},
    };
    {
        TestAppSession test_session(app_session, nullptr, DeleteApp{false});
        SyncTestFile config(test_session.app()->current_user(), partition, schema);
        auto realm = Realm::get_shared_realm(config);

        CppContext c(realm);
        realm->begin_transaction();
        auto target_obj = Object::create(
            c, realm, "Target", std::any(AnyDict{{valid_pk_name, target_id}, {"value", static_cast<int64_t>(1234)}}));
        mixed_list_values.push_back(Mixed(target_obj.get_obj().get_link()));

        Object::create(c, realm, "TopLevel",
                       std::any(AnyDict{
                           {valid_pk_name, obj_id},
                           {"mixed_array", mixed_list_values},
                       }),
                       CreatePolicy::ForceCreate);
        realm->commit_transaction();
        CHECK(!wait_for_upload(*realm));
    }

    {
        TestAppSession test_session(app_session);
        SyncTestFile config(test_session.app()->current_user(), partition, schema);
        auto realm = Realm::get_shared_realm(config);

        CHECK(!wait_for_download(*realm));
        CppContext c(realm);
        auto obj = Object::get_for_primary_key(c, realm, "TopLevel", std::any{obj_id});
        auto list = util::any_cast<List&&>(obj.get_property_value<std::any>(c, "mixed_array"));
        for (size_t idx = 0; idx < list.size(); ++idx) {
            Mixed mixed = list.get_any(idx);
            if (idx == 3) {
                CHECK(mixed.is_type(type_TypedLink));
                auto link = mixed.get<ObjLink>();
                auto link_table = realm->read_group().get_table(link.get_table_key());
                CHECK(link_table->get_name() == "class_Target");
                auto link_obj = link_table->get_object(link.get_obj_key());
                CHECK(link_obj.get_primary_key() == target_id);
            }
            else {
                CHECK(mixed == util::any_cast<Mixed>(mixed_list_values[idx]));
            }
        }
    }
}

TEST_CASE("app: roundtrip values", "[sync][pbs][app][baas]") {
    const std::string valid_pk_name = "_id";

    Schema schema{
        {"TopLevel",
         {
             {valid_pk_name, PropertyType::ObjectId, Property::IsPrimary{true}},
             {"decimal", PropertyType::Decimal | PropertyType::Nullable},
         }},
    };

    auto server_app_config = minimal_app_config("roundtrip_values", schema);
    auto app_session = create_app(server_app_config);
    auto partition = random_string(100);

    Decimal128 large_significand = Decimal128(70) / Decimal128(1.09);
    auto obj_id = ObjectId::gen();
    {
        TestAppSession test_session(app_session, nullptr, DeleteApp{false});
        SyncTestFile config(test_session.app()->current_user(), partition, schema);
        auto realm = Realm::get_shared_realm(config);

        CppContext c(realm);
        realm->begin_transaction();
        Object::create(c, realm, "TopLevel",
                       util::Any(AnyDict{
                           {valid_pk_name, obj_id},
                           {"decimal", large_significand},
                       }),
                       CreatePolicy::ForceCreate);
        realm->commit_transaction();
        CHECK(!wait_for_upload(*realm, std::chrono::seconds(600)));
    }

    {
        TestAppSession test_session(app_session);
        SyncTestFile config(test_session.app()->current_user(), partition, schema);
        auto realm = Realm::get_shared_realm(config);

        CHECK(!wait_for_download(*realm));
        CppContext c(realm);
        auto obj = Object::get_for_primary_key(c, realm, "TopLevel", util::Any{obj_id});
        auto val = obj.get_column_value<Decimal128>("decimal");
        CHECK(val == large_significand);
    }
}

TEST_CASE("app: upgrade from local to synced realm", "[sync][pbs][app][upgrade][baas]") {
    const std::string valid_pk_name = "_id";

    Schema schema{
        {"origin",
         {{valid_pk_name, PropertyType::Int, Property::IsPrimary{true}},
          {"link", PropertyType::Object | PropertyType::Nullable, "target"},
          {"embedded_link", PropertyType::Object | PropertyType::Nullable, "embedded"}}},
        {"target",
         {{valid_pk_name, PropertyType::String, Property::IsPrimary{true}},
          {"value", PropertyType::Int},
          {"name", PropertyType::String}}},
        {"other_origin",
         {{valid_pk_name, PropertyType::ObjectId, Property::IsPrimary{true}},
          {"array", PropertyType::Array | PropertyType::Object, "other_target"}}},
        {"other_target",
         {{valid_pk_name, PropertyType::UUID, Property::IsPrimary{true}}, {"value", PropertyType::Int}}},
        {"embedded", ObjectSchema::ObjectType::Embedded, {{"name", PropertyType::String | PropertyType::Nullable}}},
    };

    /*             Create local realm             */
    TestFile local_config;
    local_config.schema = schema;
    auto local_realm = Realm::get_shared_realm(local_config);
    {
        auto origin = local_realm->read_group().get_table("class_origin");
        auto target = local_realm->read_group().get_table("class_target");
        auto other_origin = local_realm->read_group().get_table("class_other_origin");
        auto other_target = local_realm->read_group().get_table("class_other_target");

        local_realm->begin_transaction();
        auto o = target->create_object_with_primary_key("Foo").set("name", "Egon");
        // 'embedded_link' property is null.
        origin->create_object_with_primary_key(47).set("link", o.get_key());
        // 'embedded_link' property is not null.
        auto obj = origin->create_object_with_primary_key(42);
        auto col_key = origin->get_column_key("embedded_link");
        obj.create_and_set_linked_object(col_key);
        other_target->create_object_with_primary_key(UUID("3b241101-e2bb-4255-8caf-4136c566a961"));
        other_origin->create_object_with_primary_key(ObjectId::gen());
        local_realm->commit_transaction();
    }

    /* Create a synced realm and upload some data */
    auto server_app_config = minimal_app_config("upgrade_from_local", schema);
    TestAppSession test_session(create_app(server_app_config));
    auto partition = random_string(100);
    auto user1 = test_session.app()->current_user();
    SyncTestFile config1(user1, partition, schema);

    auto r1 = Realm::get_shared_realm(config1);

    auto origin = r1->read_group().get_table("class_origin");
    auto target = r1->read_group().get_table("class_target");
    auto other_origin = r1->read_group().get_table("class_other_origin");
    auto other_target = r1->read_group().get_table("class_other_target");

    r1->begin_transaction();
    auto o = target->create_object_with_primary_key("Baa").set("name", "Børge");
    origin->create_object_with_primary_key(47).set("link", o.get_key());
    other_target->create_object_with_primary_key(UUID("01234567-89ab-cdef-edcb-a98765432101"));
    other_origin->create_object_with_primary_key(ObjectId::gen());
    r1->commit_transaction();
    CHECK(!wait_for_upload(*r1));

    /* Copy local realm data over in a synced one*/
    create_user_and_log_in(test_session.app());
    auto user2 = test_session.app()->current_user();
    REQUIRE(user1 != user2);

    SyncTestFile config2(user1, partition, schema);

    SharedRealm r2;
    SECTION("Copy before connecting to server") {
        local_realm->convert(config2);
        r2 = Realm::get_shared_realm(config2);
    }

    SECTION("Open synced realm first") {
        r2 = Realm::get_shared_realm(config2);
        CHECK(!wait_for_download(*r2));
        local_realm->convert(config2);
        CHECK(!wait_for_upload(*r2));
    }

    CHECK(!wait_for_download(*r2));
    advance_and_notify(*r2);
    Group& g = r2->read_group();
    // g.to_json(std::cout);
    REQUIRE(g.get_table("class_origin")->size() == 2);
    REQUIRE(g.get_table("class_target")->size() == 2);
    REQUIRE(g.get_table("class_other_origin")->size() == 2);
    REQUIRE(g.get_table("class_other_target")->size() == 2);

    CHECK(!wait_for_upload(*r2));
    CHECK(!wait_for_download(*r1));
    advance_and_notify(*r1);
    // r1->read_group().to_json(std::cout);
}

TEST_CASE("app: set new embedded object", "[sync][pbs][app][baas]") {
    const std::string valid_pk_name = "_id";

    Schema schema{
        {"TopLevel",
         {
             {valid_pk_name, PropertyType::ObjectId, Property::IsPrimary{true}},
             {"array_of_objs", PropertyType::Object | PropertyType::Array, "TopLevel_array_of_objs"},
             {"embedded_obj", PropertyType::Object | PropertyType::Nullable, "TopLevel_embedded_obj"},
             {"embedded_dict", PropertyType::Object | PropertyType::Dictionary | PropertyType::Nullable,
              "TopLevel_embedded_dict"},
         }},
        {"TopLevel_array_of_objs",
         ObjectSchema::ObjectType::Embedded,
         {
             {"array", PropertyType::Int | PropertyType::Array},
         }},
        {"TopLevel_embedded_obj",
         ObjectSchema::ObjectType::Embedded,
         {
             {"array", PropertyType::Int | PropertyType::Array},
         }},
        {"TopLevel_embedded_dict",
         ObjectSchema::ObjectType::Embedded,
         {
             {"array", PropertyType::Int | PropertyType::Array},
         }},
    };

    auto server_app_config = minimal_app_config("set_new_embedded_object", schema);
    TestAppSession test_session(create_app(server_app_config));
    auto partition = random_string(100);

    auto array_of_objs_id = ObjectId::gen();
    auto embedded_obj_id = ObjectId::gen();
    auto dict_obj_id = ObjectId::gen();

    {
        SyncTestFile config(test_session.app()->current_user(), partition, schema);
        auto realm = Realm::get_shared_realm(config);

        CppContext c(realm);
        realm->begin_transaction();
        auto array_of_objs =
            Object::create(c, realm, "TopLevel",
                           std::any(AnyDict{
                               {valid_pk_name, array_of_objs_id},
                               {"array_of_objs", AnyVector{AnyDict{{"array", AnyVector{INT64_C(1), INT64_C(2)}}}}},
                           }),
                           CreatePolicy::ForceCreate);

        auto embedded_obj =
            Object::create(c, realm, "TopLevel",
                           std::any(AnyDict{
                               {valid_pk_name, embedded_obj_id},
                               {"embedded_obj", AnyDict{{"array", AnyVector{INT64_C(1), INT64_C(2)}}}},
                           }),
                           CreatePolicy::ForceCreate);

        auto dict_obj = Object::create(
            c, realm, "TopLevel",
            std::any(AnyDict{
                {valid_pk_name, dict_obj_id},
                {"embedded_dict", AnyDict{{"foo", AnyDict{{"array", AnyVector{INT64_C(1), INT64_C(2)}}}}}},
            }),
            CreatePolicy::ForceCreate);

        realm->commit_transaction();
        {
            realm->begin_transaction();
            embedded_obj.set_property_value(c, "embedded_obj",
                                            std::any(AnyDict{{
                                                "array",
                                                AnyVector{INT64_C(3), INT64_C(4)},
                                            }}),
                                            CreatePolicy::UpdateAll);
            realm->commit_transaction();
        }

        {
            realm->begin_transaction();
            List array(array_of_objs, array_of_objs.get_object_schema().property_for_name("array_of_objs"));
            CppContext c2(realm, &array.get_object_schema());
            array.set(c2, 0, std::any{AnyDict{{"array", AnyVector{INT64_C(5), INT64_C(6)}}}});
            realm->commit_transaction();
        }

        {
            realm->begin_transaction();
            object_store::Dictionary dict(dict_obj, dict_obj.get_object_schema().property_for_name("embedded_dict"));
            CppContext c2(realm, &dict.get_object_schema());
            dict.insert(c2, "foo", std::any{AnyDict{{"array", AnyVector{INT64_C(7), INT64_C(8)}}}});
            realm->commit_transaction();
        }
        CHECK(!wait_for_upload(*realm));
    }

    {
        SyncTestFile config(test_session.app()->current_user(), partition, schema);
        auto realm = Realm::get_shared_realm(config);

        CHECK(!wait_for_download(*realm));
        CppContext c(realm);
        {
            auto obj = Object::get_for_primary_key(c, realm, "TopLevel", std::any{embedded_obj_id});
            auto embedded_obj = util::any_cast<Object&&>(obj.get_property_value<std::any>(c, "embedded_obj"));
            auto array_list = util::any_cast<List&&>(embedded_obj.get_property_value<std::any>(c, "array"));
            CHECK(array_list.size() == 2);
            CHECK(array_list.get<int64_t>(0) == int64_t(3));
            CHECK(array_list.get<int64_t>(1) == int64_t(4));
        }

        {
            auto obj = Object::get_for_primary_key(c, realm, "TopLevel", std::any{array_of_objs_id});
            auto embedded_list = util::any_cast<List&&>(obj.get_property_value<std::any>(c, "array_of_objs"));
            CppContext c2(realm, &embedded_list.get_object_schema());
            auto embedded_array_obj = util::any_cast<Object&&>(embedded_list.get(c2, 0));
            auto array_list = util::any_cast<List&&>(embedded_array_obj.get_property_value<std::any>(c2, "array"));
            CHECK(array_list.size() == 2);
            CHECK(array_list.get<int64_t>(0) == int64_t(5));
            CHECK(array_list.get<int64_t>(1) == int64_t(6));
        }

        {
            auto obj = Object::get_for_primary_key(c, realm, "TopLevel", std::any{dict_obj_id});
            object_store::Dictionary dict(obj, obj.get_object_schema().property_for_name("embedded_dict"));
            CppContext c2(realm, &dict.get_object_schema());
            auto embedded_obj = util::any_cast<Object&&>(dict.get(c2, "foo"));
            auto array_list = util::any_cast<List&&>(embedded_obj.get_property_value<std::any>(c2, "array"));
            CHECK(array_list.size() == 2);
            CHECK(array_list.get<int64_t>(0) == int64_t(7));
            CHECK(array_list.get<int64_t>(1) == int64_t(8));
        }
    }
}

TEST_CASE("app: make distributable client file", "[sync][pbs][app][baas]") {
    TestAppSession session;
    auto app = session.app();

    auto schema = get_default_schema();
    SyncTestFile original_config(app->current_user(), bson::Bson("foo"), schema);
    create_user_and_log_in(app);
    SyncTestFile target_config(app->current_user(), bson::Bson("foo"), schema);

    // Create realm file without client file id
    {
        auto realm = Realm::get_shared_realm(original_config);

        // Write some data
        realm->begin_transaction();
        CppContext c;
        Object::create(c, realm, "Person",
                       std::any(realm::AnyDict{{"_id", std::any(ObjectId::gen())},
                                               {"age", INT64_C(64)},
                                               {"firstName", std::string("Paul")},
                                               {"lastName", std::string("McCartney")}}));
        realm->commit_transaction();
        wait_for_upload(*realm);
        wait_for_download(*realm);

        realm->convert(target_config);

        // Write some additional data
        realm->begin_transaction();
        Object::create(c, realm, "Dog",
                       std::any(realm::AnyDict{{"_id", std::any(ObjectId::gen())},
                                               {"breed", std::string("stabyhoun")},
                                               {"name", std::string("albert")},
                                               {"realm_id", std::string("foo")}}));
        realm->commit_transaction();
        wait_for_upload(*realm);
    }
    // Starting a new session based on the copy
    {
        auto realm = Realm::get_shared_realm(target_config);
        REQUIRE(realm->read_group().get_table("class_Person")->size() == 1);
        REQUIRE(realm->read_group().get_table("class_Dog")->size() == 0);

        // Should be able to download the object created in the source Realm
        // after writing the copy
        wait_for_download(*realm);
        realm->refresh();
        REQUIRE(realm->read_group().get_table("class_Person")->size() == 1);
        REQUIRE(realm->read_group().get_table("class_Dog")->size() == 1);

        // Check that we can continue committing to this realm
        realm->begin_transaction();
        CppContext c;
        Object::create(c, realm, "Dog",
                       std::any(realm::AnyDict{{"_id", std::any(ObjectId::gen())},
                                               {"breed", std::string("bulldog")},
                                               {"name", std::string("fido")},
                                               {"realm_id", std::string("foo")}}));
        realm->commit_transaction();
        wait_for_upload(*realm);
    }
    // Original Realm should be able to read the object which was written to the copy
    {
        auto realm = Realm::get_shared_realm(original_config);
        REQUIRE(realm->read_group().get_table("class_Person")->size() == 1);
        REQUIRE(realm->read_group().get_table("class_Dog")->size() == 1);

        wait_for_download(*realm);
        realm->refresh();
        REQUIRE(realm->read_group().get_table("class_Person")->size() == 1);
        REQUIRE(realm->read_group().get_table("class_Dog")->size() == 2);
    }
}

TEST_CASE("app: sync integration", "[sync][pbs][app][baas]") {
    auto logger = util::Logger::get_default_logger();

    const auto schema = get_default_schema();

    auto get_dogs = [](SharedRealm r) -> Results {
        wait_for_upload(*r, std::chrono::seconds(10));
        wait_for_download(*r, std::chrono::seconds(10));
        return Results(r, r->read_group().get_table("class_Dog"));
    };

    auto create_one_dog = [](SharedRealm r) {
        r->begin_transaction();
        CppContext c;
        Object::create(c, r, "Dog",
                       std::any(AnyDict{{"_id", std::any(ObjectId::gen())},
                                        {"breed", std::string("bulldog")},
                                        {"name", std::string("fido")}}),
                       CreatePolicy::ForceCreate);
        r->commit_transaction();
    };

    TestAppSession session;
    auto app = session.app();
    const auto partition = random_string(100);

    SECTION("Add Objects") {
        {
            SyncTestFile config(app->current_user(), partition, schema);
            auto r = Realm::get_shared_realm(config);

            REQUIRE(get_dogs(r).size() == 0);
            create_one_dog(r);
            REQUIRE(get_dogs(r).size() == 1);
        }

        {
            create_user_and_log_in(app);
            SyncTestFile config(app->current_user(), partition, schema);
            auto r = Realm::get_shared_realm(config);
            Results dogs = get_dogs(r);
            REQUIRE(dogs.size() == 1);
            REQUIRE(dogs.get(0).get<String>("breed") == "bulldog");
            REQUIRE(dogs.get(0).get<String>("name") == "fido");
        }
    }

    SECTION("MemOnly durability") {
        {
            SyncTestFile config(app->current_user(), partition, schema);
            config.in_memory = true;
            config.encryption_key = std::vector<char>();

            REQUIRE(config.options().durability == DBOptions::Durability::MemOnly);
            auto r = Realm::get_shared_realm(config);

            REQUIRE(get_dogs(r).size() == 0);
            create_one_dog(r);
            REQUIRE(get_dogs(r).size() == 1);
        }

        {
            create_user_and_log_in(app);
            SyncTestFile config(app->current_user(), partition, schema);
            config.in_memory = true;
            config.encryption_key = std::vector<char>();
            auto r = Realm::get_shared_realm(config);
            Results dogs = get_dogs(r);
            REQUIRE(dogs.size() == 1);
            REQUIRE(dogs.get(0).get<String>("breed") == "bulldog");
            REQUIRE(dogs.get(0).get<String>("name") == "fido");
        }
    }

    SECTION("Fast clock on client") {
        {
            SyncTestFile config(app->current_user(), partition, schema);
            auto r = Realm::get_shared_realm(config);

            REQUIRE(get_dogs(r).size() == 0);
            create_one_dog(r);
            REQUIRE(get_dogs(r).size() == 1);
        }

        auto transport = std::make_shared<HookedTransport<>>();
        TestAppSession hooked_session(session.app_session(), transport, DeleteApp{false});
        auto app = hooked_session.app();
        std::shared_ptr<User> user = app->current_user();
        REQUIRE(user);
        REQUIRE(!user->access_token_refresh_required());
        // Make the User behave as if the client clock is 31 minutes fast, so the token looks expired locally
        // (access tokens have an lifetime of 30 minutes today).
        user->set_seconds_to_adjust_time_for_testing(31 * 60);
        REQUIRE(user->access_token_refresh_required());

        // This assumes that we make an http request for the new token while
        // already in the WaitingForAccessToken state.
        bool seen_waiting_for_access_token = false;
        transport->request_hook = [&](const Request&) -> std::optional<Response> {
            auto user = app->current_user();
            REQUIRE(user);
            for (auto& session : app->sync_manager()->get_all_sessions_for(*user)) {
                // Prior to the fix for #4941, this callback would be called from an infinite loop, always in the
                // WaitingForAccessToken state.
                if (session->state() == SyncSession::State::WaitingForAccessToken) {
                    REQUIRE(!seen_waiting_for_access_token);
                    seen_waiting_for_access_token = true;
                }
            }
            return std::nullopt;
        };
        SyncTestFile config(user, partition, schema);
        auto r = Realm::get_shared_realm(config);
        REQUIRE(seen_waiting_for_access_token);
        Results dogs = get_dogs(r);
        REQUIRE(dogs.size() == 1);
        REQUIRE(dogs.get(0).get<String>("breed") == "bulldog");
        REQUIRE(dogs.get(0).get<String>("name") == "fido");
    }

    SECTION("Expired Tokens") {
        sync::AccessToken token;
        {
            std::shared_ptr<User> user = app->current_user();
            SyncTestFile config(user, partition, schema);
            auto r = Realm::get_shared_realm(config);

            REQUIRE(get_dogs(r).size() == 0);
            create_one_dog(r);

            REQUIRE(get_dogs(r).size() == 1);
            sync::AccessToken::ParseError error_state = realm::sync::AccessToken::ParseError::none;
            sync::AccessToken::parse(user->access_token(), token, error_state, nullptr);
            REQUIRE(error_state == sync::AccessToken::ParseError::none);
            REQUIRE(token.timestamp);
            REQUIRE(token.expires);
            REQUIRE(token.timestamp < token.expires);
            std::chrono::system_clock::time_point now = std::chrono::system_clock::now();
            token.expires = std::chrono::system_clock::to_time_t(now - 30s);
            REQUIRE(token.expired(now));
        }

        auto transport = std::make_shared<HookedTransport<>>();
        TestAppSession hooked_session(session.app_session(), transport, DeleteApp{false});
        auto app = hooked_session.app();
        std::shared_ptr<User> user = app->current_user();
        REQUIRE(user);
        REQUIRE(!user->access_token_refresh_required());
        // Set a bad access token, with an expired time. This will trigger a refresh initiated by the client.
        user->update_data_for_testing([&token](UserData& data) {
            data.access_token = RealmJWT(encode_fake_jwt("fake_access_token", token.expires, token.timestamp));
        });
        REQUIRE(user->access_token_refresh_required());

        SECTION("Expired Access Token is Refreshed") {
            // This assumes that we make an http request for the new token while
            // already in the WaitingForAccessToken state.
            bool seen_waiting_for_access_token = false;
            transport->request_hook = [&](const Request&) -> std::optional<Response> {
                auto user = app->current_user();
                REQUIRE(user);
                for (auto& session : app->sync_manager()->get_all_sessions_for(*user)) {
                    if (session->state() == SyncSession::State::WaitingForAccessToken) {
                        REQUIRE(!seen_waiting_for_access_token);
                        seen_waiting_for_access_token = true;
                    }
                }
                return std::nullopt;
            };
            SyncTestFile config(user, partition, schema);
            auto r = Realm::get_shared_realm(config);
            REQUIRE(seen_waiting_for_access_token);
            Results dogs = get_dogs(r);
            REQUIRE(dogs.size() == 1);
            REQUIRE(dogs.get(0).get<String>("breed") == "bulldog");
            REQUIRE(dogs.get(0).get<String>("name") == "fido");
        }

        SECTION("User is logged out if the refresh request is denied") {
            REQUIRE(user->is_logged_in());
            size_t hook_count = 0;
            transport->response_hook = [&](const Request& request, Response& response) {
                auto user = app->current_user();
                if (hook_count++ == 0) {
                    // the initial request should have a current user and log it out
                    REQUIRE(user);
                    REQUIRE(user->is_logged_in());
                }
                else {
                    INFO(request.url);
                    // any later requests (eg. redirect) won't have a current user
                    REQUIRE(!user);
                }
                // simulate the server denying the refresh
                if (request.url.find("/session") != std::string::npos) {
                    response.http_status_code = 401;
                    response.body = "fake: refresh token could not be refreshed";
                }
            };
            SyncTestFile config(user, partition, schema);
            std::atomic<bool> sync_error_handler_called{false};
            config.sync_config->error_handler = [&](std::shared_ptr<SyncSession>, SyncError error) {
                sync_error_handler_called.store(true);
                REQUIRE(error.status.code() == ErrorCodes::AuthError);
                REQUIRE_THAT(std::string{error.status.reason()},
                             Catch::Matchers::StartsWith("Unable to refresh the user access token"));
            };
            auto r = Realm::get_shared_realm(config);
            timed_wait_for([&] {
                return sync_error_handler_called.load();
            });
            // the failed refresh logs out the user
            REQUIRE(!user->is_logged_in());
        }

        SECTION("User is left logged out if logged out while the refresh is in progress") {
            REQUIRE(user->is_logged_in());
            transport->request_hook = [&](const Request&) -> std::optional<Response> {
                user->log_out();
                return std::nullopt;
            };
            SyncTestFile config(user, partition, schema);
            std::atomic<bool> sync_error_handler_called{false};
            config.sync_config->error_handler = [&](std::shared_ptr<SyncSession>, SyncError error) {
                sync_error_handler_called.store(true);
                REQUIRE(error.status.code() == ErrorCodes::AuthError);
                REQUIRE_THAT(std::string{error.status.reason()},
                             Catch::Matchers::StartsWith("Unable to refresh the user access token"));
            };
            auto r = Realm::get_shared_realm(config);
            timed_wait_for([&] {
                return sync_error_handler_called.load();
            });
            REQUIRE_FALSE(user->is_logged_in());
            REQUIRE(user->state() == SyncUser::State::LoggedOut);
        }

        SECTION("Requests that receive an error are retried on a backoff") {
            using namespace std::chrono;
            std::vector<time_point<steady_clock>> response_times;
            std::atomic<bool> did_receive_valid_token{false};
            constexpr size_t num_error_responses = 6;

            transport->response_hook = [&](const Request& request, Response& response) {
                // simulate the server experiencing an internal server error
                if (request.url.find("/session") != std::string::npos) {
                    if (response_times.size() >= num_error_responses) {
                        did_receive_valid_token.store(true);
                        return;
                    }
                    response.http_status_code = 500;
                }
            };
            transport->request_hook = [&](const Request& request) -> std::optional<Response> {
                if (!did_receive_valid_token.load() && request.url.find("/session") != std::string::npos) {
                    response_times.push_back(steady_clock::now());
                }
                return std::nullopt;
            };
            SyncTestFile config(user, partition, schema);
            config.sync_config->error_handler = [&](std::shared_ptr<SyncSession>, SyncError error) {
                REQUIRE(error.status.code() == ErrorCodes::AuthError);
                REQUIRE_THAT(std::string{error.status.reason()},
                             Catch::Matchers::StartsWith("Unable to refresh the user access token"));
            };
            auto r = Realm::get_shared_realm(config);
            create_one_dog(r);
            timed_wait_for(
                [&] {
                    return did_receive_valid_token.load();
                },
                30s);
            REQUIRE(user->is_logged_in());
            REQUIRE(response_times.size() >= num_error_responses);
            std::vector<uint64_t> delay_times;
            for (size_t i = 1; i < response_times.size(); ++i) {
                delay_times.push_back(duration_cast<milliseconds>(response_times[i] - response_times[i - 1]).count());
            }

            // sync delays start at 1000ms minus a random number of up to 25%.
            // the subsequent delay is double the previous one minus a random 25% again.
            // this calculation happens in Connection::initiate_reconnect_wait()
            bool increasing_delay = true;
            for (size_t i = 1; i < delay_times.size(); ++i) {
                if (delay_times[i - 1] >= delay_times[i]) {
                    increasing_delay = false;
                }
            }
            // fail if the first delay isn't longer than half a second
            if (delay_times.size() <= 1 || delay_times[1] < 500) {
                increasing_delay = false;
            }
            if (!increasing_delay) {
                std::cerr << "delay times are not increasing: ";
                for (auto& delay : delay_times) {
                    std::cerr << delay << ", ";
                }
                std::cerr << std::endl;
            }
            REQUIRE(increasing_delay);
        }
    }

    SECTION("Invalid refresh token") {
        auto& app_session = session.app_session();
        std::mutex mtx;
        auto verify_error_on_sync_with_invalid_refresh_token = [&](std::shared_ptr<User> user, Realm::Config config) {
            REQUIRE(user);
            REQUIRE(app_session.admin_api.verify_access_token(user->access_token(), app_session.server_app_id));

            // requesting a new access token fails because the refresh token used for this request is revoked
            user->refresh_custom_data([&](Optional<AppError> error) {
                REQUIRE(error);
                REQUIRE(error->additional_status_code == 401);
                REQUIRE(error->code() == ErrorCodes::InvalidSession);
            });

            // Set a bad access token. This will force a request for a new access token when the sync session opens
            // this is only necessary because the server doesn't actually revoke previously issued access tokens
            // instead allowing their session to time out as normal. So this simulates the access token expiring.
            // see:
            // https://github.com/10gen/baas/blob/05837cc3753218dfaf89229c6930277ef1616402/api/common/auth.go#L1380-L1386
            user->update_data_for_testing([](UserData& data) {
                data.access_token = RealmJWT(encode_fake_jwt("fake_access_token"));
            });
            REQUIRE(!app_session.admin_api.verify_access_token(user->access_token(), app_session.server_app_id));

            auto [sync_error_promise, sync_error] = util::make_promise_future<SyncError>();
            config.sync_config->error_handler =
                [promise = util::CopyablePromiseHolder(std::move(sync_error_promise))](std::shared_ptr<SyncSession>,
                                                                                       SyncError error) mutable {
                    promise.get_promise().emplace_value(std::move(error));
                };

            auto transport = static_cast<SynchronousTestTransport*>(session.transport());
            transport->block(); // don't let the token refresh happen until we're ready for it
            auto r = Realm::get_shared_realm(config);
            auto session = app->sync_manager()->get_existing_session(config.path);
            REQUIRE(user->is_logged_in());
            REQUIRE(!sync_error.is_ready());
            {
                std::atomic<bool> called{false};
                session->wait_for_upload_completion([&](Status stat) {
                    std::lock_guard lock(mtx);
                    called.store(true);
                    REQUIRE(stat.code() == ErrorCodes::InvalidSession);
                });
                transport->unblock();
                timed_wait_for([&] {
                    return called.load();
                });
                std::lock_guard lock(mtx);
                REQUIRE(called);
            }

            auto sync_error_res = wait_for_future(std::move(sync_error)).get();
            REQUIRE(sync_error_res.status == ErrorCodes::AuthError);
            REQUIRE_THAT(std::string{sync_error_res.status.reason()},
                         Catch::Matchers::StartsWith("Unable to refresh the user access token"));

            // the failed refresh logs out the user
            std::lock_guard lock(mtx);
            REQUIRE(!user->is_logged_in());
        };

        SECTION("Disabled user results in a sync error") {
            auto creds = create_user_and_log_in(app);
            auto user = app->current_user();
            REQUIRE(user);
            SyncTestFile config(user, partition, schema);
            REQUIRE(app_session.admin_api.verify_access_token(user->access_token(), app_session.server_app_id));
            app_session.admin_api.disable_user_sessions(app->current_user()->user_id(), app_session.server_app_id);

            verify_error_on_sync_with_invalid_refresh_token(user, config);

            // logging in again doesn't fix things while the account is disabled
            auto error = failed_log_in(app, creds);
            REQUIRE(error.code() == ErrorCodes::UserDisabled);

            // admin enables user sessions again which should allow the session to continue
            app_session.admin_api.enable_user_sessions(user->user_id(), app_session.server_app_id);

            // logging in now works properly
            log_in(app, creds);

            // still referencing the same user
            REQUIRE(user == app->current_user());
            REQUIRE(user->is_logged_in());

            {
                // check that there are no errors initiating a session now by making sure upload/download succeeds
                auto r = Realm::get_shared_realm(config);
                Results dogs = get_dogs(r);
            }
        }

        SECTION("Revoked refresh token results in a sync error") {
            auto creds = create_user_and_log_in(app);
            auto user = app->current_user();
            SyncTestFile config(user, partition, schema);
            REQUIRE(app_session.admin_api.verify_access_token(user->access_token(), app_session.server_app_id));
            app_session.admin_api.revoke_user_sessions(user->user_id(), app_session.server_app_id);
            // revoking a user session only affects the refresh token, so the access token should still continue to
            // work.
            REQUIRE(app_session.admin_api.verify_access_token(user->access_token(), app_session.server_app_id));

            verify_error_on_sync_with_invalid_refresh_token(user, config);

            // logging in again succeeds and generates a new and valid refresh token
            log_in(app, creds);

            // still referencing the same user and now the user is logged in
            REQUIRE(user == app->current_user());
            REQUIRE(user->is_logged_in());

            // new requests for an access token succeed again
            user->refresh_custom_data([&](Optional<AppError> error) {
                REQUIRE_FALSE(error);
            });

            {
                // check that there are no errors initiating a new sync session by making sure upload/download
                // succeeds
                auto r = Realm::get_shared_realm(config);
                Results dogs = get_dogs(r);
            }
        }

        SECTION("Revoked refresh token on an anonymous user results in a sync error") {
            app->current_user()->log_out();
            auto anon_user = log_in(app);
            REQUIRE(app->current_user() == anon_user);
            SyncTestFile config(anon_user, partition, schema);
            REQUIRE(app_session.admin_api.verify_access_token(anon_user->access_token(), app_session.server_app_id));
            app_session.admin_api.revoke_user_sessions(anon_user->user_id(), app_session.server_app_id);
            // revoking a user session only affects the refresh token, so the access token should still continue to
            // work.
            REQUIRE(app_session.admin_api.verify_access_token(anon_user->access_token(), app_session.server_app_id));

            verify_error_on_sync_with_invalid_refresh_token(anon_user, config);

            // the user has been logged out, and current user is reset
            REQUIRE(!app->current_user());
            REQUIRE(!anon_user->is_logged_in());
            REQUIRE(anon_user->state() == SyncUser::State::Removed);

            // new requests for an access token do not work for anon users
            anon_user->refresh_custom_data([&](Optional<AppError> error) {
                REQUIRE(error);
                REQUIRE(error->reason() ==
                        util::format("Cannot initiate a refresh on user '%1' because the user has been removed",
                                     anon_user->user_id()));
            });

            REQUIRE_EXCEPTION(
                Realm::get_shared_realm(config), ClientUserNotFound,
                util::format("Cannot start a sync session for user '%1' because this user has been removed.",
                             anon_user->user_id()));
        }

        SECTION("Opening a Realm with a removed email user results produces an exception") {
            auto creds = create_user_and_log_in(app);
            auto email_user = app->current_user();
            const std::string user_ident = email_user->user_id();
            REQUIRE(email_user);
            SyncTestFile config(email_user, partition, schema);
            REQUIRE(email_user->is_logged_in());
            {
                // sync works on a valid user
                auto r = Realm::get_shared_realm(config);
                Results dogs = get_dogs(r);
            }
            app->remove_user(email_user, [](util::Optional<AppError> err) {
                REQUIRE(!err);
            });
            REQUIRE_FALSE(email_user->is_logged_in());
            REQUIRE(email_user->state() == SyncUser::State::Removed);

            // should not be able to open a synced Realm with an invalid user
            REQUIRE_EXCEPTION(
                Realm::get_shared_realm(config), ClientUserNotFound,
                util::format("Cannot start a sync session for user '%1' because this user has been removed.",
                             user_ident));

            std::shared_ptr<User> new_user_instance = log_in(app, creds);
            // the previous instance is still invalid
            REQUIRE_FALSE(email_user->is_logged_in());
            REQUIRE(email_user->state() == SyncUser::State::Removed);
            // but the new instance will work and has the same server issued ident
            REQUIRE(new_user_instance);
            REQUIRE(new_user_instance->is_logged_in());
            REQUIRE(new_user_instance->user_id() == user_ident);
            {
                // sync works again if the same user is logged back in
                config.sync_config->user = new_user_instance;
                auto r = Realm::get_shared_realm(config);
                Results dogs = get_dogs(r);
            }
        }
    }

    SECTION("large write transactions which would be too large if batched") {
        SyncTestFile config(app->current_user(), partition, schema);

        std::mutex mutex;
        bool done = false;
        auto r = Realm::get_shared_realm(config);
        r->sync_session()->pause();

        // Create 26 MB worth of dogs in 26 transactions, which should work but
        // will result in an error from the server if the changesets are batched
        // for upload.
        CppContext c;
        for (auto i = 'a'; i < 'z'; ++i) {
            r->begin_transaction();
            Object::create(c, r, "Dog",
                           std::any(AnyDict{{"_id", std::any(ObjectId::gen())},
                                            {"breed", std::string("bulldog")},
                                            {"name", random_string(1024 * 1024)}}),
                           CreatePolicy::ForceCreate);
            r->commit_transaction();
        }
        r->sync_session()->wait_for_upload_completion([&](Status status) {
            std::lock_guard lk(mutex);
            REQUIRE(status.is_ok());
            done = true;
        });
        r->sync_session()->resume();

        // If we haven't gotten an error in more than 5 minutes, then something has gone wrong
        // and we should fail the test.
        timed_wait_for(
            [&] {
                std::lock_guard lk(mutex);
                return done;
            },
            std::chrono::minutes(5));
    }

    SECTION("too large sync message error handling") {
        SyncTestFile config(app->current_user(), partition, schema);

        auto pf = util::make_promise_future<SyncError>();
        config.sync_config->error_handler =
            [sp = util::CopyablePromiseHolder(std::move(pf.promise))](auto, SyncError error) mutable {
                sp.get_promise().emplace_value(std::move(error));
            };
        auto r = Realm::get_shared_realm(config);

        // Create 26 MB worth of dogs in a single transaction - this should all get put into one changeset
        // and get uploaded at once, which for now is an error on the server.
        r->begin_transaction();
        CppContext c;
        for (auto i = 'a'; i < 'z'; ++i) {
            Object::create(c, r, "Dog",
                           std::any(AnyDict{{"_id", std::any(ObjectId::gen())},
                                            {"breed", std::string("bulldog")},
                                            {"name", random_string(1024 * 1024)}}),
                           CreatePolicy::ForceCreate);
        }
        r->commit_transaction();

#if defined(TEST_TIMEOUT_EXTRA) && TEST_TIMEOUT_EXTRA > 0
        // It may take 30 minutes to transfer 16MB at 10KB/s
        auto delay = std::chrono::minutes(35);
#else
        auto delay = std::chrono::minutes(5);
#endif

        auto error = wait_for_future(std::move(pf.future), delay).get();
        REQUIRE(error.status == ErrorCodes::LimitExceeded);
        REQUIRE(error.status.reason() ==
                "Sync websocket closed because the server received a message that was too large: "
                "read limited at 16777217 bytes");
        REQUIRE(error.is_client_reset_requested());
        REQUIRE(error.server_requests_action == sync::ProtocolErrorInfo::Action::ClientReset);
    }

    SECTION("freezing realm does not resume session") {
        SyncTestFile config(app->current_user(), partition, schema);
        auto realm = Realm::get_shared_realm(config);
        wait_for_download(*realm);

        auto state = realm->sync_session()->state();
        REQUIRE(state == SyncSession::State::Active);

        realm->sync_session()->pause();
        state = realm->sync_session()->state();
        REQUIRE(state == SyncSession::State::Paused);

        realm->read_group();

        {
            auto frozen = realm->freeze();
            REQUIRE(realm->sync_session() == realm->sync_session());
            REQUIRE(realm->sync_session()->state() == SyncSession::State::Paused);
        }

        {
            auto frozen = Realm::get_frozen_realm(config, realm->read_transaction_version());
            REQUIRE(realm->sync_session() == realm->sync_session());
            REQUIRE(realm->sync_session()->state() == SyncSession::State::Paused);
        }
    }

    SECTION("pausing a session does not hold the DB open") {
        auto logger = util::Logger::get_default_logger();
        SyncTestFile config(app->current_user(), partition, schema);
        DBRef dbref;
        std::shared_ptr<SyncSession> sync_sess_ext_ref;
        {
            auto realm = Realm::get_shared_realm(config);
            wait_for_download(*realm);

            auto state = realm->sync_session()->state();
            REQUIRE(state == SyncSession::State::Active);

            sync_sess_ext_ref = realm->sync_session()->external_reference();
            dbref = TestHelper::get_db(*realm);
            // An active PBS realm should have one ref each for:
            // - RealmCoordinator
            // - SyncSession
            // - MigrationStore
            // - SessionWrapper
            // - local dbref
            logger->trace("DBRef ACTIVE use count: %1", dbref.use_count());
            REQUIRE(dbref.use_count() >= 5);

            realm->sync_session()->pause();
            state = realm->sync_session()->state();
            REQUIRE(state == SyncSession::State::Paused);
            logger->trace("DBRef PAUSING called use count: %1", dbref.use_count());
        }

        // Closing the realm should leave one ref each for:
        // - SyncSession
        // - MigrationStore
        // - local dbref
        REQUIRE_THAT(
            [&] {
                logger->trace("DBRef PAUSED use count: %1", dbref.use_count());
                return dbref.use_count() < 4;
            },
            ReturnsTrueWithinTimeLimit{});

        // Releasing the external reference should leave one ref for:
        // - local dbref
        sync_sess_ext_ref.reset();
        REQUIRE_THAT(
            [&] {
                logger->trace("DBRef TEARDOWN use count: %1", dbref.use_count());
                return dbref.use_count() == 1;
            },
            ReturnsTrueWithinTimeLimit{});
    }

    SECTION("validation") {
        SyncTestFile config(app->current_user(), partition, schema);

        SECTION("invalid partition error handling") {
            config.sync_config->partition_value = "not a bson serialized string";
            std::atomic<bool> error_did_occur = false;
            config.sync_config->error_handler = [&error_did_occur](std::shared_ptr<SyncSession>, SyncError error) {
                CHECK(error.status.reason().find(
                          "Illegal Realm path (BIND): serialized partition 'not a bson serialized "
                          "string' is invalid") != std::string::npos);
                error_did_occur.store(true);
            };
            auto r = Realm::get_shared_realm(config);
            auto session = app->sync_manager()->get_existing_session(r->config().path);
            timed_wait_for([&] {
                return error_did_occur.load();
            });
            REQUIRE(error_did_occur.load());
        }

        SECTION("invalid pk schema error handling") {
            const std::string invalid_pk_name = "my_primary_key";
            auto it = config.schema->find("Dog");
            REQUIRE(it != config.schema->end());
            REQUIRE(it->primary_key_property());
            REQUIRE(it->primary_key_property()->name == "_id");
            it->primary_key_property()->name = invalid_pk_name;
            it->primary_key = invalid_pk_name;
            REQUIRE_THROWS_CONTAINING(Realm::get_shared_realm(config),
                                      "The primary key property on a synchronized Realm must be named '_id' but "
                                      "found 'my_primary_key' for type 'Dog'");
        }

        SECTION("missing pk schema error handling") {
            auto it = config.schema->find("Dog");
            REQUIRE(it != config.schema->end());
            REQUIRE(it->primary_key_property());
            it->primary_key_property()->is_primary = false;
            it->primary_key = "";
            REQUIRE(!it->primary_key_property());
            REQUIRE_THROWS_CONTAINING(Realm::get_shared_realm(config),
                                      "There must be a primary key property named '_id' on a synchronized "
                                      "Realm but none was found for type 'Dog'");
        }
    }

    SECTION("get_file_ident") {
        SyncTestFile config(app->current_user(), partition, schema);
        config.sync_config->client_resync_mode = ClientResyncMode::RecoverOrDiscard;
        auto r = Realm::get_shared_realm(config);
        wait_for_download(*r);

        auto first_ident = r->sync_session()->get_file_ident();
        REQUIRE(first_ident.ident != 0);
        REQUIRE(first_ident.salt != 0);

        reset_utils::trigger_client_reset(session.app_session(), r);
        r->sync_session()->restart_session();
        wait_for_download(*r);

        REQUIRE(first_ident.ident != r->sync_session()->get_file_ident().ident);
        REQUIRE(first_ident.salt != r->sync_session()->get_file_ident().salt);
    }
}

TEST_CASE("app: sync logs contain baas coid", "[sync][app][baas]") {
    class InMemoryLogger : public util::Logger {
    public:
        void do_log(const util::LogCategory& cat, Level level, const std::string& msg) final
        {
            auto formatted_line = util::format("%1 %2 %3", cat.get_name(), level, msg);
            std::lock_guard lk(mtx);
            log_messages.emplace_back(std::move(formatted_line));
        }

        std::vector<std::string> get_log_messages()
        {
            std::lock_guard lk(mtx);
            std::vector<std::string> ret;
            std::swap(ret, log_messages);
            return ret;
        }

        std::mutex mtx;
        std::vector<std::string> log_messages;
    };

    auto in_mem_logger = std::make_shared<InMemoryLogger>();
    TestAppSession app_session(get_runtime_app_session(), nullptr, DeleteApp{false}, ReconnectMode::normal, nullptr,
                               in_mem_logger);

    const auto partition = random_string(100);
    SyncTestFile config(app_session.app()->current_user(), partition, util::none);
    auto realm = successfully_async_open_realm(config);
    auto sync_session = realm->sync_session();
    auto coid = SyncSession::OnlyForTesting::get_appservices_connection_id(*sync_session);

    auto transition_log_msg =
        util::format("Connection[1] Connected to app services with request id: \"%1\". Further log entries for this "
                     "connection will be prefixed with \"Connection[1:%1]\" instead of \"Connection[1]\"",
                     coid);
    auto bind_send_msg = util::format("Connection[1:%1] Session[1]: Sending: BIND", coid);
    auto ping_send_msg = util::format("Connection[1:%1] Will emit a ping in", coid);

    auto log_messages = in_mem_logger->get_log_messages();
    REQUIRE_THAT(log_messages, AnyMatch(ContainsSubstring(transition_log_msg)));
    REQUIRE_THAT(log_messages, AnyMatch(ContainsSubstring(bind_send_msg)));
    REQUIRE_THAT(log_messages, AnyMatch(ContainsSubstring(ping_send_msg)));
}


TEST_CASE("app: trailing slash in base url", "[sync][app]") {
    auto logger = util::Logger::get_default_logger();

    const auto schema = get_default_schema();

    SyncServer server({});
    auto transport = std::make_shared<HookedTransport<UnitTestTransport>>();
    auto socket_provider = std::make_shared<HookedSocketProvider>(logger, "");
    OfflineAppSession::Config oas_config(transport);
    oas_config.base_url = util::format("http://localhost:%1/", server.port());
    oas_config.socket_provider = socket_provider;
    OfflineAppSession oas(oas_config);
    AutoVerifiedEmailCredentials creds;
    auto app = oas.app();
    const auto partition = random_string(100);

    transport->request_hook = [&](const Request& req) -> std::optional<Response> {
        if (req.url.find("/location") == std::string::npos) {
            return std::nullopt;
        }

        REQUIRE(req.url == util::format("http://localhost:%1/api/client/v2.0/app/app_id/location", server.port()));
        return Response{
            200,
            0,
            {},
            nlohmann::json(nlohmann::json::object({
                               {"hostname", util::format("http://localhost:%1", server.port())},
                               {"ws_hostname", util::format("ws://localhost:%1", server.port())},
                               {"sync_route", util::format("ws://localhost:%1/realm-sync", server.port())},
                           }))
                .dump(),
        };
    };

    SyncTestFile realm_config(oas, "test");

    auto r = Realm::get_shared_realm(realm_config);
    REQUIRE(!wait_for_download(*r));
}

TEST_CASE("app: redirect handling", "[sync][pbs][app]") {
    auto logger = util::Logger::get_default_logger();

    const auto schema = get_default_schema();

    auto transport = std::make_shared<HookedTransport<UnitTestTransport>>();
    auto socket_provider = std::make_shared<HookedSocketProvider>(logger, "");
    OfflineAppSession::Config oas_config(transport);
    oas_config.base_url = "http://original.invalid:9090";
    oas_config.socket_provider = socket_provider;
    OfflineAppSession oas(oas_config);
    AutoVerifiedEmailCredentials creds;
    auto app = oas.app();
    const auto partition = random_string(100);

    SECTION("invalid redirect response reports and error") {
        int request_count = 0;

        // This will fail due to no Location header
        transport->request_hook = [&](const Request& request) -> std::optional<Response> {
            logger->trace("request.url (%1): %2", request_count, request.url);
            REQUIRE(request_count++ == 0);
            return Response{301, 0, {{"Content-Type", "application/json"}}, "Some body data"};
        };
        app->provider_client<app::App::UsernamePasswordProviderClient>().register_email(
            creds.email, creds.password, [&](util::Optional<app::AppError> error) {
                REQUIRE(error);
                REQUIRE(error->is_client_error());
                REQUIRE(error->code() == ErrorCodes::ClientRedirectError);
                REQUIRE(error->reason() == "Redirect response missing location header");
            });

        // This will fail due to empty Location header
        transport->request_hook = [&](const Request& request) -> std::optional<Response> {
            logger->trace("request.url (%1): %2", request_count, request.url);
            REQUIRE(request_count++ == 1);
            return Response{301, 0, {{"Location", ""}, {"Content-Type", "application/json"}}, "Some body data"};
        };

        app->provider_client<app::App::UsernamePasswordProviderClient>().register_email(
            creds.email, creds.password, [&](util::Optional<app::AppError> error) {
                REQUIRE(error);
                REQUIRE(error->is_client_error());
                REQUIRE(error->code() == ErrorCodes::ClientRedirectError);
                REQUIRE(error->reason() == "Redirect response missing location header");
            });
    }

    SECTION("valid redirect response") {
        int request_count = 0;
        const std::string second_host = "http://second.invalid:9091";
        const std::string third_host = "http://third.invalid:9092";

        transport->request_hook = [&](const Request& request) -> std::optional<Response> {
            logger->trace("Received request[%1]: %2", request_count, request.url);
            switch (request_count++) {
                case 0:
                    REQUIRE_THAT(request.url, ContainsSubstring("/location"));
                    REQUIRE_THAT(request.url, ContainsSubstring(*oas_config.base_url));
                    return Response{301, 0, {{"Location", second_host}, {"Content-Type", "application/json"}}, ""};

                case 1:
                    REQUIRE_THAT(request.url, ContainsSubstring("/location"));
                    REQUIRE_THAT(request.url, ContainsSubstring(second_host));
                    return Response{301, 0, {{"Location", third_host}, {"Content-Type", "application/json"}}, ""};

                case 2:
                    REQUIRE_THAT(request.url, ContainsSubstring("/location"));
                    REQUIRE_THAT(request.url, ContainsSubstring(third_host));
                    return Response{301, 0, {{"Location", second_host}, {"Content-Type", "application/json"}}, ""};

                case 3:
                    REQUIRE_THAT(request.url, ContainsSubstring("/location"));
                    REQUIRE_THAT(request.url, ContainsSubstring(second_host));
                    return std::nullopt;

                default:
                    // some.fake.url is the location reported by UnitTestTransport
                    REQUIRE_THAT(request.url, ContainsSubstring("https://some.fake.url"));
                    return std::nullopt;
            }
        };

        // This will be successful after a couple of retries due to the redirect response
        app->provider_client<app::App::UsernamePasswordProviderClient>().register_email(
            creds.email, creds.password, [&](util::Optional<app::AppError> error) {
                REQUIRE(!error);
            });
    }

    SECTION("too many redirects eventually reports an error") {
        int request_count = 0;
        transport->request_hook = [&](const Request& request) -> std::optional<Response> {
            logger->trace("request.url (%1): %2", request_count, request.url);
            REQUIRE(request_count < 21);
            ++request_count;
            return Response{request_count % 2 == 1 ? 308 : 301,
                            0,
                            {{"Location", "http://somehost:9090"}, {"Content-Type", "application/json"}},
                            "Some body data"};
        };

        app->log_in_with_credentials(app::AppCredentials::username_password(creds.email, creds.password),
                                     [&](std::shared_ptr<realm::SyncUser> user, util::Optional<app::AppError> error) {
                                         REQUIRE(!user);
                                         REQUIRE(error);
                                         REQUIRE(error->is_client_error());
                                         REQUIRE(error->code() == ErrorCodes::ClientTooManyRedirects);
                                         REQUIRE(error->reason() == "number of redirections exceeded 20");
                                     });
        REQUIRE(request_count == 21);
    }

    SECTION("server in maintenance reports error") {
        transport->request_hook = [&](const Request&) -> std::optional<Response> {
            nlohmann::json maintenance_error = {{"error_code", "MaintenanceInProgress"},
                                                {"error", "This service is currently undergoing maintenance"},
                                                {"link", "https://link.to/server_logs"}};
            return Response{500, 0, {{"Content-Type", "application/json"}}, maintenance_error.dump()};
        };

        app->log_in_with_credentials(realm::app::AppCredentials::username_password(creds.email, creds.password),
                                     [&](std::shared_ptr<realm::SyncUser> user, util::Optional<app::AppError> error) {
                                         REQUIRE(!user);
                                         REQUIRE(error);
                                         REQUIRE(error->is_service_error());
                                         REQUIRE(error->code() == ErrorCodes::MaintenanceInProgress);
                                         REQUIRE(error->reason() ==
                                                 "This service is currently undergoing maintenance");
                                         REQUIRE(error->link_to_server_logs == "https://link.to/server_logs");
                                         REQUIRE(*error->additional_status_code == 500);
                                     });
    }

    SECTION("websocket redirects update existing session") {
        SyncServer server({});

        transport->request_hook = [&](const Request& req) -> std::optional<Response> {
            if (req.url.find("/location") != std::string::npos) {
                return Response{
                    200,
                    0,
                    {},
                    nlohmann::json({
                                       {"hostname", "http://some.fake.url"},
                                       {"ws_hostname", "ws://ws.some.fake.url"},
                                       {"sync_route", "ws://some.fake.url/realm-sync"},
                                   })
                        .dump(),
                };
            }
            return std::nullopt;
        };

        // The location info is fake, so we need to override it with the actual
        // server endpoint
        socket_provider->websocket_endpoint_resolver = [&](sync::WebSocketEndpoint& ep) {
            ep.address = "127.0.0.1";
            ep.port = server.port();
        };

        SyncTestFile realm_config(oas, "test");

        std::mutex logout_mutex;
        std::condition_variable logout_cv;
        bool logged_out = false;
        realm_config.sync_config->error_handler = [&](std::shared_ptr<SyncSession>, SyncError error) {
            if (error.status == ErrorCodes::AuthError) {
                {
                    std::unique_lock lk(logout_mutex);
                    logged_out = true;
                }
                logout_cv.notify_one();
                return;
            }
            util::format(std::cerr, "An unexpected sync error was caught by the default SyncTestFile handler: '%1'\n",
                         error.status);
            abort();
        };

        auto r = Realm::get_shared_realm(realm_config);
        REQUIRE(!wait_for_download(*r));
        auto sync_session = r->sync_session();
        sync_session->pause();
        SyncManager::OnlyForTesting::voluntary_disconnect_all_connections(*oas.sync_manager());

        int connect_count = 0;
        socket_provider->websocket_connect_func = [&]() -> std::optional<SocketProviderError> {
            // Report a 308 response the first time we try to reconnect the websocket,
            // which should result in App performing a location update.
            // The actual Location header isn't used when we get a redirect on
            // the websocket, so we don't need to supply it here
            if (connect_count++ > 0)
                return std::nullopt;
            return sync::HTTPStatus::PermanentRedirect;
        };

        SECTION("valid websocket redirect") {
            socket_provider->websocket_endpoint_resolver = [&](sync::WebSocketEndpoint& ep) {
                logger->trace("resolve attempt %1: %2", connect_count, ep.address);
                // First call happens after the call to the above hook which will
                // force a 308 response. Second call happens after the redirect
                // has been handled.
                REQUIRE(connect_count <= 2);
                if (connect_count == 2) {
                    REQUIRE(ep.address == "ws.invalid");
                }

                // Overriding the handshake result happens after dns resolution,
                // so we need to set it to a valid endpoint for even the first call
                ep.address = "127.0.0.1";
                ep.port = server.port();
            };

            int request_count = 0;
            transport->request_hook = [&](const Request& request) -> std::optional<Response> {
                logger->trace("request.url (%1): %2", request_count, request.url);
                ++request_count;

                // First request should be a location request against the original URL
                if (request_count == 1) {
                    REQUIRE_THAT(request.url, ContainsSubstring("some.fake.url"));
                    REQUIRE_THAT(request.url, ContainsSubstring("/location"));
                    return Response{static_cast<int>(sync::HTTPStatus::PermanentRedirect),
                                    0,
                                    {{"Location", "http://asdf.invalid"}},
                                    ""};
                }

                // Second request should be a location request against the new URL
                if (request_count == 2) {
                    REQUIRE_THAT(request.url, ContainsSubstring("/location"));
                    REQUIRE_THAT(request.url, ContainsSubstring("asdf.invalid"));
                    return Response{200,
                                    0,
                                    {},
                                    nlohmann::json({
                                                       {"hostname", "http://http.invalid"},
                                                       {"ws_hostname", "ws://ws.invalid"},
                                                       {"sync_route", "ws://ws.invalid/realm-sync"},
                                                   })
                                        .dump()};
                }

                // Rest of the requests get handled normally
                return std::nullopt;
            };

            sync_session->resume();
            REQUIRE(!wait_for_download(*r));
            REQUIRE(request_count > 1);
            REQUIRE(realm_config.sync_config->user->is_logged_in());

            // Verify session is using the updated server url from the redirect
            auto server_url = sync_session->full_realm_url();
            REQUIRE_THAT(server_url, ContainsSubstring("ws.invalid"));
        }

        SECTION("websocket redirect into auth error logs out user") {
            int request_count = 0;
            transport->request_hook = [&](const Request& request) -> std::optional<Response> {
                logger->trace("request.url (%1): %2", request_count, request.url);
                ++request_count;

                if (request_count == 1) {
                    // First request should be a location request against the original URL
                    REQUIRE_THAT(request.url, ContainsSubstring("some.fake.url"));
                    REQUIRE_THAT(request.url, ContainsSubstring("/location"));
                    return Response{static_cast<int>(sync::HTTPStatus::PermanentRedirect),
                                    0,
                                    {{"Location", "http://asdf.invalid"}},
                                    ""};
                }

                // Second request should be a location request against the new URL
                if (request_count == 2) {
                    REQUIRE_THAT(request.url, ContainsSubstring("/location"));
                    REQUIRE_THAT(request.url, ContainsSubstring("asdf.invalid"));
                    return Response{200,
                                    0,
                                    {},
                                    nlohmann::json({
                                                       {"hostname", "http://http.invalid"},
                                                       {"ws_hostname", "ws://ws.invalid"},
                                                   })
                                        .dump()};
                }

                // Third request should be for an acccess token, which we reject
                REQUIRE(request_count == 3);
                REQUIRE_THAT(request.url, ContainsSubstring("auth/session"));
                return Response{static_cast<int>(sync::HTTPStatus::Unauthorized), 0, {}, ""};
            };

            sync_session->resume();
            REQUIRE(wait_for_download(*r));
            std::unique_lock lk(logout_mutex);
            auto result = logout_cv.wait_for(lk, std::chrono::seconds(15), [&]() {
                return logged_out;
            });
            REQUIRE(result);
            REQUIRE_FALSE(realm_config.sync_config->user->is_logged_in());
        }

        SECTION("too many websocket redirects logs out user") {
            int request_count = 0;
            const int max_http_redirects = 20; // from app.cpp in object-store
            transport->request_hook = [&](const Request& request) -> std::optional<Response> {
                logger->trace("request.url (%1): %2", request_count, request.url);

                // The test should never request anything other than /location
                // even though the user is set to the logged-out state as trying
                // to log out on the server needs to go through /location first too
                REQUIRE_THAT(request.url, ContainsSubstring("/location"));
                REQUIRE(request_count <= max_http_redirects);

                // First request should be a location request against the original URL
                // and rest should use the redirect url
                if (request_count++ == 0) {
                    REQUIRE_THAT(request.url, ContainsSubstring("some.fake.url"));
                }
                else {
                    REQUIRE_THAT(request.url, ContainsSubstring("asdf.invalid"));
                }
                // Keep returning the redirected response
                return Response{static_cast<int>(sync::HTTPStatus::MovedPermanently),
                                0,
                                {{"Location", "http://asdf.invalid"}},
                                ""};
            };

            sync_session->resume();
            REQUIRE(wait_for_download(*r));
            std::unique_lock lk(logout_mutex);
            auto result = logout_cv.wait_for(lk, std::chrono::seconds(15), [&]() {
                return logged_out;
            });
            REQUIRE(result);
            REQUIRE_FALSE(realm_config.sync_config->user->is_logged_in());
        }
    }
}

TEST_CASE("app: base_url", "[sync][app][base_url]") {
    struct BaseUrlTransport : UnitTestTransport {
        std::string expected_url;
        std::optional<std::string_view> redirect_url;
        bool location_requested = false;
        bool location_returns_error = false;

        void reset(std::string_view expect_url, std::optional<std::string_view> redir_url = std::nullopt)
        {
            expected_url = std::string(expect_url);
            redirect_url = redir_url;
            location_requested = false;
            location_returns_error = false;
        }

        void send_request_to_server(const Request& request,
                                    util::UniqueFunction<void(const Response&)>&& completion) override
        {
            if (request.url.find("/location") != std::string::npos) {
                CHECK(request.method == HttpMethod::get);
                CHECK_THAT(request.url, ContainsSubstring(expected_url));
                location_requested = true;
                if (location_returns_error) {
                    completion(app::Response{static_cast<int>(sync::HTTPStatus::NotFound), 0, {}, "404 not found"});
                    return;
                }
                if (redirect_url) {
                    // Update the expected url to be the redirect url
                    expected_url = std::string(*redirect_url);
                    redirect_url.reset();

                    completion(app::Response{static_cast<int>(sync::HTTPStatus::PermanentRedirect),
                                             0,
                                             {{"location", expected_url}},
                                             "308 permanent redirect"});
                    return;
                }
                auto ws_url = App::create_ws_host_url(expected_url);
                completion(
                    app::Response{static_cast<int>(sync::HTTPStatus::Ok),
                                  0,
                                  {},
                                  util::format("{\"deployment_model\":\"GLOBAL\",\"location\":\"US-VA\",\"hostname\":"
                                               "\"%1\",\"ws_hostname\":\"%2\"}",
                                               expected_url, ws_url)});
                return;
            }

            UnitTestTransport::send_request_to_server(request, std::move(completion));
        }
    };

    auto logger = util::Logger::get_default_logger();

    auto redir_transport = std::make_shared<BaseUrlTransport>();
    auto get_config_with_base_url = [&](std::optional<std::string> base_url = std::nullopt) {
        OfflineAppSession::Config config(redir_transport);
        config.base_url = base_url;
        return config;
    };

    SECTION("Test App::create_ws_host_url") {
        auto result = App::create_ws_host_url("blah");
        CHECK(result == "blah");
        result = App::create_ws_host_url("http://localhost:9090");
        CHECK(result == "ws://localhost:9090");
        result = App::create_ws_host_url("https://localhost:9090");
        CHECK(result == "wss://localhost:9090");
        result = App::create_ws_host_url("https://localhost:9090/some/extra/stuff");
        CHECK(result == "wss://localhost:9090/some/extra/stuff");
        result = App::create_ws_host_url("http://172.0.0.1:9090");
        CHECK(result == "ws://172.0.0.1:9090");
        result = App::create_ws_host_url("https://172.0.0.1:9090");
        CHECK(result == "wss://172.0.0.1:9090");
        // Old default base url
        result = App::create_ws_host_url("http://realm.mongodb.com");
        CHECK(result == "ws://ws.realm.mongodb.com");
        result = App::create_ws_host_url("https://realm.mongodb.com");
        CHECK(result == "wss://ws.realm.mongodb.com");
        result = App::create_ws_host_url("https://realm.mongodb.com/some/extra/stuff");
        CHECK(result == "wss://ws.realm.mongodb.com/some/extra/stuff");
        result = App::create_ws_host_url("https://us-east-1.aws.realm.mongodb.com");
        CHECK(result == "wss://ws.us-east-1.aws.realm.mongodb.com");
        result = App::create_ws_host_url("https://us-east-1.aws.realm.mongodb.com");
        CHECK(result == "wss://ws.us-east-1.aws.realm.mongodb.com");
        result = App::create_ws_host_url("https://us-east-1.aws.realm.mongodb.com/some/extra/stuff");
        CHECK(result == "wss://ws.us-east-1.aws.realm.mongodb.com/some/extra/stuff");
        // New default base url
        result = App::create_ws_host_url("http://services.cloud.mongodb.com");
        CHECK(result == "ws://ws.services.cloud.mongodb.com");
        result = App::create_ws_host_url("https://services.cloud.mongodb.com");
        CHECK(result == "wss://ws.services.cloud.mongodb.com");
        result = App::create_ws_host_url("https://services.cloud.mongodb.com/some/extra/stuff");
        CHECK(result == "wss://ws.services.cloud.mongodb.com/some/extra/stuff");
        result = App::create_ws_host_url("http://us-east-1.aws.services.cloud.mongodb.com");
        CHECK(result == "ws://us-east-1.aws.ws.services.cloud.mongodb.com");
        result = App::create_ws_host_url("https://us-east-1.aws.services.cloud.mongodb.com");
        CHECK(result == "wss://us-east-1.aws.ws.services.cloud.mongodb.com");
        result = App::create_ws_host_url("https://us-east-1.aws.services.cloud.mongodb.com/some/extra/stuff");
        CHECK(result == "wss://us-east-1.aws.ws.services.cloud.mongodb.com/some/extra/stuff");
    }

    SECTION("Test app config baseurl") {
        {
            // First time through, base_url is empty; https://services.cloud.mongodb.com is expected
            redir_transport->reset(App::default_base_url());
            auto config = get_config_with_base_url();
            OfflineAppSession oas(config);
            auto app = oas.app();

            // Location is not requested until first app services request
            CHECK(!redir_transport->location_requested);
            // Initial hostname and ws hostname use base url, but aren't used until location is updated
            CHECK(app->get_host_url() == App::default_base_url());
            CHECK(app->get_ws_host_url() == App::create_ws_host_url(App::default_base_url()));

            oas.make_user();
            CHECK(redir_transport->location_requested);
            CHECK(app->get_base_url() == App::default_base_url());
            CHECK(app->get_host_url() == App::default_base_url());
            CHECK(app->get_ws_host_url() == App::create_ws_host_url(App::default_base_url()));
        }
        {
            // Second time through, base_url is set to https://alternate.someurl.fake is expected
            redir_transport->reset("https://alternate.someurl.fake");
            auto config = get_config_with_base_url("https://alternate.someurl.fake");
            OfflineAppSession oas(config);
            auto app = oas.app();

            // Location is not requested until first app services request
            CHECK(!redir_transport->location_requested);
            // Initial hostname and ws hostname use base url, but aren't used until location is updated
            CHECK(app->get_host_url() == "https://alternate.someurl.fake");
            CHECK(app->get_ws_host_url() == "wss://alternate.someurl.fake");

            oas.make_user();
            CHECK(redir_transport->location_requested);
            CHECK(app->get_base_url() == "https://alternate.someurl.fake");
            CHECK(app->get_host_url() == "https://alternate.someurl.fake");
            CHECK(app->get_ws_host_url() == "wss://alternate.someurl.fake");
        }
        {
            // Third time through, base_url is not set, expect https://services.cloud.mongodb.com,
            // since metadata is no longer used
            std::string expected_url = std::string(App::default_base_url());
            std::string expected_wsurl = App::create_ws_host_url(App::default_base_url());
            redir_transport->reset(expected_url);
            auto config = get_config_with_base_url();
            OfflineAppSession oas(config);
            auto app = oas.app();

            // Location is not requested until first app services request
            CHECK(!redir_transport->location_requested);
            // Initial hostname and ws hostname use base url, but aren't used until location is updated
            CHECK(app->get_host_url() == expected_url);
            CHECK(app->get_ws_host_url() == expected_wsurl);

            oas.make_user();
            CHECK(redir_transport->location_requested);
            CHECK(app->get_base_url() == expected_url);
            CHECK(app->get_host_url() == expected_url);
            CHECK(app->get_ws_host_url() == expected_wsurl);
        }
        {
            // Fourth time through, base_url is set to https://some-other.someurl.fake, with a redirect
            redir_transport->reset("https://some-other.someurl.fake", "http://redirect.someurl.fake");
            auto config = get_config_with_base_url("https://some-other.someurl.fake");
            OfflineAppSession oas(config);
            auto app = oas.app();

            // Location is not requested until first app services request
            CHECK(!redir_transport->location_requested);
            // Initial hostname and ws hostname use base url, but aren't used until location is updated
            CHECK(app->get_host_url() == "https://some-other.someurl.fake");
            CHECK(app->get_ws_host_url() == "wss://some-other.someurl.fake");

            oas.make_user();
            CHECK(redir_transport->location_requested);
            // Base URL is still set to the original value
            CHECK(app->get_base_url() == "https://some-other.someurl.fake");
            // Hostname and ws hostname use the redirect URL values
            CHECK(app->get_host_url() == "http://redirect.someurl.fake");
            CHECK(app->get_ws_host_url() == "ws://redirect.someurl.fake");
        }
    }

    SECTION("Test update_baseurl") {
        redir_transport->reset("https://alternate.someurl.fake");
        auto config = get_config_with_base_url("https://alternate.someurl.fake");
        OfflineAppSession oas(config);
        auto app = oas.app();

        // Location is not requested until first app services request
        CHECK(!redir_transport->location_requested);

        oas.make_user();
        CHECK(redir_transport->location_requested);
        CHECK(app->get_base_url() == "https://alternate.someurl.fake");
        CHECK(app->get_host_url() == "https://alternate.someurl.fake");
        CHECK(app->get_ws_host_url() == "wss://alternate.someurl.fake");

        redir_transport->reset(App::default_base_url());

        // Revert the base URL to the default URL value using the empty string
        app->update_base_url("", [](util::Optional<app::AppError> error) {
            CHECK(!error);
        });
        CHECK(redir_transport->location_requested);
        CHECK(app->get_base_url() == App::default_base_url());
        CHECK(app->get_host_url() == App::default_base_url());
        CHECK(app->get_ws_host_url() == App::create_ws_host_url(App::default_base_url()));
        oas.make_user();
    }

    SECTION("Test update_baseurl with redirect") {
        redir_transport->reset("https://alternate.someurl.fake");
        auto config = get_config_with_base_url("https://alternate.someurl.fake");
        OfflineAppSession oas(config);
        auto app = oas.app();

        // Location is not requested until first app services request
        CHECK(!redir_transport->location_requested);

        oas.make_user();
        CHECK(redir_transport->location_requested);
        CHECK(app->get_base_url() == "https://alternate.someurl.fake");
        CHECK(app->get_host_url() == "https://alternate.someurl.fake");
        CHECK(app->get_ws_host_url() == "wss://alternate.someurl.fake");

        redir_transport->reset("http://some-other.someurl.fake", "https://redirect.otherurl.fake");

        app->update_base_url("http://some-other.someurl.fake", [](util::Optional<app::AppError> error) {
            CHECK(!error);
        });
        CHECK(redir_transport->location_requested);
        CHECK(app->get_base_url() == "http://some-other.someurl.fake");
        CHECK(app->get_host_url() == "https://redirect.otherurl.fake");
        CHECK(app->get_ws_host_url() == "wss://redirect.otherurl.fake");
        // Expected URL is still "https://redirect.otherurl.fake" after redirect
        oas.make_user();
    }

    SECTION("Test update_baseurl returns error") {
        redir_transport->reset("http://alternate.someurl.fake");
        auto config = get_config_with_base_url("http://alternate.someurl.fake");
        OfflineAppSession oas(config);
        auto app = oas.app();

        // Location is not requested until first app services request
        CHECK(!redir_transport->location_requested);

        oas.make_user();
        CHECK(redir_transport->location_requested);
        CHECK(app->get_base_url() == "http://alternate.someurl.fake");
        CHECK(app->get_host_url() == "http://alternate.someurl.fake");
        CHECK(app->get_ws_host_url() == "ws://alternate.someurl.fake");

        redir_transport->reset("https://some-other.someurl.fake");
        redir_transport->location_returns_error = true;

        app->update_base_url("https://some-other.someurl.fake", [](util::Optional<app::AppError> error) {
            CHECK(error);
        });
        CHECK(redir_transport->location_requested);
        // Verify original url values are still being used
        CHECK(app->get_base_url() == "http://alternate.someurl.fake");
        CHECK(app->get_host_url() == "http://alternate.someurl.fake");
        CHECK(app->get_ws_host_url() == "ws://alternate.someurl.fake");
    }

    // Verify new sync session updates location when created with cached user
    SECTION("Verify new sync session updates location") {
        bool use_ssl = GENERATE(true, false);
        std::string initial_host = "alternate.someurl.fake";
        unsigned initial_port = use_ssl ? 443 : 80;
        std::string expected_host = "redirect.someurl.fake";
        unsigned expected_port = 8081;
        std::string init_url = util::format("http%1://%2", use_ssl ? "s" : "", initial_host);
        std::string init_wsurl = util::format("ws%1://%2", use_ssl ? "s" : "", initial_host);
        std::string redir_url = util::format("http%1://%2:%3", use_ssl ? "s" : "", expected_host, expected_port);
        std::string redir_wsurl = util::format("ws%1://%2:%3", use_ssl ? "s" : "", expected_host, expected_port);

        auto socket_provider = std::make_shared<HookedSocketProvider>(logger, "some user agent");
        socket_provider->websocket_connect_func = []() -> std::optional<SocketProviderError> {
            return SocketProviderError(sync::websocket::WebSocketError::websocket_connection_failed, "404 not found");
        };

        auto config = get_config_with_base_url(init_url);
        config.metadata_mode = AppConfig::MetadataMode::NoEncryption;
        config.socket_provider = socket_provider;
        config.storage_path = util::make_temp_dir();
        config.delete_storage = false; // persist the current user

        // Log in to get a cached user
        {
            redir_transport->reset(init_url);
            OfflineAppSession oas(config);
            auto app = oas.app();

            {
                auto [sync_route, verified] = app->sync_manager()->sync_route();
                CHECK_THAT(sync_route, ContainsSubstring(app::App::create_ws_host_url(init_url)));
                CHECK_FALSE(verified);
            }

            oas.make_user();
            CHECK(redir_transport->location_requested);
            CHECK(app->get_base_url() == init_url);
            CHECK(app->get_host_url() == init_url);
            CHECK(app->get_ws_host_url() == init_wsurl);
            auto [sync_route, verified] = app->sync_manager()->sync_route();
            CHECK_THAT(sync_route, ContainsSubstring(app::App::create_ws_host_url(init_url)));
            CHECK_THAT(sync_route, ContainsSubstring(init_wsurl));
            CHECK(verified);
        }

        // the next instance can clean up the files
        config.delete_storage = true;
        // Recreate the app using the cached user and start a sync session, which will is set to fail on connect
        SECTION("Sync Session fails on connect after updating location") {
            enum class TestState { start, session_started };
            TestingStateMachine<TestState> state(TestState::start);
            redir_transport->reset(init_url, redir_url);

            OfflineAppSession oas(config);
            auto app = oas.app();

            // Verify the default sync route, which has not been verified
            {
                auto [sync_route, verified] = app->sync_manager()->sync_route();
                CHECK_THAT(sync_route, ContainsSubstring(app::App::create_ws_host_url(init_url)));
                CHECK_FALSE(verified);
            }
            REQUIRE(app->current_user());

            std::atomic<int> connect_attempts = 0;
            socket_provider->endpoint_verify_func = [&](const sync::WebSocketEndpoint& ep) {
                // First connection attempt is to the originally specified endpoint. Since
                // it hasn't been verified, we swallow the error and do a location update,
                // which will then try to connect to the redir target
                auto attempt = connect_attempts++;
                if (attempt == 0) {
                    CHECK(ep.address == initial_host);
                    CHECK(ep.port == initial_port);
                    CHECK(ep.is_ssl == use_ssl);
                }
                else {
                    CHECK(ep.address == expected_host);
                    CHECK(ep.port == expected_port);
                    CHECK(ep.is_ssl == use_ssl);
                }
            };

            RealmConfig r_config;
            r_config.path = app->config().base_file_path + "/fakerealm.realm";
            r_config.sync_config = std::make_shared<SyncConfig>(app->current_user(), SyncConfig::FLXSyncEnabled{});
            r_config.sync_config->error_handler = [&](std::shared_ptr<SyncSession>, SyncError error) mutable {
                // Websocket is forcing a 404 failure so it won't actually start
                logger->debug("Received expected error: %1", error.status);
                CHECK(!error.status.is_ok());
                CHECK(error.status.code() == ErrorCodes::SyncConnectFailed);
                CHECK(!error.is_fatal);
                state.transition_to(TestState::session_started);
            };
            auto realm = Realm::get_shared_realm(r_config);
            state.wait_for(TestState::session_started);

            CHECK(redir_transport->location_requested);
            CHECK(app->get_base_url() == init_url);
            CHECK(app->get_host_url() == redir_url);
            CHECK(app->get_ws_host_url() == redir_wsurl);
            auto [sync_route, verified] = app->sync_manager()->sync_route();
            CHECK_THAT(sync_route, ContainsSubstring(redir_wsurl));
            CHECK(verified);
        }

        SECTION("Sync Session retries after initial location failure") {
            enum class TestState { start, location_failed, session_started };
            TestingStateMachine<TestState> state(TestState::start);
            const int retry_count = GENERATE(1, 3);

            redir_transport->reset(init_url);
            redir_transport->location_returns_error = true;

            OfflineAppSession oas(config);
            auto app = oas.app();
            REQUIRE(app->current_user());
            // Verify the default sync route, which has not been verified
            {
                auto [sync_route, verified] = app->sync_manager()->sync_route();
                CHECK_THAT(sync_route, ContainsSubstring(app::App::create_ws_host_url(init_url)));
                CHECK_FALSE(verified);
            }

            socket_provider->endpoint_verify_func = [&](const sync::WebSocketEndpoint& ep) {
                CHECK(ep.address == initial_host);
                CHECK(ep.port == initial_port);
                CHECK(ep.is_ssl == use_ssl);
            };

            socket_provider->websocket_connect_func = [&, request_count =
                                                              0]() mutable -> std::optional<SocketProviderError> {
                if (request_count == 0) {
                    // First connection attempt is to the unverified initial URL
                    // since we have a valid access token but have never successfully
                    // connected. This failing will trigger a location update.
                    CHECK_FALSE(redir_transport->location_requested);
                }
                else {
                    // All attempts after the first should have requested location
                    CHECK(redir_transport->location_requested);
                    redir_transport->location_requested = false;
                }

                // Until we allow a location request to succeed we should keep
                // getting the original unverified route
                if (redir_transport->location_returns_error) {
                    CHECK(app->get_base_url() == init_url);
                    CHECK(app->get_host_url() == init_url);
                    CHECK(app->get_ws_host_url() == init_wsurl);
                    {
                        auto [sync_route, verified] = app->sync_manager()->sync_route();
                        CHECK_THAT(sync_route, ContainsSubstring(app::App::create_ws_host_url(init_url)));
                        CHECK_FALSE(verified);
                    }
                }

                // After the chosen number of attempts let the location request succeed
                if (request_count++ >= retry_count) {
                    redir_transport->reset(init_url, redir_url);
                    socket_provider->endpoint_verify_func = [&](const sync::WebSocketEndpoint& ep) {
                        CHECK(ep.address == expected_host);
                        CHECK(ep.port == expected_port);
                        CHECK(ep.is_ssl == use_ssl);
                        state.transition_to(TestState::location_failed);
                    };
                }

                return SocketProviderError(sync::websocket::WebSocketError::websocket_connection_failed,
                                           "404 not found");
            };

            RealmConfig r_config;
            r_config.path = app->config().base_file_path + "/fakerealm.realm";
            r_config.sync_config = std::make_shared<SyncConfig>(app->current_user(), SyncConfig::FLXSyncEnabled{});
            r_config.sync_config->error_handler = [&](std::shared_ptr<SyncSession>, SyncError error) mutable {
                // An error will only be reported if the websocket fails after updating the location and access token
                logger->debug("Received expected error: %1", error.status);
                CHECK(!error.status.is_ok());
                CHECK(error.status.code() == ErrorCodes::SyncConnectFailed);
                CHECK(!error.is_fatal);
                state.transition_with([&](TestState cur_state) -> std::optional<TestState> {
                    if (cur_state == TestState::location_failed) {
                        // This time, the session was being started, and the location was successful
                        // Websocket is forcing a 404 failure so it won't actually start
                        return TestState::session_started;
                    }
                    return std::nullopt;
                });
            };
            auto realm = Realm::get_shared_realm(r_config);
            state.wait_for(TestState::session_started);

            CHECK(app->get_base_url() == init_url);
            CHECK(app->get_host_url() == redir_url);
            CHECK(app->get_ws_host_url() == redir_wsurl);
            auto [sync_route, verified] = app->sync_manager()->sync_route();
            CHECK_THAT(sync_route, ContainsSubstring(redir_wsurl));
            CHECK(verified);
        }
    }
}

TEST_CASE("app: custom user data integration tests", "[sync][app][user][function][baas]") {
    TestAppSession session;
    auto app = session.app();
    auto user = app->current_user();

    SECTION("custom user data happy path") {
        bool processed = false;
        app->call_function("updateUserData", {bson::BsonDocument({{"favorite_color", "green"}})},
                           [&](auto response, auto error) {
                               CHECK(error == none);
                               CHECK(response);
                               CHECK(*response == true);
                               processed = true;
                           });
        CHECK(processed);
        processed = false;
        app->refresh_custom_data(user, [&](auto) {
            processed = true;
        });
        CHECK(processed);
        auto data = *user->custom_data();
        CHECK(data["favorite_color"] == "green");
    }
}

TEST_CASE("app: jwt login and metadata tests", "[sync][app][user][metadata][function][baas]") {
    TestAppSession session;
    auto app = session.app();
    auto jwt = create_jwt(session.app()->app_id());

    SECTION("jwt happy path") {
        bool processed = false;
        bool logged_in_once = false;

        auto token = app->subscribe([&logged_in_once, &app](auto&) {
            REQUIRE(!logged_in_once);
            auto user = app->current_user();
            auto metadata = user->user_profile();

            // Ensure that the JWT metadata fields are available when the callback is fired on login.
            CHECK(metadata["name"] == "Foo Bar");
            logged_in_once = true;
        });

        std::shared_ptr<User> user = log_in(app, AppCredentials::custom(jwt));

        app->call_function(user, "updateUserData", {bson::BsonDocument({{"name", "Not Foo Bar"}})},
                           [&](auto response, auto error) {
                               CHECK(error == none);
                               CHECK(response);
                               CHECK(*response == true);
                               processed = true;
                           });
        CHECK(processed);
        processed = false;
        app->refresh_custom_data(user, [&](auto) {
            processed = true;
        });
        CHECK(processed);
        auto metadata = user->user_profile();
        auto custom_data = *user->custom_data();
        CHECK(custom_data["name"] == "Not Foo Bar");
        CHECK(metadata["name"] == "Foo Bar");

        REQUIRE(logged_in_once);

        app->unsubscribe(token);
    }
}

namespace cf = realm::collection_fixtures;
TEMPLATE_TEST_CASE("app: collections of links integration", "[sync][pbs][app][collections][baas]", cf::ListOfObjects,
                   cf::ListOfMixedLinks, cf::SetOfObjects, cf::SetOfMixedLinks, cf::DictionaryOfObjects,
                   cf::DictionaryOfMixedLinks)
{
    const std::string valid_pk_name = "_id";
    const auto partition = random_string(100);
    TestType test_type("collection", "dest");
    Schema schema = {{"source",
                      {{valid_pk_name, PropertyType::Int | PropertyType::Nullable, true},
                       {"realm_id", PropertyType::String | PropertyType::Nullable},
                       test_type.property()}},
                     {"dest",
                      {
                          {valid_pk_name, PropertyType::Int | PropertyType::Nullable, true},
                          {"realm_id", PropertyType::String | PropertyType::Nullable},
                      }}};
    auto server_app_config = minimal_app_config("collections_of_links", schema);
    TestAppSession test_session(create_app(server_app_config));

    auto wait_for_num_objects_to_equal = [](realm::SharedRealm r, const std::string& table_name, size_t count) {
        timed_sleeping_wait_for([&]() -> bool {
            r->refresh();
            TableRef dest = r->read_group().get_table(table_name);
            size_t cur_count = dest->size();
            return cur_count == count;
        });
    };
    auto wait_for_num_outgoing_links_to_equal = [&](realm::SharedRealm r, Obj obj, size_t count) {
        timed_sleeping_wait_for([&]() -> bool {
            r->refresh();
            return test_type.size_of_collection(obj) == count;
        });
    };

    CppContext c;
    auto create_one_source_object = [&](realm::SharedRealm r, int64_t val, std::vector<ObjLink> links = {}) {
        r->begin_transaction();
        auto object = Object::create(
            c, r, "source",
            std::any(realm::AnyDict{{valid_pk_name, std::any(val)}, {"realm_id", std::string(partition)}}),
            CreatePolicy::ForceCreate);

        for (auto link : links) {
            auto& obj = object.get_obj();
            test_type.add_link(obj, link);
        }
        r->commit_transaction();
        return object;
    };

    auto create_one_dest_object = [&](realm::SharedRealm r, int64_t val) -> ObjLink {
        r->begin_transaction();
        auto obj = Object::create(
            c, r, "dest",
            std::any(realm::AnyDict{{valid_pk_name, std::any(val)}, {"realm_id", std::string(partition)}}),
            CreatePolicy::ForceCreate);
        r->commit_transaction();
        return ObjLink{obj.get_obj().get_table()->get_key(), obj.get_obj().get_key()};
    };

    auto require_links_to_match_ids = [&](std::vector<Obj> links, std::vector<int64_t> expected) {
        std::vector<int64_t> actual;
        for (auto obj : links) {
            actual.push_back(obj.get<Int>(valid_pk_name));
        }
        std::sort(actual.begin(), actual.end());
        std::sort(expected.begin(), expected.end());
        REQUIRE(actual == expected);
    };

    SECTION("integration testing") {
        auto app = test_session.app();
        SyncTestFile config1(app->current_user(), partition, schema); // uses the current user created above
        config1.automatic_change_notifications = false;
        auto r1 = realm::Realm::get_shared_realm(config1);
        Results r1_source_objs = realm::Results(r1, r1->read_group().get_table("class_source"));

        create_user_and_log_in(app);                                  // changes the current user
        SyncTestFile config2(app->current_user(), partition, schema); // uses the user created above
        config2.automatic_change_notifications = false;
        auto r2 = realm::Realm::get_shared_realm(config2);
        Results r2_source_objs = realm::Results(r2, r2->read_group().get_table("class_source"));

        constexpr int64_t source_pk = 0;
        constexpr int64_t dest_pk_1 = 1;
        constexpr int64_t dest_pk_2 = 2;
        constexpr int64_t dest_pk_3 = 3;
        Object object;

        { // add a container collection with three valid links
            REQUIRE(r1_source_objs.size() == 0);
            ObjLink dest1 = create_one_dest_object(r1, dest_pk_1);
            ObjLink dest2 = create_one_dest_object(r1, dest_pk_2);
            ObjLink dest3 = create_one_dest_object(r1, dest_pk_3);
            object = create_one_source_object(r1, source_pk, {dest1, dest2, dest3});
            REQUIRE(r1_source_objs.size() == 1);
            REQUIRE(r1_source_objs.get(0).get<Int>(valid_pk_name) == source_pk);
            REQUIRE(r1_source_objs.get(0).get<String>("realm_id") == partition);
            require_links_to_match_ids(test_type.get_links(r1_source_objs.get(0)), {dest_pk_1, dest_pk_2, dest_pk_3});
        }

        size_t expected_coll_size = 3;
        std::vector<int64_t> remaining_dest_object_ids;
        { // erase one of the destination objects
            wait_for_num_objects_to_equal(r2, "class_source", 1);
            wait_for_num_objects_to_equal(r2, "class_dest", 3);
            REQUIRE(r2_source_objs.size() == 1);
            REQUIRE(r2_source_objs.get(0).get<Int>(valid_pk_name) == source_pk);
            REQUIRE(test_type.size_of_collection(r2_source_objs.get(0)) == 3);
            auto linked_objects = test_type.get_links(r2_source_objs.get(0));
            require_links_to_match_ids(linked_objects, {dest_pk_1, dest_pk_2, dest_pk_3});
            r2->begin_transaction();
            linked_objects[0].remove();
            r2->commit_transaction();
            remaining_dest_object_ids = {linked_objects[1].template get<Int>(valid_pk_name),
                                         linked_objects[2].template get<Int>(valid_pk_name)};
            expected_coll_size = test_type.will_erase_removed_object_links() ? 2 : 3;
            REQUIRE(test_type.size_of_collection(r2_source_objs.get(0)) == expected_coll_size);
        }

        { // remove a link from the collection
            wait_for_num_objects_to_equal(r1, "class_dest", 2);
            REQUIRE(r1_source_objs.size() == 1);
            REQUIRE(test_type.size_of_collection(r1_source_objs.get(0)) == expected_coll_size);
            auto linked_objects = test_type.get_links(r1_source_objs.get(0));
            require_links_to_match_ids(linked_objects, remaining_dest_object_ids);
            r1->begin_transaction();
            auto obj = r1_source_objs.get(0);
            test_type.remove_link(obj,
                                  ObjLink{linked_objects[0].get_table()->get_key(), linked_objects[0].get_key()});
            r1->commit_transaction();
            --expected_coll_size;
            remaining_dest_object_ids = {linked_objects[1].template get<Int>(valid_pk_name)};
            REQUIRE(test_type.size_of_collection(r1_source_objs.get(0)) == expected_coll_size);
        }
        bool coll_cleared = false;
        advance_and_notify(*r1);
        auto collection = test_type.get_collection(r1, r1_source_objs.get(0));
        auto token = collection.add_notification_callback([&coll_cleared](CollectionChangeSet c) {
            coll_cleared = c.collection_was_cleared;
        });

        { // clear the collection
            REQUIRE(r2_source_objs.size() == 1);
            REQUIRE(r2_source_objs.get(0).get<Int>(valid_pk_name) == source_pk);
            wait_for_num_outgoing_links_to_equal(r2, r2_source_objs.get(0), expected_coll_size);
            auto linked_objects = test_type.get_links(r2_source_objs.get(0));
            require_links_to_match_ids(linked_objects, remaining_dest_object_ids);
            r2->begin_transaction();
            test_type.clear_collection(r2_source_objs.get(0));
            r2->commit_transaction();
            expected_coll_size = 0;
            REQUIRE(test_type.size_of_collection(r2_source_objs.get(0)) == expected_coll_size);
        }

        { // expect an empty collection
            REQUIRE(!coll_cleared);
            REQUIRE(r1_source_objs.size() == 1);
            wait_for_num_outgoing_links_to_equal(r1, r1_source_objs.get(0), expected_coll_size);
            advance_and_notify(*r1);
            REQUIRE(coll_cleared);
        }
    }
}

TEMPLATE_TEST_CASE("app: partition types", "[sync][pbs][app][partition][baas]", cf::Int, cf::String, cf::OID,
                   cf::UUID, cf::BoxedOptional<cf::Int>, cf::UnboxedOptional<cf::String>, cf::BoxedOptional<cf::OID>,
                   cf::BoxedOptional<cf::UUID>)
{
    const std::string valid_pk_name = "_id";
    const std::string partition_key_col_name = "partition_key_prop";
    const std::string table_name = "class_partition_test_type";
    auto partition_property = Property(partition_key_col_name, TestType::property_type);
    Schema schema = {{Group::table_name_to_class_name(table_name),
                      {
                          {valid_pk_name, PropertyType::Int, true},
                          partition_property,
                      }}};
    auto server_app_config = minimal_app_config("partition_types_app_name", schema);
    server_app_config.partition_key = partition_property;
    TestAppSession test_session(create_app(server_app_config));
    auto app = test_session.app();

    auto wait_for_num_objects_to_equal = [](realm::SharedRealm r, const std::string& table_name, size_t count) {
        timed_sleeping_wait_for([&]() -> bool {
            r->refresh();
            TableRef dest = r->read_group().get_table(table_name);
            size_t cur_count = dest->size();
            return cur_count == count;
        });
    };
    using T = typename TestType::Type;
    CppContext c;
    auto create_object = [&](realm::SharedRealm r, int64_t val, std::any partition) {
        r->begin_transaction();
        auto object = Object::create(
            c, r, Group::table_name_to_class_name(table_name),
            std::any(realm::AnyDict{{valid_pk_name, std::any(val)}, {partition_key_col_name, partition}}),
            CreatePolicy::ForceCreate);
        r->commit_transaction();
    };

    auto get_bson = [](T val) -> bson::Bson {
        if constexpr (std::is_same_v<T, StringData>) {
            return val.is_null() ? bson::Bson(util::none) : bson::Bson(val);
        }
        else if constexpr (TestType::is_optional) {
            return val ? bson::Bson(*val) : bson::Bson(util::none);
        }
        else {
            return bson::Bson(val);
        }
    };

    SECTION("can round trip an object") {
        auto values = TestType::values();
        auto user1 = app->current_user();
        create_user_and_log_in(app);
        auto user2 = app->current_user();
        REQUIRE(user1);
        REQUIRE(user2);
        REQUIRE(user1 != user2);
        for (T partition_value : values) {
            SyncTestFile config1(user1, get_bson(partition_value), schema); // uses the current user created above
            auto r1 = realm::Realm::get_shared_realm(config1);
            Results r1_source_objs = realm::Results(r1, r1->read_group().get_table(table_name));

            SyncTestFile config2(user2, get_bson(partition_value), schema); // uses the user created above
            auto r2 = realm::Realm::get_shared_realm(config2);
            Results r2_source_objs = realm::Results(r2, r2->read_group().get_table(table_name));

            const int64_t pk_value = random_int();
            {
                REQUIRE(r1_source_objs.size() == 0);
                create_object(r1, pk_value, TestType::to_any(partition_value));
                REQUIRE(r1_source_objs.size() == 1);
                REQUIRE(r1_source_objs.get(0).get<T>(partition_key_col_name) == partition_value);
                REQUIRE(r1_source_objs.get(0).get<Int>(valid_pk_name) == pk_value);
            }
            {
                wait_for_num_objects_to_equal(r2, table_name, 1);
                REQUIRE(r2_source_objs.size() == 1);
                REQUIRE(r2_source_objs.size() == 1);
                REQUIRE(r2_source_objs.get(0).get<T>(partition_key_col_name) == partition_value);
                REQUIRE(r2_source_objs.get(0).get<Int>(valid_pk_name) == pk_value);
            }
        }
    }
}

TEST_CASE("app: full-text compatible with sync", "[sync][app][baas]") {
    const std::string valid_pk_name = "_id";

    Schema schema{
        {"TopLevel",
         {
             {valid_pk_name, PropertyType::ObjectId, Property::IsPrimary{true}},
             {"full_text", Property::IsFulltextIndexed{true}},
         }},
    };

    auto server_app_config = minimal_app_config("full_text", schema);
    auto app_session = create_app(server_app_config);
    const auto partition = random_string(100);
    TestAppSession test_session(app_session, nullptr);
    SyncTestFile config(test_session.app()->current_user(), partition, schema);
    SharedRealm realm;
    SECTION("sync open") {
        INFO("realm opened without async open");
        realm = Realm::get_shared_realm(config);
    }
    SECTION("async open") {
        INFO("realm opened with async open");
        auto async_open_task = Realm::get_synchronized_realm(config);

        auto realm_future = async_open_task->start();
        realm = Realm::get_shared_realm(std::move(realm_future.get()));
    }

    CppContext c(realm);
    auto obj_id_1 = ObjectId::gen();
    auto obj_id_2 = ObjectId::gen();
    realm->begin_transaction();
    Object::create(c, realm, "TopLevel", std::any(AnyDict{{"_id", obj_id_1}, {"full_text", "Hello, world!"s}}));
    Object::create(c, realm, "TopLevel", std::any(AnyDict{{"_id", obj_id_2}, {"full_text", "Hello, everyone!"s}}));
    realm->commit_transaction();

    auto table = realm->read_group().get_table("class_TopLevel");
    REQUIRE(table->search_index_type(table->get_column_key("full_text")) == IndexType::Fulltext);
    Results world_results(realm, Query(table).fulltext(table->get_column_key("full_text"), "world"));
    REQUIRE(world_results.size() == 1);
    REQUIRE(world_results.get<Obj>(0).get_primary_key() == Mixed{obj_id_1});
}

#endif // REALM_ENABLE_AUTH_TESTS

TEST_CASE("app: custom error handling", "[sync][app][custom errors]") {
    class CustomErrorTransport : public GenericNetworkTransport {
    public:
        CustomErrorTransport(int code, const std::string& message)
            : m_code(code)
            , m_message(message)
        {
        }

        void send_request_to_server(const Request&, util::UniqueFunction<void(const Response&)>&& completion) override
        {
            completion(Response{0, m_code, HttpHeaders(), m_message});
        }

    private:
        int m_code;
        std::string m_message;
    };

    SECTION("custom code and message is sent back") {
        OfflineAppSession offline_session({std::make_shared<CustomErrorTransport>(1001, "Boom!")});
        auto error = failed_log_in(offline_session.app());
        CHECK(error.is_custom_error());
        CHECK(*error.additional_status_code == 1001);
        CHECK(error.reason() == "Boom!");
    }
}

// MARK: - Unit Tests

static const std::string bad_access_token = "lolwut";
static const std::string dummy_device_id = "123400000000000000000000";

TEST_CASE("subscribable unit tests", "[sync][app]") {
    struct Foo : public Subscribable<Foo> {
        void event()
        {
            emit_change_to_subscribers(*this);
        }
    };

    auto foo = Foo();

    SECTION("subscriber receives events") {
        auto event_count = 0;
        auto token = foo.subscribe([&event_count](auto&) {
            event_count++;
        });

        foo.event();
        foo.event();
        foo.event();

        CHECK(event_count == 3);
    }

    SECTION("subscriber can unsubscribe") {
        auto event_count = 0;
        auto token = foo.subscribe([&event_count](auto&) {
            event_count++;
        });

        foo.event();
        CHECK(event_count == 1);

        foo.unsubscribe(token);
        foo.event();
        CHECK(event_count == 1);
    }

    SECTION("subscriber is unsubscribed on dtor") {
        auto event_count = 0;
        {
            auto token = foo.subscribe([&event_count](auto&) {
                event_count++;
            });

            foo.event();
            CHECK(event_count == 1);
        }
        foo.event();
        CHECK(event_count == 1);
    }

    SECTION("multiple subscribers receive events") {
        auto event_count = 0;
        {
            auto token1 = foo.subscribe([&event_count](auto&) {
                event_count++;
            });
            auto token2 = foo.subscribe([&event_count](auto&) {
                event_count++;
            });

            foo.event();
            CHECK(event_count == 2);
        }
        foo.event();
        CHECK(event_count == 2);
    }
}

TEST_CASE("app: login_with_credentials unit_tests", "[sync][app][user]") {
    auto transport = std::make_shared<UnitTestTransport>();
    OfflineAppSession::Config config{transport};
    transport->set_profile(profile_0);

    SECTION("login_anonymous good") {
        config.storage_path = util::make_temp_dir();
        config.metadata_mode = AppConfig::MetadataMode::NoEncryption;
        {
            config.delete_storage = false;
            OfflineAppSession oas(config);
            auto app = oas.app();
            auto user = log_in(app);

            REQUIRE(user->identities().size() == 1);
            CHECK(user->identities()[0].id == UnitTestTransport::identity_0_id);
            UserProfile user_profile = user->user_profile();

            CHECK(user_profile.name() == profile_0_name);
            CHECK(user_profile.first_name() == profile_0_first_name);
            CHECK(user_profile.last_name() == profile_0_last_name);
            CHECK(user_profile.email() == profile_0_email);
            CHECK(user_profile.picture_url() == profile_0_picture_url);
            CHECK(user_profile.gender() == profile_0_gender);
            CHECK(user_profile.birthday() == profile_0_birthday);
            CHECK(user_profile.min_age() == profile_0_min_age);
            CHECK(user_profile.max_age() == profile_0_max_age);
        }
        // assert everything is stored properly between runs
        {
            config.delete_storage = true; // clean up after this session
            OfflineAppSession oas(config);
            auto app = oas.app();
            REQUIRE(app->all_users().size() == 1);
            auto user = app->all_users()[0];
            REQUIRE(user->identities().size() == 1);
            CHECK(user->identities()[0].id == UnitTestTransport::identity_0_id);
            UserProfile user_profile = user->user_profile();

            CHECK(user_profile.name() == profile_0_name);
            CHECK(user_profile.first_name() == profile_0_first_name);
            CHECK(user_profile.last_name() == profile_0_last_name);
            CHECK(user_profile.email() == profile_0_email);
            CHECK(user_profile.picture_url() == profile_0_picture_url);
            CHECK(user_profile.gender() == profile_0_gender);
            CHECK(user_profile.birthday() == profile_0_birthday);
            CHECK(user_profile.min_age() == profile_0_min_age);
            CHECK(user_profile.max_age() == profile_0_max_age);
        }
    }

    SECTION("login_anonymous bad") {
        struct transport : UnitTestTransport {
            void send_request_to_server(const Request& request,
                                        util::UniqueFunction<void(const Response&)>&& completion) override
            {
                if (request.url.find("/login") != std::string::npos) {
                    completion({200, 0, {}, user_json(bad_access_token).dump()});
                }
                else {
                    UnitTestTransport::send_request_to_server(request, std::move(completion));
                }
            }
        };

        config.transport = instance_of<transport>;
        OfflineAppSession oas(config);
        auto error = failed_log_in(oas.app());
        CHECK(error.reason() == std::string("Could not log in user: received malformed JWT"));
        CHECK(error.code_string() == "BadToken");
        CHECK(error.is_json_error());
        CHECK(error.code() == ErrorCodes::BadToken);
    }

    SECTION("login_anonynous multiple users") {
        OfflineAppSession oas(config);
        auto app = oas.app();

        auto user1 = log_in(app);
        auto user2 = log_in(app, AppCredentials::anonymous(false));
        CHECK(user1 != user2);
    }
}

TEST_CASE("app: UserAPIKeyProviderClient unit_tests", "[sync][app][user][api key]") {
    OfflineAppSession oas({std::make_shared<UnitTestTransport>()});
    auto client = oas.app()->provider_client<App::UserAPIKeyProviderClient>();

    auto logged_in_user = oas.make_user();
    bool processed = false;
    ObjectId obj_id(UnitTestTransport::api_key_id.c_str());

    SECTION("create api key") {
        client.create_api_key(UnitTestTransport::api_key_name, logged_in_user,
                              [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
                                  REQUIRE_FALSE(error);
                                  CHECK(user_api_key.disabled == false);
                                  CHECK(user_api_key.id.to_string() == UnitTestTransport::api_key_id);
                                  CHECK(user_api_key.key == UnitTestTransport::api_key);
                                  CHECK(user_api_key.name == UnitTestTransport::api_key_name);
                              });
    }

    SECTION("fetch api key") {
        client.fetch_api_key(obj_id, logged_in_user, [&](App::UserAPIKey user_api_key, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(user_api_key.disabled == false);
            CHECK(user_api_key.id.to_string() == UnitTestTransport::api_key_id);
            CHECK(user_api_key.name == UnitTestTransport::api_key_name);
        });
    }

    SECTION("fetch api keys") {
        client.fetch_api_keys(logged_in_user,
                              [&](std::vector<App::UserAPIKey> user_api_keys, Optional<AppError> error) {
                                  REQUIRE_FALSE(error);
                                  CHECK(user_api_keys.size() == 2);
                                  for (auto user_api_key : user_api_keys) {
                                      CHECK(user_api_key.disabled == false);
                                      CHECK(user_api_key.id.to_string() == UnitTestTransport::api_key_id);
                                      CHECK(user_api_key.name == UnitTestTransport::api_key_name);
                                  }
                                  processed = true;
                              });
        CHECK(processed);
    }
}

TEST_CASE("app: user_semantics", "[sync][app][user]") {
    OfflineAppSession oas;
    auto app = oas.app();

    const auto login_user_email_pass = [=] {
        return log_in(app, AppCredentials::username_password("bob", "thompson"));
    };
    const auto login_user_anonymous = [=] {
        return log_in(app, AppCredentials::anonymous());
    };

    CHECK(!app->current_user());

    int event_processed = 0;
    auto token = app->subscribe([&](auto&) {
        event_processed++;
        // Read the current user to verify that doing so does not deadlock
        app->current_user();
    });

    SECTION("current user is populated") {
        const auto user1 = login_user_anonymous();
        CHECK(app->current_user()->user_id() == user1->user_id());
        CHECK(event_processed == 1);
    }

    SECTION("current user is updated on login") {
        const auto user1 = login_user_anonymous();
        CHECK(app->current_user()->user_id() == user1->user_id());
        const auto user2 = login_user_email_pass();
        CHECK(app->current_user()->user_id() == user2->user_id());
        CHECK(user1->user_id() != user2->user_id());
        CHECK(event_processed == 2);
    }

    SECTION("current user is updated to last used user on logout") {
        const auto user1 = login_user_anonymous();
        CHECK(app->current_user()->user_id() == user1->user_id());
        CHECK(app->all_users()[0]->state() == SyncUser::State::LoggedIn);

        const auto user2 = login_user_email_pass();
        CHECK(app->all_users()[0]->state() == SyncUser::State::LoggedIn);
        CHECK(app->all_users()[1]->state() == SyncUser::State::LoggedIn);
        CHECK(app->current_user()->user_id() == user2->user_id());
        CHECK(user1 != user2);

        // should reuse existing session
        const auto user3 = login_user_anonymous();
        CHECK(user3 == user1);

        auto user_events_processed = 0;
        auto _ = user3->subscribe([&user_events_processed](auto&) {
            user_events_processed++;
        });

        app->log_out([](auto) {});
        CHECK(user_events_processed == 1);
        REQUIRE(app->current_user());
        CHECK(app->current_user()->user_id() == user2->user_id());

        CHECK(app->all_users().size() == 1);
        CHECK(app->all_users()[0]->state() == SyncUser::State::LoggedIn);

        CHECK(event_processed == 4);
    }

    SECTION("anon users are removed on logout") {
        const auto user1 = login_user_anonymous();
        CHECK(app->current_user()->user_id() == user1->user_id());
        CHECK(app->all_users()[0]->state() == SyncUser::State::LoggedIn);

        const auto user2 = login_user_anonymous();
        CHECK(app->all_users()[0]->state() == SyncUser::State::LoggedIn);
        CHECK(app->all_users().size() == 1);
        CHECK(app->current_user()->user_id() == user2->user_id());
        CHECK(user1->user_id() == user2->user_id());

        app->log_out([](auto) {});
        CHECK(app->all_users().size() == 0);

        CHECK(event_processed == 3);
    }

    SECTION("logout user") {
        auto user1 = login_user_email_pass();
        auto user2 = login_user_anonymous();

        // Anonymous users are special
        app->log_out(user2, [](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });
        CHECK(user2->state() == SyncUser::State::Removed);

        // Other users can be LoggedOut
        app->log_out(user1, [](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });
        CHECK(user1->state() == SyncUser::State::LoggedOut);

        // Logging out already logged out users does nothing
        app->log_out(user1, [](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });
        CHECK(user1->state() == SyncUser::State::LoggedOut);

        app->log_out(user2, [](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });
        CHECK(user2->state() == SyncUser::State::Removed);

        CHECK(event_processed == 4);
    }

    SECTION("unsubscribed observers no longer process events") {
        app->unsubscribe(token);

        const auto user1 = login_user_anonymous();
        CHECK(app->current_user()->user_id() == user1->user_id());
        CHECK(app->all_users()[0]->state() == SyncUser::State::LoggedIn);

        const auto user2 = login_user_anonymous();
        CHECK(app->all_users()[0]->state() == SyncUser::State::LoggedIn);
        CHECK(app->all_users().size() == 1);
        CHECK(app->current_user()->user_id() == user2->user_id());
        CHECK(user1->user_id() == user2->user_id());

        app->log_out([](auto) {});
        CHECK(app->all_users().size() == 0);

        CHECK(event_processed == 0);
    }
}

namespace {
struct ErrorCheckingTransport : public GenericNetworkTransport {
    ErrorCheckingTransport(Response* r)
        : m_response(r)
    {
    }
    void send_request_to_server(const Request& request,
                                util::UniqueFunction<void(const Response&)>&& completion) override
    {
        // Make sure to return a valid location response
        if (request.url.find("/location") != std::string::npos) {
            completion(Response{200,
                                0,
                                {{"content-type", "application/json"}},
                                "{\"deployment_model\":\"GLOBAL\",\"location\":\"US-VA\",\"hostname\":"
                                "\"http://some.fake.url\",\"ws_hostname\":\"ws://some.fake.url\"}"});
            return;
        }

        completion(Response(*m_response));
    }

private:
    Response* m_response;
};
} // namespace

TEST_CASE("app: response error handling", "[sync][app]") {
    std::string response_body = nlohmann::json({{"access_token", good_access_token},
                                                {"refresh_token", good_access_token},
                                                {"user_id", "Brown Bear"},
                                                {"device_id", "Panda Bear"}})
                                    .dump();

    Response response{200, 0, {{"Content-Type", "text/plain"}}, response_body};

    OfflineAppSession oas({std::make_shared<ErrorCheckingTransport>(&response)});
    auto app = oas.app();

    SECTION("http 404") {
        response.http_status_code = 404;
        auto error = failed_log_in(app);
        CHECK(!error.is_json_error());
        CHECK(!error.is_custom_error());
        CHECK(!error.is_service_error());
        CHECK(error.is_http_error());
        CHECK(*error.additional_status_code == 404);
        CHECK_THAT(std::string(error.reason()), ContainsSubstring("http error code considered fatal"));
    }
    SECTION("http 500") {
        response.http_status_code = 500;
        auto error = failed_log_in(app);
        CHECK(!error.is_json_error());
        CHECK(!error.is_custom_error());
        CHECK(!error.is_service_error());
        CHECK(error.is_http_error());
        CHECK(*error.additional_status_code == 500);
        CHECK_THAT(std::string(error.reason()), ContainsSubstring("http error code considered fatal"));
        CHECK(error.link_to_server_logs.empty());
    }

    SECTION("custom error code") {
        response.custom_status_code = 42;
        response.body = "Custom error message";
        auto error = failed_log_in(app);
        CHECK(!error.is_http_error());
        CHECK(!error.is_json_error());
        CHECK(!error.is_service_error());
        CHECK(error.is_custom_error());
        CHECK(*error.additional_status_code == 42);
        CHECK(error.reason() == std::string("Custom error message"));
        CHECK(error.link_to_server_logs.empty());
    }

    SECTION("session error code") {
        response.headers = HttpHeaders{{"Content-Type", "application/json"}};
        response.http_status_code = 400;
        response.body = nlohmann::json({{"error_code", "MongoDBError"},
                                        {"error", "a fake MongoDB error message!"},
                                        {"access_token", good_access_token},
                                        {"refresh_token", good_access_token},
                                        {"user_id", "Brown Bear"},
                                        {"device_id", "Panda Bear"},
                                        {"link", "http://...whatever the server passes us"}})
                            .dump();
        auto error = failed_log_in(app);
        CHECK(!error.is_http_error());
        CHECK(!error.is_json_error());
        CHECK(!error.is_custom_error());
        CHECK(error.is_service_error());
        CHECK(error.code() == ErrorCodes::MongoDBError);
        CHECK(error.reason() == std::string("a fake MongoDB error message!"));
        CHECK(error.link_to_server_logs == std::string("http://...whatever the server passes us"));
    }

    SECTION("json error code") {
        response.body = "this: is not{} a valid json body!";
        auto error = failed_log_in(app);
        CHECK(!error.is_http_error());
        CHECK(error.is_json_error());
        CHECK(!error.is_custom_error());
        CHECK(!error.is_service_error());
        CHECK(error.code() == ErrorCodes::MalformedJson);
        CHECK(error.reason() ==
              std::string("[json.exception.parse_error.101] parse error at line 1, column 2: syntax error "
                          "while parsing value - invalid literal; last read: 'th'"));
        CHECK(error.code_string() == "MalformedJson");
    }
}

TEST_CASE("app: switch user", "[sync][app][user]") {
    OfflineAppSession oas;
    auto app = oas.app();

    bool processed = false;

    SECTION("switch user expect success") {
        CHECK(app->all_users().size() == 0);

        // Log in user 1
        auto user_a = log_in(app, AppCredentials::username_password("test@10gen.com", "password"));
        CHECK(app->current_user() == user_a);

        // Log in user 2
        auto user_b = log_in(app, AppCredentials::username_password("test2@10gen.com", "password"));
        CHECK(app->current_user() == user_b);

        CHECK(app->all_users().size() == 2);

        app->switch_user(user_a);
        CHECK(app->current_user() == user_a);

        app->switch_user(user_b);

        CHECK(app->current_user() == user_b);
        processed = true;
        CHECK(processed);
    }

    SECTION("cannot switch to a logged out user") {
        CHECK(app->all_users().size() == 0);

        // Log in user 1
        auto user_a = log_in(app, AppCredentials::username_password("test@10gen.com", "password"));
        CHECK(app->current_user() == user_a);

        app->log_out([&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        CHECK(app->current_user() == nullptr);
        CHECK(user_a->state() == SyncUser::State::LoggedOut);

        // Log in user 2
        auto user_b = log_in(app, AppCredentials::username_password("test2@10gen.com", "password"));
        CHECK(app->current_user() == user_b);
        CHECK(app->all_users().size() == 2);

        REQUIRE_THROWS_AS(app->switch_user(user_a), AppError);
        CHECK(app->current_user() == user_b);
    }
}

TEST_CASE("app: remove user", "[sync][app][user]") {
    OfflineAppSession oas;
    auto app = oas.app();

    SECTION("remove anonymous user") {
        CHECK(app->all_users().size() == 0);

        // Log in user 1
        auto user_a = log_in(app);
        CHECK(user_a->state() == SyncUser::State::LoggedIn);

        app->log_out(user_a, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
            // a logged out anon user will be marked as Removed, not LoggedOut
            CHECK(user_a->state() == SyncUser::State::Removed);
        });
        CHECK(app->all_users().empty());

        app->remove_user(user_a, [&](Optional<AppError> error) {
            CHECK(error->reason() == "User has already been removed");
            CHECK(app->all_users().size() == 0);
        });

        // Log in user 2
        auto user_b = log_in(app);
        CHECK(app->current_user() == user_b);
        CHECK(user_b->state() == SyncUser::State::LoggedIn);
        CHECK(app->all_users().size() == 1);

        app->remove_user(user_b, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
            CHECK(app->all_users().size() == 0);
        });

        CHECK(app->current_user() == nullptr);

        // check both handles are no longer valid
        CHECK(user_a->state() == SyncUser::State::Removed);
        CHECK(user_b->state() == SyncUser::State::Removed);
    }

    SECTION("remove user with credentials") {
        CHECK(app->all_users().size() == 0);
        CHECK(app->current_user() == nullptr);

        auto user = log_in(app, AppCredentials::username_password("email", "pass"));

        CHECK(user->state() == SyncUser::State::LoggedIn);

        app->log_out(user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        CHECK(user->state() == SyncUser::State::LoggedOut);

        app->remove_user(user, [&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });
        CHECK(app->all_users().size() == 0);

        Optional<AppError> error;
        app->remove_user(user, [&](Optional<AppError> err) {
            error = err;
        });
        CHECK(error->code() > 0);
        CHECK(app->all_users().size() == 0);
        CHECK(user->state() == SyncUser::State::Removed);
    }
}

TEST_CASE("app: link_user", "[sync][app][user]") {
    OfflineAppSession oas;
    auto app = oas.app();

    auto email = util::format("realm_tests_do_autoverify%1@%2.com", random_string(10), random_string(10));
    auto password = random_string(10);

    auto custom_credentials = AppCredentials::facebook("a_token");
    auto email_pass_credentials = AppCredentials::username_password(email, password);

    auto sync_user = log_in(app, email_pass_credentials);
    REQUIRE(sync_user->identities().size() == 2);
    CHECK(sync_user->identities()[0].provider_type == IdentityProviderUsernamePassword);

    SECTION("successful link") {
        bool processed = false;
        app->link_user(sync_user, custom_credentials, [&](std::shared_ptr<SyncUser> user, Optional<AppError> error) {
            REQUIRE_FALSE(error);
            REQUIRE(user);
            CHECK(user->user_id() == sync_user->user_id());
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("link_user should fail when logged out") {
        app->log_out([&](Optional<AppError> error) {
            REQUIRE_FALSE(error);
        });

        bool processed = false;
        app->link_user(sync_user, custom_credentials, [&](std::shared_ptr<SyncUser> user, Optional<AppError> error) {
            CHECK(error->reason() == "The specified user is not logged in.");
            CHECK(!user);
            processed = true;
        });
        CHECK(processed);
    }
}

TEST_CASE("app: auth providers", "[sync][app][user]") {
    SECTION("auth providers facebook") {
        auto credentials = AppCredentials::facebook("a_token");
        CHECK(credentials.provider() == AuthProvider::FACEBOOK);
        CHECK(credentials.provider_as_string() == IdentityProviderFacebook);
        CHECK(credentials.serialize_as_bson() ==
              bson::BsonDocument{{"provider", "oauth2-facebook"}, {"accessToken", "a_token"}});
    }

    SECTION("auth providers anonymous") {
        auto credentials = AppCredentials::anonymous();
        CHECK(credentials.provider() == AuthProvider::ANONYMOUS);
        CHECK(credentials.provider_as_string() == IdentityProviderAnonymous);
        CHECK(credentials.serialize_as_bson() == bson::BsonDocument{{"provider", "anon-user"}});
    }

    SECTION("auth providers anonymous no reuse") {
        auto credentials = AppCredentials::anonymous(false);
        CHECK(credentials.provider() == AuthProvider::ANONYMOUS_NO_REUSE);
        CHECK(credentials.provider_as_string() == IdentityProviderAnonymous);
        CHECK(credentials.serialize_as_bson() == bson::BsonDocument{{"provider", "anon-user"}});
    }

    SECTION("auth providers google authCode") {
        auto credentials = AppCredentials::google(AuthCode("a_token"));
        CHECK(credentials.provider() == AuthProvider::GOOGLE);
        CHECK(credentials.provider_as_string() == IdentityProviderGoogle);
        CHECK(credentials.serialize_as_bson() ==
              bson::BsonDocument{{"provider", "oauth2-google"}, {"authCode", "a_token"}});
    }

    SECTION("auth providers google idToken") {
        auto credentials = AppCredentials::google(IdToken("a_token"));
        CHECK(credentials.provider() == AuthProvider::GOOGLE);
        CHECK(credentials.provider_as_string() == IdentityProviderGoogle);
        CHECK(credentials.serialize_as_bson() ==
              bson::BsonDocument{{"provider", "oauth2-google"}, {"id_token", "a_token"}});
    }

    SECTION("auth providers apple") {
        auto credentials = AppCredentials::apple("a_token");
        CHECK(credentials.provider() == AuthProvider::APPLE);
        CHECK(credentials.provider_as_string() == IdentityProviderApple);
        CHECK(credentials.serialize_as_bson() ==
              bson::BsonDocument{{"provider", "oauth2-apple"}, {"id_token", "a_token"}});
    }

    SECTION("auth providers custom") {
        auto credentials = AppCredentials::custom("a_token");
        CHECK(credentials.provider() == AuthProvider::CUSTOM);
        CHECK(credentials.provider_as_string() == IdentityProviderCustom);
        CHECK(credentials.serialize_as_bson() ==
              bson::BsonDocument{{"provider", "custom-token"}, {"token", "a_token"}});
    }

    SECTION("auth providers username password") {
        auto credentials = AppCredentials::username_password("user", "pass");
        CHECK(credentials.provider() == AuthProvider::USERNAME_PASSWORD);
        CHECK(credentials.provider_as_string() == IdentityProviderUsernamePassword);
        CHECK(credentials.serialize_as_bson() ==
              bson::BsonDocument{{"provider", "local-userpass"}, {"username", "user"}, {"password", "pass"}});
    }

    SECTION("auth providers function") {
        bson::BsonDocument function_params{{"name", "mongo"}};
        auto credentials = AppCredentials::function(function_params);
        CHECK(credentials.provider() == AuthProvider::FUNCTION);
        CHECK(credentials.provider_as_string() == IdentityProviderFunction);
        CHECK(credentials.serialize_as_bson() == bson::BsonDocument{{"name", "mongo"}});
    }

    SECTION("auth providers api key") {
        auto credentials = AppCredentials::api_key("a key");
        CHECK(credentials.provider() == AuthProvider::API_KEY);
        CHECK(credentials.provider_as_string() == IdentityProviderAPIKey);
        CHECK(credentials.serialize_as_bson() == bson::BsonDocument{{"provider", "api-key"}, {"key", "a key"}});
        CHECK(enum_from_provider_type(provider_type_from_enum(AuthProvider::API_KEY)) == AuthProvider::API_KEY);
    }
}

TEST_CASE("app: refresh access token unit tests", "[sync][app][user][token]") {
    SECTION("refresh custom data happy path") {
        static bool session_route_hit = false;

        struct transport : UnitTestTransport {
            void send_request_to_server(const Request& request,
                                        util::UniqueFunction<void(const Response&)>&& completion) override
            {
                if (request.url.find("/session") != std::string::npos) {
                    session_route_hit = true;
                    nlohmann::json json{{"access_token", good_access_token}};
                    completion({200, 0, {}, json.dump()});
                }
                else {
                    UnitTestTransport::send_request_to_server(request, std::move(completion));
                }
            }
        };
        OfflineAppSession oas(OfflineAppSession::Config{instance_of<transport>});
        auto app = oas.app();
        oas.make_user();

        bool processed = false;
        app->refresh_custom_data(app->current_user(), [&](const Optional<AppError>& error) {
            REQUIRE_FALSE(error);
            CHECK(session_route_hit);
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("refresh custom data sad path") {
        static bool session_route_hit = false;

        struct transport : UnitTestTransport {
            void send_request_to_server(const Request& request,
                                        util::UniqueFunction<void(const Response&)>&& completion) override
            {
                if (request.url.find("/session") != std::string::npos) {
                    session_route_hit = true;
                    nlohmann::json json{{"access_token", bad_access_token}};
                    completion({200, 0, {}, json.dump()});
                }
                else {
                    UnitTestTransport::send_request_to_server(request, std::move(completion));
                }
            }
        };

        OfflineAppSession oas(OfflineAppSession::Config{instance_of<transport>});
        auto app = oas.app();
        oas.make_user();

        bool processed = false;
        app->refresh_custom_data(app->current_user(), [&](const Optional<AppError>& error) {
            CHECK(error->reason() == "malformed JWT");
            CHECK(error->code() == ErrorCodes::BadToken);
            CHECK(session_route_hit);
            processed = true;
        });
        CHECK(processed);
    }

    SECTION("refresh token ensure flow is correct") {
        /*
         Expected flow:
         Login - this gets access and refresh tokens
         Get profile - throw back a 401 error
         Refresh token - get a new token for the user
         Get profile - get the profile with the new token
         */
        struct transport : GenericNetworkTransport {
            enum class TestState { unknown, location, login, profile_1, refresh, profile_2 };
            TestingStateMachine<TestState> state{TestState::unknown};
            void send_request_to_server(const Request& request,
                                        util::UniqueFunction<void(const Response&)>&& completion) override
            {
                if (request.url.find("/login") != std::string::npos) {
                    CHECK(state.get() == TestState::location);
                    state.transition_to(TestState::login);
                    completion({200, 0, {}, user_json(good_access_token).dump()});
                }
                else if (request.url.find("/profile") != std::string::npos) {
                    auto item = AppUtils::find_header("Authorization", request.headers);
                    CHECK(item);
                    auto access_token = item->second;
                    // simulated bad token request
                    if (access_token.find(good_access_token2) != std::string::npos) {
                        CHECK(state.get() == TestState::refresh);
                        state.transition_to(TestState::profile_2);
                        completion({200, 0, {}, user_profile_json().dump()});
                    }
                    else if (access_token.find(good_access_token) != std::string::npos) {
                        CHECK(state.get() == TestState::login);
                        state.transition_to(TestState::profile_1);
                        completion({401, 0, {}});
                    }
                }
                else if (request.url.find("/session") != std::string::npos && request.method == HttpMethod::post) {
                    CHECK(state.get() == TestState::profile_1);
                    state.transition_to(TestState::refresh);
                    nlohmann::json json{{"access_token", good_access_token2}};
                    completion({200, 0, {}, json.dump()});
                }
                else if (request.url.find("/location") != std::string::npos) {
                    CHECK(state.get() == TestState::unknown);
                    state.transition_to(TestState::location);
                    CHECK(request.method == HttpMethod::get);
                    completion({200,
                                0,
                                {},
                                "{\"deployment_model\":\"GLOBAL\",\"location\":\"US-VA\",\"hostname\":"
                                "\"http://localhost:9090\",\"ws_hostname\":\"ws://localhost:9090\"}"});
                }
                else {
                    FAIL("Unexpected request in test code" + request.url);
                }
            }
        };

        OfflineAppSession oas(OfflineAppSession::Config{instance_of<transport>});
        auto app = oas.app();
        REQUIRE(log_in(app));
    }
}

TEST_CASE("app: app released during async operation", "[app][user]") {
    struct Transport : public UnitTestTransport {
        std::string endpoint_to_hook;
        std::optional<Request> stored_request;
        util::UniqueFunction<void(const Response&)> stored_completion;

        void send_request_to_server(const Request& request,
                                    util::UniqueFunction<void(const Response&)>&& completion) override
        {
            // Store the completion handler for the chosen endpoint so that we can
            // invoke it after releasing the test's references to the App to
            // verify that it doesn't crash
            if (request.url.find(endpoint_to_hook) != std::string::npos) {
                REQUIRE_FALSE(stored_request);
                REQUIRE_FALSE(stored_completion);
                stored_request = request;
                stored_completion = std::move(completion);
                return;
            }

            UnitTestTransport::send_request_to_server(request, std::move(completion));
        }

        bool has_stored() const
        {
            return !!stored_completion;
        }

        void send_stored()
        {
            REQUIRE(stored_request);
            REQUIRE(stored_completion);
            UnitTestTransport::send_request_to_server(*stored_request, std::move(stored_completion));
            stored_request.reset();
            stored_completion = nullptr;
        }
    };
    auto transport = std::make_shared<Transport>();
    test_util::TestDirGuard base_path(util::make_temp_dir(), false);
    AppConfig app_config;
    set_app_config_defaults(app_config, transport);
    app_config.base_file_path = base_path;

    SECTION("login") {
        transport->endpoint_to_hook = GENERATE("/location", "/login", "/profile");
        bool called = false;
        {
            auto app = App::get_app(App::CacheMode::Disabled, app_config);
            app->log_in_with_credentials(AppCredentials::anonymous(),
                                         [&](std::shared_ptr<SyncUser> user, util::Optional<AppError> error) mutable {
                                             REQUIRE_FALSE(error);
                                             REQUIRE(user);
                                             REQUIRE(user->is_logged_in());
                                             called = true;
                                         });
            REQUIRE(transport->has_stored());
        }
        REQUIRE_FALSE(called);
        transport->send_stored();
        REQUIRE(called);
    }

    SECTION("access token refresh") {
        transport->endpoint_to_hook = "/auth/session";
        SECTION("directly via user") {
            bool completion_called = false;
            {
                auto app = App::get_app(App::CacheMode::Disabled, app_config);
                create_user_and_log_in(app);
                app->current_user()->refresh_custom_data([&](std::optional<app::AppError> error) {
                    REQUIRE_FALSE(error);
                    completion_called = true;
                });
                REQUIRE(transport->has_stored());
            }

            REQUIRE_FALSE(completion_called);
            transport->send_stored();
            REQUIRE(completion_called);
        }

        SECTION("via sync session") {
            {
                auto app = App::get_app(App::CacheMode::Disabled, app_config);
                create_user_and_log_in(app);
                auto user = app->current_user();
                SyncTestFile config(user, bson::Bson("test"));
                // give the user an expired access token so that the first use will try to refresh it
                user->update_data_for_testing([](auto& data) {
                    data.access_token = RealmJWT(encode_fake_jwt("token", 123, 456));
                });
                REQUIRE_FALSE(transport->stored_completion);
                auto realm = Realm::get_shared_realm(config);
                REQUIRE(transport->has_stored());
            }
            transport->send_stored();
        }
    }

    REQUIRE_FALSE(transport->has_stored());
}

TEST_CASE("app: make_streaming_request", "[sync][app][streaming]") {
    constexpr uint64_t timeout_ms = 60000; // this is the default
    OfflineAppSession oas({std::make_shared<UnitTestTransport>(timeout_ms)});
    auto app = oas.app();

    auto user = log_in(app);

    using Headers = decltype(Request().headers);

    const auto url_prefix = "https://some.fake.url/api/client/v2.0/app/app_id/functions/call?baas_request="sv;
    const auto get_request_args = [&](const Request& req) {
        REQUIRE(req.url.substr(0, url_prefix.size()) == url_prefix);
        auto args = req.url.substr(url_prefix.size());
        if (auto amp = args.find('&'); amp != std::string::npos) {
            args.resize(amp);
        }

        auto vec = util::base64_decode_to_vector(util::uri_percent_decode(args));
        REQUIRE(!!vec);
        auto parsed = bson::parse({vec->data(), vec->size()});
        REQUIRE(parsed.type() == bson::Bson::Type::Document);
        auto out = parsed.operator const bson::BsonDocument&();
        CHECK(out.size() == 3);
        return out;
    };

    const auto make_request = [&](std::shared_ptr<User> user, auto&&... args) {
        auto req = app->make_streaming_request(user, "func", bson::BsonArray{args...}, {"svc"});
        CHECK(req.method == HttpMethod::get);
        CHECK(req.body == "");
        CHECK(req.headers == Headers{{"Accept", "text/event-stream"}});
        CHECK(req.timeout_ms == timeout_ms);

        auto req_args = get_request_args(req);
        CHECK(req_args["name"] == "func");
        CHECK(req_args["service"] == "svc");
        CHECK(req_args["arguments"] == bson::BsonArray{args...});

        return req;
    };

    SECTION("no args") {
        auto req = make_request(nullptr);
        CHECK(req.url.find('&') == std::string::npos);
    }
    SECTION("args") {
        auto req = make_request(nullptr, "arg1", "arg2");
        CHECK(req.url.find('&') == std::string::npos);
    }
    SECTION("percent encoding") {
        // These force the base64 encoding to have + and / bytes and = padding, all of which are uri encoded.
        auto req = make_request(nullptr, ">>>>>?????");

        CHECK(req.url.find('&') == std::string::npos);
        CHECK_THAT(req.url, ContainsSubstring("%2B"));     // + (from >)
        CHECK_THAT(req.url, ContainsSubstring("%2F"));     // / (from ?)
        CHECK_THAT(req.url, ContainsSubstring("%3D"));     // = (tail padding)
        CHECK(req.url.rfind("%3D") == req.url.size() - 3); // = (tail padding)
    }
    SECTION("with user") {
        auto req = make_request(user, "arg1", "arg2");

        auto amp = req.url.find('&');
        REQUIRE(amp != std::string::npos);
        auto tail = req.url.substr(amp);
        REQUIRE(tail == ("&baas_at=" + user->access_token()));
    }
}

TEST_CASE("app: sync_user_profile unit tests", "[sync][app][user]") {
    SECTION("with empty map") {
        auto profile = UserProfile(bson::BsonDocument());
        CHECK(profile.name() == util::none);
        CHECK(profile.email() == util::none);
        CHECK(profile.picture_url() == util::none);
        CHECK(profile.first_name() == util::none);
        CHECK(profile.last_name() == util::none);
        CHECK(profile.gender() == util::none);
        CHECK(profile.birthday() == util::none);
        CHECK(profile.min_age() == util::none);
        CHECK(profile.max_age() == util::none);
    }
    SECTION("with full map") {
        auto profile = UserProfile(bson::BsonDocument({
            {"first_name", "Jan"},
            {"last_name", "Jaanson"},
            {"name", "Jan Jaanson"},
            {"email", "jan.jaanson@jaanson.com"},
            {"gender", "none"},
            {"birthday", "January 1, 1970"},
            {"min_age", "0"},
            {"max_age", "100"},
            {"picture_url", "some"},
        }));
        CHECK(profile.name() == "Jan Jaanson");
        CHECK(profile.email() == "jan.jaanson@jaanson.com");
        CHECK(profile.picture_url() == "some");
        CHECK(profile.first_name() == "Jan");
        CHECK(profile.last_name() == "Jaanson");
        CHECK(profile.gender() == "none");
        CHECK(profile.birthday() == "January 1, 1970");
        CHECK(profile.min_age() == "0");
        CHECK(profile.max_age() == "100");
    }
}

TEST_CASE("app: shared instances", "[sync][app]") {
    test_util::TestDirGuard test_dir(util::make_temp_dir(), false);

    AppConfig base_config;
    set_app_config_defaults(base_config, instance_of<UnitTestTransport>);
    base_config.base_file_path = test_dir;

    auto config1 = base_config;
    config1.app_id = "app1";

    auto config2 = base_config;
    config2.app_id = "app1";
    config2.base_url = std::string(App::default_base_url());

    auto config3 = base_config;
    config3.app_id = "app2";

    auto config4 = base_config;
    config4.app_id = "app2";
    config4.base_url = "http://localhost:9090";

    // should all point to same underlying app
    auto app1_1 = App::get_app(app::App::CacheMode::Enabled, config1);
    auto app1_2 = App::get_app(app::App::CacheMode::Enabled, config1);
    auto app1_3 = App::get_cached_app(config1.app_id, config1.base_url);
    auto app1_4 = App::get_app(app::App::CacheMode::Enabled, config2);
    auto app1_5 = App::get_cached_app(config1.app_id);

    CHECK(app1_1 == app1_2);
    CHECK(app1_1 == app1_3);
    CHECK(app1_1 == app1_4);
    CHECK(app1_1 == app1_5);

    // config3 and config4 should point to different apps
    auto app2_1 = App::get_app(app::App::CacheMode::Enabled, config3);
    auto app2_2 = App::get_cached_app(config3.app_id, config3.base_url);
    auto app2_3 = App::get_app(app::App::CacheMode::Enabled, config4);
    auto app2_4 = App::get_cached_app(config3.app_id);
    auto app2_5 = App::get_cached_app(config4.app_id, "https://some.different.url");

    CHECK(app2_1 == app2_2);
    CHECK(app2_1 != app2_3);
    CHECK(app2_4 != nullptr);
    CHECK(app2_5 == nullptr);

    CHECK(app1_1 != app2_1);
    CHECK(app1_1 != app2_3);
    CHECK(app1_1 != app2_4);
}

realm / realm-core / jorgen.edelbo_402

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous