package org.infinispan.test.integration.remote;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.Map;

import org.infinispan.Version;
import org.infinispan.test.integration.memcached.SimpleMemcachedClient;
import org.infinispan.test.integration.util.ITestUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import net.spy.memcached.DefaultConnectionFactory;

/**
 * Tests for Memcached endpoint.
 *
 * @author Michal Linhard
 * @author Martin Gencur
 * @author Jozef Vilkolak
 */
public abstract class AbstractMemcacheClientIT {

   private SimpleMemcachedClient mc;

   // they are deleted in the setUp and tearDown methods
   static final String KEY_A = "a";
   static final String KEY_B = "b";
   static final String KEY_C = "c";

   static final String ENCODING = "UTF-8";

   @Before
   public void initialize() throws IOException {
      mc = createMemcachedClient();
      mc.delete(KEY_A);
      mc.delete(KEY_B);
      mc.delete(KEY_C);
      assertNull(mc.get(KEY_A));
      assertNull(mc.get(KEY_B));
      assertNull(mc.get(KEY_C));
   }

   @After
   public void cleanUp() throws IOException {
      mc.delete(KEY_A);
      mc.delete(KEY_B);
      mc.delete(KEY_C);
      mc.close();
   }

   @Test
   public void testSetGet() throws Exception {
      mc.set(KEY_A, "A");
      assertEquals("A", mc.get(KEY_A));
   }

   @Test
   public void testSetGetNewLineChars() throws Exception {
      mc.set(KEY_A, "A\r\nA");
      assertEquals("A\r\nA", mc.get(KEY_A));
   }

   @Test
   public void testSetGetObject() throws Exception {
      ByteArrayOutputStream bout = new ByteArrayOutputStream();
      ObjectOutputStream oo = new ObjectOutputStream(bout);
      oo.writeObject(new TestSerializable("CONTENT"));
      oo.flush();

      byte[] byteData = bout.toByteArray();
      mc.writeln("set " + KEY_A + " 0 0 " + byteData.length);
      mc.flush();
      mc.write(byteData);
      mc.write("\r\n".getBytes(ENCODING));
      mc.flush();
      assertEquals("STORED", mc.readln());
      byte[] bytesBack = mc.getBytes(KEY_A);
      assertEquals(byteData.length, bytesBack.length);
      ObjectInputStream oin = new ObjectInputStream(new ByteArrayInputStream(bytesBack));
      TestSerializable ts = (TestSerializable) oin.readObject();
      assertEquals("CONTENT", ts.getContent());
   }

   @Test
   public void testSetGetFlags() throws Exception {
      mc.writeln("set " + KEY_A + " 1234 0 1");
      mc.writeln("A");
      mc.flush();
      assertEquals("STORED", mc.readln());
      mc.writeln("get " + KEY_A);
      mc.flush();
      assertEquals("VALUE " + KEY_A + " 1234 1", mc.readln());
      assertEquals("A", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testSecondSet() throws Exception {
      mc.set(KEY_A, "A");
      assertEquals("A", mc.get(KEY_A));
      mc.set(KEY_A, "B");
      assertEquals("B", mc.get(KEY_A));
   }

   @Test
   public void testGetMultipleKeys() throws Exception {
      mc.set(KEY_A, "A");
      mc.set(KEY_B, "B");
      mc.writeln("get " + KEY_A + " " + KEY_B);
      mc.flush();
      assertEquals("VALUE " + KEY_A + " 0 1", mc.readln());
      assertEquals("A", mc.readln());
      assertEquals("VALUE " + KEY_B + " 0 1", mc.readln());
      assertEquals("B", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testAdd() throws Exception {
      String data = "testAdd";
      mc.writeln("add " + KEY_A + " 0 0 " + data.getBytes(ENCODING).length);
      mc.writeln(data);
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals(data, mc.get(KEY_A));
      // second add
      mc.writeln("add " + KEY_A + " 0 0 " + data.getBytes(ENCODING).length);
      mc.writeln(data);
      mc.flush();
      assertEquals("NOT_STORED", mc.readln());
   }

   @Test
   public void testReplace() throws Exception {
      mc.set(KEY_A, "testAdd");
      assertEquals("testAdd", mc.get(KEY_A));
      // replace
      mc.writeln("replace " + KEY_A + " 0 0 " + "replacement".getBytes(ENCODING).length);
      mc.writeln("replacement");
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("replacement", mc.get(KEY_A));
   }

   @Test
   public void testReplaceNonExistent() throws Exception {
      mc.writeln("replace " + KEY_A + " 0 0 " + "replacement".getBytes(ENCODING).length);
      mc.writeln("replacement");
      mc.flush();
      assertEquals("NOT_STORED", mc.readln());
   }

   @Test
   public void testAppend() throws Exception {
      mc.set(KEY_A, "Hello");
      assertEquals("Hello", mc.get(KEY_A));
      mc.writeln("append " + KEY_A + " 0 0 " + ", World!".getBytes(ENCODING).length);
      mc.writeln(", World!");
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("Hello, World!", mc.get(KEY_A));
   }

   @Test
   public void testAppendNonExistent() throws Exception {
      mc.writeln("append " + KEY_A + " 0 0 " + ", World!".getBytes(ENCODING).length);
      mc.writeln(", World!");
      mc.flush();
      assertEquals("NOT_STORED", mc.readln());
   }

   @Test
   public void testPrepend() throws Exception {
      mc.set(KEY_A, "World!");
      assertEquals("World!", mc.get(KEY_A));
      mc.writeln("prepend " + KEY_A + " 0 0 " + "Hello, ".getBytes(ENCODING).length);
      mc.writeln("Hello, ");
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("Hello, World!", mc.get(KEY_A));
   }

   @Test
   public void testPrependNonExistent() throws Exception {
      mc.writeln("prepend " + KEY_A + " 0 0 " + "Hello, ".getBytes(ENCODING).length);
      mc.writeln("Hello, ");
      mc.flush();
      assertEquals("NOT_STORED", mc.readln());
   }

   @Test
   public void testCas() throws Exception {
      mc.set(KEY_A, "A");
      mc.writeln("gets " + KEY_A);
      mc.flush();
      String[] valueline = mc.readln().split(" ");
      assertEquals("VALUE", valueline[0]);
      assertEquals(KEY_A, valueline[1]);
      assertEquals("0", valueline[2]); // flags
      assertEquals("1", valueline[3]); // number of bytes in str "A"
      assertEquals("A", mc.readln());
      String casId = valueline[4];
      assertEquals("END", mc.readln());
      mc.writeln("cas " + KEY_A + " 0 0 1 " + casId);
      mc.writeln("B");
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("B", mc.get(KEY_A));
      mc.writeln("gets " + KEY_A);
      mc.flush();
      valueline = mc.readln().split(" ");
      assertFalse(casId.equals(valueline[4]));
      assertEquals("B", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testCasExists() throws Exception {
      mc.set(KEY_A, "A");
      mc.writeln("gets " + KEY_A);
      mc.flush();
      String[] valueline = mc.readln().split(" ");
      assertEquals("VALUE", valueline[0]);
      assertEquals(KEY_A, valueline[1]);
      assertEquals("0", valueline[2]); // flags
      assertEquals("1", valueline[3]); // number of bytes in str "A"
      assertEquals("A", mc.readln());
      String casId = valueline[4];
      assertEquals("END", mc.readln());
      mc.writeln("cas " + KEY_A + " 0 0 1 1" + casId); // note appended 1 before casId
      mc.writeln("B");
      mc.flush();
      assertEquals("EXISTS", mc.readln());
      assertEquals("A", mc.get(KEY_A));
   }

   @Test
   public void testGetNotFound() throws Exception {
      mc.writeln("get " + KEY_A);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testCasNotFound() throws Exception {
      mc.writeln("cas " + KEY_A + " 0 0 1 1");
      mc.writeln("B");
      mc.flush();
      assertEquals("NOT_FOUND", mc.readln());
   }

   @Test
   public void testExpTime() throws Exception {
      mc.writeln("set " + KEY_A + " 0 2 1");
      mc.writeln("A");
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("A", mc.get(KEY_A));
      ITestUtils.sleepForSecs(3);
      assertNull(mc.get(KEY_A));
   }

   /**
    * NOTE: This is randomly failing test on various platforms.
    *
    * @throws Exception
    */
   @Test
   public void testExpTimeMaxSeconds() throws Exception {
      // set exp.time to max number treated as seconds
      // 2592000 = 60*60*24*30
      mc.writeln("set " + KEY_A + " 0 2592000 1");
      mc.writeln("A");
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("A", mc.get(KEY_A));
      ITestUtils.sleepForSecs(2);
      assertEquals("A", mc.get(KEY_A));
   }

   @Test
   public void testExpTimeAbsolutePast() throws Exception {
      // set exp.time to min number treated as unix time
      // 2592001 = 60*60*24*30 + 1
      // corresponding to Sat, 31 Jan 1970 00:00:01 GMT
      mc.writeln("set " + KEY_A + " 0 2592001 1");
      mc.writeln("A");
      mc.flush();
      assertEquals("STORED", mc.readln());
      // should expire immediately (the next millisecond)
      // the sleep here is only to make sure that System.currentTimeMillis()
      // increases before next get
      Thread.sleep(5);
      assertNull(mc.get(KEY_A));
   }

   @Test
   public void testExpTimeAbsoluteFuture() throws Exception {
      long now = mc.getServerTime();
      mc.writeln("set " + KEY_A + " 0 " + (now + 2) + " 1");
      mc.writeln("A");
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("A", mc.get(KEY_A));
      ITestUtils.sleepForSecs(2);
      assertNull(mc.get(KEY_A));
   }

   private String key250() {
      String key250 = "";
      for (int i = 0; i < 250; i++) {
         key250 += "a";
      }
      return key250;
   }

   @Test
   public void testKeyLonger250() throws Exception {
      String key250 = key250();
      mc.delete(key250);
      String key251 = key250 + "a";
      // byte length is the same, because we're using "UTF-8 safe" char 'a'
      assertEquals(key250.length(), key250.getBytes(ENCODING).length);
      assertEquals(key251.length(), key251.getBytes(ENCODING).length);
      mc.set(key250, "A");
      assertEquals("A", mc.get(key250));
      mc.writeln("set " + key251 + " 0 0 1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testKeyLonger250Get() throws Exception {
      mc.writeln("get " + key250() + "a");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testKeyLonger250Gets() throws Exception {
      mc.writeln("gets " + key250() + "a");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testKeyLonger250Multiget() throws Exception {
      String key250 = key250();
      String key251 = key250 + "a";
      // byte length is the same, because we're using "UTF-8 safe" char 'a'
      assertEquals(key250.length(), key250.getBytes(ENCODING).length);
      assertEquals(key251.length(), key251.getBytes(ENCODING).length);
      mc.set(key250, "A");
      assertEquals("A", mc.get(key250));
      mc.writeln("get " + key250 + " " + key251);
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
      assertEquals("DELETED", mc.delete(key250));
   }

   @Test
   public void testKeyLonger250Add() throws Exception {
      String key250 = key250();
      mc.delete(key250);
      String key251 = key250 + "a";
      // byte length is the same, because we're using "UTF-8 safe" char 'a'
      assertEquals(key250.length(), key250.getBytes(ENCODING).length);
      assertEquals(key251.length(), key251.getBytes(ENCODING).length);
      mc.writeln("add " + key250 + " 0 0 1");
      mc.writeln("A");
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("A", mc.get(key250));
      mc.writeln("add " + key251 + " 0 0 1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testKeyLonger250Replace() throws Exception {
      mc.writeln("replace " + key250() + "a 0 0 1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testKeyLonger250Append() throws Exception {
      mc.writeln("append " + key250() + "a 0 0 1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testKeyLonger250Prepend() throws Exception {
      mc.writeln("prepend " + key250() + "a 0 0 1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testKeyLonger250Cas() throws Exception {
      mc.writeln("prepend " + key250() + "a 0 0 1 1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testKeyLonger250Incr() throws Exception {
      mc.writeln("incr " + key250() + "a 1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testKeyLonger250Decr() throws Exception {
      mc.writeln("decr " + key250() + "a 1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testKeyLonger250Delete() throws Exception {
      mc.writeln("delete " + key250() + "a");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testDelete() throws Exception {
      mc.set(KEY_A, "A");
      assertEquals("A", mc.get(KEY_A));
      assertEquals("DELETED", mc.delete(KEY_A));
      assertNull(mc.get(KEY_A));
   }

   @Test
   public void testDeleteNonExistent() throws Exception {
      assertEquals("NOT_FOUND", mc.delete(KEY_A));
   }

   @Test
   public void testIncr() throws Exception {
      mc.set(KEY_A, "0");
      mc.writeln("incr " + KEY_A + " 1");
      mc.flush();
      assertEquals("1", mc.readln());
   }

   @Test
   public void testDecr() throws Exception {
      mc.set(KEY_A, "1");
      mc.writeln("decr " + KEY_A + " 1");
      mc.flush();
      assertEquals("0", mc.readln());
   }

   @Test
   public void testIncrNotFound() throws Exception {
      mc.writeln("incr " + KEY_A + " 1");
      mc.flush();
      assertEquals("NOT_FOUND", mc.readln());
   }

   @Test
   public void testDecrNotFound() throws Exception {
      mc.writeln("decr " + KEY_A + " 1");
      mc.flush();
      assertEquals("NOT_FOUND", mc.readln());
   }

   @Test
   public void testIncr64BitMax() throws Exception {
      mc.set(KEY_A, "18446744073709551614");
      mc.writeln("incr " + KEY_A + " 1");
      mc.flush();
      assertEquals("18446744073709551615", mc.readln());
      mc.writeln("incr " + KEY_A + " 1");
      mc.flush();
      assertEquals("0", mc.readln());
   }

   @Test
   public void testDecrZero() throws Exception {
      mc.set(KEY_A, "0");
      mc.writeln("decr " + KEY_A + " 1");
      mc.flush();
      assertEquals("0", mc.readln());
   }

   @Test
   public void testIncrBigIncrement() throws Exception {
      mc.set(KEY_A, "0");
      mc.writeln("incr " + KEY_A + " 18446744073709551615");
      mc.flush();
      assertEquals("18446744073709551615", mc.readln());
   }

   @Test
   public void testIncrBigDecrement() throws Exception {
      mc.set(KEY_A, "18446744073709551615");
      mc.writeln("decr " + KEY_A + " 18446744073709551615");
      mc.flush();
      assertEquals("0", mc.readln());
   }

   @Test
   public void testUnsupportedStats() throws Exception {
      Map<String, String> stats = mc.getStats();
      assertEquals(stats.get("pid"), "0");
      assertEquals(stats.get("pointer_size"), "0");
      assertEquals(stats.get("rusage_user"), "0");
      assertEquals(stats.get("rusage_system"), "0");
      assertEquals(stats.get("bytes"), "0");
      assertEquals(stats.get("curr_connections"), "0");
      assertEquals(stats.get("total_connections"), "0");
      assertEquals(stats.get("connection_structures"), "0");
      assertEquals(stats.get("auth_cmds"), "0");
      assertEquals(stats.get("auth_errors"), "0");
      assertEquals(stats.get("limit_maxbytes"), "0");
      assertEquals(stats.get("conn_yields"), "0");
      assertEquals(stats.get("threads"), "0");
   }

   @Test
   public void testStatsBytes() throws Exception {
      Map<String, String> stats = mc.getStats();
      int bytesRead = new Integer(stats.get("bytes_read"));
      int bytesWritten = new Integer(stats.get("bytes_written"));

      mc.set("testKey", "testValue");
      mc.get("testKey");
      mc.delete("testKey");

      stats = mc.getStats();
      assertTrue("Bytes read didnt increase.", bytesRead < Integer.parseInt(stats.get("bytes_read")));
      assertTrue("Bytes written didnt increase.", bytesWritten < Integer.parseInt(stats.get("bytes_written")));
   }

   @Test
   public void testStatsTime() throws Exception {
      Map<String, String> stats = mc.getStats();
      int uptime = new Integer(stats.get("uptime"));
      int time = new Integer(stats.get("time"));
      ITestUtils.sleepForSecs(1);
      stats = mc.getStats();
      assertTrue(uptime < new Integer(stats.get("uptime")));
      assertTrue(time < new Integer(stats.get("time")));
   }

   @Test
   public void testStatsVersion() throws Exception {
      Map<String, String> stats = mc.getStats();
      String version = stats.get("version");
      assertNotNull(version);
      assertTrue(version.startsWith(Version.getMajorMinor()));
   }

   @Test
   public void testStatsGetSetItemCount() throws Exception {
      Map<String, String> stats = mc.getStats();
      int cmd_set = new Integer(stats.get("cmd_set"));
      int cmd_get = new Integer(stats.get("cmd_get"));
      int get_hits = new Integer(stats.get("get_hits"));
      int get_misses = new Integer(stats.get("get_misses"));
      int curr_items = new Integer(stats.get("curr_items"));
      int total_items = new Integer(stats.get("total_items"));
      mc.set(KEY_A, "A");
      assertEquals("A", mc.get(KEY_A));
      assertNull(mc.get(KEY_B));
      stats = mc.getStats();
      int cmd_set_new = new Integer(stats.get("cmd_set"));
      int cmd_get_new = new Integer(stats.get("cmd_get"));
      int get_hits_new = new Integer(stats.get("get_hits"));
      int get_misses_new = new Integer(stats.get("get_misses"));
      int curr_items_new = new Integer(stats.get("curr_items"));
      int total_items_new = new Integer(stats.get("total_items"));
      assertEquals(cmd_get + 2, cmd_get_new);
      assertEquals(cmd_set + 1, cmd_set_new);
      assertEquals(get_hits + 1, get_hits_new);
      assertEquals(get_misses + 1, get_misses_new);
      assertEquals(curr_items + 1, curr_items_new);
      assertEquals(total_items + 1, total_items_new);
   }

   @Test
   public void testStatsDelete() throws Exception {
      Map<String, String> stats = mc.getStats();
      int delete_misses = new Integer(stats.get("delete_misses"));
      int delete_hits = new Integer(stats.get("delete_hits"));
      mc.set(KEY_A, "A");
      assertEquals("DELETED", mc.delete(KEY_A));
      assertEquals("NOT_FOUND", mc.delete(KEY_B));
      stats = mc.getStats();
      int delete_misses_new = new Integer(stats.get("delete_misses"));
      int delete_hits_new = new Integer(stats.get("delete_hits"));
      assertEquals(delete_misses + 1, delete_misses_new);
      assertEquals(delete_hits + 1, delete_hits_new);
   }

   @Test
   public void testStatsIncrDecr() throws Exception {
      Map<String, String> stats = mc.getStats();
      int incr_misses = new Integer(stats.get("incr_misses"));
      int incr_hits = new Integer(stats.get("incr_hits"));
      int decr_misses = new Integer(stats.get("decr_misses"));
      int decr_hits = new Integer(stats.get("decr_hits"));

      mc.set(KEY_A, "0");
      mc.writeln("incr " + KEY_A + " 1");
      mc.flush();
      assertEquals("1", mc.readln());
      mc.writeln("decr " + KEY_A + " 1");
      mc.flush();
      assertEquals("0", mc.readln());
      mc.writeln("incr " + KEY_B + " 1");
      mc.flush();
      assertEquals("NOT_FOUND", mc.readln());
      mc.writeln("decr " + KEY_B + " 1");
      mc.flush();
      assertEquals("NOT_FOUND", mc.readln());

      stats = mc.getStats();
      int incr_misses_new = new Integer(stats.get("incr_misses"));
      int incr_hits_new = new Integer(stats.get("incr_hits"));
      int decr_misses_new = new Integer(stats.get("decr_misses"));
      int decr_hits_new = new Integer(stats.get("decr_hits"));
      assertEquals(incr_misses + 1, incr_misses_new);
      assertEquals(incr_hits + 1, incr_hits_new);
      assertEquals(decr_misses + 1, decr_misses_new);
      assertEquals(decr_hits + 1, decr_hits_new);
   }

   @Test
   public void testStatsCas() throws Exception {
      Map<String, String> stats = mc.getStats();
      int cas_misses = new Integer(stats.get("cas_misses"));
      int cas_hits = new Integer(stats.get("cas_hits"));

      // cas hit
      mc.set(KEY_A, "A");
      mc.writeln("gets " + KEY_A);
      mc.flush();
      String[] valueline = mc.readln().split(" ");
      assertEquals("VALUE", valueline[0]);
      assertEquals(KEY_A, valueline[1]);
      assertEquals("0", valueline[2]); // flags
      assertEquals("1", valueline[3]); // number of bytes in str "A"
      assertEquals("A", mc.readln());
      String casId = valueline[4];
      assertEquals("END", mc.readln());
      mc.writeln("cas " + KEY_A + " 0 0 1 " + casId);
      mc.writeln("B");
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("B", mc.get(KEY_A));

      // cas miss
      mc.writeln("cas " + KEY_B + " 0 0 1 1");
      mc.writeln("B");
      mc.flush();
      assertEquals("NOT_FOUND", mc.readln());

      stats = mc.getStats();
      int cas_misses_new = new Integer(stats.get("cas_misses"));
      int cas_hits_new = new Integer(stats.get("cas_hits"));
      assertEquals(cas_misses + 1, cas_misses_new);
      assertEquals(cas_hits + 1, cas_hits_new);

      int cas_badval = new Integer(stats.get("cas_badval"));

      // cas bad value
      mc.writeln("cas " + KEY_A + " 0 0 1 1" + casId); // note appended 1 before casId
      mc.writeln("C");
      mc.flush();
      assertEquals("EXISTS", mc.readln());

      stats = mc.getStats();
      int cas_badval_new = new Integer(stats.get("cas_badval"));
      assertEquals(cas_badval + 1, cas_badval_new);
   }

   @Test
   public void testBogusCommand() throws Exception {
      mc.writeln("boguscommand");
      mc.flush();
      assertStartsWith(mc.readln(), "ERROR");
   }

   @Test
   public void testBogusCommandArgs() throws Exception {
      mc.writeln("boguscommand arg1 arg2 arg3");
      mc.flush();
      assertStartsWith(mc.readln(), "ERROR");
   }

   @Test
   public void testBogusCommandPipeline() throws Exception {
      mc.writeln("boguscommand");
      mc.writeln("delete " + KEY_A);
      mc.flush();
      assertStartsWith(mc.readln(), "ERROR");
      assertEquals("NOT_FOUND", mc.readln());
   }

   @Test
   public void testCasParsing1() throws Exception {
      mc.writeln("cas bad blah 0 0 0");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testCasParsing2() throws Exception {
      mc.writeln("cas bad 0 blah 0 0");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testCasParsing3() throws Exception {
      mc.writeln("cas bad 0 0 blah 0");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testCasParsing4() throws Exception {
      mc.writeln("cas bad 0 0 0 blah");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   // missing cas unique value
   @Test
   public void testCasParsing6() throws Exception {
      mc.writeln("cas bad 0 0 6");
      mc.flush();
      assertStartsWith(mc.readln(), "SERVER_ERROR");
   }

   @Test
   public void testCasUniqueIs64Bit() throws Exception {
      // casid should be 64 bit value, in our case signed
      // ie. should have range Long.MIN_VALUE .. Long.MAX_VALUE
      mc.writeln("cas a 0 0 1 " + Long.MAX_VALUE);
      mc.writeln("a");
      mc.flush();
      assertEquals("NOT_FOUND", mc.readln());
      mc.writeln("cas a 0 0 1 " + Long.MIN_VALUE);
      mc.writeln("a");
      mc.flush();
      assertEquals("NOT_FOUND", mc.readln());
   }

   @Test
   public void testSetFlagsRange() throws Exception {
      mc.writeln("set a 0 0 1");
      mc.writeln("a");
      mc.flush();
      assertEquals("STORED", mc.readln());
      mc.writeln("set a 4294967295 0 1");
      mc.writeln("a");
      mc.flush();
      assertEquals("STORED", mc.readln());
      mc.writeln("set a -1 0 1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
      mc.writeln("set a 4294967296 0 1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testVerbosityUnsupported() throws Exception {
      mc.writeln("verbosity 0");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testQuit() throws Exception {
      mc.writeln("quit");
      mc.flush();
      assertEquals("", mc.readln());
   }

   @Test
   public void testFlushAll() throws Exception {
      mc.set(KEY_A, "valA");
      mc.set(KEY_B, "valB");
      mc.set(KEY_C, "valC");
      assertEquals("valA", mc.get(KEY_A));
      assertEquals("valB", mc.get(KEY_B));
      assertEquals("valC", mc.get(KEY_C));
      mc.writeln("flush_all");
      mc.flush();
      assertEquals("OK", mc.readln());
      assertNull(mc.get(KEY_A));
      assertNull(mc.get(KEY_B));
      assertNull(mc.get(KEY_C));
   }

   @Test
   public void testFlushAllDelayed() throws Exception {
      mc.set(KEY_A, "valA");
      mc.set(KEY_B, "valB");
      mc.set(KEY_C, "valC");
      assertEquals("valA", mc.get(KEY_A));
      assertEquals("valB", mc.get(KEY_B));
      assertEquals("valC", mc.get(KEY_C));
      mc.writeln("flush_all 1");
      mc.flush();
      assertEquals("OK", mc.readln());
      assertEquals("valA", mc.get(KEY_A));
      assertEquals("valB", mc.get(KEY_B));
      assertEquals("valC", mc.get(KEY_C));
      ITestUtils.eventually(new ITestUtils.Condition() {
         @Override
         public boolean isSatisfied() throws Exception {
            return mc.get(KEY_A) == null && mc.get(KEY_B) == null && mc.get(KEY_C) == null;
         }
      }, 20000, 40);
   }

   @Test
   public void testFlushAllDelayedUnixTime() throws Exception {
      mc.set(KEY_A, "valA");
      mc.set(KEY_B, "valB");
      mc.set(KEY_C, "valC");
      assertEquals("valA", mc.get(KEY_A));
      assertEquals("valB", mc.get(KEY_B));
      assertEquals("valC", mc.get(KEY_C));
      long t = mc.getServerTime();
      assertTrue(t > 0);
      mc.writeln("flush_all " + (t + 2));
      mc.flush();
      assertEquals("OK", mc.readln());
      assertEquals("valA", mc.get(KEY_A));
      assertEquals("valB", mc.get(KEY_B));
      assertEquals("valC", mc.get(KEY_C));
      ITestUtils.eventually(new ITestUtils.Condition() {
         @Override
         public boolean isSatisfied() throws Exception {
            return mc.get(KEY_A) == null && mc.get(KEY_B) == null && mc.get(KEY_C) == null;
         }
      }, 20000, 40);
   }

   @Test
   public void testPipeliningSet() throws Exception {
      mc.writeln("set " + KEY_A + " 0 0 1");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningAdd() throws Exception {
      mc.writeln("add " + KEY_A + " 0 0 1");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningReplace() throws Exception {
      mc.set(KEY_A, "b");
      mc.writeln("replace " + KEY_A + " 0 0 1");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningAppend() throws Exception {
      mc.set(KEY_A, "a");
      mc.writeln("append " + KEY_A + " 0 0 1");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningPrepend() throws Exception {
      mc.set(KEY_A, "a");
      mc.writeln("prepend " + KEY_A + " 0 0 1");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("STORED", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningCas() throws Exception {
      mc.writeln("cas " + KEY_A + " 0 0 1 0");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("NOT_FOUND", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningGet() throws Exception {
      mc.writeln("get " + KEY_A);
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningGetMulti() throws Exception {
      mc.writeln("get " + KEY_A + " " + KEY_C);
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningGets() throws Exception {
      mc.writeln("gets " + KEY_A);
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningGetsMulti() throws Exception {
      mc.writeln("gets " + KEY_A + " " + KEY_C);
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningDelete() throws Exception {
      mc.writeln("delete " + KEY_A);
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("NOT_FOUND", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningIncr() throws Exception {
      mc.set(KEY_A, "1");
      mc.writeln("incr " + KEY_A + " 1");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("2", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningDecr() throws Exception {
      mc.set(KEY_A, "2");
      mc.writeln("decr " + KEY_A + " 1");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("1", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test(timeout = 5000)
   public void testPipeliningStats() throws Exception {
      mc.writeln("stats");
      mc.writeln("get " + KEY_B);
      mc.flush();
      String line = null;
      do {
         line = mc.readln();
      } while (!line.equals("END"));
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningVersion() throws Exception {
      mc.writeln("version");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertNotNull(mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningVerbosity() throws Exception {
      mc.writeln("verbosity 0");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningFlushAll() throws Exception {
      mc.writeln("flush_all");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("OK", mc.readln());
      assertEquals("END", mc.readln());
   }

   @Test
   public void testPipeliningFlushAllDelayed() throws Exception {
      mc.writeln("flush_all 1");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("OK", mc.readln());
      assertEquals("END", mc.readln());
      mc.set(KEY_A, "thisWillBeFlushed");
      assertEquals("thisWillBeFlushed", mc.get(KEY_A));
      ITestUtils.eventually(new ITestUtils.Condition() {
         @Override
         public boolean isSatisfied() throws Exception {
            return mc.get(KEY_A) == null;
         }
      }, 20000, 40);
   }

   @Test
   public void testStatsArgs() throws Exception {
      mc.writeln("stats args");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testNegativeItemSize() throws Exception {
      mc.writeln("set a 0 0 -1");
      mc.flush();
      assertStartsWith(mc.readln(), "CLIENT_ERROR");
   }

   @Test
   public void testNoReplySet() throws Exception {
      mc.writeln("set " + KEY_A + " 0 0 1 noreply");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testNoReplyAdd() throws Exception {
      mc.writeln("add " + KEY_A + " 0 0 1 noreply");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testNoReplyReplace() throws Exception {
      mc.set(KEY_A, "b");
      mc.writeln("replace " + KEY_A + " 0 0 1 noreply");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testNoReplyAppend() throws Exception {
      mc.set(KEY_A, "a");
      mc.writeln("append " + KEY_A + " 0 0 1 noreply");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testNoReplyPrepend() throws Exception {
      mc.set(KEY_A, "a");
      mc.writeln("prepend " + KEY_A + " 0 0 1 noreply");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testNoReplyCas() throws Exception {
      mc.writeln("cas " + KEY_A + " 0 0 1 0 noreply");
      mc.writeln("a");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testNoReplyDelete() throws Exception {
      mc.writeln("delete " + KEY_A + " noreply");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testNoReplyIncr() throws Exception {
      mc.set(KEY_A, "1");
      mc.writeln("incr " + KEY_A + " 1 noreply");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testNoReplyDecr() throws Exception {
      mc.set(KEY_A, "2");
      mc.writeln("decr " + KEY_A + " 1 noreply");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testNoReplyFlushAll() throws Exception {
      mc.writeln("flush_all noreply");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
   }

   @Test
   public void testNoReplyFlushAllDelayed() throws Exception {
      mc.writeln("flush_all 1 noreply");
      mc.writeln("get " + KEY_B);
      mc.flush();
      assertEquals("END", mc.readln());
      mc.set(KEY_A, "thisWillBeFlushed");
      assertEquals("thisWillBeFlushed", mc.get(KEY_A));
      ITestUtils.eventually(new ITestUtils.Condition() {
         @Override
         public boolean isSatisfied() throws Exception {
            return mc.get(KEY_A) == null;
         }
      }, 20000, 40);
   }

   public static SimpleMemcachedClient createMemcachedClient() {
      DefaultConnectionFactory d = new DefaultConnectionFactory() {
         @Override
         public long getOperationTimeout() {
            return 60_000;
         }
      };
      try {
         return new SimpleMemcachedClient(ENCODING, "127.0.0.1", 11221, 10000);
      } catch (IOException e) {
         throw new IllegalStateException("Cannot create MemcachedClient", e);
      }
   }

   public static class TestSerializable implements Serializable {
      private String content;

      public TestSerializable(String content) {
         super();
         this.content = content;
      }

      public String getContent() {
         return content;
      }
   }

   private void assertStartsWith(String str, String prefix) {
      assertTrue("String \"" + str + "\" doesn't start with expected prefix \"" + prefix + "\"", str.startsWith(prefix));
   }
}
