Merge pull request #60 from jerith/split-out-tests

Better test structure

2 files changed, 203 insertions, 1 deletions

diff --git a/.travis.yml b/.travis.yml
index ef41904..fe3c745 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -6,7 +6,13 @@ python:
   - 3.3
   - 3.4
   - pypy
-  - 3.3
+matrix:
+  allow_failures:
+    - python: 3.2
+    - python: 3.3
+    - python: 3.4
+services:
+  - memcached
 install: python setup.py install
 before_script: pip install nose
 script: nosetests
diff --git a/tests/test_memcache.py b/tests/test_memcache.py
new file mode 100644
index 0000000..c12c528
--- /dev/null
+++ b/tests/test_memcache.py
@@ -0,0 +1,196 @@
+from __future__ import print_function
+
+from unittest import TestCase
+
+from memcache import Client, SERVER_MAX_KEY_LENGTH
+
+try:
+    _str_cls = basestring
+except NameError:
+    _str_cls = str
+
+
+def to_s(val):
+    if not isinstance(val, _str_cls):
+        return "%s (%s)" % (val, type(val))
+    return "%s" % val
+
+
+class FooStruct(object):
+
+    def __init__(self):
+        self.bar = "baz"
+
+    def __str__(self):
+        return "A FooStruct"
+
+    def __eq__(self, other):
+        if isinstance(other, FooStruct):
+            return self.bar == other.bar
+        return 0
+
+
+class TestMemcache(TestCase):
+    def setUp(self):
+        # TODO: unix socket server stuff
+        servers = ["127.0.0.1:11211"]
+        self.mc = Client(servers, debug=1)
+        pass
+
+    def check_setget(self, key, val, noreply=False):
+        self.mc.set(key, val, noreply=noreply)
+        newval = self.mc.get(key)
+        self.assertEqual(newval, val)
+
+    def test_setget(self):
+        self.check_setget("a_string", "some random string")
+        self.check_setget("a_string_2", "some random string", noreply=True)
+        self.check_setget("an_integer", 42)
+        self.check_setget("an_integer_2", 42, noreply=True)
+
+    def test_delete(self):
+        self.check_setget("long", int(1 << 30))
+        result = self.mc.delete("long")
+        self.assertEqual(result, True)
+        self.assertEqual(self.mc.get("long"), None)
+
+    def test_get_multi(self):
+        self.check_setget("gm_a_string", "some random string")
+        self.check_setget("gm_an_integer", 42)
+        self.assertEqual(
+            self.mc.get_multi(["gm_a_string", "gm_an_integer"]),
+            {"gm_an_integer": 42, "gm_a_string": "some random string"})
+
+    def test_get_unknown_value(self):
+        self.assertEqual(self.mc.get("unknown_value"), None)
+
+    def test_setget_foostruct(self):
+        f = FooStruct()
+        self.check_setget("foostruct", f)
+        self.check_setget("foostruct_2", f, noreply=True)
+
+    def test_incr(self):
+        self.check_setget("i_an_integer", 42)
+        self.assertEqual(self.mc.incr("i_an_integer", 1), 43)
+
+    def test_incr_noreply(self):
+        self.check_setget("i_an_integer_2", 42)
+        self.assertEqual(self.mc.incr("i_an_integer_2", 1, noreply=True), None)
+        self.assertEqual(self.mc.get("i_an_integer_2"), 43)
+
+    def test_decr(self):
+        self.check_setget("i_an_integer", 42)
+        self.assertEqual(self.mc.decr("i_an_integer", 1), 41)
+
+    def test_decr_noreply(self):
+        self.check_setget("i_an_integer_2", 42)
+        self.assertEqual(self.mc.decr("i_an_integer_2", 1, noreply=True), None)
+        self.assertEqual(self.mc.get("i_an_integer_2"), 41)
+
+    def test_sending_spaces(self):
+        try:
+            self.mc.set("this has spaces", 1)
+        except Client.MemcachedKeyCharacterError as err:
+            self.assertTrue("characters not allowed" in err.args[0])
+        else:
+            self.fail(
+                "Expected Client.MemcachedKeyCharacterError, nothing raised")
+
+    def test_sending_control_characters(self):
+        try:
+            self.mc.set("this\x10has\x11control characters\x02", 1)
+        except Client.MemcachedKeyCharacterError as err:
+            self.assertTrue("characters not allowed" in err.args[0])
+        else:
+            self.fail(
+                "Expected Client.MemcachedKeyCharacterError, nothing raised")
+
+    def test_sending_key_too_long(self):
+        try:
+            self.mc.set('a' * SERVER_MAX_KEY_LENGTH + 'a', 1)
+        except Client.MemcachedKeyLengthError as err:
+            self.assertTrue("length is >" in err.args[0])
+        else:
+            self.fail(
+                "Expected Client.MemcachedKeyLengthError, nothing raised")
+
+        # These should work.
+        self.mc.set('a' * SERVER_MAX_KEY_LENGTH, 1)
+        self.mc.set('a' * SERVER_MAX_KEY_LENGTH, 1, noreply=True)
+
+
+if __name__ == "__main__":
+    # failures = 0
+    # print("Testing docstrings...")
+    # _doctest()
+    # print("Running tests:")
+    # print()
+    # serverList = [["127.0.0.1:11211"]]
+    # if '--do-unix' in sys.argv:
+    #     serverList.append([os.path.join(os.getcwd(), 'memcached.socket')])
+
+    # for servers in serverList:
+    #     mc = Client(servers, debug=1)
+    if False:
+
+        print("Testing sending a unicode-string key...", end=" ")
+        try:
+            x = mc.set(u'keyhere', 1)
+        except Client.MemcachedStringEncodingError as msg:
+            print("OK", end=" ")
+        else:
+            print("FAIL", end=" ")
+            failures += 1
+        try:
+            x = mc.set((u'a'*SERVER_MAX_KEY_LENGTH).encode('utf-8'), 1)
+        except Client.MemcachedKeyError:
+            print("FAIL", end=" ")
+            failures += 1
+        else:
+            print("OK", end=" ")
+        s = pickle.loads('V\\u4f1a\np0\n.')
+        try:
+            x = mc.set((s * SERVER_MAX_KEY_LENGTH).encode('utf-8'), 1)
+        except Client.MemcachedKeyLengthError:
+            print("OK")
+        else:
+            print("FAIL")
+            failures += 1
+
+        print("Testing using a value larger than the memcached value limit...")
+        print('NOTE: "MemCached: while expecting[...]" is normal...')
+        x = mc.set('keyhere', 'a'*SERVER_MAX_VALUE_LENGTH)
+        if mc.get('keyhere') is None:
+            print("OK", end=" ")
+        else:
+            print("FAIL", end=" ")
+            failures += 1
+        x = mc.set('keyhere', 'a'*SERVER_MAX_VALUE_LENGTH + 'aaa')
+        if mc.get('keyhere') is None:
+            print("OK")
+        else:
+            print("FAIL")
+            failures += 1
+
+        print("Testing set_multi() with no memcacheds running", end=" ")
+        mc.disconnect_all()
+        errors = mc.set_multi({'keyhere': 'a', 'keythere': 'b'})
+        if errors != []:
+            print("FAIL")
+            failures += 1
+        else:
+            print("OK")
+
+        print("Testing delete_multi() with no memcacheds running", end=" ")
+        mc.disconnect_all()
+        ret = mc.delete_multi({'keyhere': 'a', 'keythere': 'b'})
+        if ret != 1:
+            print("FAIL")
+            failures += 1
+        else:
+            print("OK")
+
+    if failures > 0:
+        print('*** THERE WERE FAILED TESTS')
+        sys.exit(1)
+    sys.exit(0)