Capture and verify logging in tests

5 files changed, 59 insertions, 7 deletions

diff --git a/setup.cfg b/setup.cfg
index 4eee543..daf1ac7 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -3,5 +3,8 @@ release = 1
 packager = Sean Reifschneider <jafo-rpms@tummy.com>
 requires = python-memcached
 
+[flake8]
+ignore = H304,H405
+
 [wheel]
 universal = 1
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/tests/__init__.py
diff --git a/tests/test_memcache.py b/tests/test_memcache.py
index 0072813..6408e81 100644
--- a/tests/test_memcache.py
+++ b/tests/test_memcache.py
@@ -5,6 +5,7 @@ import unittest
 import six
 
 from memcache import Client, SERVER_MAX_KEY_LENGTH, SERVER_MAX_VALUE_LENGTH  # noqa: H301
+from .utils import captured_stderr
 
 
 class FooStruct(object):
@@ -135,7 +136,13 @@ class TestMemcache(unittest.TestCase):
         self.assertEqual(self.mc.get(key), value)
 
         value = 'a' * SERVER_MAX_VALUE_LENGTH
-        self.assertFalse(self.mc.set(key, value))
+        with captured_stderr() as log:
+            self.assertIs(self.mc.set(key, value), False)
+        self.assertEqual(
+            log.getvalue(),
+            "MemCached: while expecting 'STORED', got unexpected response "
+            "'SERVER_ERROR object too large for cache'\n"
+        )
         # This test fails if the -I option is used on the memcached server
         self.assertTrue(self.mc.get(key) is None)
 
@@ -155,16 +162,26 @@ class TestMemcache(unittest.TestCase):
         """Testing set_multi() with no memcacheds running."""
 
         self.mc.disconnect_all()
-        for server in self.mc.servers:
-            server.mark_dead('test')
+        with captured_stderr() as log:
+            for server in self.mc.servers:
+                server.mark_dead('test')
+        self.assertIn('Marking dead.', log.getvalue())
         errors = self.mc.set_multi({'key1': 'a', 'key2': 'b'})
         self.assertEqual(sorted(errors), ['key1', 'key2'])
 
     def test_disconnect_all_delete_multi(self):
         """Testing delete_multi() with no memcacheds running."""
         self.mc.disconnect_all()
-        ret = self.mc.delete_multi({'keyhere': 'a', 'keythere': 'b'})
+        with captured_stderr() as output:
+            ret = self.mc.delete_multi({'keyhere': 'a', 'keythere': 'b'})
         self.assertEqual(ret, 1)
+        self.assertEqual(
+            output.getvalue(),
+            "MemCached: while expecting 'DELETED', got unexpected response "
+            "'NOT_FOUND'\n"
+            "MemCached: while expecting 'DELETED', got unexpected response "
+            "'NOT_FOUND'\n"
+        )
 
 
 if __name__ == '__main__':
diff --git a/tests/test_setmulti.py b/tests/test_setmulti.py
index 3cc1591..756afe8 100644
--- a/tests/test_setmulti.py
+++ b/tests/test_setmulti.py
@@ -13,6 +13,8 @@ import socket
 import sys
 import unittest
 
+from .utils import captured_stderr
+
 sys.path.append('..')
 import memcache    # noqa: E402
 
@@ -60,10 +62,10 @@ class test_Memcached_Set_Multi(unittest.TestCase):
 
     def test_Socket_Disconnect(self):
         mapping = {'foo': 'FOO', 'bar': 'BAR'}
-        bad_keys = self.mc.set_multi(mapping)
-
+        with captured_stderr() as log:
+            bad_keys = self.mc.set_multi(mapping)
+        self.assertIn('connection closed in readline().', log.getvalue())
         self.assertEqual(sorted(bad_keys), ['bar', 'foo'])
-
         if DEBUG:
             print('set_multi({0!r}) -> {1!r}'.format(mapping, bad_keys))
 
diff --git a/tests/utils.py b/tests/utils.py
new file mode 100644
index 0000000..a10a795
--- /dev/null
+++ b/tests/utils.py
@@ -0,0 +1,30 @@
+from contextlib import contextmanager
+import sys
+
+from six import StringIO
+
+
+@contextmanager
+def captured_output(stream_name):
+    """Return a context manager used by captured_stdout/stdin/stderr
+    that temporarily replaces the sys stream *stream_name* with a StringIO.
+
+    This function and the following ``captured_std*`` are copied
+    from CPython's ``test.support`` module.
+    """
+    orig_stdout = getattr(sys, stream_name)
+    setattr(sys, stream_name, StringIO())
+    try:
+        yield getattr(sys, stream_name)
+    finally:
+        setattr(sys, stream_name, orig_stdout)
+
+
+def captured_stderr():
+    """Capture the output of sys.stderr:
+
+       with captured_stderr() as stderr:
+           print('hello', file=sys.stderr)
+       self.assertEqual(stderr.getvalue(), 'hello\n')
+    """
+    return captured_output('stderr')