cloud/test/util_test.cpp

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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 "cpp/util.h"

#include <aws/core/client/ClientConfiguration.h>
#include <bthread/countdown_event.h>

#include <stdexcept>
#include <string>
#include <string_view>
#include <tuple>
#include <vector>

#include "common/config.h"
#include "common/logging.h"
#include "common/simple_thread_pool.h"
#include "common/string_util.h"
#include "cpp/aws_common.h"
#include "cpp/sync_point.h"
#include "gtest/gtest.h"
#include "recycler/recycler.h"
#include "recycler/s3_accessor.h"
#include "recycler/sync_executor.h"

using namespace doris::cloud;

int main(int argc, char** argv) {
    const auto* conf_file = "doris_cloud.conf";
    if (!config::init(conf_file, true)) {
        std::cerr << "failed to init config file, conf=" << conf_file << std::endl;
        return -1;
    }
    if (!::init_glog("util")) {
        std::cerr << "failed to init glog" << std::endl;
        return -1;
    }

    ::testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}

TEST(StringUtilTest, test_string_strip) {
    // clang-format off
    //                       str         expect          to_drop
    std::vector<std::tuple<std::string, std::string, std::string>> leading_inputs {
        {""       , ""      , ""    },
        {""       , ""      , "/"   },
        {"/"      , ""      , "/"   },
        {"\t////" , ""      , "/ \t"},
        {"/a///"  , "a///"  , "/"   },
        {"/a/b/c/", "a/b/c/", "/"   },
        {"a/b/c/" , "a/b/c/", "/"   },
        {"a/b/c/" , "/b/c/" , "a"   },
    };
    int idx = 0;
    for (auto&& i : leading_inputs) {
        doris::cloud::strip_leading(std::get<0>(i), std::get<2>(i));
        EXPECT_EQ(std::get<0>(i), std::get<1>(i)) << " index=" << idx;
        ++idx;
    }

    idx = 0;
    std::vector<std::tuple<std::string, std::string, std::string>> trailing_inputs {
        {""       , ""      , ""    },
        {"/"      , ""      , "/"   },
        {"////\t" , ""      , "/ \t"},
        {"/a///"  , "/a"    ,  "/"  },
        {"/a/b/c/", "/a/b/c", "/"   },
        {"a/b/c/" , "a/b/c" , "/"   },
        {"a/b/c"  , "a/b/c" , "/"   },
        {"a/b/c"  , "a/b/"  , "c"   },
    };
    for (auto&& i : trailing_inputs) {
        doris::cloud::strip_trailing(std::get<0>(i), std::get<2>(i));
        EXPECT_EQ(std::get<0>(i), std::get<1>(i)) << " index=" << idx;
        ++idx;
    }

    idx = 0;
    std::vector<std::tuple<std::string, std::string>> trim_inputs {
        {""              , ""     },
        {""              , ""     },
        {"/"             , ""     },
        {"\t////"        , ""     },
        {"/a ///"        , "a"   },
        {"/a/b/c/"       , "a/b/c"},
        {"a/b/c/"        , "a/b/c"},
        {"a/b/c"         , "a/b/c"},
        {"\t/bbc///"     , "bbc"  },
        {"ab c"          , "ab c" },
        {"\t  /a/b/c \t/", "a/b/c"},
    };
    for (auto&& i : trim_inputs) {
        doris::cloud::trim(std::get<0>(i));
        EXPECT_EQ(std::get<0>(i), std::get<1>(i)) << " index=" << idx;
        ++idx;
    }

    // clang-format on
}

template <typename... Func>
auto task_wrapper(Func... funcs) -> std::function<int()> {
    return [funcs...]() {
        return [](std::initializer_list<int> numbers) {
            int i = 0;
            for (int num : numbers) {
                if (num != 0) {
                    i = num;
                }
            }
            return i;
        }({funcs()...});
    };
}

TEST(UtilTest, stage_wrapper) {
    std::function<int()> func1 = []() { return 0; };
    std::function<int()> func2 = []() { return -1; };
    std::function<int()> func3 = []() { return 0; };
    auto f = task_wrapper(func1, func2, func3);
    ASSERT_EQ(-1, f());

    f = task_wrapper(func1, func3);
    ASSERT_EQ(0, f());
}

template <typename T>
void test_cancel_after_tasks_started(T normal_value, T cancel_value,
                                     std::function<bool(const T&)> is_cancel) {
    auto pool = std::make_shared<SimpleThreadPool>(3);
    pool->start();

    bthread::CountdownEvent normal_tasks_started(2);
    bthread::CountdownEvent cancellation_observed(1);
    SyncExecutor<T> sync_executor(pool, "cancel after tasks started", [&](const T& value) {
        bool cancelled = is_cancel(value);
        if (cancelled) {
            cancellation_observed.signal();
        }
        return cancelled;
    });

    auto normal_task = [&, normal_value]() {
        normal_tasks_started.signal();
        cancellation_observed.wait();
        return normal_value;
    };
    auto cancel_task = [&, cancel_value]() {
        normal_tasks_started.wait();
        return cancel_value;
    };

    sync_executor.add(normal_task);
    sync_executor.add(normal_task);
    sync_executor.add(cancel_task);

    bool finished = true;
    auto results = sync_executor.when_all(&finished);
    ASSERT_FALSE(finished);
    ASSERT_EQ(3, results.size());
    EXPECT_EQ(normal_value, results[0]);
    EXPECT_EQ(normal_value, results[1]);
    EXPECT_EQ(cancel_value, results[2]);
}

TEST(UtilTest, delay) {
    test_cancel_after_tasks_started<int>(1, -1, [](const int& value) { return value == -1; });
    test_cancel_after_tasks_started<std::string_view>(
            "fake", "", [](const std::string_view value) { return value.empty(); });
}

TEST(UtilTest, normal) {
    auto s3_producer_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
    s3_producer_pool->start();
    // test normal execute
    {
        SyncExecutor<int> sync_executor(s3_producer_pool, "normal test",
                                        [](int k) { return k == -1; });
        auto f1 = []() { return 1; };
        sync_executor.add(f1);
        sync_executor.add(f1);
        sync_executor.add(f1);
        bool finished = true;
        std::vector<int> res = sync_executor.when_all(&finished);
        ASSERT_EQ(3, res.size());
        ASSERT_EQ(finished, true);
        std::for_each(res.begin(), res.end(), [](auto&& n) { ASSERT_EQ(1, n); });
    }
    // test when error happen
    {
        SyncExecutor<int> sync_executor(s3_producer_pool, "normal test",
                                        [](int k) { return k == -1; });
        auto f1 = []() { return 1; };
        sync_executor._stop_token = true;
        sync_executor.add(f1);
        sync_executor.add(f1);
        sync_executor.add(f1);
        bool finished = true;
        std::vector<int> res = sync_executor.when_all(&finished);
        ASSERT_EQ(finished, false);
        ASSERT_EQ(0, res.size());
    }
    {
        SyncExecutor<int> sync_executor(s3_producer_pool, "normal test",
                                        [](int k) { return k == -1; });
        auto f1 = []() { return 1; };
        auto cancel = []() { return -1; };
        sync_executor.add(f1);
        sync_executor.add(f1);
        sync_executor.add(f1);
        sync_executor.add(cancel);
        bool finished = true;
        std::vector<int> res = sync_executor.when_all(&finished);
        ASSERT_EQ(finished, false);
    }
    // test string_view
    {
        SyncExecutor<std::string_view> sync_executor(
                s3_producer_pool, "normal test",
                [](const std::string_view k) { return k.empty(); });
        std::string s = "Hello World";
        auto f1 = [&s]() { return std::string_view(s); };
        sync_executor.add(f1);
        sync_executor.add(f1);
        sync_executor.add(f1);
        bool finished = true;
        std::vector<std::string_view> res = sync_executor.when_all(&finished);
        ASSERT_EQ(3, res.size());
        std::for_each(res.begin(), res.end(), [&s](auto&& n) { ASSERT_EQ(s, n); });
    }
}

TEST(UtilTest, test_add_after_when_all) {
    auto f = []() {
        auto pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
        pool->start();
        SyncExecutor<int> sync_executor(pool, "test add after when all: inside",
                                        [](int k) { return k != 0; });
        auto f1 = []() { return 0; };
        sync_executor.add(f1);
        bool finished = true;
        std::vector<int> res = sync_executor.when_all(&finished);
        sync_executor.add(f1);
        res = sync_executor.when_all(&finished);
        EXPECT_EQ(1, res.size());
        EXPECT_EQ(finished, true);
        std::for_each(res.begin(), res.end(), [](auto&& n) { EXPECT_EQ(0, n); });
        return 0;
    };

    auto s3_producer_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
    s3_producer_pool->start();
    SyncExecutor<int> s3_sync_executor(s3_producer_pool, "test add after when all: outside",
                                       [](int k) { return k != 0; });
    s3_sync_executor.add(f);
    bool finished = true;
    std::vector<int> res = s3_sync_executor.when_all(&finished);
    EXPECT_EQ(1, res.size());
    EXPECT_EQ(finished, true);
    std::for_each(res.begin(), res.end(), [](auto&& n) { EXPECT_EQ(0, n); });
}

TEST(UtilTest, exception) {
    auto s3_producer_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
    s3_producer_pool->start();
    {
        SyncExecutor<int> sync_executor(s3_producer_pool, "exception test",
                                        [](int k) { return k != 0; });
        auto f = []() {
            throw(std::runtime_error("test exception"));
            return 1;
        };
        sync_executor.add(f);
        bool finished = true;
        std::vector<int> res = sync_executor.when_all(&finished);
        EXPECT_EQ(0, res.size());
        EXPECT_EQ(finished, false);
        std::for_each(res.begin(), res.end(), [](auto&& n) { EXPECT_EQ(1, n); });
    }
}

TEST(UtilTest, test_sync_executor) {
    auto f = []() {
        sleep(1);
        auto pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
        pool->start();
        SyncExecutor<int> sync_executor(pool, "test sync executor: inside",
                                        [](int k) { return k != 0; });
        auto f1 = []() { return 0; };
        sync_executor.add(f1);
        bool finished = true;
        std::vector<int> res = sync_executor.when_all(&finished);
        sync_executor.add(f1);
        res = sync_executor.when_all(&finished);
        EXPECT_EQ(1, res.size());
        EXPECT_EQ(finished, true);
        std::for_each(res.begin(), res.end(), [](auto&& n) { EXPECT_EQ(0, n); });
        return 0;
    };
    std::mutex go_mutex;

    auto* sp = doris::SyncPoint::get_instance();
    sp->set_call_back("SyncExecutor::when_all.set_wait_time", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        [[maybe_unused]] auto max_wait_time = *doris::try_any_cast<size_t*>(args[0]);
        max_wait_time = 100;
    });

    auto s3_producer_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
    s3_producer_pool->start();
    SyncExecutor<int> s3_sync_executor(s3_producer_pool, "test sync executor: outside",
                                       [](int k) { return k != 0; });
    s3_sync_executor.add(f);
    bool finished = true;
    std::vector<int> res = s3_sync_executor.when_all(&finished);
    EXPECT_EQ(1, res.size());
    EXPECT_EQ(finished, true);
    std::for_each(res.begin(), res.end(), [](auto&& n) { EXPECT_EQ(0, n); });
}

TEST(UtilTest, test_split) {
    auto path = doris::get_valid_ca_cert_path(doris::cloud::split(config::ca_cert_file_paths, ';'));
    LOG(INFO) << "config:" << config::ca_cert_file_paths << " path:" << path;
    ASSERT_FALSE(path.empty());
}

TEST(UtilTest, test_normalize_http_uri) {
    // ===== Basic functionality with HTTPS protocol =====
    EXPECT_EQ(doris::normalize_http_uri("https://example.com/path"), "https://example.com/path");
    EXPECT_EQ(doris::normalize_http_uri("https://example.com//path"), "https://example.com/path");
    EXPECT_EQ(doris::normalize_http_uri("https://example.com///path"), "https://example.com/path");

    // ===== Basic functionality with HTTP protocol =====
    EXPECT_EQ(doris::normalize_http_uri("http://example.com/path"), "http://example.com/path");
    EXPECT_EQ(doris::normalize_http_uri("http://example.com//path"), "http://example.com/path");
    EXPECT_EQ(doris::normalize_http_uri("http://example.com///path"), "http://example.com/path");

    // ===== Multiple consecutive slashes in different positions =====
    EXPECT_EQ(doris::normalize_http_uri("https://host.com//bucket//prefix"),
              "https://host.com/bucket/prefix");
    EXPECT_EQ(doris::normalize_http_uri("https://host.com///bucket///prefix///"),
              "https://host.com/bucket/prefix/");
    EXPECT_EQ(doris::normalize_http_uri("https://host.com////bucket////prefix////file"),
              "https://host.com/bucket/prefix/file");

    // ===== Azure blob storage specific URLs =====
    EXPECT_EQ(doris::normalize_http_uri("https://account.blob.core.windows.net//container"),
              "https://account.blob.core.windows.net/container");
    EXPECT_EQ(
            doris::normalize_http_uri("https://account.blob.core.windows.net///container//prefix"),
            "https://account.blob.core.windows.net/container/prefix");
    EXPECT_EQ(doris::normalize_http_uri(
                      "https://account.blob.core.windows.net////container///prefix///file.txt"),
              "https://account.blob.core.windows.net/container/prefix/file.txt");

    // ===== URLs without protocol =====
    EXPECT_EQ(doris::normalize_http_uri("example.com//path"), "example.com/path");
    EXPECT_EQ(doris::normalize_http_uri("host.com///bucket//prefix"), "host.com/bucket/prefix");
    EXPECT_EQ(doris::normalize_http_uri("//path//to//file"), "/path/to/file");

    // ===== Edge cases =====
    // Empty string
    EXPECT_EQ(doris::normalize_http_uri(""), "");

    // Only protocol
    EXPECT_EQ(doris::normalize_http_uri("https://"), "https://");
    EXPECT_EQ(doris::normalize_http_uri("http://"), "http://");

    // Only slashes
    EXPECT_EQ(doris::normalize_http_uri("//"), "/");
    EXPECT_EQ(doris::normalize_http_uri("///"), "/");
    EXPECT_EQ(doris::normalize_http_uri("////"), "/");

    // Single character paths
    EXPECT_EQ(doris::normalize_http_uri("https://a"), "https://a");
    EXPECT_EQ(doris::normalize_http_uri("https://a/"), "https://a/");
    EXPECT_EQ(doris::normalize_http_uri("https://a//"), "https://a/");

    // ===== Protocol preservation =====
    // Ensure protocol :// is never modified
    EXPECT_EQ(doris::normalize_http_uri("https://example.com"), "https://example.com");
    EXPECT_EQ(doris::normalize_http_uri("http://example.com"), "http://example.com");

    // Even with extra slashes after protocol
    EXPECT_EQ(doris::normalize_http_uri("https:///example.com"), "https://example.com");
    EXPECT_EQ(doris::normalize_http_uri("http:///example.com"), "http://example.com");

    // Mixed case protocol (though unusual)
    EXPECT_EQ(doris::normalize_http_uri("HTTP://example.com//path"), "HTTP://example.com/path");
    EXPECT_EQ(doris::normalize_http_uri("HTTPS://example.com//path"), "HTTPS://example.com/path");

    // ===== Trailing slashes =====
    EXPECT_EQ(doris::normalize_http_uri("https://example.com/path/"), "https://example.com/path/");
    EXPECT_EQ(doris::normalize_http_uri("https://example.com/path//"), "https://example.com/path/");
    EXPECT_EQ(doris::normalize_http_uri("https://example.com/path///"),
              "https://example.com/path/");
    EXPECT_EQ(doris::normalize_http_uri("https://example.com/path////"),
              "https://example.com/path/");

    // ===== Complex real-world scenarios =====
    // Simulating common configuration mistakes
    EXPECT_EQ(doris::normalize_http_uri("https://endpoint.com///bucket//prefix//file.txt"),
              "https://endpoint.com/bucket/prefix/file.txt");

    // User configured endpoint with trailing slash + bucket with leading slash
    EXPECT_EQ(doris::normalize_http_uri("https://endpoint.com///bucket"),
              "https://endpoint.com/bucket");

    // Multiple slashes everywhere
    EXPECT_EQ(
            doris::normalize_http_uri("https://host.com////bucket////prefix////subfolder////file"),
            "https://host.com/bucket/prefix/subfolder/file");

    // ===== Special characters in path =====
    EXPECT_EQ(
            doris::normalize_http_uri("https://example.com//path-with-dash//file_with_underscore"),
            "https://example.com/path-with-dash/file_with_underscore");
    EXPECT_EQ(doris::normalize_http_uri("https://example.com//path.with.dots//file@special"),
              "https://example.com/path.with.dots/file@special");
    EXPECT_EQ(doris::normalize_http_uri("https://example.com//bucket123//prefix456//file789"),
              "https://example.com/bucket123/prefix456/file789");

    // ===== URLs with query parameters and fragments =====
    EXPECT_EQ(doris::normalize_http_uri("https://example.com//path?query=value"),
              "https://example.com/path?query=value");
    EXPECT_EQ(doris::normalize_http_uri("https://example.com//path#fragment"),
              "https://example.com/path#fragment");
    EXPECT_EQ(doris::normalize_http_uri("https://example.com//path?query=value#fragment"),
              "https://example.com/path?query=value#fragment");
}

TEST(UtilTest, test_set_s3_client_default_http_scheme) {
    doris::cloud::S3Environment::getInstance();
    Aws::Client::ClientConfiguration client_config =
            doris::cloud::S3Environment::getClientConfiguration();

    client_config.endpointOverride = "example.com:9000";
    doris::set_s3_client_default_http_scheme(client_config, "http");
    EXPECT_EQ(client_config.endpointOverride, "example.com:9000");
    EXPECT_EQ(client_config.scheme, Aws::Http::Scheme::HTTP);

    doris::set_s3_client_default_http_scheme(client_config, "https");
    EXPECT_EQ(client_config.endpointOverride, "example.com:9000");
    EXPECT_EQ(client_config.scheme, Aws::Http::Scheme::HTTPS);

    client_config.endpointOverride = "http://example.com:9000";
    client_config.scheme = Aws::Http::Scheme::HTTP;
    doris::set_s3_client_default_http_scheme(client_config, "https");
    EXPECT_EQ(client_config.endpointOverride, "http://example.com:9000");
    EXPECT_EQ(client_config.scheme, Aws::Http::Scheme::HTTP);

    client_config.endpointOverride = "https://example.com:9000";
    client_config.scheme = Aws::Http::Scheme::HTTPS;
    doris::set_s3_client_default_http_scheme(client_config, "http");
    EXPECT_EQ(client_config.endpointOverride, "https://example.com:9000");
    EXPECT_EQ(client_config.scheme, Aws::Http::Scheme::HTTPS);
}

TEST(UtilTest, test_long_normalize_http_uri) {
    std::string longPath = "https://example.com";
    for (int i = 0; i < 100; i++) {
        longPath += "//segment" + std::to_string(i);
    }

    std::string expected = "https://example.com";
    for (int i = 0; i < 100; i++) {
        expected += "/segment" + std::to_string(i);
    }

    EXPECT_EQ(doris::normalize_http_uri(longPath), expected);
}