text:
#define FIO_INCLUDE_LINKED_LIST
#define FIO_INCLUDE_STR
// #define DEBUG 1
#include <fio.h>

#include <fiobj.h>

#include <redis_engine.h>
#include <resp_parser.h>

#define REDIS_READ_BUFFER 8192
/* *****************************************************************************
The Redis Engine and Callbacks Object
***************************************************************************** */

typedef struct {
  fio_pubsub_engine_s en;
  struct redis_engine_internal_s {
    fio_protocol_s protocol;
    intptr_t uuid;
    resp_parser_s parser;
    void (*on_message)(struct redis_engine_internal_s *parser, FIOBJ msg);
    FIOBJ str;
    FIOBJ ary;
    uint32_t ary_count;
    uint16_t buf_pos;
    uint16_t nesting;
  } pub_data, sub_data;
  subscription_s *publication_forwarder;
  subscription_s *cmd_forwarder;
  subscription_s *cmd_reply;
  char *address;
  char *port;
  char *auth;
  FIOBJ last_ch;
  size_t auth_len;
  size_t ref;
  fio_ls_embd_s queue;
  fio_lock_i lock;
  fio_lock_i lock_connection;
  uint8_t ping_int;
  volatile uint8_t pub_sent;
  volatile uint8_t flag;
  uint8_t buf[];
} redis_engine_s;

typedef struct {
  fio_ls_embd_s node;
  void (*callback)(fio_pubsub_engine_s *e, FIOBJ reply, void *udata);
  void *udata;
  size_t cmd_len;
  uint8_t cmd[];
} redis_commands_s;

/** converts from a publishing protocol to an `redis_engine_s`. */
#define pub2redis(pr) FIO_LS_EMBD_OBJ(redis_engine_s, pub_data, (pr))
/** converts from a subscribing protocol to an `redis_engine_s`. */
#define sub2redis(pr) FIO_LS_EMBD_OBJ(redis_engine_s, sub_data, (pr))

/** converts from a `resp_parser_s` to the internal data structure. */
#define parser2data(prsr)                                                      \
  FIO_LS_EMBD_OBJ(struct redis_engine_internal_s, parser, (prsr))

/* releases any resources used by an internal engine*/
static inline void redis_internal_reset(struct redis_engine_internal_s *i) {
  i->buf_pos = 0;
  i->parser = (resp_parser_s){.obj_countdown = 0, .expecting = 0};
  fiobj_free((FIOBJ)fio_ct_if(i->ary == FIOBJ_INVALID, (uintptr_t)i->str,
                              (uintptr_t)i->ary));
  i->str = FIOBJ_INVALID;
  i->ary = FIOBJ_INVALID;
  i->ary_count = 0;
  i->nesting = 0;
  i->uuid = -1;
}

/** cleans up and frees the engine data. */
static inline void redis_free(redis_engine_s *r) {
  if (fio_atomic_sub(&r->ref, 1))
    return;
  FIO_LOG_DEBUG("freeing redis engine for %s:%s", r->address, r->port);
  redis_internal_reset(&r->pub_data);
  redis_internal_reset(&r->sub_data);
  fiobj_free(r->last_ch);
  while (fio_ls_embd_any(&r->queue)) {
    fio_free(
        FIO_LS_EMBD_OBJ(redis_commands_s, node, fio_ls_embd_pop(&r->queue)));
  }
  fio_unsubscribe(r->publication_forwarder);
  r->publication_forwarder = NULL;
  fio_unsubscribe(r->cmd_forwarder);
  r->cmd_forwarder = NULL;
  fio_unsubscribe(r->cmd_reply);
  r->cmd_reply = NULL;
  fio_free(r);
}

/* *****************************************************************************
Simple RESP formatting
***************************************************************************** */

inline static void fiobj2resp___internal(FIOBJ dest, FIOBJ obj) {
  fio_str_info_s s;
  switch (FIOBJ_TYPE(obj)) {
  case FIOBJ_T_NULL:
    fiobj_str_write(dest, "$-1\r\n", 5);
    break;
  case FIOBJ_T_ARRAY:
    fiobj_str_write(dest, "*", 1);
    fiobj_str_write_i(dest, fiobj_ary_count(obj));
    fiobj_str_write(dest, "\r\n", 2);
    break;
  case FIOBJ_T_HASH:
    fiobj_str_write(dest, "*", 1);
    fiobj_str_write_i(dest, fiobj_hash_count(obj) * 2);
    fiobj_str_write(dest, "\r\n", 2);
    break;
  case FIOBJ_T_TRUE:
    fiobj_str_write(dest, "$4\r\ntrue\r\n", 10);
    break;
  case FIOBJ_T_FALSE:
    fiobj_str_write(dest, "$4\r\nfalse\r\n", 11);
    break;
#if 0
    /* Numbers aren't as good for commands as one might think... */
  case FIOBJ_T_NUMBER:
    fiobj_str_write(dest, ":", 1);
    fiobj_str_write_i(dest, fiobj_obj2num(obj));
    fiobj_str_write(dest, "\r\n", 2);
    break;
#else
  case FIOBJ_T_NUMBER: /* overflow */
#endif
  case FIOBJ_T_FLOAT:   /* overflow */
  case FIOBJ_T_UNKNOWN: /* overflow */
  case FIOBJ_T_STRING:  /* overflow */
  case FIOBJ_T_DATA:
    s = fiobj_obj2cstr(obj);
    fiobj_str_write(dest, "$", 1);
    fiobj_str_write_i(dest, s.len);
    fiobj_str_write(dest, "\r\n", 2);
    fiobj_str_write(dest, s.data, s.len);
    fiobj_str_write(dest, "\r\n", 2);
    break;
  }
}

static int fiobj2resp_task(FIOBJ o, void *dest_) {
  if (fiobj_hash_key_in_loop())
    fiobj2resp___internal((FIOBJ)dest_, fiobj_hash_key_in_loop());
  fiobj2resp___internal((FIOBJ)dest_, o);
  return 0;
}

/**
 * Converts FIOBJ objects into a RESP string (client mode).
 */
static FIOBJ fiobj2resp(FIOBJ dest, FIOBJ obj) {
  fiobj_each2(obj, fiobj2resp_task, (void *)dest);
  return dest;
}

/**
 * Converts FIOBJ objects into a RESP string (client mode).
 *
 * Don't call `fiobj_free`, object will self-destruct.
 */
static inline FIOBJ fiobj2resp_tmp(FIOBJ obj) {
  return fiobj2resp(fiobj_str_tmp(), obj);
}

/* *****************************************************************************
RESP parser callbacks
***************************************************************************** */

/** a local static callback, called when a parser / protocol error occurs. */
static int resp_on_parser_error(resp_parser_s *parser) {
  struct redis_engine_internal_s *i = parser2data(parser);
  FIO_LOG_ERROR("(redis) parser error - attempting to restart connection.\n");
  fio_close(i->uuid);
  return -1;
}

/** a local static callback, called when the RESP message is complete. */
static int resp_on_message(resp_parser_s *parser) {
  struct redis_engine_internal_s *i = parser2data(parser);
  FIOBJ msg = i->ary ? i->ary : i->str;
  i->on_message(i, msg);
  /* cleanup */
  fiobj_free(msg);
  i->ary = FIOBJ_INVALID;
  i->str = FIOBJ_INVALID;
  return 0;
}

/** a local helper to add parsed objects to the data store. */
static inline void resp_add_obj(struct redis_engine_internal_s *dest, FIOBJ o) {
  if (dest->ary) {
    fiobj_ary_push(dest->ary, o);
    --dest->ary_count;
    if (!dest->ary_count && dest->nesting) {
      FIOBJ tmp = fiobj_ary_shift(dest->ary);
      dest->ary_count = fiobj_obj2num(tmp);
      fiobj_free(tmp);
      dest->ary = fiobj_ary_shift(dest->ary);
      --dest->nesting;
    }
  }
  dest->str = o;
}

/** a local static callback, called when a Number object is parsed. */
static int resp_on_number(resp_parser_s *parser, int64_t num) {
  struct redis_engine_internal_s *data = parser2data(parser);
  resp_add_obj(data, fiobj_num_new(num));
  return 0;
}
/** a local static callback, called when a OK message is received. */
static int resp_on_okay(resp_parser_s *parser) {
  struct redis_engine_internal_s *data = parser2data(parser);
  resp_add_obj(data, fiobj_true());
  return 0;
}
/** a local static callback, called when NULL is received. */
static int resp_on_null(resp_parser_s *parser) {
  struct redis_engine_internal_s *data = parser2data(parser);
  resp_add_obj(data, fiobj_null());
  return 0;
}

/**
 * a local static callback, called when a String should be allocated.
 *
 * `str_len` is the expected number of bytes that will fill the final string
 * object, without any NUL byte marker (the string might be binary).
 *
 * If this function returns any value besides 0, parsing is stopped.
 */
static int resp_on_start_string(resp_parser_s *parser, size_t str_len) {
  struct redis_engine_internal_s *data = parser2data(parser);
  resp_add_obj(data, fiobj_str_buf(str_len));
  return 0;
}
/** a local static callback, called as String objects are streamed. */
static int resp_on_string_chunk(resp_parser_s *parser, void *data, size_t len) {
  struct redis_engine_internal_s *i = parser2data(parser);
  fiobj_str_write(i->str, data, len);
  return 0;
}
/** a local static callback, called when a String object had finished
 * streaming.
 */
static int resp_on_end_string(resp_parser_s *parser) {
  return 0;
  (void)parser;
}

/** a local static callback, called an error message is received. */
static int resp_on_err_msg(resp_parser_s *parser, void *data, size_t len) {
  struct redis_engine_internal_s *i = parser2data(parser);
  resp_add_obj(i, fiobj_str_new(data, len));
  return 0;
}

/**
 * a local static callback, called when an Array should be allocated.
 *
 * `array_len` is the expected number of objects that will fill the Array
 * object.
 *
 * There's no `resp_on_end_array` callback since the RESP protocol assumes the
 * message is finished along with the Array (`resp_on_message` is called).
 * However, just in case a non-conforming client/server sends nested Arrays,
 * the callback should test against possible overflow or nested Array endings.
 *
 * If this function returns any value besides 0, parsing is stopped.
 */
static int resp_on_start_array(resp_parser_s *parser, size_t array_len) {
  struct redis_engine_internal_s *i = parser2data(parser);
  if (i->ary) {
    ++i->nesting;
    FIOBJ tmp = fiobj_ary_new2(array_len + 2);
    fiobj_ary_push(tmp, fiobj_num_new(i->ary_count));
    fiobj_ary_push(tmp, fiobj_num_new(i->ary));
    i->ary = tmp;
  } else {
    i->ary = fiobj_ary_new2(array_len + 2);
  }
  i->ary_count = array_len;
  return 0;
}

/* *****************************************************************************
Publication and Command Handling
***************************************************************************** */

/* the deferred callback handler */
static void redis_perform_callback(void *e, void *cmd_) {
  redis_commands_s *cmd = cmd_;
  FIOBJ reply = (FIOBJ)cmd->node.next;
  if (cmd->callback)
    cmd->callback(e, reply, cmd->udata);
  fiobj_free(reply);
  FIO_LOG_DEBUG("Handled: %s\n", cmd->cmd);
  fio_free(cmd);
}

/* send command within lock, to ensure flag integrity */
static void redis_send_next_command_unsafe(redis_engine_s *r) {
  if (!r->pub_sent && fio_ls_embd_any(&r->queue)) {
    r->pub_sent = 1;
    redis_commands_s *cmd =
        FIO_LS_EMBD_OBJ(redis_commands_s, node, r->queue.next);
    fio_write2(r->pub_data.uuid, .data.buffer = cmd->cmd,
               .length = cmd->cmd_len, .after.dealloc = FIO_DEALLOC_NOOP);
    FIO_LOG_DEBUG("(redis %d) Sending (%zu bytes):\n%s\n", (int)getpid(),
                  cmd->cmd_len, cmd->cmd);
  }
}

/* attach a command to the queue */
static void redis_attach_cmd(redis_engine_s *r, redis_commands_s *cmd) {
  fio_lock(&r->lock);
  fio_ls_embd_push(&r->queue, &cmd->node);
  redis_send_next_command_unsafe(r);
  fio_unlock(&r->lock);
}

/** a local static callback, called when the RESP message is complete. */
static void resp_on_pub_message(struct redis_engine_internal_s *i, FIOBJ msg) {
  redis_engine_s *r = pub2redis(i);
  // #if DEBUG
  if (FIO_LOG_LEVEL >= FIO_LOG_LEVEL_DEBUG) {
    FIOBJ json = fiobj_obj2json(msg, 1);
    FIO_LOG_DEBUG("Redis reply:\n%s\n", fiobj_obj2cstr(json).data);
    fiobj_free(json);
  }
  // #endif
  /* publishing / command parser */
  fio_lock(&r->lock);
  fio_ls_embd_s *node = fio_ls_embd_shift(&r->queue);
  r->pub_sent = 0;
  redis_send_next_command_unsafe(r);
  fio_unlock(&r->lock);
  if (!node) {
    /* TODO: possible ping? from server?! not likely... */
    FIO_LOG_WARNING("(redis %d) received a reply when no command was sent.",
                    (int)getpid());
    return;
  }
  node->next = (void *)fiobj_dup(msg);
  fio_defer(redis_perform_callback, &r->en,
            FIO_LS_EMBD_OBJ(redis_commands_s, node, node));
}

/* *****************************************************************************
Subscription Message Handling
***************************************************************************** */

/** a local static callback, called when the RESP message is complete. */
static void resp_on_sub_message(struct redis_engine_internal_s *i, FIOBJ msg) {
  redis_engine_s *r = sub2redis(i);
  /* subscriotion parser */
  if (FIOBJ_TYPE(msg) != FIOBJ_T_ARRAY) {
    if (FIOBJ_TYPE(msg) != FIOBJ_T_STRING || fiobj_obj2cstr(msg).len != 4 ||
        fiobj_obj2cstr(msg).data[0] != 'P') {
      FIO_LOG_WARNING("(redis) unexpected data format in "
                      "subscription stream (%zu bytes):\n     %s\n",
                      fiobj_obj2cstr(msg).len, fiobj_obj2cstr(msg).data);
    }
  } else {
    // FIOBJ *ary = fiobj_ary2ptr(msg);
    // for (size_t i = 0; i < fiobj_ary_count(msg); ++i) {
    //   fio_str_info_s tmp = fiobj_obj2cstr(ary[i]);
    //   fprintf(stderr, "(%lu) %s\n", (unsigned long)i, tmp.data);
    // }
    fio_str_info_s tmp = fiobj_obj2cstr(fiobj_ary_index(msg, 0));
    if (tmp.len == 7) { /* "message"  */
      fiobj_free(r->last_ch);
      r->last_ch = fiobj_dup(fiobj_ary_index(msg, 1));
      fio_publish(.channel = fiobj_obj2cstr(r->last_ch),
                  .message = fiobj_obj2cstr(fiobj_ary_index(msg, 2)),
                  .engine = FIO_PUBSUB_CLUSTER);
    } else if (tmp.len == 8) { /* "pmessage" */
      if (!fiobj_iseq(r->last_ch, fiobj_ary_index(msg, 2)))
        fio_publish(.channel = fiobj_obj2cstr(fiobj_ary_index(msg, 2)),
                    .message = fiobj_obj2cstr(fiobj_ary_index(msg, 3)),
                    .engine = FIO_PUBSUB_CLUSTER);
    }
  }
}

/* *****************************************************************************
Connection Callbacks (fio_protocol_s) and Engine
***************************************************************************** */

/** defined later - connects to Redis */
static void redis_connect(void *r, void *i);

#define defer_redis_connect(r, i)                                              \
  do {                                                                         \
    fio_atomic_add(&(r)->ref, 1);                                              \
    fio_defer(redis_connect, (r), (i));                                        \
  } while (0);

/** Called when a data is available, but will not run concurrently */
static void redis_on_data(intptr_t uuid, fio_protocol_s *pr) {
  struct redis_engine_internal_s *internal =
      (struct redis_engine_internal_s *)pr;
  uint8_t *buf;
  if (internal->on_message == resp_on_sub_message) {
    buf = sub2redis(pr)->buf + REDIS_READ_BUFFER;
  } else {
    buf = pub2redis(pr)->buf;
  }
  ssize_t i = fio_read(uuid, buf + internal->buf_pos,
                       REDIS_READ_BUFFER - internal->buf_pos);
  if (i <= 0)
    return;

  internal->buf_pos += i;
  i = resp_parse(&internal->parser, buf, internal->buf_pos);
  if (i) {
    memmove(buf, buf + internal->buf_pos - i, i);
  }
  internal->buf_pos = i;
}

/** Called when the connection was closed, but will not run concurrently */
static void redis_on_close(intptr_t uuid, fio_protocol_s *pr) {
  struct redis_engine_internal_s *internal =
      (struct redis_engine_internal_s *)pr;
  redis_internal_reset(internal);
  redis_engine_s *r;
  if (internal->on_message == resp_on_sub_message) {
    r = sub2redis(pr);
    fiobj_free(r->last_ch);
    r->last_ch = FIOBJ_INVALID;
    if (r->flag) {
      /* reconnection for subscription connection. */
      if (uuid != -1) {
        FIO_LOG_WARNING("(redis %d) subscription connection lost. "
                        "Reconnecting...",
                        (int)getpid());
      }
      fio_atomic_sub(&r->ref, 1);
      defer_redis_connect(r, internal);
    } else {
      redis_free(r);
    }
  } else {
    r = pub2redis(pr);
    if (r->flag && uuid != -1) {
      FIO_LOG_WARNING("(redis %d) publication connection lost. "
                      "Reconnecting...",
                      (int)getpid());
    }
    r->pub_sent = 0;
    fio_close(r->sub_data.uuid);
    redis_free(r);
  }
  (void)uuid;
}

/** Called before the facil.io reactor is shut down. */
static uint8_t redis_on_shutdown(intptr_t uuid, fio_protocol_s *pr) {
  fio_write2(uuid, .data.buffer = "*1\r\n$4\r\nQUIT\r\n", .length = 14,
             .after.dealloc = FIO_DEALLOC_NOOP);
  return 0;
  (void)pr;
}

/** Called on connection timeout. */
static void redis_sub_ping(intptr_t uuid, fio_protocol_s *pr) {
  fio_write2(uuid, .data.buffer = "*1\r\n$4\r\nPING\r\n", .length = 14,
             .after.dealloc = FIO_DEALLOC_NOOP);
  (void)pr;
}

/** Called on connection timeout. */
static void redis_pub_ping(intptr_t uuid, fio_protocol_s *pr) {
  redis_engine_s *r = pub2redis(pr);
  if (fio_ls_embd_any(&r->queue)) {
    FIO_LOG_WARNING("(redis) Redis server unresponsive, disconnecting.");
    fio_close(uuid);
    return;
  }
  redis_commands_s *cmd = fio_malloc(sizeof(*cmd) + 15);
  FIO_ASSERT_ALLOC(cmd);
  *cmd = (redis_commands_s){.cmd_len = 14};
  memcpy(cmd->cmd, "*1\r\n$4\r\nPING\r\n\0", 15);
  redis_attach_cmd(r, cmd);
}

/* *****************************************************************************
Connecting to Redis
***************************************************************************** */

static void redis_on_auth(fio_pubsub_engine_s *e, FIOBJ reply, void *udata) {
  if (FIOBJ_TYPE_IS(reply, FIOBJ_T_TRUE)) {
    fio_str_info_s s = fiobj_obj2cstr(reply);
    FIO_LOG_WARNING("(redis) Authentication FAILED."
                    "        %.*s",
                    (int)s.len, s.data);
  }
  (void)e;
  (void)udata;
}

static void redis_on_connect(intptr_t uuid, void *i_) {
  struct redis_engine_internal_s *i = i_;
  redis_engine_s *r;
  i->uuid = uuid;

  if (i->on_message == resp_on_sub_message) {
    r = sub2redis(i);
    if (r->auth_len) {
      fio_write2(uuid, .data.buffer = r->auth, .length = r->auth_len,
                 .after.dealloc = FIO_DEALLOC_NOOP);
    }
    fio_pubsub_reattach(&r->en);
    if (r->pub_data.uuid == -1) {
      defer_redis_connect(r, &r->pub_data);
    }
    FIO_LOG_INFO("(redis %d) subscription connection established.",
                 (int)getpid());
  } else {
    r = pub2redis(i);
    if (r->auth_len) {
      redis_commands_s *cmd = fio_malloc(sizeof(*cmd) + r->auth_len);
      FIO_ASSERT_ALLOC(cmd);
      *cmd =
          (redis_commands_s){.cmd_len = r->auth_len, .callback = redis_on_auth};
      memcpy(cmd->cmd, r->auth, r->auth_len);
      fio_lock(&r->lock);
      r->pub_sent = 0;
      fio_ls_embd_unshift(&r->queue, &cmd->node);
      redis_send_next_command_unsafe(r);
      fio_unlock(&r->lock);
    } else {
      fio_lock(&r->lock);
      r->pub_sent = 0;
      redis_send_next_command_unsafe(r);
      fio_unlock(&r->lock);
    }
    FIO_LOG_INFO("(redis %d) publication connection established.",
                 (int)getpid());
  }

  i->protocol.rsv = 0;
  fio_attach(uuid, &i->protocol);
  fio_timeout_set(uuid, r->ping_int);

  return;
}

static void redis_on_connect_failed(intptr_t uuid, void *i_) {
  struct redis_engine_internal_s *i = i_;
  i->uuid = -1;
  i->protocol.on_close(-1, &i->protocol);
  (void)uuid;
}

static void redis_connect(void *r_, void *i_) {
  redis_engine_s *r = r_;
  struct redis_engine_internal_s *i = i_;
  fio_lock(&r->lock_connection);
  if (r->flag == 0 || i->uuid != -1 || !fio_is_running()) {
    fio_unlock(&r->lock_connection);
    redis_free(r);
    return;
  }
  // fio_atomic_add(&r->ref, 1);
  i->uuid = fio_connect(.address = r->address, .port = r->port,
                        .on_connect = redis_on_connect, .udata = i,
                        .on_fail = redis_on_connect_failed);
  fio_unlock(&r->lock_connection);
}

/* *****************************************************************************
Engine / Bridge Callbacks (Root Process)
***************************************************************************** */

static void redis_on_subscribe_root(const fio_pubsub_engine_s *eng,
                                    fio_str_info_s channel,
                                    fio_match_fn match) {
  redis_engine_s *r = (redis_engine_s *)eng;
  if (r->sub_data.uuid != -1) {
    FIOBJ cmd = fiobj_str_buf(96 + channel.len);
    if (match == FIO_MATCH_GLOB)
      fiobj_str_write(cmd, "*2\r\n$10\r\nPSUBSCRIBE\r\n$", 22);
    else
      fiobj_str_write(cmd, "*2\r\n$9\r\nSUBSCRIBE\r\n$", 20);
    fiobj_str_write_i(cmd, channel.len);
    fiobj_str_write(cmd, "\r\n", 2);
    fiobj_str_write(cmd, channel.data, channel.len);
    fiobj_str_write(cmd, "\r\n", 2);
    // {
    //   fio_str_info_s s = fiobj_obj2cstr(cmd);
    //   fprintf(stderr, "(%d) Sending Subscription (%p):\n%s\n", getpid(),
    //           (void *)r->sub_data.uuid, s.data);
    // }
    fiobj_send_free(r->sub_data.uuid, cmd);
  }
}

static void redis_on_unsubscribe_root(const fio_pubsub_engine_s *eng,
                                      fio_str_info_s channel,
                                      fio_match_fn match) {
  redis_engine_s *r = (redis_engine_s *)eng;
  if (r->sub_data.uuid != -1) {
    fio_str_s *cmd = fio_str_new2();
    fio_str_capa_assert(cmd, 96 + channel.len);
    if (match == FIO_MATCH_GLOB)
      fio_str_write(cmd, "*2\r\n$12\r\nPUNSUBSCRIBE\r\n$", 24);
    else
      fio_str_write(cmd, "*2\r\n$11\r\nUNSUBSCRIBE\r\n$", 23);
    fio_str_write_i(cmd, channel.len);
    fio_str_write(cmd, "\r\n", 2);
    fio_str_write(cmd, channel.data, channel.len);
    fio_str_write(cmd, "\r\n", 2);
    // {
    //   fio_str_info_s s = fio_str_info(cmd);
    //   fprintf(stderr, "(%d) Cancel Subscription (%p):\n%s\n", getpid(),
    //           (void *)r->sub_data.uuid, s.data);
    // }
    fio_str_send_free2(r->sub_data.uuid, cmd);
  }
}

static void redis_on_publish_root(const fio_pubsub_engine_s *eng,
                                  fio_str_info_s channel, fio_str_info_s msg,
                                  uint8_t is_json) {
  redis_engine_s *r = (redis_engine_s *)eng;
  redis_commands_s *cmd = fio_malloc(sizeof(*cmd) + channel.len + msg.len + 96);
  FIO_ASSERT_ALLOC(cmd);
  *cmd = (redis_commands_s){.cmd_len = 0};
  memcpy(cmd->cmd, "*3\r\n$7\r\nPUBLISH\r\n$", 18);
  char *buf = (char *)cmd->cmd + 18;
  buf += fio_ltoa((void *)buf, channel.len, 10);
  *buf++ = '\r';
  *buf++ = '\n';
  memcpy(buf, channel.data, channel.len);
  buf += channel.len;
  *buf++ = '\r';
  *buf++ = '\n';
  *buf++ = '$';
  buf += fio_ltoa(buf, msg.len, 10);
  *buf++ = '\r';
  *buf++ = '\n';
  memcpy(buf, msg.data, msg.len);
  buf += msg.len;
  *buf++ = '\r';
  *buf++ = '\n';
  *buf = 0;
  FIO_LOG_DEBUG("(%d) Publishing:\n%s", (int)getpid(), cmd->cmd);
  cmd->cmd_len = (uintptr_t)buf - (uintptr_t)(cmd + 1);
  redis_attach_cmd(r, cmd);
  return;
  (void)is_json;
}

/* *****************************************************************************
Engine / Bridge Stub Callbacks (Child Process)
***************************************************************************** */

static void redis_on_mock_subscribe_child(const fio_pubsub_engine_s *eng,
                                          fio_str_info_s channel,
                                          fio_match_fn match) {
  /* do nothing, root process is notified about (un)subscriptions by facil.io */
  (void)eng;
  (void)channel;
  (void)match;
}

static void redis_on_publish_child(const fio_pubsub_engine_s *eng,
                                   fio_str_info_s channel, fio_str_info_s msg,
                                   uint8_t is_json) {
  /* attach engine data to channel (prepend) */
  fio_str_s tmp = FIO_STR_INIT;
  /* by using fio_str_s, short names are allocated on the stack */
  fio_str_info_s tmp_info = fio_str_resize(&tmp, channel.len + 8);
  fio_u2str64(tmp_info.data, (uintptr_t)eng);
  memcpy(tmp_info.data + 8, channel.data, channel.len);
  /* forward publication request to Root */
  fio_publish(.filter = -1, .channel = tmp_info, .message = msg,
              .engine = FIO_PUBSUB_ROOT, .is_json = is_json);
  fio_str_free(&tmp);
  (void)eng;
}

/* *****************************************************************************
Root Publication Handler
***************************************************************************** */

/* listens to filter -1 and publishes and messages */
static void redis_on_internal_publish(fio_msg_s *msg) {
  if (msg->channel.len < 8)
    return; /* internal error, unexpected data */
  void *en = (void *)(uintptr_t)fio_str2u64(msg->channel.data);
  if (en != msg->udata1)
    return; /* should be delivered by a different engine */
  /* step after the engine data */
  msg->channel.len -= 8;
  msg->channel.data += 8;
  /* forward to publishing */
  FIO_LOG_DEBUG("Forwarding to engine %p, on channel %s", msg->udata1,
                msg->channel.data);
  redis_on_publish_root(msg->udata1, msg->channel, msg->msg, msg->is_json);
}

/* *****************************************************************************
Sending commands using the Root connection
***************************************************************************** */

/* callback from the Redis reply */
static void redis_forward_reply(fio_pubsub_engine_s *e, FIOBJ reply,
                                void *udata) {
  uint8_t *data = udata;
  fio_pubsub_engine_s *engine = (fio_pubsub_engine_s *)(uintptr_t)fio_str2u64(data + 0);
  void *callback = (void *)(uintptr_t)fio_str2u64(data + 8);
  if (engine != e || !callback) {
    FIO_LOG_DEBUG("Redis reply not forwarded (callback: %p)", callback);
    return;
  }
  int32_t pid = (int32_t)fio_str2u32(data + 24);
  FIOBJ rp = fiobj_obj2json(reply, 0);
  fio_publish(.filter = (-10 - (int32_t)pid), .channel.data = (char *)data,
              .channel.len = 28, .message = fiobj_obj2cstr(rp), .is_json = 1);
  fiobj_free(rp);
}

/* listens to channel -2 for commands that need to be sent (only ROOT) */
static void redis_on_internal_cmd(fio_msg_s *msg) {
  // void*(void *)fio_str2u64(msg->msg.data);
  fio_pubsub_engine_s *engine =
      (fio_pubsub_engine_s *)(uintptr_t)fio_str2u64(msg->channel.data + 0);
  if (engine != msg->udata1) {
    return;
  }
  redis_commands_s *cmd = fio_malloc(sizeof(*cmd) + msg->msg.len + 1 + 28);
  FIO_ASSERT_ALLOC(cmd);
  *cmd = (redis_commands_s){.callback = redis_forward_reply,
                            .udata = (cmd->cmd + msg->msg.len + 1),
                            .cmd_len = msg->msg.len};
  memcpy(cmd->cmd, msg->msg.data, msg->msg.len);
  memcpy(cmd->cmd + msg->msg.len + 1, msg->channel.data, 28);
  redis_attach_cmd((redis_engine_s *)engine, cmd);
  // fprintf(stderr, " *** Attached CMD (%d) ***\n%s\n", getpid(), cmd->cmd);
}

/* Listens on filter `-10 -getpid()` for incoming reply data */
static void redis_on_internal_reply(fio_msg_s *msg) {
  fio_pubsub_engine_s *engine =
      (fio_pubsub_engine_s *)(uintptr_t)fio_str2u64(msg->channel.data + 0);
  if (engine != msg->udata1) {
    FIO_LOG_DEBUG("Redis reply not forwarded (engine mismatch: %p != %p)",
                  (void *)engine, msg->udata1);
    return;
  }
  FIOBJ reply;
  fiobj_json2obj(&reply, msg->msg.data, msg->msg.len);
  void (*callback)(fio_pubsub_engine_s *, FIOBJ, void *) = (void (*)(
      fio_pubsub_engine_s *, FIOBJ, void *))(uintptr_t)fio_str2u64(msg->channel.data + 8);
  void *udata = (void *)(uintptr_t)fio_str2u64(msg->channel.data + 16);
  callback(engine, reply, udata);
  fiobj_free(reply);
}

/* publishes a Redis command to Root's filter -2 */
intptr_t redis_engine_send(fio_pubsub_engine_s *engine, FIOBJ command,
                           void (*callback)(fio_pubsub_engine_s *e, FIOBJ reply,
                                            void *udata),
                           void *udata) {
  if ((uintptr_t)engine < 4) {
    FIO_LOG_WARNING("(redis send) trying to use one of the core engines");
    return -1;
  }
  // if(fio_is_master()) {
  // FIOBJ resp = fiobj2resp_tmp(fio_str_info_s obj1, FIOBJ obj2);
  // TODO...
  // } else {
  /* forward publication request to Root */
  fio_str_s tmp = FIO_STR_INIT;
  fio_str_info_s ti = fio_str_resize(&tmp, 28);
  /* combine metadata */
  fio_u2str64(ti.data + 0, (uintptr_t)engine);
  fio_u2str64(ti.data + 8, (uintptr_t)callback);
  fio_u2str64(ti.data + 16, (uintptr_t)udata);
  fio_u2str32(ti.data + 24, (uint32_t)getpid());
  FIOBJ cmd = fiobj2resp_tmp(command);
  fio_publish(.filter = -2, .channel = ti, .message = fiobj_obj2cstr(cmd),
              .engine = FIO_PUBSUB_ROOT, .is_json = 0);
  fio_str_free(&tmp);
  // }
  return 0;
}

/* *****************************************************************************
Redis Engine Creation
***************************************************************************** */

static void redis_on_facil_start(void *r_) {
  redis_engine_s *r = r_;
  r->flag = 1;
  if (!fio_is_valid(r->sub_data.uuid)) {
    defer_redis_connect(r, &r->sub_data);
  }
}
static void redis_on_facil_shutdown(void *r_) {
  redis_engine_s *r = r_;
  r->flag = 0;
}

static void redis_on_engine_fork(void *r_) {
  redis_engine_s *r = r_;
  r->flag = 0;
  r->lock = FIO_LOCK_INIT;
  fio_force_close(r->sub_data.uuid);
  r->sub_data.uuid = -1;
  fio_force_close(r->pub_data.uuid);
  r->pub_data.uuid = -1;
  while (fio_ls_embd_any(&r->queue)) {
    redis_commands_s *cmd =
        FIO_LS_EMBD_OBJ(redis_commands_s, node, fio_ls_embd_pop(&r->queue));
    fio_free(cmd);
  }
  r->en = (fio_pubsub_engine_s){
      .subscribe = redis_on_mock_subscribe_child,
      .unsubscribe = redis_on_mock_subscribe_child,
      .publish = redis_on_publish_child,
  };
  fio_unsubscribe(r->publication_forwarder);
  r->publication_forwarder = NULL;
  fio_unsubscribe(r->cmd_forwarder);
  r->cmd_forwarder = NULL;
  fio_unsubscribe(r->cmd_reply);
  r->cmd_reply =
      fio_subscribe(.filter = -10 - (int32_t)getpid(),
                    .on_message = redis_on_internal_reply, .udata1 = r);
}

fio_pubsub_engine_s *redis_engine_create
FIO_IGNORE_MACRO(struct redis_engine_create_args args) {
  if (getpid() != fio_parent_pid()) {
    FIO_LOG_FATAL("(redis) Redis engine initialization can only "
                  "be performed in the Root process.");
    kill(0, SIGINT);
    fio_stop();
    return NULL;
  }
  if (!args.address.len && args.address.data)
    args.address.len = strlen(args.address.data);
  if (!args.port.len && args.port.data)
    args.port.len = strlen(args.port.data);
  if (!args.auth.len && args.auth.data) {
    args.auth.len = strlen(args.auth.data);
  }

  if (!args.address.data || !args.address.len) {
    args.address = (fio_str_info_s){.len = 9, .data = (char *)"localhost"};
  }
  if (!args.port.data || !args.port.len) {
    args.port = (fio_str_info_s){.len = 4, .data = (char *)"6379"};
  }
  redis_engine_s *r =
      fio_malloc(sizeof(*r) + args.port.len + 1 + args.address.len + 1 +
                 args.auth.len + 1 + (REDIS_READ_BUFFER * 2));
  FIO_ASSERT_ALLOC(r);
  *r = (redis_engine_s){
      .en =
          {
              .subscribe = redis_on_subscribe_root,
              .unsubscribe = redis_on_unsubscribe_root,
              .publish = redis_on_publish_root,
          },
      .pub_data =
          {
              .protocol =
                  {
                      .on_data = redis_on_data,
                      .on_close = redis_on_close,
                      .on_shutdown = redis_on_shutdown,
                      .ping = redis_pub_ping,
                  },
              .uuid = -1,
              .on_message = resp_on_pub_message,
          },
      .sub_data =
          {
              .protocol =
                  {
                      .on_data = redis_on_data,
                      .on_close = redis_on_close,
                      .on_shutdown = redis_on_shutdown,
                      .ping = redis_sub_ping,
                  },
              .on_message = resp_on_sub_message,
              .uuid = -1,
          },
      .publication_forwarder =
          fio_subscribe(.filter = -1, .udata1 = r,
                        .on_message = redis_on_internal_publish),
      .cmd_forwarder = fio_subscribe(.filter = -2, .udata1 = r,
                                     .on_message = redis_on_internal_cmd),
      .cmd_reply =
          fio_subscribe(.filter = -10 - (uint32_t)getpid(), .udata1 = r,
                        .on_message = redis_on_internal_reply),
      .address = ((char *)(r + 1) + (REDIS_READ_BUFFER * 2)),
      .port =
          ((char *)(r + 1) + (REDIS_READ_BUFFER * 2) + args.address.len + 1),
      .auth = ((char *)(r + 1) + (REDIS_READ_BUFFER * 2) + args.address.len +
               args.port.len + 2),
      .auth_len = args.auth.len,
      .ref = 1,
      .queue = FIO_LS_INIT(r->queue),
      .lock = FIO_LOCK_INIT,
      .lock_connection = FIO_LOCK_INIT,
      .ping_int = args.ping_interval,
      .flag = 1,
  };
  memcpy(r->address, args.address.data, args.address.len);
  memcpy(r->port, args.port.data, args.port.len);
  if (args.auth.len)
    memcpy(r->auth, args.auth.data, args.auth.len);
  fio_pubsub_attach(&r->en);
  redis_on_facil_start(r);
  fio_state_callback_add(FIO_CALL_IN_CHILD, redis_on_engine_fork, r);
  fio_state_callback_add(FIO_CALL_ON_SHUTDOWN, redis_on_facil_shutdown, r);
  /* if restarting */
  fio_state_callback_add(FIO_CALL_PRE_START, redis_on_facil_start, r);

  FIO_LOG_DEBUG("Redis engine initialized %p", (void *)r);
  return &r->en;
}

/* *****************************************************************************
Redis Engine Destruction
***************************************************************************** */

void redis_engine_destroy(fio_pubsub_engine_s *engine) {
  redis_engine_s *r = (redis_engine_s *)engine;
  r->flag = 0;
  fio_pubsub_detach(&r->en);
  fio_state_callback_remove(FIO_CALL_IN_CHILD, redis_on_engine_fork, r);
  fio_state_callback_remove(FIO_CALL_ON_SHUTDOWN, redis_on_facil_shutdown, r);
  fio_state_callback_remove(FIO_CALL_PRE_START, redis_on_facil_start, r);
  FIO_LOG_DEBUG("Redis engine destroyed %p", (void *)r);
  redis_free(r);
}


text:
package org.openfact.core.security;

import org.openfact.core.models.PermissionType;
import org.openfact.core.models.RoleMembershipModel;
import org.openfact.core.models.RoleModel;

import java.io.Serializable;
import java.util.*;

public class IndexedPermissions implements Serializable {

    private Set<String> organizations = new HashSet<>();
    private Set<String> qualifiedPermissions = new HashSet<>();
    private Map<PermissionType, Set<String>> permissionToOrganizationsMap = new HashMap<>();

    public IndexedPermissions(Set<RoleMembershipModel> memberships) {
        index(memberships);
    }

    private void index(Set<RoleMembershipModel> memberships) {
        for (RoleMembershipModel membership : memberships) {
            RoleModel role = membership.getRole();
            for (PermissionType permission : role.getPermissions()) {
                String organizationId= membership.getOrganization().getId();
                String qualifiedPermission = createQualifiedPermissionKey(permission, organizationId);
                organizations.add(organizationId);
                qualifiedPermissions.add(qualifiedPermission);
                Set<String> orgs = permissionToOrganizationsMap.computeIfAbsent(permission, k -> new HashSet<>());
                orgs.add(organizationId);
            }
        }
    }

    /**
     * Returns true if the qualified permission exists.
     * @param permissionName the permission name
     * @param organizationId the org qualifier
     * @return true if has qualified permission
     */
    public boolean hasQualifiedPermission(PermissionType permissionName, String organizationId) {
        String key = createQualifiedPermissionKey(permissionName, organizationId);
        return qualifiedPermissions.contains(key);
    }

    public boolean isMemberOf(String organizationId) {
        return organizations.contains(organizationId);
    }

    public Set<String> getOrgQualifiers(PermissionType permissionName) {
        Set<String> orgs = permissionToOrganizationsMap.get(permissionName);
        if (orgs == null) {
            orgs = Collections.emptySet();
        }
        return Collections.unmodifiableSet(orgs);
    }

    protected String createQualifiedPermissionKey(PermissionType permissionName, String organizationId) {
        return permissionName.name() + "||" + organizationId;
    }

}

redpajama_set_name: RedPajamaGithub


text:
namespace bot_flash_cards_blip_sdk_csharp
{
    using System;
    using Take.Blip.Client.Console;

    class Program
    {
        static int Main(string[] args) => ConsoleRunner.RunAsync(args).GetAwaiter().GetResult();
    }
}


text:
package com.github.datasnap.tests;

import java.util.HashMap;
import java.util.Map;

import com.github.datasnap.controller.*;
import com.github.datasnap.events.IEvent;
import com.github.datasnap.events.GlobalPositionEvent;
import com.github.datasnap.propertiesandevents.*;
import com.github.datasnap.utils.Defaults;

public class GlobalPositionTest {

	public static void main(String[] args) {
		DataSnap dataSnap = new DataSnap();
		dataSnap.initialize();

		String[]organizationIds = {Defaults.ORGANISATION_ID};
		String[]projectIds = {Defaults.PROJECT_ID};

		String eventType = "global_position_sighting";
		Place place = new Place();
		place.setPlaceId("placeId");
		place.setLastPlace("placeid-3");
		User user = new User();
		PropId propId = new PropId();
		propId.setMobileDeviceIosIdfa("1a847de9f24b18eee3fac634b833b7887b32dea3");
		propId.setGlobalDistinctId("userid1234");
		Beacon beacon = new Beacon();
		String beaconid = "SHDG-28AHD";
		beacon.setIdentifier(beaconid);

		GlobalPosition globalPosition = new GlobalPosition();
		// globalPosition.setLocation("coordinates88.21337");
		globalPosition.setAccuracy("10");
		globalPosition.setAltitude("13.00");
		globalPosition.setCourse("77.34375");
		globalPosition.setSpeed("4.0");

		DeviceInfo deviceInfo = new DeviceInfo();
		deviceInfo.setCreated("2014-08-22 14:48:02 +0000");
		Device device = new Device();
		device.setUserAgent("Mozilla/5.0 (Linux; U; Android 4.0.3; ko-kr; LG-L160L Build/IML74K) AppleWebkit/534.30 (KHTML, like Gecko) Version/4.0 Mobile Safari/534.30");
		device.setIpAddress("127.1.1.1");
		device.setPlatform("ios");
		device.setOsVersion("7.0");
		device.setModel("iPhone5");
		device.setManufacturer("Apple");
		device.setName("hashed device name");
		device.setVendorId("63A7355F-5AF2-4E20-BE55-C3E80D0305B1");
		device.setCarrierName("Verizon");
		device.setCountryCode("1");
		device.setNetworkCode("327");
		deviceInfo.setDevice(device);

		Beacon beacon2 = new Beacon();
		String beaconid2 = "SHDG-test";
		beacon2.setIdentifier(beaconid2);

		Map<String, Object> additionalProperties = new HashMap<String, Object>();
		additionalProperties.put("beacontest", beacon2);
		additionalProperties.put("beacontest2", beacon2);

		IEvent event = new GlobalPositionEvent(eventType, organizationIds, projectIds, user,
				globalPosition, deviceInfo, additionalProperties);
		dataSnap.addEvent(event);
	}

}


text:
'use strict';

/*
 * Make sure two EPOLLONESHOT events can be handled.
 *
 * This test expects a newline as input on stdin.
 */
const Epoll = require('../').Epoll;
const util = require('./util');

const stdin = 0; // fd for stdin

let eventCount = 0;

const epoll = new Epoll((err, fd, events) => {
  eventCount += 1;

  if (eventCount === 1) {
    setTimeout(_ => {
      epoll.modify(fd, Epoll.EPOLLIN | Epoll.EPOLLONESHOT);
    }, 100);
  } else if (eventCount === 2) {
    setTimeout(_ => {
      util.read(fd); // read stdin (the newline)
      epoll.remove(fd).close();
    }, 100);
  } else {
    console.log('*** Error: unexpected event');
  }
});

epoll.add(stdin, Epoll.EPOLLIN | Epoll.EPOLLONESHOT);


text:
package com.googlecode.mp4parser.stuff;

import org.mp4parser.IsoFile;
import org.mp4parser.boxes.iso14496.part12.HandlerBox;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.List;

public class VideoAudioChecker {

    public static void main(String[] args) throws IOException {
        String f = VideoAudioChecker.class.getProtectionDomain().getCodeSource().getLocation().getFile() + "/count-video.mp4";
        IsoFile isoFile = new IsoFile(new FileInputStream(f).getChannel());
        System.err.println(getType(isoFile));
    }

    public static TYPE getType(IsoFile isoFile) {

        List<HandlerBox> handlerBoxes =
                isoFile.getBoxes(HandlerBox.class, true);
        for (HandlerBox handlerBox : handlerBoxes) {
            if ("vide".equals(handlerBox.getHandlerType())) {
                return TYPE.VIDEO;
            } else if ("soun".equals(handlerBox.getHandlerType())) {
                return TYPE.AUDIO;
            } else {
                System.err.println("unknown");
            }

        }
        return TYPE.AUDIO;
    }

    private enum TYPE {
        AUDIO,
        VIDEO
    }

}


text:
package org.apache.hadoop.hbase.snapshot;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.IOException;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.master.HMaster;
import org.apache.hadoop.hbase.master.snapshot.SnapshotManager;
import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.SnapshotDescription;
import org.apache.hadoop.hbase.regionserver.ConstantSizeRegionSplitPolicy;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.testclassification.RegionServerTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.FSUtils;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;

/**
 * Test creating/using/deleting snapshots from the client
 * <p>
 * This is an end-to-end test for the snapshot utility
 *
 * TODO This is essentially a clone of TestSnapshotFromClient.  This is worth refactoring this
 * because there will be a few more flavors of snapshots that need to run these tests.
 */
@Category({RegionServerTests.class, LargeTests.class})
public class TestFlushSnapshotFromClient {
  private static final Log LOG = LogFactory.getLog(TestFlushSnapshotFromClient.class);
  private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();
  private static final int NUM_RS = 2;
  private static final byte[] TEST_FAM = Bytes.toBytes("fam");
  private static final TableName TABLE_NAME = TableName.valueOf("test");
  private final int DEFAULT_NUM_ROWS = 100;

  /**
   * Setup the config for the cluster
   * @throws Exception on failure
   */
  @BeforeClass
  public static void setupCluster() throws Exception {
    // Uncomment the following lines if more verbosity is needed for
    // debugging (see HBASE-12285 for details).
    //((Log4JLogger)RpcServer.LOG).getLogger().setLevel(Level.ALL);
    //((Log4JLogger)AbstractRpcClient.LOG).getLogger().setLevel(Level.ALL);
    //((Log4JLogger)ScannerCallable.LOG).getLogger().setLevel(Level.ALL);
    setupConf(UTIL.getConfiguration());
    UTIL.startMiniCluster(NUM_RS);
  }

  private static void setupConf(Configuration conf) {
    // disable the ui
    conf.setInt("hbase.regionsever.info.port", -1);
    // change the flush size to a small amount, regulating number of store files
    conf.setInt("hbase.hregion.memstore.flush.size", 25000);
    // so make sure we get a compaction when doing a load, but keep around some
    // files in the store
    conf.setInt("hbase.hstore.compaction.min", 10);
    conf.setInt("hbase.hstore.compactionThreshold", 10);
    // block writes if we get to 12 store files
    conf.setInt("hbase.hstore.blockingStoreFiles", 12);
    // Enable snapshot
    conf.setBoolean(SnapshotManager.HBASE_SNAPSHOT_ENABLED, true);
    conf.set(HConstants.HBASE_REGION_SPLIT_POLICY_KEY,
      ConstantSizeRegionSplitPolicy.class.getName());
  }

  @Before
  public void setup() throws Exception {
    SnapshotTestingUtils.createTable(UTIL, TABLE_NAME, TEST_FAM);
  }

  @After
  public void tearDown() throws Exception {
    UTIL.deleteTable(TABLE_NAME);

    SnapshotTestingUtils.deleteAllSnapshots(UTIL.getHBaseAdmin());
    SnapshotTestingUtils.deleteArchiveDirectory(UTIL);
  }

  @AfterClass
  public static void cleanupTest() throws Exception {
    try {
      UTIL.shutdownMiniCluster();
    } catch (Exception e) {
      LOG.warn("failure shutting down cluster", e);
    }
  }

  /**
   * Test simple flush snapshotting a table that is online
   * @throws Exception
   */
  @Test (timeout=300000)
  public void testFlushTableSnapshot() throws Exception {
    Admin admin = UTIL.getHBaseAdmin();
    // make sure we don't fail on listing snapshots
    SnapshotTestingUtils.assertNoSnapshots(admin);

    // put some stuff in the table
    SnapshotTestingUtils.loadData(UTIL, TABLE_NAME, DEFAULT_NUM_ROWS, TEST_FAM);

    LOG.debug("FS state before snapshot:");
    FSUtils.logFileSystemState(UTIL.getTestFileSystem(),
        FSUtils.getRootDir(UTIL.getConfiguration()), LOG);

    // take a snapshot of the enabled table
    String snapshotString = "offlineTableSnapshot";
    byte[] snapshot = Bytes.toBytes(snapshotString);
    admin.snapshot(snapshotString, TABLE_NAME, SnapshotDescription.Type.FLUSH);
    LOG.debug("Snapshot completed.");

    // make sure we have the snapshot
    List<SnapshotDescription> snapshots = SnapshotTestingUtils.assertOneSnapshotThatMatches(admin,
      snapshot, TABLE_NAME);

    // make sure its a valid snapshot
    FileSystem fs = UTIL.getHBaseCluster().getMaster().getMasterFileSystem().getFileSystem();
    Path rootDir = UTIL.getHBaseCluster().getMaster().getMasterFileSystem().getRootDir();
    LOG.debug("FS state after snapshot:");
    FSUtils.logFileSystemState(UTIL.getTestFileSystem(),
        FSUtils.getRootDir(UTIL.getConfiguration()), LOG);

    SnapshotTestingUtils.confirmSnapshotValid(snapshots.get(0), TABLE_NAME, TEST_FAM, rootDir,
        admin, fs);
  }

   /**
   * Test snapshotting a table that is online without flushing
   * @throws Exception
   */
  @Test(timeout=30000)
  public void testSkipFlushTableSnapshot() throws Exception {
    Admin admin = UTIL.getHBaseAdmin();
    // make sure we don't fail on listing snapshots
    SnapshotTestingUtils.assertNoSnapshots(admin);

    // put some stuff in the table
    Table table = UTIL.getConnection().getTable(TABLE_NAME);
    UTIL.loadTable(table, TEST_FAM);

    LOG.debug("FS state before snapshot:");
    FSUtils.logFileSystemState(UTIL.getTestFileSystem(),
        FSUtils.getRootDir(UTIL.getConfiguration()), LOG);

    // take a snapshot of the enabled table
    String snapshotString = "skipFlushTableSnapshot";
    byte[] snapshot = Bytes.toBytes(snapshotString);
    admin.snapshot(snapshotString, TABLE_NAME, SnapshotDescription.Type.SKIPFLUSH);
    LOG.debug("Snapshot completed.");

    // make sure we have the snapshot
    List<SnapshotDescription> snapshots = SnapshotTestingUtils.assertOneSnapshotThatMatches(admin,
        snapshot, TABLE_NAME);

    // make sure its a valid snapshot
    FileSystem fs = UTIL.getHBaseCluster().getMaster().getMasterFileSystem().getFileSystem();
    Path rootDir = UTIL.getHBaseCluster().getMaster().getMasterFileSystem().getRootDir();
    LOG.debug("FS state after snapshot:");
    FSUtils.logFileSystemState(UTIL.getTestFileSystem(),
        FSUtils.getRootDir(UTIL.getConfiguration()), LOG);

    SnapshotTestingUtils.confirmSnapshotValid(snapshots.get(0), TABLE_NAME, TEST_FAM, rootDir,
        admin, fs);

    admin.deleteSnapshot(snapshot);
    snapshots = admin.listSnapshots();
    SnapshotTestingUtils.assertNoSnapshots(admin);
  }

  /**
   * Test simple flush snapshotting a table that is online
   * @throws Exception
   */
  @Test (timeout=300000)
  public void testFlushTableSnapshotWithProcedure() throws Exception {
    Admin admin = UTIL.getHBaseAdmin();
    // make sure we don't fail on listing snapshots
    SnapshotTestingUtils.assertNoSnapshots(admin);

    // put some stuff in the table
    SnapshotTestingUtils.loadData(UTIL, TABLE_NAME, DEFAULT_NUM_ROWS, TEST_FAM);

    LOG.debug("FS state before snapshot:");
    FSUtils.logFileSystemState(UTIL.getTestFileSystem(),
        FSUtils.getRootDir(UTIL.getConfiguration()), LOG);

    // take a snapshot of the enabled table
    String snapshotString = "offlineTableSnapshot";
    byte[] snapshot = Bytes.toBytes(snapshotString);
    Map<String, String> props = new HashMap<String, String>();
    props.put("table", TABLE_NAME.getNameAsString());
    admin.execProcedure(SnapshotManager.ONLINE_SNAPSHOT_CONTROLLER_DESCRIPTION,
        snapshotString, props);

    LOG.debug("Snapshot completed.");

    // make sure we have the snapshot
    List<SnapshotDescription> snapshots = SnapshotTestingUtils.assertOneSnapshotThatMatches(admin,
      snapshot, TABLE_NAME);

    // make sure its a valid snapshot
    FileSystem fs = UTIL.getHBaseCluster().getMaster().getMasterFileSystem().getFileSystem();
    Path rootDir = UTIL.getHBaseCluster().getMaster().getMasterFileSystem().getRootDir();
    LOG.debug("FS state after snapshot:");
    FSUtils.logFileSystemState(UTIL.getTestFileSystem(),
        FSUtils.getRootDir(UTIL.getConfiguration()), LOG);

    SnapshotTestingUtils.confirmSnapshotValid(snapshots.get(0), TABLE_NAME, TEST_FAM, rootDir,
        admin, fs);
  }

  @Test (timeout=300000)
  public void testSnapshotFailsOnNonExistantTable() throws Exception {
    Admin admin = UTIL.getHBaseAdmin();
    // make sure we don't fail on listing snapshots
    SnapshotTestingUtils.assertNoSnapshots(admin);
    TableName tableName = TableName.valueOf("_not_a_table");

    // make sure the table doesn't exist
    boolean fail = false;
    do {
    try {
      admin.getTableDescriptor(tableName);
      fail = true;
      LOG.error("Table:" + tableName + " already exists, checking a new name");
      tableName = TableName.valueOf(tableName+"!");
    } catch (TableNotFoundException e) {
      fail = false;
      }
    } while (fail);

    // snapshot the non-existant table
    try {
      admin.snapshot("fail", tableName, SnapshotDescription.Type.FLUSH);
      fail("Snapshot succeeded even though there is not table.");
    } catch (SnapshotCreationException e) {
      LOG.info("Correctly failed to snapshot a non-existant table:" + e.getMessage());
    }
  }

  @Test(timeout = 300000)
  public void testAsyncFlushSnapshot() throws Exception {
    Admin admin = UTIL.getHBaseAdmin();
    SnapshotDescription snapshot = SnapshotDescription.newBuilder().setName("asyncSnapshot")
        .setTable(TABLE_NAME.getNameAsString())
        .setType(SnapshotDescription.Type.FLUSH)
        .build();

    // take the snapshot async
    admin.takeSnapshotAsync(snapshot);

    // constantly loop, looking for the snapshot to complete
    HMaster master = UTIL.getMiniHBaseCluster().getMaster();
    SnapshotTestingUtils.waitForSnapshotToComplete(master, snapshot, 200);
    LOG.info(" === Async Snapshot Completed ===");
    FSUtils.logFileSystemState(UTIL.getTestFileSystem(),
      FSUtils.getRootDir(UTIL.getConfiguration()), LOG);
    // make sure we get the snapshot
    SnapshotTestingUtils.assertOneSnapshotThatMatches(admin, snapshot);
  }

  @Test (timeout=300000)
  public void testSnapshotStateAfterMerge() throws Exception {
    int numRows = DEFAULT_NUM_ROWS;
    Admin admin = UTIL.getHBaseAdmin();
    // make sure we don't fail on listing snapshots
    SnapshotTestingUtils.assertNoSnapshots(admin);
    // load the table so we have some data
    SnapshotTestingUtils.loadData(UTIL, TABLE_NAME, numRows, TEST_FAM);

    // Take a snapshot
    String snapshotBeforeMergeName = "snapshotBeforeMerge";
    admin.snapshot(snapshotBeforeMergeName, TABLE_NAME, SnapshotDescription.Type.FLUSH);

    // Clone the table
    TableName cloneBeforeMergeName = TableName.valueOf("cloneBeforeMerge");
    admin.cloneSnapshot(snapshotBeforeMergeName, cloneBeforeMergeName);
    SnapshotTestingUtils.waitForTableToBeOnline(UTIL, cloneBeforeMergeName);

    // Merge two regions
    List<HRegionInfo> regions = admin.getTableRegions(TABLE_NAME);
    Collections.sort(regions, new Comparator<HRegionInfo>() {
      public int compare(HRegionInfo r1, HRegionInfo r2) {
        return Bytes.compareTo(r1.getStartKey(), r2.getStartKey());
      }
    });

    int numRegions = admin.getTableRegions(TABLE_NAME).size();
    int numRegionsAfterMerge = numRegions - 2;
    admin.mergeRegions(regions.get(1).getEncodedNameAsBytes(),
        regions.get(2).getEncodedNameAsBytes(), true);
    admin.mergeRegions(regions.get(5).getEncodedNameAsBytes(),
        regions.get(6).getEncodedNameAsBytes(), true);

    // Verify that there's one region less
    waitRegionsAfterMerge(numRegionsAfterMerge);
    assertEquals(numRegionsAfterMerge, admin.getTableRegions(TABLE_NAME).size());

    // Clone the table
    TableName cloneAfterMergeName = TableName.valueOf("cloneAfterMerge");
    admin.cloneSnapshot(snapshotBeforeMergeName, cloneAfterMergeName);
    SnapshotTestingUtils.waitForTableToBeOnline(UTIL, cloneAfterMergeName);

    SnapshotTestingUtils.verifyRowCount(UTIL, TABLE_NAME, numRows);
    SnapshotTestingUtils.verifyRowCount(UTIL, cloneBeforeMergeName, numRows);
    SnapshotTestingUtils.verifyRowCount(UTIL, cloneAfterMergeName, numRows);

    // test that we can delete the snapshot
    UTIL.deleteTable(cloneAfterMergeName);
    UTIL.deleteTable(cloneBeforeMergeName);
  }

  @Test (timeout=300000)
  public void testTakeSnapshotAfterMerge() throws Exception {
    int numRows = DEFAULT_NUM_ROWS;
    Admin admin = UTIL.getHBaseAdmin();
    // make sure we don't fail on listing snapshots
    SnapshotTestingUtils.assertNoSnapshots(admin);
    // load the table so we have some data
    SnapshotTestingUtils.loadData(UTIL, TABLE_NAME, numRows, TEST_FAM);

    // Merge two regions
    List<HRegionInfo> regions = admin.getTableRegions(TABLE_NAME);
    Collections.sort(regions, new Comparator<HRegionInfo>() {
      public int compare(HRegionInfo r1, HRegionInfo r2) {
        return Bytes.compareTo(r1.getStartKey(), r2.getStartKey());
      }
    });

    int numRegions = admin.getTableRegions(TABLE_NAME).size();
    int numRegionsAfterMerge = numRegions - 2;
    admin.mergeRegions(regions.get(1).getEncodedNameAsBytes(),
        regions.get(2).getEncodedNameAsBytes(), true);
    admin.mergeRegions(regions.get(5).getEncodedNameAsBytes(),
        regions.get(6).getEncodedNameAsBytes(), true);

    waitRegionsAfterMerge(numRegionsAfterMerge);
    assertEquals(numRegionsAfterMerge, admin.getTableRegions(TABLE_NAME).size());

    // Take a snapshot
    String snapshotName = "snapshotAfterMerge";
    SnapshotTestingUtils.snapshot(admin, snapshotName, TABLE_NAME.getNameAsString(),
      SnapshotDescription.Type.FLUSH, 3);

    // Clone the table
    TableName cloneName = TableName.valueOf("cloneMerge");
    admin.cloneSnapshot(snapshotName, cloneName);
    SnapshotTestingUtils.waitForTableToBeOnline(UTIL, cloneName);

    SnapshotTestingUtils.verifyRowCount(UTIL, TABLE_NAME, numRows);
    SnapshotTestingUtils.verifyRowCount(UTIL, cloneName, numRows);

    // test that we can delete the snapshot
    UTIL.deleteTable(cloneName);
  }

  /**
   * Basic end-to-end test of simple-flush-based snapshots
   */
  @Test (timeout=300000)
  public void testFlushCreateListDestroy() throws Exception {
    LOG.debug("------- Starting Snapshot test -------------");
    Admin admin = UTIL.getHBaseAdmin();
    // make sure we don't fail on listing snapshots
    SnapshotTestingUtils.assertNoSnapshots(admin);
    // load the table so we have some data
    SnapshotTestingUtils.loadData(UTIL, TABLE_NAME, DEFAULT_NUM_ROWS, TEST_FAM);

    String snapshotName = "flushSnapshotCreateListDestroy";
    FileSystem fs = UTIL.getHBaseCluster().getMaster().getMasterFileSystem().getFileSystem();
    Path rootDir = UTIL.getHBaseCluster().getMaster().getMasterFileSystem().getRootDir();
    SnapshotTestingUtils.createSnapshotAndValidate(admin, TABLE_NAME, Bytes.toString(TEST_FAM),
      snapshotName, rootDir, fs, true);
  }

  /**
   * Demonstrate that we reject snapshot requests if there is a snapshot already running on the
   * same table currently running and that concurrent snapshots on different tables can both
   * succeed concurretly.
   */
  @Test(timeout=300000)
  public void testConcurrentSnapshottingAttempts() throws IOException, InterruptedException {
    final TableName TABLE2_NAME = TableName.valueOf(TABLE_NAME + "2");

    int ssNum = 20;
    Admin admin = UTIL.getHBaseAdmin();
    // make sure we don't fail on listing snapshots
    SnapshotTestingUtils.assertNoSnapshots(admin);
    // create second testing table
    SnapshotTestingUtils.createTable(UTIL, TABLE2_NAME, TEST_FAM);
    // load the table so we have some data
    SnapshotTestingUtils.loadData(UTIL, TABLE_NAME, DEFAULT_NUM_ROWS, TEST_FAM);
    SnapshotTestingUtils.loadData(UTIL, TABLE2_NAME, DEFAULT_NUM_ROWS, TEST_FAM);

    final CountDownLatch toBeSubmitted = new CountDownLatch(ssNum);
    // We'll have one of these per thread
    class SSRunnable implements Runnable {
      SnapshotDescription ss;
      SSRunnable(SnapshotDescription ss) {
        this.ss = ss;
      }

      @Override
      public void run() {
        try {
          Admin admin = UTIL.getHBaseAdmin();
          LOG.info("Submitting snapshot request: " + ClientSnapshotDescriptionUtils.toString(ss));
          admin.takeSnapshotAsync(ss);
        } catch (Exception e) {
          LOG.info("Exception during snapshot request: " + ClientSnapshotDescriptionUtils.toString(
              ss)
              + ".  This is ok, we expect some", e);
        }
        LOG.info("Submitted snapshot request: " + ClientSnapshotDescriptionUtils.toString(ss));
        toBeSubmitted.countDown();
      }
    };

    // build descriptions
    SnapshotDescription[] descs = new SnapshotDescription[ssNum];
    for (int i = 0; i < ssNum; i++) {
      SnapshotDescription.Builder builder = SnapshotDescription.newBuilder();
      builder.setTable(((i % 2) == 0 ? TABLE_NAME : TABLE2_NAME).getNameAsString());
      builder.setName("ss"+i);
      builder.setType(SnapshotDescription.Type.FLUSH);
      descs[i] = builder.build();
    }

    // kick each off its own thread
    for (int i=0 ; i < ssNum; i++) {
      new Thread(new SSRunnable(descs[i])).start();
    }

    // wait until all have been submitted
    toBeSubmitted.await();

    // loop until all are done.
    while (true) {
      int doneCount = 0;
      for (SnapshotDescription ss : descs) {
        try {
          if (admin.isSnapshotFinished(ss)) {
            doneCount++;
          }
        } catch (Exception e) {
          LOG.warn("Got an exception when checking for snapshot " + ss.getName(), e);
          doneCount++;
        }
      }
      if (doneCount == descs.length) {
        break;
      }
      Thread.sleep(100);
    }

    // dump for debugging
    logFSTree(FSUtils.getRootDir(UTIL.getConfiguration()));

    List<SnapshotDescription> taken = admin.listSnapshots();
    int takenSize = taken.size();
    LOG.info("Taken " + takenSize + " snapshots:  " + taken);
    assertTrue("We expect at least 1 request to be rejected because of we concurrently" +
        " issued many requests", takenSize < ssNum && takenSize > 0);

    // Verify that there's at least one snapshot per table
    int t1SnapshotsCount = 0;
    int t2SnapshotsCount = 0;
    for (SnapshotDescription ss : taken) {
      if (TableName.valueOf(ss.getTable()).equals(TABLE_NAME)) {
        t1SnapshotsCount++;
      } else if (TableName.valueOf(ss.getTable()).equals(TABLE2_NAME)) {
        t2SnapshotsCount++;
      }
    }
    assertTrue("We expect at least 1 snapshot of table1 ", t1SnapshotsCount > 0);
    assertTrue("We expect at least 1 snapshot of table2 ", t2SnapshotsCount > 0);

    UTIL.deleteTable(TABLE2_NAME);
  }

  private void logFSTree(Path root) throws IOException {
    FSUtils.logFileSystemState(UTIL.getDFSCluster().getFileSystem(), root, LOG);
  }

  private void waitRegionsAfterMerge(final long numRegionsAfterMerge)
      throws IOException, InterruptedException {
    Admin admin = UTIL.getHBaseAdmin();
    // Verify that there's one region less
    long startTime = System.currentTimeMillis();
    while (admin.getTableRegions(TABLE_NAME).size() != numRegionsAfterMerge) {
      // This may be flaky... if after 15sec the merge is not complete give up
      // it will fail in the assertEquals(numRegionsAfterMerge).
      if ((System.currentTimeMillis() - startTime) > 15000)
        break;
      Thread.sleep(100);
    }
    SnapshotTestingUtils.waitForTableToBeOnline(UTIL, TABLE_NAME);
  }
}


text:
/**
 *
 */
package me.learn.personal.month7;

import java.util.Stack;

/**
 * Title 591 :
 *
 * Date : Mar 6, 2021
 *
 * @author bramanarayan
 *
 */
public class TagValidator {

	public boolean isValid(String input) {

		if (!input.startsWith("<"))
			return false;

		if (input.length() < 2 || !(input.charAt(1) >= 'A' && input.charAt(1) <= 'Z'))
			return false;

		Stack<String> stack = new Stack<>();
		String currentStr = "";

		// Iterator
		for (int i = 0; i < input.length(); i++) {

			currentStr += input.charAt(i);

			if (currentStr.startsWith("<![CDATA[")) {  // CDATA TAG
				if (currentStr.endsWith("]]>")) {
					currentStr = "";
				}
			} else if (currentStr.startsWith("</")) { // END TAG

				if (currentStr.endsWith(">")) {
					String tag = currentStr.substring(2, currentStr.length() - 1);

					if (stack.isEmpty() || !stack.peek().equals("<" + tag + ">")) // if stack top do not match START tag
						return false;

					stack.pop();
					currentStr = "";

					if (i < input.length() - 1 && stack.isEmpty())
						return false;
				}
			} else if (currentStr.startsWith("<")) {  // START TAG

				if (currentStr.endsWith(">")) {
					if (!valid_tag(currentStr))
						return false;
					stack.push(currentStr);
					currentStr = "";
				}
			} else {
				currentStr = ""; // TEXT WITHIN TAGS - so ignore, I dont care
			}
		}
		return stack.isEmpty() && currentStr.length() == 0;
	}

	// check the conditions if its a valid tag or not
	boolean valid_tag(String tag) {
		if (!tag.startsWith("<") || !tag.endsWith(">"))
			return false;
		if (tag.length() < 3 || tag.length() > 11)
			return false;
		for (int i = 1; i < tag.length() - 1; i++) {
			char c = tag.charAt(i);
			if (!Character.isUpperCase(c))
				return false;
		}
		return true;
	}

}


text:
"""
Verifies basic OAUTH functionality in AMO.

Sample request_token query:
    /en-US/firefox/oauth/request_token/?
        oauth_consumer_key=GYKEp7m5fJpj9j8Vjz&
        oauth_nonce=A7A79B47-B571-4D70-AA6C-592A0555E94B&
        oauth_signature_method=HMAC-SHA1&
        oauth_timestamp=1282950712&
        oauth_version=1.0

With headers:

    Authorization: OAuth realm="",
    oauth_consumer_key="GYKEp7m5fJpj9j8Vjz",
    oauth_signature_method="HMAC-SHA1",
    oauth_signature="JBCA4ah%2FOQC0lLWV8aChGAC+15s%3D",
    oauth_timestamp="1282950995",
    oauth_nonce="1008F707-37E6-4ABF-8322-C6B658771D88",
    oauth_version="1.0"
"""
import json
import os
import time
import urllib
import urlparse

from django import forms
from django.conf import settings
from django.core import mail
from django.test.client import (encode_multipart, Client, FakePayload,
                                BOUNDARY, MULTIPART_CONTENT, RequestFactory)

import oauth2 as oauth
from mock import Mock, patch
from nose.tools import eq_
from piston.models import Consumer

from olympia import amo
from olympia.amo.helpers import absolutify
from olympia.amo.tests import TestCase
from olympia.amo.urlresolvers import reverse
from olympia.api.authentication import AMOOAuthAuthentication
from olympia.addons.models import Addon, AddonUser
from olympia.devhub.models import ActivityLog, SubmitStep
from olympia.files.models import File
from olympia.translations.models import Translation
from olympia.users.models import UserProfile
from olympia.versions.models import AppVersion, Version

def _get_args(consumer, token=None, callback=False, verifier=None):
    d = dict(oauth_consumer_key=consumer.key,
             oauth_nonce=oauth.generate_nonce(),
             oauth_signature_method='HMAC-SHA1',
             oauth_timestamp=int(time.time()),
             oauth_version='1.0')

    if callback:
        d['oauth_callback'] = 'http://testserver/foo'

    if verifier:
        d['oauth_verifier'] = verifier

    return d

def get_absolute_url(url):
    if isinstance(url, tuple):
        url = reverse(url[0], args=url[1:])
    else:
        url = reverse(url)

    return 'http://%s%s' % ('api', url)

def data_keys(d):
    # Form keys and values MUST be part of the signature.
    # File keys MUST be part of the signature.
    # But file values MUST NOT be included as part of the signature.
    return dict([k, '' if isinstance(v, file) else v] for k, v in d.items())

class OAuthClient(Client):
    """OauthClient can make magically signed requests."""
    signature_method = oauth.SignatureMethod_HMAC_SHA1()

    def get(self, url, consumer=None, token=None, callback=False,
            verifier=None, params=None):
        url = get_absolute_url(url)
        if params:
            url = '%s?%s' % (url, urllib.urlencode(params))
        req = oauth.Request(method='GET', url=url,
                            parameters=_get_args(consumer, callback=callback,
                                                 verifier=verifier))
        req.sign_request(self.signature_method, consumer, token)
        return super(OAuthClient, self).get(
            req.to_url(), HTTP_HOST='api', HTTP_AUTHORIZATION='OAuth realm=""',
            **req)

    def delete(self, url, consumer=None, token=None, callback=False,
               verifier=None):
        url = get_absolute_url(url)
        req = oauth.Request(method='DELETE', url=url,
                            parameters=_get_args(consumer, callback=callback,
                                                 verifier=verifier))
        req.sign_request(self.signature_method, consumer, token)
        return super(OAuthClient, self).delete(
            req.to_url(), HTTP_HOST='api', HTTP_AUTHORIZATION='OAuth realm=""',
            **req)

    def post(self, url, consumer=None, token=None, callback=False,
             verifier=None, data={}):
        url = get_absolute_url(url)
        params = _get_args(consumer, callback=callback, verifier=verifier)
        params.update(data_keys(data))
        req = oauth.Request(method='POST', url=url, parameters=params)
        req.sign_request(self.signature_method, consumer, token)
        return super(OAuthClient, self).post(
            req.to_url(), HTTP_HOST='api', HTTP_AUTHORIZATION='OAuth realm=""',
            data=data, headers=req.to_header())

    def put(self, url, consumer=None, token=None, callback=False,
            verifier=None, data={}, content_type=MULTIPART_CONTENT, **kwargs):
        """
        Send a resource to the server using PUT.
        """
        # If data has come from JSON remove unicode keys.
        data = dict([(str(k), v) for k, v in data.items()])
        url = get_absolute_url(url)
        params = _get_args(consumer, callback=callback, verifier=verifier)
        params.update(data_keys(data))

        req = oauth.Request(method='PUT', url=url, parameters=params)
        req.sign_request(self.signature_method, consumer, token)
        post_data = encode_multipart(BOUNDARY, data)

        parsed = urlparse.urlparse(url)
        query_string = urllib.urlencode(req, doseq=True)
        r = {
            'CONTENT_LENGTH': len(post_data),
            'CONTENT_TYPE': content_type,
            'PATH_INFO': urllib.unquote(parsed[2]),
            'QUERY_STRING': query_string,
            'REQUEST_METHOD': 'PUT',
            'wsgi.input': FakePayload(post_data),
            'HTTP_HOST': 'api',
            'HTTP_AUTHORIZATION': 'OAuth realm=""',
        }
        r.update(req)

        response = self.request(**r)
        return response

oclient = OAuthClient()
token_keys = ('oauth_token_secret', 'oauth_token',)

def get_token_from_response(response):
    data = urlparse.parse_qs(response.content)
    for key in token_keys:
        assert key in data.keys(), '%s not in %s' % (key, data.keys())

    return oauth.Token(key=data['oauth_token'][0],
                       secret=data['oauth_token_secret'][0])

def get_request_token(consumer, callback=False):
    r = oclient.get('oauth.request_token', consumer, callback=callback)
    return get_token_from_response(r)

def get_access_token(consumer, token, authorize=True, verifier=None):
    r = oclient.get('oauth.access_token', consumer, token, verifier=verifier)

    if authorize:
        return get_token_from_response(r)
    else:
        eq_(r.status_code, 401)

class BaseOAuth(TestCase):
    fixtures = ['base/users', 'base/appversion', 'base/licenses']

    def setUp(self):
        super(BaseOAuth, self).setUp()
        self.editor = UserProfile.objects.get(email='editor@mozilla.com')
        self.admin = UserProfile.objects.get(email='admin@mozilla.com')
        consumers = []
        for status in ('accepted', 'pending', 'canceled', ):
            c = Consumer(name='a', status=status, user=self.editor)
            c.generate_random_codes()
            c.save()
            consumers.append(c)
        self.accepted_consumer = consumers[0]
        self.pending_consumer = consumers[1]
        self.canceled_consumer = consumers[2]
        self.token = None

class TestBaseOAuth(BaseOAuth):

    def test_accepted(self):
        self.assertRaises(AssertionError, get_request_token,
                          self.accepted_consumer)

    def test_accepted_callback(self):
        get_request_token(self.accepted_consumer, callback=True)

    def test_request_token_pending(self):
        get_request_token(self.pending_consumer, callback=True)

    def test_request_token_cancelled(self):
        get_request_token(self.canceled_consumer, callback=True)

    def test_request_token_fake(self):
        """Try with a phony consumer key"""
        c = Mock()
        c.key = 'yer'
        c.secret = 'mom'
        r = oclient.get('oauth.request_token', c, callback=True)
        eq_(r.content, 'Invalid Consumer.')

    @patch('piston.authentication.oauth.OAuthAuthentication.is_authenticated')
    def _test_auth(self, pk, is_authenticated, two_legged=True):
        request = RequestFactory().get('/en-US/firefox/2/api/2/user/',
                                       data={'authenticate_as': pk})
        request.user = None

        def alter_request(*args, **kw):
            request.user = self.admin
            return True
        is_authenticated.return_value = True
        is_authenticated.side_effect = alter_request

        auth = AMOOAuthAuthentication()
        auth.two_legged = two_legged
        auth.is_authenticated(request)
        return request

    def test_login_nonexistant(self):
        eq_(self.admin, self._test_auth(9999).user)

    def test_login_deleted(self):
        # If _test_auth returns self.admin, that means the user was
        # not altered to the user set in authenticate_as.
        self.editor.update(deleted=True)
        pk = self.editor.pk
        eq_(self.admin, self._test_auth(pk).user)

    def test_login_unconfirmed(self):
        self.editor.update(confirmationcode='something')
        pk = self.editor.pk
        eq_(self.admin, self._test_auth(pk).user)

    def test_login_works(self):
        pk = self.editor.pk
        eq_(self.editor, self._test_auth(pk).user)

    def test_login_three_legged(self):
        pk = self.editor.pk
        eq_(self.admin, self._test_auth(pk, two_legged=False).user)

class TestUser(BaseOAuth):

    def test_user(self):
        r = oclient.get('api.user', self.accepted_consumer, self.token)
        eq_(json.loads(r.content)['email'], 'editor@mozilla.com')

    def test_user_lookup(self):
        partner = UserProfile.objects.get(email='partner@mozilla.com')
        c = Consumer(name='p', status='accepted', user=partner)
        c.generate_random_codes()
        c.save()
        r = oclient.get('api.user', c, None,
                        params={'email': 'admin@mozilla.com'})
        eq_(r.status_code, 200)
        eq_(json.loads(r.content)['email'], 'admin@mozilla.com')

    def test_failed_user_lookup(self):
        partner = UserProfile.objects.get(email='partner@mozilla.com')
        c = Consumer(name='p', status='accepted', user=partner)
        c.generate_random_codes()
        c.save()
        r = oclient.get('api.user', c, None,
                        params={'email': 'not_a_user@mozilla.com'})
        eq_(r.status_code, 404)

    def test_forbidden_user_lookup(self, response_code=401):
        r = oclient.get('api.user', self.accepted_consumer, self.token,
                        params={'email': 'admin@mozilla.com'})
        eq_(r.status_code, response_code)

class TestDRFUser(TestUser):

    def setUp(self):
        super(TestDRFUser, self).setUp()
        self.create_switch('drf')

    def test_forbidden_user_lookup(self):
        super(TestDRFUser, self).test_forbidden_user_lookup(response_code=403)

def activitylog_count(type=None):
    qs = ActivityLog.objects
    if type:
        qs = qs.filter(action=type.id)
    return qs.count()

class TestAddon(BaseOAuth):
    created_http_status = 200
    permission_denied_http_status = 401

    def setUp(self):
        super(TestAddon, self).setUp()
        path = 'src/olympia/files/fixtures/files/extension.xpi'
        xpi = os.path.join(settings.ROOT, path)
        f = open(xpi)

        self.create_data = dict(builtin=0,
                                name='FREEDOM',
                                text='This is FREE!',
                                platform='mac',
                                xpi=f)

        path = 'src/olympia/files/fixtures/files/extension-0.2.xpi'
        self.version_data = dict(builtin=2, platform='windows',
                                 xpi=open(os.path.join(settings.ROOT, path)))
        self.update_data = dict(name='fu',
                                default_locale='fr',
                                homepage='mozilla.com',
                                support_email='go@away.com',
                                support_url='http://google.com/',
                                description='awesome',
                                summary='sucks',
                                developer_comments='i made it for you',
                                eula='love it',
                                privacy_policy='aybabtu',
                                the_reason='for shits',
                                the_future='is gone',
                                view_source=1,
                                prerelease=1,
                                binary=False,
                                site_specific=1)

    def make_create_request(self, data):
        return oclient.post('api.addons', self.accepted_consumer, self.token,
                            data=data)

    def create_addon(self):
        current_count = activitylog_count(amo.LOG.CREATE_ADDON)
        r = self.make_create_request(self.create_data)
        eq_(r.status_code, self.created_http_status, r.content)
        # 1 new add-on
        eq_(activitylog_count(amo.LOG.CREATE_ADDON), current_count + 1)
        return json.loads(r.content)

    def test_create_no_user(self):
        # The user in TwoLeggedAuth is set to the consumer user.
        # If there isn't one, we should get a challenge back.
        self.accepted_consumer.user = None
        self.accepted_consumer.save()
        r = self.make_create_request(self.create_data)
        eq_(r.status_code, 401)

    def test_create_user_altered(self):
        data = self.create_data
        data['authenticate_as'] = self.editor.pk
        r = self.make_create_request(data)
        eq_(r.status_code, self.created_http_status, r.content)

        id = json.loads(r.content)['id']
        ad = Addon.objects.get(pk=id)
        eq_(len(ad.authors.all()), 1)
        eq_(ad.authors.all()[0].pk, self.editor.pk)

    def test_create(self):
        # License (req'd): MIT, GPLv2, GPLv3, LGPLv2.1, LGPLv3, MIT, BSD, Other
        # Custom License (if other, req'd)
        # XPI file... (req'd)
        # Platform (All by default): 'mac', 'all', 'bsd', 'linux', 'solaris',
        #   'windows'

        data = self.create_addon()
        id = data['id']
        name = data['name']
        eq_(name, 'xpi name')
        assert Addon.objects.get(pk=id)

    def create_no_license(self):
        data = self.create_data.copy()
        del data['builtin']
        return self.make_create_request(data)

    def test_create_no_license(self):
        r = self.create_no_license()
        eq_(r.status_code, self.created_http_status, r.content)
        eq_(Addon.objects.count(), 1)

    def test_create_no_license_step(self):
        r = self.create_no_license()
        eq_(r.status_code, self.created_http_status, r.content)
        id = json.loads(r.content)['id']
        eq_(SubmitStep.objects.get(addon=id).step, 5)

    def test_create_no_license_url(self):
        self.create_no_license()
        self.client.login(username='editor@mozilla.com', password='password')
        res = self.client.get(reverse('devhub.submit.resume',
                                      args=['xpi-name']))
        self.assert3xx(res, reverse('devhub.submit.5', args=['xpi-name']))

    def test_create_no_license_status(self):
        self.create_no_license()
        eq_(Addon.objects.get(slug='xpi-name').status, 0)

    def test_create_status(self):
        r = self.make_create_request(self.create_data)
        eq_(r.status_code, self.created_http_status, r.content)
        eq_(Addon.objects.get(slug='xpi-name').status, 0)
        eq_(Addon.objects.count(), 1)

    def test_create_slug(self):
        r = self.make_create_request(self.create_data)
        content = json.loads(r.content)
        eq_(content['slug'], 'xpi-name')
        eq_(content['resource_uri'],
            absolutify(reverse('addons.detail', args=['xpi-name'])))

    def test_delete(self):
        data = self.create_addon()
        id = data['id']
        # Force it to be public so an email gets sent.
        Addon.objects.filter(id=id).update(highest_status=amo.STATUS_PUBLIC)

        r = oclient.delete(('api.addon', id), self.accepted_consumer,
                           self.token)
        eq_(r.status_code, 204, r.content)
        eq_(Addon.objects.filter(pk=id).count(), 0, "Didn't delete.")

        eq_(len(mail.outbox), 1)

    def test_update(self):
        # create an addon
        data = self.create_addon()
        id = data['id']

        current_count = activitylog_count()
        r = oclient.put(('api.addon', id), self.accepted_consumer, self.token,
                        data=self.update_data)
        eq_(r.status_code, 200, r.content)

        # EDIT_PROPERTIES
        eq_(activitylog_count(), current_count + 1)

        a = Addon.objects.get(pk=id)
        for field, expected in self.update_data.iteritems():
            value = getattr(a, field)
            if isinstance(value, Translation):
                value = unicode(value)

            eq_(value, expected,
                "'%s' didn't match: got '%s' instead of '%s'"
                % (field, getattr(a, field), expected))

    @patch('olympia.api.handlers.AddonForm.is_valid')
    def test_update_fail(self, is_valid):
        data = self.create_addon()
        id = data['id']
        is_valid.return_value = False
        r = oclient.put(('api.addon', id), self.accepted_consumer, self.token,
                        data=self.update_data)
        eq_(r.status_code, 400, r.content)

    def test_update_nonexistant(self):
        r = oclient.put(('api.addon', 0), self.accepted_consumer, self.token,
                        data={})
        eq_(r.status_code, 410, r.content)

    @patch('olympia.api.handlers.XPIForm.clean_xpi')
    def test_xpi_failure(self, f):
        f.side_effect = forms.ValidationError('F')
        r = self.make_create_request(self.create_data)
        eq_(r.status_code, 400)

    def test_fake_license(self):
        data = self.create_data.copy()
        data['builtin'] = 'fff'

        r = self.make_create_request(data)
        eq_(r.status_code, 400, r.content)
        eq_(r.content, 'Bad Request: Invalid data provided: '
            'Select a valid choice. fff is not one of the available choices. '
            '(builtin)')

    @patch('zipfile.ZipFile.infolist')
    def test_bad_zip(self, infolist):
        fake = Mock()
        fake.filename = '..'
        infolist.return_value = [fake]
        r = self.make_create_request(self.create_data)
        eq_(r.status_code, 400, r.content)

    @patch('olympia.versions.models.AppVersion.objects.get')
    def test_bad_appversion(self, get):
        get.side_effect = AppVersion.DoesNotExist()
        data = self.create_addon()
        assert data, "We didn't get data."

    def test_wrong_guid(self):
        data = self.create_addon()
        id = data['id']
        addon = Addon.objects.get(pk=id)
        addon.guid = 'XXX'
        addon.save()

        # Upload new version of file
        r = oclient.post(('api.versions', id,), self.accepted_consumer,
                         self.token, data=self.version_data)
        eq_(r.status_code, 400)
        eq_(r.content, 'Bad Request: Add-on did not validate: '
            "Add-on ID doesn't match add-on.")

    def test_duplicate_guid(self):
        self.create_addon()
        data = self.create_data.copy()
        data['xpi'] = self.version_data['xpi']
        r = self.make_create_request(data)
        eq_(r.status_code, 400)
        eq_(r.content, 'Bad Request: Add-on did not validate: '
            'Duplicate add-on ID found.')

    def test_create_version(self):
        # Create an addon and let's use this for the new version.
        data = self.create_addon()
        id = data['id']

        log_count = activitylog_count()

        # Upload new version of file
        r = oclient.post(('api.versions', id,), self.accepted_consumer,
                         self.token, data=self.version_data)

        eq_(r.status_code, 200, r.content)

        # verify we've logged a new version and a new app version
        eq_(log_count + 2, activitylog_count())

        # validate that the addon has 2 versions
        a = Addon.objects.get(pk=id)
        eq_(a.versions.all().count(), 2)

        # validate the version number
        v = a.versions.get(version='0.2')
        eq_(v.version, '0.2')
        # validate any new version data
        eq_(amo.PLATFORMS[v.files.get().platform].shortname, 'windows')

    def test_create_version_bad_license(self):
        data = self.create_addon()
        id = data['id']
        data = self.version_data.copy()
        data['builtin'] = 'fu'
        r = oclient.post(('api.versions', id,), self.accepted_consumer,
                         self.token, data=data)

        eq_(r.status_code, 400, r.content)

    def test_create_version_no_license(self):
        data = self.create_addon()
        id = data['id']
        data = self.version_data.copy()
        del data['builtin']
        r = oclient.post(('api.versions', id,), self.accepted_consumer,
                         self.token, data=data)

        eq_(r.status_code, 200, r.content)
        data = json.loads(r.content)
        id = data['id']
        v = Version.objects.get(pk=id)
        assert not v.license

    def create_for_update(self):
        data = self.create_addon()
        id = data['id']
        a = Addon.objects.get(pk=id)
        v = a.versions.get()
        eq_(v.version, '0.1')
        return a, v, 'src/olympia/files/fixtures/files/extension-0.2.xpi'

    def test_update_version_no_license(self):
        a, v, path = self.create_for_update()
        data = dict(release_notes='fukyeah', platform='windows',
                    xpi=open(os.path.join(settings.ROOT, path)))
        r = oclient.put(('api.version', a.id, v.id), self.accepted_consumer,
                        self.token, data=data, content_type=MULTIPART_CONTENT)
        eq_(r.status_code, 200, r.content)
        v = a.versions.get()
        eq_(v.version, '0.2')
        eq_(v.license, None)

    def test_update_version_bad_license(self):
        a, v, path = self.create_for_update()
        data = dict(release_notes='fukyeah', builtin=3, platform='windows',
                    xpi=open(os.path.join(settings.ROOT, path)))
        r = oclient.put(('api.version', a.id, v.id), self.accepted_consumer,
                        self.token, data=data, content_type=MULTIPART_CONTENT)
        eq_(r.status_code, 400, r.content)

    def test_update_version(self):
        a, v, path = self.create_for_update()
        data = dict(release_notes='fukyeah', builtin=2, platform='windows',
                    xpi=open(os.path.join(settings.ROOT, path)))
        log_count = activitylog_count()
        # upload new version
        r = oclient.put(('api.version', a.id, v.id), self.accepted_consumer,
                        self.token, data=data, content_type=MULTIPART_CONTENT)
        eq_(r.status_code, 200, r.content[:1000])

        # verify we've logged a version update and a new app version
        eq_(activitylog_count(), log_count + 2)
        # verify data
        v = a.versions.get()
        eq_(v.version, '0.2')
        eq_(str(v.releasenotes), 'fukyeah')
        eq_(str(v.license.builtin), '2')

    def test_update_version_bad_xpi(self):
        data = self.create_addon()
        id = data['id']

        # verify version
        a = Addon.objects.get(pk=id)
        v = a.versions.get()

        eq_(v.version, '0.1')

        data = dict(release_notes='fukyeah', platform='windows')

        # upload new version
        r = oclient.put(('api.version', id, v.id), self.accepted_consumer,
                        self.token, data=data, content_type=MULTIPART_CONTENT)
        eq_(r.status_code, 400)

    def test_update_version_bad_id(self):
        r = oclient.put(('api.version', 0, 0), self.accepted_consumer,
                        self.token, data={}, content_type=MULTIPART_CONTENT)
        eq_(r.status_code, 410, r.content)

    def test_get_version(self):
        data = self.create_addon()
        a = Addon.objects.get(pk=data['id'])
        r = oclient.get(('api.version', data['id'], a.versions.get().id),
                        self.accepted_consumer, self.token)
        eq_(r.status_code, 200)

    def test_get_version_statuses(self):
        data = self.create_addon()
        a = Addon.objects.get(pk=data['id'])
        r = oclient.get(('api.version', data['id'], a.versions.get().id),
                        self.accepted_consumer, self.token)
        eq_(json.loads(r.content)['statuses'],
            [[File.objects.all()[0].pk, 1]])

    @patch(
        'olympia.api.authorization.AllowRelatedAppOwner.has_object_permission')
    @patch('olympia.api.authorization.AllowAppOwner.has_object_permission')
    @patch('olympia.access.acl.action_allowed')
    @patch('olympia.access.acl.check_addon_ownership')
    def test_not_my_addon(self, addon_ownership, action_allowed,
                          app_owner, related_app_owner):
        data = self.create_addon()
        id = data['id']
        a = Addon.objects.get(pk=id)
        v = a.versions.get()
        # The first one is for piston, the 3 next ones are for DRF.
        addon_ownership.return_value = False
        action_allowed.return_value = False
        app_owner.return_value = False
        related_app_owner.return_value = False

        r = oclient.put(('api.version', id, v.id), self.accepted_consumer,
                        self.token, data={}, content_type=MULTIPART_CONTENT)
        eq_(r.status_code, self.permission_denied_http_status, r.content)

        r = oclient.put(('api.addon', id), self.accepted_consumer, self.token,
                        data=self.update_data)
        eq_(r.status_code, self.permission_denied_http_status, r.content)

    def test_delete_version(self):
        data = self.create_addon()
        id = data['id']

        a = Addon.objects.get(pk=id)
        v = a.versions.get()

        log_count = activitylog_count()
        r = oclient.delete(('api.version', id, v.id), self.accepted_consumer,
                           self.token)
        eq_(activitylog_count(), log_count + 1)

        eq_(r.status_code, 204, r.content)
        eq_(a.versions.count(), 0)

    def test_retrieve_versions(self):
        data = self.create_addon()
        id = data['id']

        a = Addon.objects.get(pk=id)
        v = a.versions.get()

        r = oclient.get(('api.versions', id), self.accepted_consumer,
                        self.token)
        eq_(r.status_code, 200, r.content)
        data = json.loads(r.content)
        for attr in ('id', 'version',):
            expect = getattr(v, attr)
            val = data[0].get(attr)
            eq_(expect, val,
                'Got "%s" was expecting "%s" for "%s".' % (val, expect, attr,))

    def test_no_addons(self):
        r = oclient.get('api.addons', self.accepted_consumer, self.token)
        eq_(json.loads(r.content)['count'], 0)

    def test_no_user(self):
        addon = Addon.objects.create(type=amo.ADDON_EXTENSION)
        AddonUser.objects.create(addon=addon, user=self.admin,
                                 role=amo.AUTHOR_ROLE_DEV)
        r = oclient.get('api.addons', self.accepted_consumer, self.token)
        eq_(json.loads(r.content)['count'], 0)

    def test_my_addons_only(self):
        for num in range(0, 2):
            addon = Addon.objects.create(type=amo.ADDON_EXTENSION)
        AddonUser.objects.create(addon=addon, user=self.editor,
                                 role=amo.AUTHOR_ROLE_DEV)
        r = oclient.get('api.addons', self.accepted_consumer, self.token,
                        params={'authenticate_as': self.editor.pk})
        j = json.loads(r.content)
        eq_(j['count'], 1)
        eq_(j['objects'][0]['id'], addon.id)

    def test_one_addon(self):
        addon = Addon.objects.create(type=amo.ADDON_EXTENSION)
        AddonUser.objects.create(addon=addon, user=self.editor,
                                 role=amo.AUTHOR_ROLE_DEV)
        r = oclient.get(('api.addon', addon.pk), self.accepted_consumer,
                        self.token, params={'authenticate_as': self.editor.pk})
        eq_(json.loads(r.content)['id'], addon.pk)

    def test_my_addons_role(self):
        addon = Addon.objects.create(type=amo.ADDON_EXTENSION)
        AddonUser.objects.create(addon=addon, user=self.editor,
                                 role=amo.AUTHOR_ROLE_VIEWER)
        r = oclient.get('api.addons', self.accepted_consumer, self.token)
        eq_(json.loads(r.content)['count'], 0)

    def test_my_addons_disabled(self):
        addon = Addon.objects.create(type=amo.ADDON_EXTENSION,
                                     status=amo.STATUS_DISABLED)
        AddonUser.objects.create(addon=addon, user=self.editor,
                                 role=amo.AUTHOR_ROLE_DEV)
        r = oclient.get('api.addons', self.accepted_consumer, self.token)
        eq_(json.loads(r.content)['count'], 0)

    def test_my_addons_deleted(self):
        addon = Addon.objects.create(type=amo.ADDON_EXTENSION,
                                     status=amo.STATUS_DELETED)
        AddonUser.objects.create(addon=addon, user=self.editor,
                                 role=amo.AUTHOR_ROLE_DEV)
        r = oclient.get('api.addons', self.accepted_consumer, self.token)
        eq_(json.loads(r.content)['count'], 0)

class TestDRFAddon(TestAddon):
    created_http_status = 201
    permission_denied_http_status = 403

    def setUp(self):
        super(TestDRFAddon, self).setUp()
        self.create_switch('drf')

    def _compare_dicts(self, drf_data, piston_data):
        """
        Given 2 dicts of data from DRF and Piston, compare keys then values.
        """
        eq_(sorted(drf_data.keys()), sorted(piston_data.keys()),
            ('Keys inexistent from Piston: {0}\n'
             'Keys inexistent from DRF: {1}').format(
                set(piston_data) - set(drf_data),
                set(drf_data) - set(piston_data)))
        for drf_item, piston_item in zip(sorted(drf_data.items()),
                                         sorted(piston_data.items())):
            eq_(drf_item[0], piston_item[0])
            eq_(drf_item[1], piston_item[1],
                ('Different representations for key "{0}": DRF={1}, Piston={2}'
                 .format(drf_item[0], drf_item[1], piston_item[1])))

    def compare_output(self, url, listed=False):
        """
        Load responses from DRF and Piston given the `url` parameter and
        compare returned data dicts, key by key. Useful to make sure
        that both responses are similar.

        Set `listed` to True for comparing responses as lists.
        """
        r = oclient.get(url, self.accepted_consumer, self.token)
        eq_(r.status_code, 200, r.content)
        drf_data = json.loads(r.content)
        self.create_switch('drf', **{'active': False})
        r = oclient.get(url, self.accepted_consumer, self.token)
        eq_(r.status_code, 200, r.content)
        piston_data = json.loads(r.content)
        if listed:
            eq_(len(drf_data), len(piston_data))
            for items in zip(drf_data, piston_data):
                self._compare_dicts(items[0], items[1])
        else:
            self._compare_dicts(drf_data, piston_data)

    def test_diff_versions(self):
        data = self.create_addon()
        self.compare_output(('api.versions', data['id']), listed=True)

    def test_diff_version(self):
        data = self.create_addon()
        addon = Addon.objects.get(pk=data['id'])
        version = addon.versions.get()
        self.compare_output(('api.version', addon.id, version.id))

    def test_diff_addons(self):
        self.create_addon()
        self.compare_output(('api.addons'))

    def test_diff_addon(self):
        data = self.create_addon()
        self.compare_output(('api.addon', data['id']))


text:
const struct MyMovieGenreAttributes MyMovieGenreAttributes = {
	.identifier = @"identifier",
	.name = @"name",
};

const struct MyMovieGenreRelationships MyMovieGenreRelationships = {
	.movies = @"movies",
};

const struct MyMovieGenreFetchedProperties MyMovieGenreFetchedProperties = {
};

@implementation MyMovieGenreID
@end

@implementation _MyMovieGenre

+ (id)insertInManagedObjectContext:(NSManagedObjectContext*)moc_ {
	NSParameterAssert(moc_);
	return [NSEntityDescription insertNewObjectForEntityForName:@"MyMovieGenre" inManagedObjectContext:moc_];
}

+ (NSString*)entityName {
	return @"MyMovieGenre";
}

+ (NSEntityDescription*)entityInManagedObjectContext:(NSManagedObjectContext*)moc_ {
	NSParameterAssert(moc_);
	return [NSEntityDescription entityForName:@"MyMovieGenre" inManagedObjectContext:moc_];
}

- (MyMovieGenreID*)objectID {
	return (MyMovieGenreID*)[super objectID];
}

+ (NSSet*)keyPathsForValuesAffectingValueForKey:(NSString*)key {
	NSSet *keyPaths = [super keyPathsForValuesAffectingValueForKey:key];

	if ([key isEqualToString:@"identifierValue"]) {
		NSSet *affectingKey = [NSSet setWithObject:@"identifier"];
		keyPaths = [keyPaths setByAddingObjectsFromSet:affectingKey];
		return keyPaths;
	}

	return keyPaths;
}

@dynamic identifier;

- (int16_t)identifierValue {
	NSNumber *result = [self identifier];
	return [result shortValue];
}

- (void)setIdentifierValue:(int16_t)value_ {
	[self setIdentifier:[NSNumber numberWithShort:value_]];
}

- (int16_t)primitiveIdentifierValue {
	NSNumber *result = [self primitiveIdentifier];
	return [result shortValue];
}

- (void)setPrimitiveIdentifierValue:(int16_t)value_ {
	[self setPrimitiveIdentifier:[NSNumber numberWithShort:value_]];
}

@dynamic name;

@dynamic movies;

@end


text:
require 'spec_helper'
require 'specinfra/helper/detect_os/freebsd'

describe Specinfra::Helper::DetectOs::Freebsd do
  freebsd = Specinfra::Helper::DetectOs::Freebsd.new(:exec)
  it 'Should return freebsd 8 when FreeBSD 8.2-RELEASE-p3 is installed.' do
    allow(freebsd).to receive(:run_command) {
      CommandResult.new(:stdout => 'FreeBSD 8.2-RELEASE-p3', :exit_status => 0)
    }
    expect(freebsd.detect).to include(
      :family  => 'freebsd',
      :release => '8'
    )
  end
  it 'Should return freebsd 9 when FreeBSD 9.3-RELEASE is installed.' do
    allow(freebsd).to receive(:run_command) {
      CommandResult.new(:stdout => 'FreeBSD 9.3-RELEASE', :exit_status => 0)
    }
    expect(freebsd.detect).to include(
      :family  => 'freebsd',
      :release => '9'
    )
  end
  it 'Should return freebsd 10 when FreeBSD 10.1-RELEASE is installed.' do
    allow(freebsd).to receive(:run_command) {
      CommandResult.new(:stdout => 'FreeBSD 10.1--RELEASE', :exit_status => 0)
    }
    expect(freebsd.detect).to include(
      :family  => 'freebsd',
      :release => '10'
    )
  end
end


text:
<?php
class ZeroValueTest extends Doctrine_Record
{
    public function setTableDefinition()
    {
        $this->hasColumn('id', 'integer', 4, array('primary' => true,  'autoincrement' => true,));
        $this->hasColumn('username', 'string', 128, array('notnull' => true,));
        $this->hasColumn('algorithm', 'string', 128, array('default' => 'sha1', 'notnull' => true,));
        $this->hasColumn('salt', 'string', 128, array('notnull' => true,));
        $this->hasColumn('password', 'string', 128, array('notnull' => true,));
        $this->hasColumn('created_at', 'timestamp', null, array());
        $this->hasColumn('last_login', 'timestamp', null, array());
        $this->hasColumn('is_active', 'boolean', null, array('default' => true, 'notnull' => true,));
        $this->hasColumn('is_super_admin', 'boolean', null, array('default' => false, 'notnull' => true,));
    }

    public function setUp()
    { }
}


text:
package org.multiverse.stms.gamma.transactions.fat;

import org.junit.Test;
import org.multiverse.stms.gamma.transactionalobjects.GammaTxnRef;

import static org.junit.Assert.assertEquals;

public class FatVariableLengthGammaTxn_hardResetTest extends FatGammaTxn_hardResetTest<FatVariableLengthGammaTxn> {

    @Override
    protected FatVariableLengthGammaTxn newTransaction() {
        return new FatVariableLengthGammaTxn(stm);
    }

    @Test
    public void testDownsizing(){
        FatVariableLengthGammaTxn tx = new FatVariableLengthGammaTxn(stm);
        for(int k=0;k<10000;k++){
            new GammaTxnRef(tx);
        }
        tx.hardReset();
        assertEquals(tx.config.minimalArrayTreeSize, tx.array.length);
    }
}