<commit_before>/*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
 * 
 *  Gearmand client and server library.
 *
 *  Copyright (C) 2011 Data Differential, http://datadifferential.com/
 *  Copyright (C) 2008 <NAME>, <NAME>
 *  All rights reserved.
 *
 *  Redistribution and use in source and binary forms, with or without
 *  modification, are permitted provided that the following conditions are
 *  met:
 *
 *      * Redistributions of source code must retain the above copyright
 *  notice, this list of conditions and the following disclaimer.
 *
 *      * 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.
 *
 *      * The names of its contributors may not be used to endorse or
 *  promote products derived from this software without specific prior
 *  written permission.
 *
 *  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.
 *
 */

#include <config.h>

#include <libtest/test.hpp>
using namespace libtest;

#include <cassert>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <iostream>
#include <vector>
#include <boost/foreach.hpp>

#define GEARMAN_CORE
#include <libgearman/gearman.h>

#include <tests/start_worker.h>

#define NAMESPACE_KEY "foo123"

#define WORKER_FUNCTION_NAME "client_test"
#define WORKER_CHUNKED_FUNCTION_NAME "reverse_test"
#define WORKER_UNIQUE_FUNCTION_NAME "unique_test"
#define WORKER_SPLIT_FUNCTION_NAME "split_worker"

#include <tests/limits.h>
#include <tests/do.h>
#include <tests/server_options.h>
#include <tests/do_background.h>
#include <tests/execute.h>
#include <tests/gearman_client_do_job_handle.h>
#include <tests/gearman_execute_partition.h>
#include <tests/protocol.h>
#include <tests/regression.h>
#include <tests/task.h>
#include <tests/unique.h>
#include <tests/workers.h>
#include <tests/workers_v1.h>

#ifndef __INTEL_COMPILER
#pragma GCC diagnostic ignored "-Wold-style-cast"
#endif

struct client_test_st
{
  gearman_client_st *_client;
  bool _clone;
  std::vector<worker_handle_st *> workers;
  const char *_worker_name;

  client_test_st() :
    _clone(true),
    _worker_name(WORKER_FUNCTION_NAME)
  { 
    _client= gearman_client_create(NULL);

    if (_client == NULL)
    {
      throw "gearman_client_create() failed";
    }
  }

  ~client_test_st()
  {
    for (std::vector<worker_handle_st *>::iterator iter= workers.begin(); iter != workers.end(); iter++)
    {
      delete *iter;
    }
    workers.clear();
    gearman_client_free(_client);
  }

  void push(worker_handle_st *arg)
  {
    workers.push_back(arg);
  }

  const char *worker_name() const
  {
    return _worker_name;
  }

  void set_worker_name(const char *arg)
  {
    _worker_name= arg;
  }

  void set_clone(bool arg)
  {
    _clone= arg;
  }

  bool clone() const
  {
    return _clone;
  }

  gearman_client_st *client()
  {
    return _client;
  }

  void reset_client()
  {
    gearman_client_free(_client);
    _client= gearman_client_create(NULL);
  }
};

class Client {
public:
  Client()
  {
    _client= gearman_client_create(NULL);

    if (_client == NULL)
    {
      throw "gearman_client_create() failed";
    }
  }

  gearman_client_st* operator&() const
  { 
    return _client;
  }

  gearman_client_st* operator->() const
  { 
    return _client;
  }

  ~Client()
  {
    gearman_client_free(_client);
  }

private:
  gearman_client_st *_client;

};

#ifndef __INTEL_COMPILER
#pragma GCC diagnostic ignored "-Wold-style-cast"
#endif

/**
  @note Just here until I fix libhashkit.
*/
static uint32_t internal_generate_hash(const char *key, size_t key_length)
{
  const char *ptr= key;
  uint32_t value= 0;

  while (key_length--)
  {
    uint32_t val= (uint32_t) *ptr++;
    value += val;
    value += (value << 10);
    value ^= (value >> 6);
  }
  value += (value << 3);
  value ^= (value >> 11);
  value += (value << 15);

  return value == 0 ? 1 : (uint32_t) value;
}

/* Prototypes */
void *client_test_temp_worker(gearman_job_st *job, void *context,
                              size_t *result_size, gearman_return_t *ret_ptr);


extern "C"
{
  static void *client_thread(void *object)
  {
    volatile gearman_return_t *ret= (volatile gearman_return_t *)object;
    gearman_client_st client;
    gearman_client_st *client_ptr;
    size_t result_size;

    client_ptr= gearman_client_create(&client);

    if (client_ptr == NULL)
    {
      *ret= GEARMAN_MEMORY_ALLOCATION_FAILURE;
      pthread_exit(0);
    }

    gearman_return_t rc= gearman_client_add_server(&client, NULL, libtest::default_port());
    if (gearman_failed(rc))
    {
      *ret= rc;
      pthread_exit(0);
    }

    gearman_client_set_timeout(&client, 400);
    for (size_t x= 0; x < 5; x++)
    {
      (void)gearman_client_do(&client, "client_test_temp", NULL, NULL, 0, &result_size, &rc);

      if (gearman_failed(rc))
      {
        *ret= rc;
        pthread_exit(0);
      }
    }
    gearman_client_free(client_ptr);

    pthread_exit(0);
  }
}

static test_return_t init_test(void *)
{
  gearman_client_st client;

  test_truth(gearman_client_create(&client));

  gearman_client_free(&client);

  return TEST_SUCCESS;
}

static test_return_t allocation_test(void *)
{
  gearman_client_st *client;

  test_truth(client= gearman_client_create(NULL));

  gearman_client_free(client);

  return TEST_SUCCESS;
}

static test_return_t clone_test(void *object)
{
  const gearman_client_st *from= (gearman_client_st *)object;
  gearman_client_st *from_with_host;
  gearman_client_st *client;

  client= gearman_client_clone(NULL, NULL);

  test_truth(client);
  test_truth(client->options.allocated);

  gearman_client_free(client);

  client= gearman_client_clone(NULL, from);
  test_truth(client);
  gearman_client_free(client);

  from_with_host= gearman_client_create(NULL);
  test_truth(from_with_host);
  gearman_client_add_server(from_with_host, "localhost", 12345);

  client= gearman_client_clone(NULL, from_with_host);
  test_truth(client);

  test_truth(client->universal.con_list);
  test_truth(gearman_client_compare(client, from_with_host));

  gearman_client_free(client);
  gearman_client_free(from_with_host);

  return TEST_SUCCESS;
}

static test_return_t option_test(void *)
{
  gearman_client_st *gear;
  gearman_client_options_t default_options;

  gear= gearman_client_create(NULL);
  test_truth(gear);
  { // Initial Allocated, no changes
    test_truth(gear->options.allocated);
    test_false(gear->options.non_blocking);
    test_false(gear->options.unbuffered_result);
    test_false(gear->options.no_new);
    test_false(gear->options.free_tasks);
  }

  /* Set up for default options */
  default_options= gearman_client_options(gear);

  /*
    We take the basic options, and push
    them back in. See if we change anything.
  */
  gearman_client_set_options(gear, default_options);
  { // Initial Allocated, no changes
    test_truth(gear->options.allocated);
    test_false(gear->options.non_blocking);
    test_false(gear->options.unbuffered_result);
    test_false(gear->options.no_new);
    test_false(gear->options.free_tasks);
  }

  /*
    We will trying to modify non-mutable options (which should not be allowed)
  */
  {
    gearman_client_remove_options(gear, GEARMAN_CLIENT_ALLOCATED);
    { // Initial Allocated, no changes
      test_truth(gear->options.allocated);
      test_false(gear->options.non_blocking);
      test_false(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
    gearman_client_remove_options(gear, GEARMAN_CLIENT_NO_NEW);
    { // Initial Allocated, no changes
      test_truth(gear->options.allocated);
      test_false(gear->options.non_blocking);
      test_false(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
  }

  /*
    We will test modifying GEARMAN_CLIENT_NON_BLOCKING in several manners.
  */
  {
    gearman_client_remove_options(gear, GEARMAN_CLIENT_NON_BLOCKING);
    { // GEARMAN_CLIENT_NON_BLOCKING set to default, by default.
      test_truth(gear->options.allocated);
      test_false(gear->options.non_blocking);
      test_false(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
    gearman_client_add_options(gear, GEARMAN_CLIENT_NON_BLOCKING);
    { // GEARMAN_CLIENT_NON_BLOCKING set to default, by default.
      test_truth(gear->options.allocated);
      test_truth(gear->options.non_blocking);
      test_false(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
    gearman_client_set_options(gear, GEARMAN_CLIENT_NON_BLOCKING);
    { // GEARMAN_CLIENT_NON_BLOCKING set to default, by default.
      test_truth(gear->options.allocated);
      test_truth(gear->options.non_blocking);
      test_false(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
    gearman_client_set_options(gear, GEARMAN_CLIENT_UNBUFFERED_RESULT);
    { // Everything is now set to false except GEARMAN_CLIENT_UNBUFFERED_RESULT, and non-mutable options
      test_truth(gear->options.allocated);
      test_false(gear->options.non_blocking);
      test_truth(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
    /*
      Reset options to default. Then add an option, and then add more options. Make sure
      the options are all additive.
    */
    {
      gearman_client_set_options(gear, default_options);
      { // See if we return to defaults
        test_truth(gear->options.allocated);
        test_false(gear->options.non_blocking);
        test_false(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_false(gear->options.free_tasks);
      }
      gearman_client_add_options(gear, GEARMAN_CLIENT_FREE_TASKS);
      { // All defaults, except timeout_return
        test_truth(gear->options.allocated);
        test_false(gear->options.non_blocking);
        test_false(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_truth(gear->options.free_tasks);
      }
      gearman_client_add_options(gear, (gearman_client_options_t)(GEARMAN_CLIENT_NON_BLOCKING|GEARMAN_CLIENT_UNBUFFERED_RESULT));
      { // GEARMAN_CLIENT_NON_BLOCKING set to default, by default.
        test_truth(gear->options.allocated);
        test_truth(gear->options.non_blocking);
        test_truth(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_truth(gear->options.free_tasks);
      }
    }
    /*
      Add an option, and then replace with that option plus a new option.
    */
    {
      gearman_client_set_options(gear, default_options);
      { // See if we return to defaults
        test_truth(gear->options.allocated);
        test_false(gear->options.non_blocking);
        test_false(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_false(gear->options.free_tasks);
      }
      gearman_client_add_options(gear, GEARMAN_CLIENT_FREE_TASKS);
      { // All defaults, except timeout_return
        test_truth(gear->options.allocated);
        test_false(gear->options.non_blocking);
        test_false(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_truth(gear->options.free_tasks);
      }
      gearman_client_add_options(gear, (gearman_client_options_t)(GEARMAN_CLIENT_FREE_TASKS|GEARMAN_CLIENT_UNBUFFERED_RESULT));
      { // GEARMAN_CLIENT_NON_BLOCKING set to default, by default.
        test_truth(gear->options.allocated);
        test_false(gear->options.non_blocking);
        test_truth(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_truth(gear->options.free_tasks);
      }
    }
  }

  gearman_client_free(gear);

  return TEST_SUCCESS;
}

static test_return_t echo_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  gearman_string_t value= { test_literal_param("This is my echo test") };

  test_compare(GEARMAN_SUCCESS, gearman_client_echo(client, gearman_string_param(value)));

  return TEST_SUCCESS;
}

static test_return_t submit_job_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  const char *worker_function= (const char *)gearman_client_context(client);
  gearman_string_t value= { test_literal_param("submit_job_test") };

  size_t result_length;
  gearman_return_t rc;
  void *job_result= gearman_client_do(client, worker_function, NULL, gearman_string_param(value), &result_length, &rc);

  test_compare_got(GEARMAN_SUCCESS, 
                   rc, gearman_client_error(client) ? gearman_client_error(client) : gearman_strerror(rc));

  test_truth(job_result);
  test_compare(gearman_size(value), result_length);

  test_memcmp(gearman_c_str(value), job_result, gearman_size(value));

  free(job_result);

  return TEST_SUCCESS;
}

static test_return_t submit_null_job_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;

  test_truth(client);

  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  size_t result_length;
  gearman_return_t rc;
  void *job_result= gearman_client_do(client, worker_function, NULL, NULL, 0,
                                      &result_length, &rc);
  test_compare_got(GEARMAN_SUCCESS, rc, gearman_client_error(client));
  test_zero(result_length);
  test_false(job_result);

  return TEST_SUCCESS;
}

static test_return_t submit_exception_job_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  size_t result_length;
  gearman_return_t rc;
  void *job_result= gearman_client_do(client, worker_function, NULL,
                                      test_literal_param("exception"),
                                      &result_length, &rc);
  test_compare_got(GEARMAN_SUCCESS, rc, gearman_client_error(client) ? gearman_client_error(client) : gearman_strerror(rc));
  test_memcmp("exception", job_result, result_length);
  free(job_result);

  return TEST_SUCCESS;
}

static test_return_t submit_warning_job_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  size_t result_length;
  gearman_return_t rc;
  void *job_result= gearman_client_do(client, worker_function, NULL,
                                      test_literal_param("warning"),
                                      &result_length, &rc);
  test_compare_got(GEARMAN_SUCCESS, rc, gearman_client_error(client) ? gearman_client_error(client) : gearman_strerror(rc));
  test_memcmp("warning", job_result, result_length);
  free(job_result);

  return TEST_SUCCESS;
}

static test_return_t submit_fail_job_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);


  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  size_t result_length;
  gearman_return_t rc;
  void *job_result= gearman_client_do(client, worker_function, NULL, "fail", 4,
                                      &result_length, &rc);
  test_compare_got(GEARMAN_WORK_FAIL, rc, gearman_client_error(client));
  test_false(job_result);
  test_false(result_length);

  return TEST_SUCCESS;
}

static test_return_t submit_multiple_do(void *object)
{
  for (uint32_t x= 0; x < 100 /* arbitrary */; x++)
  {
    uint32_t option= random() %3;

    switch (option)
    {
    case 0:
      test_compare(TEST_SUCCESS, submit_null_job_test(object));
      break;
    case 1:
      test_compare(TEST_SUCCESS, submit_job_test(object));
      break;
    default:
    case 2:
      test_compare(TEST_SUCCESS, submit_fail_job_test(object));
      break;
    }
  }

  return TEST_SUCCESS;
}

static test_return_t gearman_client_job_status_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  gearman_string_t value= { test_literal_param("background_test") };

  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  gearman_job_handle_t job_handle;
  test_compare_got(GEARMAN_SUCCESS,
                   gearman_client_do_background(client, worker_function, NULL, gearman_string_param(value), job_handle), 
                   gearman_client_error(client));

  gearman_return_t ret;
  bool is_known;
  do
  {
    bool is_running;
    uint32_t numerator;
    uint32_t denominator;

    test_compare_got(GEARMAN_SUCCESS,
                     ret= gearman_client_job_status(client, job_handle, &is_known, &is_running, &numerator, &denominator),
                     gearman_client_error(client));
  } while (gearman_continue(ret) and is_known);

  return TEST_SUCCESS;
}

static void* test_malloc_fn(size_t size, void *context)
{
  bool *malloc_check= (bool *)context;
  *malloc_check= true;
  return malloc(size);
}

static void test_free_fn(void *ptr, void *context)
{
  bool *free_check= (bool *)context;
  *free_check= true;
  return free(ptr);
}

static test_return_t gearman_client_set_workload_malloc_fn_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  bool malloc_check= false;
  gearman_client_set_workload_malloc_fn(client, test_malloc_fn, &malloc_check);

  test_compare(TEST_SUCCESS, submit_job_test(object));
  test_compare(true, malloc_check);

  return TEST_SUCCESS;
}

static test_return_t gearman_client_set_workload_free_fn_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  bool free_check= false;
  gearman_client_set_workload_free_fn(client, test_free_fn, &free_check);

  test_compare(TEST_SUCCESS, submit_job_test(object));
  test_compare(true, free_check);

  return TEST_SUCCESS;
}

struct _alloc_test_st {
  int64_t count;
  int64_t total;

  _alloc_test_st():
    count(0),
    total(0)
  { }

  void add() 
  {
    count++;
    total++;
  }

  void subtract() 
  {
    count--;
  }

  bool success() // count is valid as 1 only with the current test
  {
    if (total and count == 1)
      return true;

    std::cerr << __func__ << ":" << __LINE__ << " Total:" <<  total << " Count:" << count << std::endl;

    return false;
  }
};

static void* test_malloc_count_fn(size_t size, void *context)
{
  _alloc_test_st *_foo= (_alloc_test_st *)context;

  _foo->add();
  
  return malloc(size);
}

static void test_free_count_fn(void *ptr, void *context)
{
  _alloc_test_st *_foo= (_alloc_test_st *)context;

  _foo->subtract();

  return free(ptr);
}


static test_return_t gearman_client_set_workload_allocators_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  _alloc_test_st _foo;

  gearman_client_set_workload_malloc_fn(client, test_malloc_count_fn, &_foo);
  gearman_client_set_workload_free_fn(client, test_free_count_fn, &_foo);

  test_compare(TEST_SUCCESS, submit_job_test(object));
  test_true(_foo.success());

  return TEST_SUCCESS;
}

static test_return_t gearman_client_job_status_with_return(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  gearman_string_t value= { test_literal_param("background_test") };

  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  gearman_job_handle_t job_handle;
  test_compare_got(GEARMAN_SUCCESS,
                   gearman_client_do_background(client, worker_function, NULL, gearman_string_param(value), job_handle), 
                   gearman_client_error(client));

  gearman_return_t ret;
  do
  {
    uint32_t numerator;
    uint32_t denominator;

    ret= gearman_client_job_status(client, job_handle, NULL, NULL, &numerator, &denominator);
  } while (gearman_continue(ret));
  test_compare(GEARMAN_SUCCESS, ret);

  return TEST_SUCCESS;
}

static test_return_t background_failure_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  gearman_job_handle_t job_handle;
  bool is_known;
  bool is_running;
  uint32_t numerator;
  uint32_t denominator;

  gearman_return_t rc= gearman_client_do_background(client, "does_not_exist", NULL,
                                                    test_literal_param("background_failure_test"),
                                                    job_handle);
  test_compare_got(GEARMAN_SUCCESS, rc, gearman_client_error(client));

  do {
    rc= gearman_client_job_status(client, job_handle, &is_known, &is_running,
                                  &numerator, &denominator);
    test_true(is_known == true and is_running == false and numerator == 0 and denominator == 0);
  } while (gearman_continue(rc)); // We do not test for is_known since the server will keep the job around until a worker comes along
  test_compare(GEARMAN_SUCCESS, rc);

  return TEST_SUCCESS;
}

static test_return_t add_servers_test(void *)
{
  gearman_client_st client, *client_ptr;

  client_ptr= gearman_client_create(&client);
  test_truth(client_ptr);

  gearman_return_t rc;
  rc= gearman_client_add_servers(&client, "localhost:4730,localhost");
  test_compare_got(GEARMAN_SUCCESS, rc, gearman_strerror(rc));

  rc= gearman_client_add_servers(&client, "old_jobserver:7003,broken:12345");
  test_compare_got(GEARMAN_GETADDRINFO, rc, gearman_strerror(rc));

  gearman_client_free(&client);

  return TEST_SUCCESS;
}

static test_return_t hostname_resolution(void *)
{
  Client client;
  assert(&client);

  test_skip(GEARMAN_SUCCESS, gearman_client_add_servers(&client, "exist.gearman.info:12345"));

  test_compare(GEARMAN_SUCCESS, client->universal.error.rc);

  test_compare_hint(GEARMAN_COULD_NOT_CONNECT,
                    gearman_client_echo(&client, test_literal_param("foo")),
                    gearman_client_error(&client));

  return TEST_SUCCESS;
}

static test_return_t gearman_client_st_id_t_TEST(void *)
{
  return TEST_SKIPPED;<commit_msg>Merging bzr://gaz.tangent.org/gearmand/build/ to Build branch<commit_after>/*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
 * 
 *  Gearmand client and server library.
 *
 *  Copyright (C) 2011 Data Differential, http://datadifferential.com/
 *  Copyright (C) 2008 <NAME>, <NAME>
 *  All rights reserved.
 *
 *  Redistribution and use in source and binary forms, with or without
 *  modification, are permitted provided that the following conditions are
 *  met:
 *
 *      * Redistributions of source code must retain the above copyright
 *  notice, this list of conditions and the following disclaimer.
 *
 *      * 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.
 *
 *      * The names of its contributors may not be used to endorse or
 *  promote products derived from this software without specific prior
 *  written permission.
 *
 *  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.
 *
 */

#include <config.h>

#include <libtest/test.hpp>
using namespace libtest;

#include <cassert>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <iostream>
#include <vector>
#include <boost/foreach.hpp>

#define GEARMAN_CORE
#include <libgearman/gearman.h>

#include <tests/start_worker.h>

#define NAMESPACE_KEY "foo123"

#define WORKER_FUNCTION_NAME "client_test"
#define WORKER_CHUNKED_FUNCTION_NAME "reverse_test"
#define WORKER_UNIQUE_FUNCTION_NAME "unique_test"
#define WORKER_SPLIT_FUNCTION_NAME "split_worker"

#include <tests/limits.h>
#include <tests/do.h>
#include <tests/server_options.h>
#include <tests/do_background.h>
#include <tests/execute.h>
#include <tests/gearman_client_do_job_handle.h>
#include <tests/gearman_execute_partition.h>
#include <tests/protocol.h>
#include <tests/regression.h>
#include <tests/task.h>
#include <tests/unique.h>
#include <tests/workers.h>
#include <tests/workers_v1.h>

#ifndef __INTEL_COMPILER
#pragma GCC diagnostic ignored "-Wold-style-cast"
#endif

struct client_test_st
{
  gearman_client_st *_client;
  bool _clone;
  std::vector<worker_handle_st *> workers;
  const char *_worker_name;

  client_test_st() :
    _clone(true),
    _worker_name(WORKER_FUNCTION_NAME)
  { 
    _client= gearman_client_create(NULL);

    if (_client == NULL)
    {
      throw "gearman_client_create() failed";
    }
  }

  ~client_test_st()
  {
    for (std::vector<worker_handle_st *>::iterator iter= workers.begin(); iter != workers.end(); iter++)
    {
      delete *iter;
    }
    workers.clear();
    gearman_client_free(_client);
  }

  void push(worker_handle_st *arg)
  {
    workers.push_back(arg);
  }

  const char *worker_name() const
  {
    return _worker_name;
  }

  void set_worker_name(const char *arg)
  {
    _worker_name= arg;
  }

  void set_clone(bool arg)
  {
    _clone= arg;
  }

  bool clone() const
  {
    return _clone;
  }

  gearman_client_st *client()
  {
    return _client;
  }

  void reset_client()
  {
    gearman_client_free(_client);
    _client= gearman_client_create(NULL);
  }
};

class Client {
public:
  Client()
  {
    _client= gearman_client_create(NULL);

    if (_client == NULL)
    {
      throw "gearman_client_create() failed";
    }
  }

  gearman_client_st* operator&() const
  { 
    return _client;
  }

  gearman_client_st* operator->() const
  { 
    return _client;
  }

  ~Client()
  {
    gearman_client_free(_client);
  }

private:
  gearman_client_st *_client;

};

#ifndef __INTEL_COMPILER
#pragma GCC diagnostic ignored "-Wold-style-cast"
#endif

/**
  @note Just here until I fix libhashkit.
*/
static uint32_t internal_generate_hash(const char *key, size_t key_length)
{
  const char *ptr= key;
  uint32_t value= 0;

  while (key_length--)
  {
    uint32_t val= (uint32_t) *ptr++;
    value += val;
    value += (value << 10);
    value ^= (value >> 6);
  }
  value += (value << 3);
  value ^= (value >> 11);
  value += (value << 15);

  return value == 0 ? 1 : (uint32_t) value;
}

/* Prototypes */
void *client_test_temp_worker(gearman_job_st *job, void *context,
                              size_t *result_size, gearman_return_t *ret_ptr);


extern "C"
{
  static void *client_thread(void *object)
  {
    volatile gearman_return_t *ret= (volatile gearman_return_t *)object;
    gearman_client_st client;
    gearman_client_st *client_ptr;
    size_t result_size;

    client_ptr= gearman_client_create(&client);

    if (client_ptr == NULL)
    {
      *ret= GEARMAN_MEMORY_ALLOCATION_FAILURE;
      pthread_exit(0);
    }

    gearman_return_t rc= gearman_client_add_server(&client, NULL, libtest::default_port());
    if (gearman_failed(rc))
    {
      *ret= rc;
      pthread_exit(0);
    }

    gearman_client_set_timeout(&client, 400);
    for (size_t x= 0; x < 5; x++)
    {
      (void)gearman_client_do(&client, "client_test_temp", NULL, NULL, 0, &result_size, &rc);

      if (gearman_failed(rc))
      {
        *ret= rc;
        pthread_exit(0);
      }
    }
    gearman_client_free(client_ptr);

    pthread_exit(0);
  }
}

static test_return_t init_test(void *)
{
  gearman_client_st client;

  test_truth(gearman_client_create(&client));

  gearman_client_free(&client);

  return TEST_SUCCESS;
}

static test_return_t allocation_test(void *)
{
  gearman_client_st *client;

  test_truth(client= gearman_client_create(NULL));

  gearman_client_free(client);

  return TEST_SUCCESS;
}

static test_return_t clone_test(void *object)
{
  const gearman_client_st *from= (gearman_client_st *)object;
  gearman_client_st *from_with_host;
  gearman_client_st *client;

  client= gearman_client_clone(NULL, NULL);

  test_truth(client);
  test_truth(client->options.allocated);

  gearman_client_free(client);

  client= gearman_client_clone(NULL, from);
  test_truth(client);
  gearman_client_free(client);

  from_with_host= gearman_client_create(NULL);
  test_truth(from_with_host);
  gearman_client_add_server(from_with_host, "localhost", 12345);

  client= gearman_client_clone(NULL, from_with_host);
  test_truth(client);

  test_truth(client->universal.con_list);
  test_truth(gearman_client_compare(client, from_with_host));

  gearman_client_free(client);
  gearman_client_free(from_with_host);

  return TEST_SUCCESS;
}

static test_return_t option_test(void *)
{
  gearman_client_st *gear;
  gearman_client_options_t default_options;

  gear= gearman_client_create(NULL);
  test_truth(gear);
  { // Initial Allocated, no changes
    test_truth(gear->options.allocated);
    test_false(gear->options.non_blocking);
    test_false(gear->options.unbuffered_result);
    test_false(gear->options.no_new);
    test_false(gear->options.free_tasks);
  }

  /* Set up for default options */
  default_options= gearman_client_options(gear);

  /*
    We take the basic options, and push
    them back in. See if we change anything.
  */
  gearman_client_set_options(gear, default_options);
  { // Initial Allocated, no changes
    test_truth(gear->options.allocated);
    test_false(gear->options.non_blocking);
    test_false(gear->options.unbuffered_result);
    test_false(gear->options.no_new);
    test_false(gear->options.free_tasks);
  }

  /*
    We will trying to modify non-mutable options (which should not be allowed)
  */
  {
    gearman_client_remove_options(gear, GEARMAN_CLIENT_ALLOCATED);
    { // Initial Allocated, no changes
      test_truth(gear->options.allocated);
      test_false(gear->options.non_blocking);
      test_false(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
    gearman_client_remove_options(gear, GEARMAN_CLIENT_NO_NEW);
    { // Initial Allocated, no changes
      test_truth(gear->options.allocated);
      test_false(gear->options.non_blocking);
      test_false(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
  }

  /*
    We will test modifying GEARMAN_CLIENT_NON_BLOCKING in several manners.
  */
  {
    gearman_client_remove_options(gear, GEARMAN_CLIENT_NON_BLOCKING);
    { // GEARMAN_CLIENT_NON_BLOCKING set to default, by default.
      test_truth(gear->options.allocated);
      test_false(gear->options.non_blocking);
      test_false(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
    gearman_client_add_options(gear, GEARMAN_CLIENT_NON_BLOCKING);
    { // GEARMAN_CLIENT_NON_BLOCKING set to default, by default.
      test_truth(gear->options.allocated);
      test_truth(gear->options.non_blocking);
      test_false(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
    gearman_client_set_options(gear, GEARMAN_CLIENT_NON_BLOCKING);
    { // GEARMAN_CLIENT_NON_BLOCKING set to default, by default.
      test_truth(gear->options.allocated);
      test_truth(gear->options.non_blocking);
      test_false(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
    gearman_client_set_options(gear, GEARMAN_CLIENT_UNBUFFERED_RESULT);
    { // Everything is now set to false except GEARMAN_CLIENT_UNBUFFERED_RESULT, and non-mutable options
      test_truth(gear->options.allocated);
      test_false(gear->options.non_blocking);
      test_truth(gear->options.unbuffered_result);
      test_false(gear->options.no_new);
      test_false(gear->options.free_tasks);
    }
    /*
      Reset options to default. Then add an option, and then add more options. Make sure
      the options are all additive.
    */
    {
      gearman_client_set_options(gear, default_options);
      { // See if we return to defaults
        test_truth(gear->options.allocated);
        test_false(gear->options.non_blocking);
        test_false(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_false(gear->options.free_tasks);
      }
      gearman_client_add_options(gear, GEARMAN_CLIENT_FREE_TASKS);
      { // All defaults, except timeout_return
        test_truth(gear->options.allocated);
        test_false(gear->options.non_blocking);
        test_false(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_truth(gear->options.free_tasks);
      }
      gearman_client_add_options(gear, (gearman_client_options_t)(GEARMAN_CLIENT_NON_BLOCKING|GEARMAN_CLIENT_UNBUFFERED_RESULT));
      { // GEARMAN_CLIENT_NON_BLOCKING set to default, by default.
        test_truth(gear->options.allocated);
        test_truth(gear->options.non_blocking);
        test_truth(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_truth(gear->options.free_tasks);
      }
    }
    /*
      Add an option, and then replace with that option plus a new option.
    */
    {
      gearman_client_set_options(gear, default_options);
      { // See if we return to defaults
        test_truth(gear->options.allocated);
        test_false(gear->options.non_blocking);
        test_false(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_false(gear->options.free_tasks);
      }
      gearman_client_add_options(gear, GEARMAN_CLIENT_FREE_TASKS);
      { // All defaults, except timeout_return
        test_truth(gear->options.allocated);
        test_false(gear->options.non_blocking);
        test_false(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_truth(gear->options.free_tasks);
      }
      gearman_client_add_options(gear, (gearman_client_options_t)(GEARMAN_CLIENT_FREE_TASKS|GEARMAN_CLIENT_UNBUFFERED_RESULT));
      { // GEARMAN_CLIENT_NON_BLOCKING set to default, by default.
        test_truth(gear->options.allocated);
        test_false(gear->options.non_blocking);
        test_truth(gear->options.unbuffered_result);
        test_false(gear->options.no_new);
        test_truth(gear->options.free_tasks);
      }
    }
  }

  gearman_client_free(gear);

  return TEST_SUCCESS;
}

static test_return_t echo_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  gearman_string_t value= { test_literal_param("This is my echo test") };

  test_compare(GEARMAN_SUCCESS, gearman_client_echo(client, gearman_string_param(value)));

  return TEST_SUCCESS;
}

static test_return_t submit_job_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  const char *worker_function= (const char *)gearman_client_context(client);
  gearman_string_t value= { test_literal_param("submit_job_test") };

  size_t result_length;
  gearman_return_t rc;
  void *job_result= gearman_client_do(client, worker_function, NULL, gearman_string_param(value), &result_length, &rc);

  test_compare_got(GEARMAN_SUCCESS, 
                   rc, gearman_client_error(client) ? gearman_client_error(client) : gearman_strerror(rc));

  test_truth(job_result);
  test_compare(gearman_size(value), result_length);

  test_memcmp(gearman_c_str(value), job_result, gearman_size(value));

  free(job_result);

  return TEST_SUCCESS;
}

static test_return_t submit_null_job_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;

  test_truth(client);

  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  size_t result_length;
  gearman_return_t rc;
  void *job_result= gearman_client_do(client, worker_function, NULL, NULL, 0,
                                      &result_length, &rc);
  test_compare_got(GEARMAN_SUCCESS, rc, gearman_client_error(client));
  test_zero(result_length);
  test_false(job_result);

  return TEST_SUCCESS;
}

static test_return_t submit_exception_job_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  size_t result_length;
  gearman_return_t rc;
  void *job_result= gearman_client_do(client, worker_function, NULL,
                                      test_literal_param("exception"),
                                      &result_length, &rc);
  test_compare_got(GEARMAN_SUCCESS, rc, gearman_client_error(client) ? gearman_client_error(client) : gearman_strerror(rc));
  test_memcmp("exception", job_result, result_length);
  free(job_result);

  return TEST_SUCCESS;
}

static test_return_t submit_warning_job_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  size_t result_length;
  gearman_return_t rc;
  void *job_result= gearman_client_do(client, worker_function, NULL,
                                      test_literal_param("warning"),
                                      &result_length, &rc);
  test_compare_got(GEARMAN_SUCCESS, rc, gearman_client_error(client) ? gearman_client_error(client) : gearman_strerror(rc));
  test_memcmp("warning", job_result, result_length);
  free(job_result);

  return TEST_SUCCESS;
}

static test_return_t submit_fail_job_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);


  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  size_t result_length;
  gearman_return_t rc;
  void *job_result= gearman_client_do(client, worker_function, NULL, "fail", 4,
                                      &result_length, &rc);
  test_compare_got(GEARMAN_WORK_FAIL, rc, gearman_client_error(client));
  test_false(job_result);
  test_false(result_length);

  return TEST_SUCCESS;
}

static test_return_t submit_multiple_do(void *object)
{
  for (uint32_t x= 0; x < 100 /* arbitrary */; x++)
  {
    uint32_t option= random() %3;

    switch (option)
    {
    case 0:
      test_compare(TEST_SUCCESS, submit_null_job_test(object));
      break;
    case 1:
      test_compare(TEST_SUCCESS, submit_job_test(object));
      break;
    default:
    case 2:
      test_compare(TEST_SUCCESS, submit_fail_job_test(object));
      break;
    }
  }

  return TEST_SUCCESS;
}

static test_return_t gearman_client_job_status_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  gearman_string_t value= { test_literal_param("background_test") };

  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  gearman_job_handle_t job_handle;
  test_compare_got(GEARMAN_SUCCESS,
                   gearman_client_do_background(client, worker_function, NULL, gearman_string_param(value), job_handle), 
                   gearman_client_error(client));

  gearman_return_t ret;
  bool is_known;
  do
  {
    bool is_running;
    uint32_t numerator;
    uint32_t denominator;

    test_compare_got(GEARMAN_SUCCESS,
                     ret= gearman_client_job_status(client, job_handle, &is_known, &is_running, &numerator, &denominator),
                     gearman_client_error(client));
  } while (gearman_continue(ret) and is_known);

  return TEST_SUCCESS;
}

static void* test_malloc_fn(size_t size, void *context)
{
  bool *malloc_check= (bool *)context;
  *malloc_check= true;
  return malloc(size);
}

static void test_free_fn(void *ptr, void *context)
{
  bool *free_check= (bool *)context;
  *free_check= true;
  return free(ptr);
}

static test_return_t gearman_client_set_workload_malloc_fn_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  bool malloc_check= false;
  gearman_client_set_workload_malloc_fn(client, test_malloc_fn, &malloc_check);

  test_compare(TEST_SUCCESS, submit_job_test(object));
  test_compare(true, malloc_check);

  return TEST_SUCCESS;
}

static test_return_t gearman_client_set_workload_free_fn_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  bool free_check= false;
  gearman_client_set_workload_free_fn(client, test_free_fn, &free_check);

  test_compare(TEST_SUCCESS, submit_job_test(object));
  test_compare(true, free_check);

  return TEST_SUCCESS;
}

struct _alloc_test_st {
  int64_t count;
  int64_t total;

  _alloc_test_st():
    count(0),
    total(0)
  { }

  void add() 
  {
    count++;
    total++;
  }

  void subtract() 
  {
    count--;
  }

  bool success() // count is valid as 1 only with the current test
  {
    if (total and count == 1)
      return true;

    std::cerr << __func__ << ":" << __LINE__ << " Total:" <<  total << " Count:" << count << std::endl;

    return false;
  }
};

static void* test_malloc_count_fn(size_t size, void *context)
{
  _alloc_test_st *_foo= (_alloc_test_st *)context;

  _foo->add();
  
  return malloc(size);
}

static void test_free_count_fn(void *ptr, void *context)
{
  _alloc_test_st *_foo= (_alloc_test_st *)context;

  _foo->subtract();

  return free(ptr);
}


static test_return_t gearman_client_set_workload_allocators_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  _alloc_test_st _foo;

  gearman_client_set_workload_malloc_fn(client, test_malloc_count_fn, &_foo);
  gearman_client_set_workload_free_fn(client, test_free_count_fn, &_foo);

  test_compare(TEST_SUCCESS, submit_job_test(object));
  test_true(_foo.success());

  return TEST_SUCCESS;
}

static test_return_t gearman_client_job_status_with_return(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  test_truth(client);

  gearman_string_t value= { test_literal_param("background_test") };

  const char *worker_function= (const char *)gearman_client_context(client);
  test_truth(worker_function);

  gearman_job_handle_t job_handle;
  test_compare_got(GEARMAN_SUCCESS,
                   gearman_client_do_background(client, worker_function, NULL, gearman_string_param(value), job_handle), 
                   gearman_client_error(client));

  gearman_return_t ret;
  do
  {
    uint32_t numerator;
    uint32_t denominator;

    ret= gearman_client_job_status(client, job_handle, NULL, NULL, &numerator, &denominator);
  } while (gearman_continue(ret));
  test_compare(GEARMAN_SUCCESS, ret);

  return TEST_SUCCESS;
}

static test_return_t background_failure_test(void *object)
{
  gearman_client_st *client= (gearman_client_st *)object;
  gearman_job_handle_t job_handle;
  bool is_known;
  bool is_running;
  uint32_t numerator;
  uint32_t denominator;

  gearman_return_t rc= gearman_client_do_background(client, "does_not_exist", NULL,
                                                    test_literal_param("background_failure_test"),
                                                    job_handle);
  test_compare_got(GEARMAN_SUCCESS, rc, gearman_client_error(client));

  do {
    rc= gearman_client_job_status(client, job_handle, &is_known, &is_running,
                                  &numerator, &denominator);
    test_true(is_known == true and is_running == false and numerator == 0 and denominator == 0);
  } while (gearman_continue(rc)); // We do not test for is_known since the server will keep the job around until a worker comes along
  test_compare(GEARMAN_SUCCESS, rc);

  return TEST_SUCCESS;
}

static test_return_t add_servers_test(void *)
{
  gearman_client_st *client= gearman_client_create(NULL);
  test_truth(client);

  test_compare_got(GEARMAN_SUCCESS,
                   gearman_client_add_servers(client, "localhost:4730,localhost"),
                   gearman_client_error(client));

  test_compare_got(GEARMAN_GETADDRINFO,
                   gearman_client_add_servers(client, "exist.gearman.info:7003,does_not_exist.gearman.info:12345"),
                   gearman_client_error(client));

  gearman_client_free(client);

  return TEST_SUCCESS;
}

static test_return_t hostname_resolution(void *)
{
  Client client;
  assert(&client);

  test_skip(GEARMAN_SUCCESS, gearman_client_add_servers(&client, "exist.gearman.info:12345"));

  test_compare(GEARMAN_SUCCESS, client->universal.error.rc);

  test_compare_hint(GEARMAN_COULD_NOT_CONNECT,
                    gearman_client_echo(&client, test_literal_param("foo")),
                    gearman_client_error(&client));

  return TEST_SUCCESS;
}

static test_return_t gearman_client_st_id_t_TEST(void *)
{
  return TEST_SKIPPED;
{'fraction_non_alphanumeric': 0.10192736950549451, 'fraction_numerical': 0.004464285714285714, 'mean_word_length': 4.088238506060937, 'pattern_counts': {'":': 2, '<': 109, '<?xml version=': 0, '>': 221, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6628519', 'n_tokens_mistral': 16313, 'n_tokens_neox': 15543, 'n_words': 2965}