path: root/fluent-bit/lib/librdkafka-2.1.0/src/rdunittest.c

/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2017 Magnus Edenhill
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#ifdef _WIN32
#define RD_UNITTEST_QPC_OVERRIDES 1
#endif

#include "rd.h"
#include "rdunittest.h"

#include "rdvarint.h"
#include "rdbuf.h"
#include "crc32c.h"
#include "rdmurmur2.h"
#include "rdfnv1a.h"
#if WITH_HDRHISTOGRAM
#include "rdhdrhistogram.h"
#endif
#include "rdkafka_int.h"
#include "rdkafka_broker.h"
#include "rdkafka_request.h"

#include "rdsysqueue.h"
#include "rdkafka_sasl_oauthbearer.h"
#if WITH_OAUTHBEARER_OIDC
#include "rdkafka_sasl_oauthbearer_oidc.h"
#endif
#include "rdkafka_msgset.h"
#include "rdkafka_txnmgr.h"

rd_bool_t rd_unittest_assert_on_failure = rd_false;
rd_bool_t rd_unittest_on_ci             = rd_false;
rd_bool_t rd_unittest_slow              = rd_false;

#if ENABLE_CODECOV
/**
 * @name Code coverage
 * @{
 */

static rd_atomic64_t rd_ut_covnrs[RD_UT_COVNR_MAX + 1];

void rd_ut_coverage(const char *file, const char *func, int line, int covnr) {
        rd_assert(covnr >= 0 && covnr <= RD_UT_COVNR_MAX);
        rd_atomic64_add(&rd_ut_covnrs[covnr], 1);
}


int64_t
rd_ut_coverage_check(const char *file, const char *func, int line, int covnr) {
        int64_t r;

        rd_assert(covnr >= 0 && covnr <= RD_UT_COVNR_MAX);

        r = rd_atomic64_get(&rd_ut_covnrs[covnr]);

        if (!r) {
                fprintf(stderr,
                        "\033[31m"
                        "RDUT: FAIL: %s:%d: %s: "
                        "Code coverage nr %d: FAIL: "
                        "code path not executed: "
                        "perform `grep -RnF 'COVERAGE(%d)' src/` to find "
                        "source location"
                        "\033[0m\n",
                        file, line, func, covnr, covnr);
                if (rd_unittest_assert_on_failure)
                        rd_assert(!*"unittest failure");
                return 0;
        }

        fprintf(stderr,
                "\033[34mRDUT: CCOV: %s:%d: %s: Code coverage nr %d: "
                "PASS (%" PRId64 " code path execution(s))\033[0m\n",
                file, line, func, covnr, r);

        return r;
}
/**@}*/

#endif /* ENABLE_CODECOV */


/**
 * @name Test rdsysqueue.h / queue.h
 * @{
 */

struct ut_tq {
        TAILQ_ENTRY(ut_tq) link;
        int v;
};

TAILQ_HEAD(ut_tq_head, ut_tq);

struct ut_tq_args {
        const char *name; /**< Descriptive test name */
        struct {
                int base; /**< Base value */
                int cnt;  /**< Number of elements to add */
                int step; /**< Value step */
        } q[3];           /**< Queue element definition */
        int qcnt;         /**< Number of defs in .q */
        int exp[16];      /**< Expected value order after join */
};

/**
 * @brief Find the previous element (insert position) for
 *        value \p val in list \p head or NULL if \p val is less than
 *        the first element in \p head.
 * @remarks \p head must be ascending sorted.
 */
static struct ut_tq *ut_tq_find_prev_pos(const struct ut_tq_head *head,
                                         int val) {
        struct ut_tq *e, *prev = NULL;

        TAILQ_FOREACH(e, head, link) {
                if (e->v > val)
                        return prev;
                prev = e;
        }

        return prev;
}

static int ut_tq_test(const struct ut_tq_args *args) {
        int totcnt                = 0;
        int fails                 = 0;
        struct ut_tq_head *tqh[3] = {NULL, NULL, NULL};
        struct ut_tq *e, *insert_after;
        int i, qi;

        RD_UT_SAY("Testing TAILQ: %s", args->name);

        /*
         * Verify TAILQ_INSERT_LIST:
         *  For each insert position test:
         *  - create two lists: tqh 0 and 1
         *  - add entries to both lists
         *  - insert list 1 into 0
         *  - verify expected order and correctness
         */

        /* Use heap allocated heads to let valgrind/asan assist
         * in detecting corruption. */

        for (qi = 0; qi < args->qcnt; qi++) {
                tqh[qi] = rd_calloc(1, sizeof(*tqh[qi]));
                TAILQ_INIT(tqh[qi]);

                for (i = 0; i < args->q[qi].cnt; i++) {
                        e    = rd_malloc(sizeof(*e));
                        e->v = args->q[qi].base + (i * args->q[qi].step);
                        TAILQ_INSERT_TAIL(tqh[qi], e, link);
                }

                totcnt += args->q[qi].cnt;
        }

        for (qi = 1; qi < args->qcnt; qi++) {
                insert_after = ut_tq_find_prev_pos(tqh[0], args->q[qi].base);
                if (!insert_after) {
                        /* Insert position is head of list,
                         * do two-step concat+move */
                        TAILQ_PREPEND(tqh[0], tqh[qi], ut_tq_head, link);
                } else {
                        TAILQ_INSERT_LIST(tqh[0], insert_after, tqh[qi],
                                          ut_tq_head, struct ut_tq *, link);
                }

                RD_UT_ASSERT(TAILQ_EMPTY(tqh[qi]), "expected empty tqh[%d]",
                             qi);
                RD_UT_ASSERT(!TAILQ_EMPTY(tqh[0]), "expected non-empty tqh[0]");

                memset(tqh[qi], (int)'A', sizeof(*tqh[qi]));
                rd_free(tqh[qi]);
        }

        RD_UT_ASSERT(TAILQ_LAST(tqh[0], ut_tq_head)->v == args->exp[totcnt - 1],
                     "TAILQ_LAST val %d, expected %d",
                     TAILQ_LAST(tqh[0], ut_tq_head)->v, args->exp[totcnt - 1]);

        /* Add sentinel value to verify that INSERT_TAIL works
         * after INSERT_LIST */
        e    = rd_malloc(sizeof(*e));
        e->v = 99;
        TAILQ_INSERT_TAIL(tqh[0], e, link);
        totcnt++;

        i = 0;
        TAILQ_FOREACH(e, tqh[0], link) {
                if (i >= totcnt) {
                        RD_UT_WARN(
                            "Too many elements in list tqh[0]: "
                            "idx %d > totcnt %d: element %p (value %d)",
                            i, totcnt, e, e->v);
                        fails++;
                } else if (e->v != args->exp[i]) {
                        RD_UT_WARN(
                            "Element idx %d/%d in tqh[0] has value %d, "
                            "expected %d",
                            i, totcnt, e->v, args->exp[i]);
                        fails++;
                } else if (i == totcnt - 1 &&
                           e != TAILQ_LAST(tqh[0], ut_tq_head)) {
                        RD_UT_WARN("TAILQ_LAST == %p, expected %p",
                                   TAILQ_LAST(tqh[0], ut_tq_head), e);
                        fails++;
                }
                i++;
        }

        /* Then scan it in reverse */
        i = totcnt - 1;
        TAILQ_FOREACH_REVERSE(e, tqh[0], ut_tq_head, link) {
                if (i < 0) {
                        RD_UT_WARN(
                            "REVERSE: Too many elements in list tqh[0]: "
                            "idx %d < 0: element %p (value %d)",
                            i, e, e->v);
                        fails++;
                } else if (e->v != args->exp[i]) {
                        RD_UT_WARN(
                            "REVERSE: Element idx %d/%d in tqh[0] has "
                            "value %d, expected %d",
                            i, totcnt, e->v, args->exp[i]);
                        fails++;
                } else if (i == totcnt - 1 &&
                           e != TAILQ_LAST(tqh[0], ut_tq_head)) {
                        RD_UT_WARN("REVERSE: TAILQ_LAST == %p, expected %p",
                                   TAILQ_LAST(tqh[0], ut_tq_head), e);
                        fails++;
                }
                i--;
        }

        RD_UT_ASSERT(TAILQ_LAST(tqh[0], ut_tq_head)->v == args->exp[totcnt - 1],
                     "TAILQ_LAST val %d, expected %d",
                     TAILQ_LAST(tqh[0], ut_tq_head)->v, args->exp[totcnt - 1]);

        while ((e = TAILQ_FIRST(tqh[0]))) {
                TAILQ_REMOVE(tqh[0], e, link);
                rd_free(e);
        }

        rd_free(tqh[0]);

        return fails;
}


static int unittest_sysqueue(void) {
        const struct ut_tq_args args[] = {
            {"empty tqh[0]",
             {{0, 0, 0}, {0, 3, 1}},
             2,
             {0, 1, 2, 99 /*sentinel*/}},
            {"prepend 1,0",
             {{10, 3, 1}, {0, 3, 1}},
             2,
             {0, 1, 2, 10, 11, 12, 99}},
            {"prepend 2,1,0",
             {
                 {10, 3, 1}, /* 10, 11, 12 */
                 {5, 3, 1},  /* 5, 6, 7 */
                 {0, 2, 1}   /* 0, 1 */
             },
             3,
             {0, 1, 5, 6, 7, 10, 11, 12, 99}},
            {"insert 1", {{0, 3, 2}, {1, 2, 2}}, 2, {0, 1, 3, 2, 4, 99}},
            {"insert 1,2",
             {
                 {0, 3, 3}, /* 0, 3, 6 */
                 {1, 2, 3}, /* 1, 4 */
                 {2, 1, 3}  /* 2 */
             },
             3,
             {0, 1, 2, 4, 3, 6, 99}},
            {"append 1",
             {{0, 5, 1}, {5, 5, 1}},
             2,
             {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 99}},
            {"append 1,2",
             {
                 {0, 5, 1}, /* 0, 1, 2, 3, 4 */
                 {5, 5, 1}, /* 5, 6, 7, 8, 9 */
                 {11, 2, 1} /* 11, 12 */
             },
             3,
             {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 11, 12, 99}},
            {
                "insert 1,0,2",
                {
                    {5, 3, 1}, /* 5, 6, 7 */
                    {0, 1, 1}, /* 0 */
                    {10, 2, 1} /* 10, 11 */
                },
                3,
                {0, 5, 6, 7, 10, 11, 99},
            },
            {
                "insert 2,0,1",
                {
                    {5, 3, 1},  /* 5, 6, 7 */
                    {10, 2, 1}, /* 10, 11 */
                    {0, 1, 1}   /* 0 */
                },
                3,
                {0, 5, 6, 7, 10, 11, 99},
            },
            {NULL}};
        int i;
        int fails = 0;

        for (i = 0; args[i].name != NULL; i++)
                fails += ut_tq_test(&args[i]);

        RD_UT_ASSERT(!fails, "See %d previous failure(s)", fails);

        RD_UT_PASS();
}

/**@}*/


/**
 * @name rd_clock() unittests
 * @{
 */

#if RD_UNITTEST_QPC_OVERRIDES

/**
 * These values are based off a machine with freq 14318180
 * which would cause the original rd_clock() calculation to overflow
 * after about 8 days.
 * Details:
 * https://github.com/confluentinc/confluent-kafka-dotnet/issues/603#issuecomment-417274540
 */

static const int64_t rd_ut_qpc_freq = 14318180;
static int64_t rd_ut_qpc_now;

BOOL rd_ut_QueryPerformanceFrequency(_Out_ LARGE_INTEGER *lpFrequency) {
        lpFrequency->QuadPart = rd_ut_qpc_freq;
        return TRUE;
}

BOOL rd_ut_QueryPerformanceCounter(_Out_ LARGE_INTEGER *lpPerformanceCount) {
        lpPerformanceCount->QuadPart = rd_ut_qpc_now * rd_ut_qpc_freq;
        return TRUE;
}

static int unittest_rdclock(void) {
        rd_ts_t t1, t2;

        /* First let "uptime" be fresh boot (0). */
        rd_ut_qpc_now = 0;
        t1            = rd_clock();
        rd_ut_qpc_now++;
        t2 = rd_clock();
        RD_UT_ASSERT(t2 == t1 + (1 * 1000000),
                     "Expected t2 %" PRId64 " to be 1s more than t1 %" PRId64,
                     t2, t1);

        /* Then skip forward to 8 days, which should trigger the
         * overflow in a faulty implementation. */
        rd_ut_qpc_now = 8 * 86400;
        t2            = rd_clock();
        RD_UT_ASSERT(t2 == t1 + (8LL * 86400 * 1000000),
                     "Expected t2 %" PRId64
                     " to be 8 days larger than t1 %" PRId64,
                     t2, t1);

        /* And make sure we can run on a system with 38 years of uptime.. */
        rd_ut_qpc_now = 38 * 365 * 86400;
        t2            = rd_clock();
        RD_UT_ASSERT(t2 == t1 + (38LL * 365 * 86400 * 1000000),
                     "Expected t2 %" PRId64
                     " to be 38 years larger than t1 %" PRId64,
                     t2, t1);

        RD_UT_PASS();
}
#endif



/**@}*/

extern int unittest_string(void);
extern int unittest_cgrp(void);
#if WITH_SASL_SCRAM
extern int unittest_scram(void);
#endif
extern int unittest_assignors(void);
extern int unittest_map(void);
#if WITH_CURL
extern int unittest_http(void);
#endif
#if WITH_OAUTHBEARER_OIDC
extern int unittest_sasl_oauthbearer_oidc(void);
#endif

int rd_unittest(void) {
        int fails = 0;
        const struct {
                const char *name;
                int (*call)(void);
        } unittests[] = {
                {"sysqueue", unittest_sysqueue},
                {"string", unittest_string},
                {"map", unittest_map},
                {"rdbuf", unittest_rdbuf},
                {"rdvarint", unittest_rdvarint},
                {"crc32c", unittest_rd_crc32c},
                {"msg", unittest_msg},
                {"murmurhash", unittest_murmur2},
                {"fnv1a", unittest_fnv1a},
#if WITH_HDRHISTOGRAM
                {"rdhdrhistogram", unittest_rdhdrhistogram},
#endif
#ifdef _WIN32
                {"rdclock", unittest_rdclock},
#endif
                {"conf", unittest_conf},
                {"broker", unittest_broker},
                {"request", unittest_request},
#if WITH_SASL_OAUTHBEARER
                {"sasl_oauthbearer", unittest_sasl_oauthbearer},
#endif
                {"aborted_txns", unittest_aborted_txns},
                {"cgrp", unittest_cgrp},
#if WITH_SASL_SCRAM
                {"scram", unittest_scram},
#endif
                {"assignors", unittest_assignors},
#if WITH_CURL
                {"http", unittest_http},
#endif
#if WITH_OAUTHBEARER_OIDC
                {"sasl_oauthbearer_oidc", unittest_sasl_oauthbearer_oidc},
#endif
                {NULL}
        };
        int i;
        const char *match = rd_getenv("RD_UT_TEST", NULL);
        int cnt           = 0;

        if (rd_getenv("RD_UT_ASSERT", NULL))
                rd_unittest_assert_on_failure = rd_true;
        if (rd_getenv("CI", NULL)) {
                RD_UT_SAY("Unittests running on CI");
                rd_unittest_on_ci = rd_true;
        }

        if (rd_unittest_on_ci || (ENABLE_DEVEL + 0)) {
                RD_UT_SAY("Unittests will not error out on slow CPUs");
                rd_unittest_slow = rd_true;
        }

        rd_kafka_global_init();

#if ENABLE_CODECOV
        for (i = 0; i < RD_UT_COVNR_MAX + 1; i++)
                rd_atomic64_init(&rd_ut_covnrs[i], 0);
#endif

        for (i = 0; unittests[i].name; i++) {
                int f;

                if (match && !strstr(unittests[i].name, match))
                        continue;

                f = unittests[i].call();
                RD_UT_SAY("unittest: %s: %4s\033[0m", unittests[i].name,
                          f ? "\033[31mFAIL" : "\033[32mPASS");
                fails += f;
                cnt++;
        }

#if ENABLE_CODECOV
#if FIXME /* This check only works if all tests that use coverage checks       \
           * are run, which we can't really know, so disable until we          \
           * know what to do with this. */
        if (!match) {
                /* Verify all code paths were covered */
                int cov_fails = 0;
                for (i = 0; i < RD_UT_COVNR_MAX + 1; i++) {
                        if (!RD_UT_COVERAGE_CHECK(i))
                                cov_fails++;
                }
                if (cov_fails > 0)
                        RD_UT_SAY("%d code coverage failure(s) (ignored)\n",
                                  cov_fails);
        }
#endif
#endif

        if (!cnt && match)
                RD_UT_WARN("No unittests matching \"%s\"", match);

        return fails;
}