r"""Tests. They want YOU!! Basic functionality. >>> _pylibmc.__version__ == pylibmc.__version__ True >>> c = _pylibmc.client([test_server]) >>> c.get("hello") >>> c.set("test_key", 123) True >>> c.get("test_key") 123 >>> c.get("test_key_2") >>> c.delete("test_key") True >>> c.get("test_key") >>> We should handle empty values just nicely. >>> c.set("foo", "") True >>> c.get("foo") '' Now this section is because most other implementations ignore zero-keys. >>> c.get("") >>> c.set("", "hi") False >>> c.delete("") False >>> Multi functionality. >>> c.set_multi({"a": 1, "b": 2, "c": 3}) [] >>> c.get_multi("abc").keys() == ["a", "c", "b"] True >>> c.delete_multi("abc") True >>> c.get_multi("abc").keys() == [] True >>> c.set_multi(dict(zip("abc", "def")), key_prefix="test_") [] >>> list(sorted(c.get_multi("abc", key_prefix="test_").iteritems())) [('a', 'd'), ('b', 'e'), ('c', 'f')] >>> c.get("test_a") 'd' >>> c.delete_multi("abc", key_prefix="test_") True >>> bool(c.get_multi("abc", key_prefix="test_")) False Zero-key-test-time! >>> c.get_multi([""]) {} >>> c.delete_multi([""]) False >>> c.set_multi({"": "hi"}) [''] >>> c.delete_multi({"a": "b"}) Traceback (most recent call last): ... TypeError: keys must be a sequence, not a mapping Timed stuff. The reason we, at UNIX times, set it two seconds in the future and then sleep for >3 is that memcached might round the time up and down and left and yeah, so you know... >>> from time import sleep, time >>> c.set("hi", "steven", 1) True >>> c.get("hi") 'steven' >>> sleep(2.1) >>> c.get("hi") >>> c.set("hi", "loretta", int(time()) + 2) True >>> c.set_multi({"hi2": "charlotta"}, 1) [] >>> c.get("hi") 'loretta' >>> c.get("hi2") 'charlotta' >>> sleep(3.1) >>> c.get("hi"), c.get("hi2") (None, None) See issue #1 ``http://github.com/lericson/pylibmc/issues/#issue/1`` -- delete should not accept a time argument. >>> c.delete("foo", 123) Traceback (most recent call last): ... TypeError: delete() takes exactly 1 argument (2 given) >>> c.delete_multi(["foo"], time=123) Traceback (most recent call last): ... TypeError: 'time' is an invalid keyword argument for this function Now for keys with funny types. >>> c.set(1, "hi") Traceback (most recent call last): ... TypeError: argument 1 must be string, not int >>> c.get(1) Traceback (most recent call last): ... TypeError: key must be an instance of str >>> c.set_multi({1: True}) Traceback (most recent call last): ... TypeError: key must be an instance of str >>> c.get_multi([1, 2]) Traceback (most recent call last): ... TypeError: key must be an instance of str This didn't use to work, but now it does. >>> c.get_multi([]) {} Getting stats is fun! >>> for (svr, stats) in c.get_stats(): ... print svr ... ks = stats.keys() ... while ks: ... cks, ks = ks[:6], ks[6:] ... print ", ".join(cks) 127.0.0.1:11211 (0) pid, total_items, uptime, version, limit_maxbytes, rusage_user bytes_read, rusage_system, cmd_get, curr_connections, threads, total_connections cmd_set, curr_items, get_misses, evictions, bytes, connection_structures bytes_written, time, pointer_size, get_hits Also test some flush all. >>> c.set("hi", "guys") True >>> c.get("hi") 'guys' >>> c.flush_all() True >>> c.get("hi") >>> Get and set booleans. >>> c.set("greta", True) True >>> c.get("greta") True >>> c.set("greta", False) True >>> c.get("greta") False >>> c.delete("greta") True Complex data types! >>> bla = Foo() >>> bla.bar = "Hello!" >>> c.set("tengil", bla) True >>> c.get("tengil").bar == bla.bar True Cloning (ethically, I don't know about it.) >>> c is not c.clone() True >>> c2 = c.clone() >>> c.set("test", "hello") True >>> c2.get("test") 'hello' >>> c2.delete("test") True >>> del c2 Per-error exceptions >>> c.incr("test") Traceback (most recent call last): ... NotFound: error 16 from memcached_increment: NOT FOUND >>> c.incr(chr(0)) Traceback (most recent call last): ... ProtocolError: error 8 from memcached_increment: PROTOCOL ERROR Behaviors. >>> c.set_behaviors({"tcp_nodelay": True, "hash": 6}) >>> list(sorted((k, v) for (k, v) in c.get_behaviors().items() ... if k in ("tcp_nodelay", "hash"))) [('hash', 6), ('tcp_nodelay', 1)] Binary protocol! >>> c = _pylibmc.client([test_server], binary=True) >>> c.set("hello", "world") True >>> c.get("hello") 'world' >>> c.delete("hello") True incr_multi >>> c.add_multi({'a': 1, 'b': 0, 'c': 4}) [] >>> c.incr_multi(('a', 'b', 'c'), delta=1) >>> list(sorted(c.get_multi(('a', 'b', 'c')).items())) [('a', 2), ('b', 1), ('c', 5)] >>> c.delete_multi(('a', 'b', 'c')) True >>> c.add_multi({'a': 1, 'b': 0, 'c': 4}, key_prefix='x') [] >>> c.incr_multi(('a', 'b', 'c'), key_prefix='x', delta=5) >>> list(sorted(c.get_multi(('a', 'b', 'c'), key_prefix='x').items())) [('a', 6), ('b', 5), ('c', 9)] >>> c.delete_multi(('a', 'b', 'c'), key_prefix='x') True >>> c.add('a', 1) True >>> c.incr_multi(('a', 'b', 'c'), key_prefix='x', delta=1) Traceback (most recent call last): ... NotFound: error 16 from memcached_increment: NOT FOUND >>> c.delete('xa') False Empty server lists are bad for your health. >>> c = _pylibmc.client([]) Traceback (most recent call last): ... MemcachedError: empty server list Python-wrapped behaviors dict >>> pc = pylibmc.Client(["%s:%d" % test_server[1:]]) >>> (pc.behaviors["hash"], pc.behaviors["distribution"]) ('default', 'modula') >>> pc.behaviors.update({"hash": "fnv1a_32", "distribution": "consistent"}) >>> (pc.behaviors["hash"], pc.behaviors["distribution"]) ('fnv1a_32', 'consistent') >>> pc = pylibmc.Client(["%s:%d" % test_server[1:]]) >>> b = pc.behaviors >>> ks = list(sorted(k for k in b.keys() if not k.startswith("_"))) >>> ks # doctest: +NORMALIZE_WHITESPACE ['buffer_requests', 'cache_lookups', 'cas', 'connect_timeout', 'distribution', 'failure_limit', 'hash', 'ketama', 'ketama_hash', 'ketama_weighted', 'no_block', 'receive_timeout', 'send_timeout', 'tcp_nodelay', 'verify_keys'] >>> b["hash"] 'default' >>> b["hash"] = 'fnv1a_32' >>> pc.behaviors["hash"] 'fnv1a_32' >>> super(pylibmc.Client, pc).get_behaviors()["hash"] 6 Ormod's Zero-byte Adventure Story >>> bc = _pylibmc.client([test_server], binary=True) >>> bc.set("\0\0", "ORMOD") True >>> bc.get_multi(["\0\0"]) {'\x00\x00': 'ORMOD'} Test server/client max length >>> mc.get('x'*250) >>> mc.get('x'*251) Traceback (most recent call last): ... ValueError: key too long, max is 250 Test CAS >>> mc.behaviors['cas'] = True >>> mc.delete('foo') and False False >>> mc.gets('foo') (None, None) >>> mc.set('foo', 'bar') True >>> foostr, cas = mc.gets('foo') >>> foostr 'bar' >>> mc.cas('foo', 'quux', cas+1) False >>> mc.cas('foo', 'baz', cas) True >>> mc.get('foo') 'baz' >>> mc.behaviors['cas'] = False >>> mc.gets('foo') Traceback (most recent call last): ... ValueError: gets without cas behavior >>> mc.cas('foo', 'bar', 1) Traceback (most recent call last): ... ValueError: cas without cas behavior """ # Used to test pickling. class Foo(object): pass # Fix up sys.path so as to include the correct build/lib.*/ directory. import sys from distutils.dist import Distribution from distutils.command.build import build build_cmd = build(Distribution({"ext_modules": True})) build_cmd.finalize_options() lib_dirn = build_cmd.build_lib sys.path.insert(0, lib_dirn) import pylibmc, _pylibmc import socket __doc__ = pylibmc.__doc__ + "\n\n" + __doc__ # {{{ Ported cmemcache tests import unittest class TestCmemcached(unittest.TestCase): def setUp(self): self.mc = pylibmc.Client(["%s:%d" % (test_server[1:])]) def testSetAndGet(self): self.mc.set("num12345", 12345) self.assertEqual(self.mc.get("num12345"), 12345) self.mc.set("str12345", "12345") self.assertEqual(self.mc.get("str12345"), "12345") def testDelete(self): self.mc.set("str12345", "12345") #delete return True on success, otherwise False ret = self.mc.delete("str12345") self.assertEqual(self.mc.get("str12345"), None) self.assertEqual(ret, True) # This test only works with old memcacheds. This has become a "client # error" in memcached. try: ret = self.mc.delete("hello world") except pylibmc.ClientError: pass else: self.assertEqual(ret, False) def testGetMulti(self): self.mc.set("a", "valueA") self.mc.set("b", "valueB") self.mc.set("c", "valueC") result = self.mc.get_multi(["a", "b", "c", "", "hello world"]) self.assertEqual(result, {'a':'valueA', 'b':'valueB', 'c':'valueC'}) def testBigGetMulti(self): count = 10 keys = ['key%d' % i for i in xrange(count)] pairs = zip(keys, ['value%d' % i for i in xrange(count)]) for key, value in pairs: self.mc.set(key, value) result = self.mc.get_multi(keys) assert result == dict(pairs) def testFunnyDelete(self): result= self.mc.delete("") self.assertEqual(result, False) def testAppend(self): self.mc.delete("a") self.mc.set("a", "I ") ret = self.mc.append("a", "Do") result = self.mc.get("a") self.assertEqual(result, "I Do") def testPrepend(self): self.mc.delete("a") self.mc.set("a", "Do") ret = self.mc.prepend("a", "I ") result = self.mc.get("a") self.assertEqual(result, "I Do") # }}} from os import environ test_server = ( _pylibmc.server_type_tcp, str(environ.get("MEMCACHED_HOST", "127.0.0.1")), int(environ.get("MEMCACHED_PORT", "11211"))) def get_version(addr): (type_, host, port) = addr if (type_ != _pylibmc.server_type_tcp): raise NotImplementedError("test server can only be on tcp for now") else: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect((host, port)) s.send("version\r\n") version = s.recv(4096) s.close() if not version.startswith("VERSION ") or not version.endswith("\r\n"): raise ValueError("unexpected version return: %r" % (version,)) else: version = version[8:-2] return version def is_alive(addr): try: return bool(get_version(addr)) except (ValueError, socket.error): return False def make_full_suite(): from doctest import DocTestFinder, DocTestSuite suite = unittest.TestSuite() loader = unittest.TestLoader() finder = DocTestFinder() for modname in (__name__, "pylibmc", "_pylibmc"): ss = (DocTestSuite(sys.modules[modname], test_finder=finder), loader.loadTestsFromName(modname)) for subsuite in ss: map(suite.addTest, subsuite._tests) return suite class _TextTestResult(unittest._TextTestResult): def startTest(self, test): # Treat doctests a little specially because we want each example to # count as a test. if hasattr(test, "_dt_test"): self.testsRun += len(test._dt_test.examples) - 1 elif hasattr(test, "countTestCases"): self.testsRun += test.countTestCases() - 1 return unittest._TextTestResult.startTest(self, test) class TextTestRunner(unittest.TextTestRunner): def _makeResult(self): return _TextTestResult(self.stream, self.descriptions, self.verbosity) class TestProgram(unittest.TestProgram): defaultTest = "make_full_suite" testRunner = TextTestRunner def __init__(self, module="__main__", argv=None, testLoader=unittest.defaultTestLoader): super(TestProgram, self).__init__( module=module, argv=argv, testLoader=testLoader, defaultTest=self.defaultTest, testRunner=self.testRunner) def runTests(self): pass def makeRunner(self): return self.testRunner(verbosity=self.verbosity) def run(self): runner = self.makeRunner() return runner.run(self.test) if __name__ == "__main__": if hasattr(_pylibmc, "__file__"): print "Starting tests with _pylibmc at", _pylibmc.__file__ else: print "Starting tests with static _pylibmc:", _pylibmc print "Reported libmemcached version:", _pylibmc.libmemcached_version print "Reported pylibmc version:", _pylibmc.__version__ print "Support compression:", _pylibmc.support_compression if not is_alive(test_server): raise SystemExit("Test server (%r) not alive." % (test_server,)) res = TestProgram().run() if not res.wasSuccessful(): sys.exit(1)