def test_zerokey(self):
     bc = make_test_client(binary=True)
     k = "\x00\x01"
     test_str = "test"
     ok_(bc.set(k, test_str))
     rk = next(iter(bc.get_multi([k])))
     eq_(k, rk)

 def test_get_with_default(self):
     mc = make_test_client(binary=True)
     key = 'get-api-test'
     mc.delete(key)
     eq_(mc.get(key), None)
     default = object()
     assert mc.get(key, default) is default

 def setUp(self):
     self.mc = make_test_client(
         cls=self.memcached_client,
         server_type=self.memcached_server_type,
         host=self.memcached_host,
         port=self.memcached_port,
     )

 def _test_get(self, key, val):
     bc = make_test_client(binary=True)
     refcountables = [key, val]
     initial_refcounts = get_refcounts(refcountables)
     bc.set(key, val)
     eq_(get_refcounts(refcountables), initial_refcounts)
     eq_(bc.get(key), val)
     eq_(get_refcounts(refcountables), initial_refcounts)

 def test_none_values(self):
     mc = make_test_client(binary=True)
     mc.set('none-test', None)
     self.assertEqual(mc.get('none-test'), None)
     self.assertEqual(mc.get('none-test', 'default'), None)
     # formerly, this would raise a KeyError, which was incorrect
     self.assertEqual(mc['none-test'], None)
     assert 'none-test' in mc

 def test_cas(self):
     k = "testkey"
     mc = make_test_client(binary=False, behaviors={"cas": True})
     ok_(mc.set(k, 0))
     while True:
         rv, cas = mc.gets(k)
         ok_(mc.cas(k, rv + 1, cas))
         if rv == 10:
             break

 def test_get_multi(self):
     bc = make_test_client(binary=True)
     keys = ["first", "second"]
     value = "first_value"
     refcountables = keys + [value]
     initial_refcounts = get_refcounts(refcountables)
     bc.set(keys[0], value)
     eq_(get_refcounts(refcountables), initial_refcounts)
     eq_(bc.get_multi(keys), {keys[0]: value})
     eq_(get_refcounts(refcountables), initial_refcounts)

    def test_override_deserialize(self):
        class MyClient(pylibmc.Client):
            ignored = []

            def deserialize(self, bytes_, flags):
                try:
                    return super(MyClient, self).deserialize(bytes_, flags)
                except Exception as error:
                    self.ignored.append(error)
                    raise pylibmc.CacheMiss

        global MyObject  # Needed by the pickling system.

        class MyObject(object):
            def __getstate__(self):
                return dict(a=1)

            def __eq__(self, other):
                return type(other) is type(self)

            def __setstate__(self, d):
                assert d["a"] == 1

        c = make_test_client(MyClient, behaviors={"cas": True})
        eq_(c.get("notathing"), None)

        refcountables = ["foo", "myobj", "noneobj", "myobj2", "cachemiss"]
        initial_refcounts = get_refcounts(refcountables)

        c["foo"] = "foo"
        c["myobj"] = MyObject()
        c["noneobj"] = None
        c["myobj2"] = MyObject()

        # Show that everything is initially regular.
        eq_(c.get("myobj"), MyObject())
        eq_(get_refcounts(refcountables), initial_refcounts)
        eq_(c.get_multi(["foo", "myobj", "noneobj", "cachemiss"]), dict(foo="foo", myobj=MyObject(), noneobj=None))
        eq_(get_refcounts(refcountables), initial_refcounts)
        eq_(c.gets("myobj2")[0], MyObject())
        eq_(get_refcounts(refcountables), initial_refcounts)

        # Show that the subclass can transform unpickling issues into a cache miss.
        del MyObject  # Break unpickling

        eq_(c.get("myobj"), None)
        eq_(get_refcounts(refcountables), initial_refcounts)
        eq_(c.get_multi(["foo", "myobj", "noneobj", "cachemiss"]), dict(foo="foo", noneobj=None))
        eq_(get_refcounts(refcountables), initial_refcounts)
        eq_(c.gets("myobj2"), (None, None))
        eq_(get_refcounts(refcountables), initial_refcounts)

        # The ignored errors are "AttributeError: test.test_client has no MyObject"
        eq_(len(MyClient.ignored), 3)
        assert all(isinstance(error, AttributeError) for error in MyClient.ignored)

 def test_get_multi_bytes_and_unicode(self):
     bc = make_test_client(binary=True)
     keys = ["third", b"fourth"]
     value = "another_value"
     kv = dict((k, value) for k in keys)
     refcountables = [keys] + [value]
     initial_refcounts = get_refcounts(refcountables)
     bc.set_multi(kv)
     eq_(get_refcounts(refcountables), initial_refcounts)
     eq_(bc.get_multi(keys)[keys[0]], value)
     eq_(get_refcounts(refcountables), initial_refcounts)

 def test_get_invalid_key(self):
     bc = make_test_client(binary=True)
     key = object()
     initial_refcount = sys.getrefcount(key)
     raised = False
     try:
         bc.get(key)
     except TypeError:
         raised = True
     assert raised
     eq_(sys.getrefcount(key), initial_refcount)

    def test_incr(self):
        bc = make_test_client(binary=True)
        keys = [b"increment_key", "increment_key_again"]
        refcountables = keys
        initial_refcounts = get_refcounts(refcountables)

        bc.set(keys[0], 1)
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.incr(keys[0])
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.set(keys[1], 5)
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.incr(keys[1])
        eq_(get_refcounts(refcountables), initial_refcounts)

 def test_zerokey(self):
     bc = make_test_client(binary=True)
     k = "\x00\x01"
     test_str = "test"
     ok_(bc.set(k, test_str))
     if PY3:
         rk = list(bc.get_multi([k]).keys())[0]
         # Keys are converted to UTF-8 strings before being
         # used, so the key that we get back will be a byte
         # string. Encode the test key to match this.
         k = k.encode('utf-8')
     else:
         rk = bc.get_multi([k]).keys()[0]
     eq_(k, rk)

    def test_cas(self):
        k = "testkey"
        val = 1138478589238
        mc = make_test_client(binary=False, behaviors={'cas': True})
        refcountables = [k, val]
        initial_refcounts = get_refcounts(refcountables)

        ok_(mc.set(k, 0))
        eq_(get_refcounts(refcountables), initial_refcounts)
        while True:
            rv, cas = mc.gets(k)
            eq_(get_refcounts(refcountables), initial_refcounts)
            ok_(mc.cas(k, rv + 1, cas))
            eq_(get_refcounts(refcountables), initial_refcounts)
            if rv == 10:
                break

    def test_nonintegers(self):
        # tuples (python_value, (expected_bytestring, expected_flags))
        SERIALIZATION_TEST_VALUES = [
            # booleans
            (True, (b"1", 16)),
            (False, (b"0", 16)),
            # bytestrings
            (b"asdf", (b"asdf", 0)),
            (b"\xb5\xb1\xbf\xed\xa9\xc2{8", (b"\xb5\xb1\xbf\xed\xa9\xc2{8", 0)),
            # objects
            (datetime.date(2015, 12, 28), (pickle.dumps(datetime.date(2015, 12, 28)), 1)),
        ]

        c = make_test_client(binary=True)
        for value, serialized_value in SERIALIZATION_TEST_VALUES:
            eq_(c.serialize(value), serialized_value)
            eq_(c.deserialize(*serialized_value), value)

 def test_get_with_default(self):
     bc = make_test_client(binary=True)
     key = b'refcountest4'
     val = 'some_value'
     default = object()
     refcountables = [key, val, default]
     initial_refcounts = get_refcounts(refcountables)
     bc.set(key, val)
     eq_(get_refcounts(refcountables), initial_refcounts)
     assert bc.get(key) == val
     eq_(get_refcounts(refcountables), initial_refcounts)
     assert bc.get(key, default) == val
     eq_(get_refcounts(refcountables), initial_refcounts)
     bc.delete(key)
     assert bc.get(key) is None
     eq_(get_refcounts(refcountables), initial_refcounts)
     assert bc.get(key, default) is default
     eq_(get_refcounts(refcountables), initial_refcounts)

    def test_set_and_delete_multi(self):
        bc = make_test_client(binary=True)
        keys = ["delone", b"deltwo", "delthree", "delfour"]
        values = [b"valone", "valtwo", object(), 2]
        refcountables = keys + values
        initial_refcounts = get_refcounts(refcountables)

        bc.set_multi(dict(zip(keys, values)))
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.delete_multi([keys[0]])
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.delete_multi([keys[1]])
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.set_multi(dict(zip(keys, values)))
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.delete_multi(keys)
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.delete_multi(keys)
        eq_(get_refcounts(refcountables), initial_refcounts)

    def test_delete(self):
        bc = make_test_client(binary=True)
        keys = ["delone", b"deltwo"]
        values = [b"valone", "valtwo"]
        refcountables = keys + values
        initial_refcounts = get_refcounts(refcountables)

        bc.set(keys[0], values[0])
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.set(keys[1], values[1])
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.delete(keys[0])
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.delete(keys[1])
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.delete(keys[0])
        eq_(get_refcounts(refcountables), initial_refcounts)
        bc.delete(keys[1])
        eq_(get_refcounts(refcountables), initial_refcounts)

    def test_override_serialize(self):
        class MyClient(pylibmc.Client):
            def serialize(self, value):
                return json.dumps(value).encode('utf-8'), 0

            def deserialize(self, bytes_, flags):
                return json.loads(bytes_.decode('utf-8'))

        c = make_test_client(MyClient)
        c['foo'] = (1, 2, 3, 4)
        # json turns tuples into lists:
        eq_(c['foo'], [1, 2, 3, 4])

        raised = False
        try:
            c['bar'] = object()
        except TypeError:
            raised = True
        assert raised

def get_status_memcached(mems_addr=None):
    '''
    Get the status of caches.

    :param mems_addr: list of memcached address IP:PORT.
    '''
    memcacheds = {}

    for mem in mems_addr:

        addr, port = mem.split(':')

        try:
            alive = bool(make_test_client(host=addr, port=port))
            memcacheds[mem] = alive
        except NotAliveError:
            memcacheds[mem] = False

    return memcacheds

    def test_nonintegers(self):
        # tuples (python_value, (expected_bytestring, expected_flags))
        SERIALIZATION_TEST_VALUES = [
            # booleans are just ints
            (True, (b'1', f_int)),
            (False, (b'0', f_int)),
            # bytestrings
            (b'asdf', (b'asdf', f_none)),
            (b'\xb5\xb1\xbf\xed\xa9\xc2{8', (b'\xb5\xb1\xbf\xed\xa9\xc2{8', f_none)),
            (b'', (b'', f_none)),
            # unicode objects
            (u'åäö', (u'åäö'.encode('utf-8'), f_text)),
            (u'', (b'', f_text)),
            # objects
            (datetime.date(2015, 12, 28), (pickle.dumps(datetime.date(2015, 12, 28),
                                                        protocol=-1), f_pickle)),
        ]

        c = make_test_client(binary=True)
        for value, serialized_value in SERIALIZATION_TEST_VALUES:
            eq_(c.serialize(value), serialized_value)
            eq_(c.deserialize(*serialized_value), value)