Merge pull request #150 from tony/clean-test-stdout

Clean test stdout

Author: tony

Makefile

@@ -4,16 +4,22 @@ test:
 watch_test:
 	if command -v entr > /dev/null; then find . -type f -not -path '*/\.*' | grep -i '.*[.]py' | entr -c make test; else make test; echo "\nInstall entr(1) to automatically run tests on file change.\n See http://entrproject.org/"; fi
 
+nose:
+	nosetests tmuxp/testsuite/*.py -m "^test*."
+
+nose_time:
+	nosetests tmuxp/testsuite/*.py -m "^test*." --with-timer --timer-top-n 15
+
 build_docs:
 	cd doc && $(MAKE) html
 
 watch_docs:
 	cd doc && $(MAKE) watch_docs
 
 flake8:
-	        flake8 tmuxp
+	flake8 tmuxp
 
 watch_flake8:
-	        if command -v entr > /dev/null; then find . -type f -not -path '*/\.*' | grep -i '.*[.][py]' | entr -c make flake8; else make flake8; echo "\nInstall entr(1) to automatically run tests on file change.\n See http://entrproject.org/"; fi
+	if command -v entr > /dev/null; then find . -type f -not -path '*/\.*' | grep -i '.*[.][py]' | entr -c make flake8; else make flake8; echo "\nInstall entr(1) to automatically run tests on file change.\n See http://entrproject.org/"; fi
 
 .PHONY: flake8

bootstrap_env.py

@@ -79,7 +79,7 @@ def which(exe=None, throw=True):
 python_bin = os.path.join(env_dir, 'bin', 'python')
 virtualenv_bin = which('virtualenv', throw=False)
 virtualenv_exists = os.path.exists(env_dir) and os.path.isfile(python_bin)
-sphinx_requirements_filepath = os.path.join(project_dir, 'doc', 'requirements.pip')
+sphinx_requirements_filepath = os.path.join(project_dir, 'requirements', 'doc.txt')
 
 
 try:

doc/requirements.txt

@@ -0,0 +1 @@
+-r ../requirements/doc.txt

requirements.pip requirements/base.txtrequirements.pip renamed to requirements/base.txt

@@ -1,4 +1,4 @@
--r ../requirements.pip
+-r ./base.txt
 docutils==0.12
 sphinx
 reportlab

doc/requirements.pip requirements/doc.txtdoc/requirements.pip renamed to requirements/doc.txt

@@ -14,7 +14,7 @@
 with open("tmuxp/__about__.py") as fp:
     exec(fp.read(), about)
 
-with open('requirements.pip') as f:
+with open('requirements/base.txt') as f:
     install_reqs = [line for line in f.read().split('\n') if line]
     tests_reqs = []
 

setup.py

@@ -12,15 +12,15 @@
 import pkgutil
 import sys
 
-try:
-    import unittest2 as unittest
-except ImportError:  # Python 2.7
-    import unittest
-
 from tmuxp import log
 from tmuxp._compat import string_types, PY2, reraise
 from tmuxp.server import Server
 
+if sys.version_info < (2, 7):
+    import unittest2 as unittest
+else:
+    import unittest
+
 t = Server()
 t.socket_name = 'tmuxp_test'
 

tmuxp/testsuite/__init__.py

@@ -1,38 +1,49 @@
 # -*- coding: utf-8 -*-
-"""Helper methods for tmuxp unittests."""
+"""Helper methods for tmuxp unittests.
+
+_CallableContext, WhateverIO, decorator and stdouts are from the case project,
+https://github.com/celery/case, license BSD 3-clause.
+"""
 
 from __future__ import (absolute_import, division, print_function,
                         unicode_literals, with_statement)
 
 import contextlib
+import functools
+import inspect
+import io
 import logging
-from random import randint
+import os
+import sys
+import tempfile
+from contextlib import contextmanager
 
 from tmuxp import exc
 from tmuxp.testsuite import t
 
-try:
+if sys.version_info < (2, 7):
     import unittest2 as unittest
-except ImportError:  # Python 2.7
+else:
     import unittest
 
-
 logger = logging.getLogger(__name__)
 
 TEST_SESSION_PREFIX = 'test tmuxp_'
 
+namer = tempfile._RandomNameSequence()
+
 
 def get_test_session_name(server, prefix=TEST_SESSION_PREFIX):
     while True:
-        session_name = prefix + str(randint(0, 9999999))
+        session_name = prefix + next(namer)
         if not t.has_session(session_name):
             break
     return session_name
 
 
 def get_test_window_name(session, prefix=TEST_SESSION_PREFIX):
     while True:
-        window_name = prefix + str(randint(0, 9999999))
+        window_name = prefix + next(namer)
         if not session.findWhere(window_name=window_name):
             break
     return window_name
@@ -194,3 +205,163 @@ def bootstrap(self):
         self.TEST_SESSION_NAME = TEST_SESSION_NAME
         self.server = t
         self.session = session
+
+
+StringIO = io.StringIO
+_SIO_write = StringIO.write
+_SIO_init = StringIO.__init__
+
+
+def update_wrapper(wrapper, wrapped, *args, **kwargs):
+    wrapper = functools.update_wrapper(wrapper, wrapped, *args, **kwargs)
+    wrapper.__wrapped__ = wrapped
+    return wrapper
+
+
+def wraps(wrapped,
+          assigned=functools.WRAPPER_ASSIGNMENTS,
+          updated=functools.WRAPPER_UPDATES):
+    return functools.partial(update_wrapper, wrapped=wrapped,
+                             assigned=assigned, updated=updated)
+
+
+class _CallableContext(object):
+
+    def __init__(self, context, cargs, ckwargs, fun):
+        self.context = context
+        self.cargs = cargs
+        self.ckwargs = ckwargs
+        self.fun = fun
+
+    def __call__(self, *args, **kwargs):
+        return self.fun(*args, **kwargs)
+
+    def __enter__(self):
+        self.ctx = self.context(*self.cargs, **self.ckwargs)
+        return self.ctx.__enter__()
+
+    def __exit__(self, *einfo):
+        if self.ctx:
+            return self.ctx.__exit__(*einfo)
+
+
+def decorator(predicate):
+    context = contextmanager(predicate)
+
+    @wraps(predicate)
+    def take_arguments(*pargs, **pkwargs):
+
+        @wraps(predicate)
+        def decorator(cls):
+            if inspect.isclass(cls):
+                orig_setup = cls.setUp
+                orig_teardown = cls.tearDown
+
+                @wraps(cls.setUp)
+                def around_setup(*args, **kwargs):
+                    try:
+                        contexts = args[0].__rb3dc_contexts__
+                    except AttributeError:
+                        contexts = args[0].__rb3dc_contexts__ = []
+                    p = context(*pargs, **pkwargs)
+                    p.__enter__()
+                    contexts.append(p)
+                    return orig_setup(*args, **kwargs)
+                around_setup.__wrapped__ = cls.setUp
+                cls.setUp = around_setup
+
+                @wraps(cls.tearDown)
+                def around_teardown(*args, **kwargs):
+                    try:
+                        contexts = args[0].__rb3dc_contexts__
+                    except AttributeError:
+                        pass
+                    else:
+                        for context in contexts:
+                            context.__exit__(*sys.exc_info())
+                    orig_teardown(*args, **kwargs)
+                around_teardown.__wrapped__ = cls.tearDown
+                cls.tearDown = around_teardown
+
+                return cls
+            else:
+                @wraps(cls)
+                def around_case(self, *args, **kwargs):
+                    with context(*pargs, **pkwargs) as context_args:
+                        context_args = context_args or ()
+                        if not isinstance(context_args, tuple):
+                            context_args = (context_args,)
+                        return cls(*(self,) + args + context_args, **kwargs)
+                return around_case
+
+        if len(pargs) == 1 and callable(pargs[0]):
+            fun, pargs = pargs[0], ()
+            return decorator(fun)
+        return _CallableContext(context, pargs, pkwargs, decorator)
+    assert take_arguments.__wrapped__
+    return take_arguments
+
+
+class WhateverIO(StringIO):
+
+    def __init__(self, v=None, *a, **kw):
+        _SIO_init(self, v.decode() if isinstance(v, bytes) else v, *a, **kw)
+
+    def write(self, data):
+        _SIO_write(self, data.decode() if isinstance(data, bytes) else data)
+
+
+@decorator
+def stdouts():
+    """Override `sys.stdout` and `sys.stderr` with `StringIO`
+    instances.
+    Decorator example::
+        @mock.stdouts
+        def test_foo(self, stdout, stderr):
+            something()
+            self.assertIn('foo', stdout.getvalue())
+    Context example::
+        with mock.stdouts() as (stdout, stderr):
+            something()
+            self.assertIn('foo', stdout.getvalue())
+    """
+    prev_out, prev_err = sys.stdout, sys.stderr
+    prev_rout, prev_rerr = sys.__stdout__, sys.__stderr__
+    mystdout, mystderr = WhateverIO(), WhateverIO()
+    sys.stdout = sys.__stdout__ = mystdout
+    sys.stderr = sys.__stderr__ = mystderr
+
+    try:
+        yield mystdout, mystderr
+    finally:
+        sys.stdout = prev_out
+        sys.stderr = prev_err
+        sys.__stdout__ = prev_rout
+        sys.__stderr__ = prev_rerr
+
+
+@decorator
+def mute():
+    """Redirect `sys.stdout` and `sys.stderr` to /dev/null, silencent them.
+    Decorator example::
+        @mute
+        def test_foo(self):
+            something()
+    Context example::
+        with mute():
+            something()
+    """
+    prev_out, prev_err = sys.stdout, sys.stderr
+    prev_rout, prev_rerr = sys.__stdout__, sys.__stderr__
+    devnull = open(os.devnull, 'w')
+    mystdout, mystderr = devnull, devnull
+    sys.stdout = sys.__stdout__ = mystdout
+    sys.stderr = sys.__stderr__ = mystderr
+
+    try:
+        yield
+    finally:
+        sys.stdout = prev_out
+        sys.stderr = prev_err
+        sys.__stdout__ = prev_rout
+        sys.__stderr__ = prev_rerr

tmuxp/testsuite/helpers.py

@@ -11,11 +11,10 @@
 
 import logging
 import unittest
-from random import randint
 
 from tmuxp import Pane, Session, Window
 from tmuxp.testsuite import t
-from tmuxp.testsuite.helpers import TEST_SESSION_PREFIX, TmuxTestCase
+from tmuxp.testsuite.helpers import TEST_SESSION_PREFIX, TmuxTestCase, namer
 
 logger = logging.getLogger(__name__)
 
@@ -99,7 +98,7 @@ class SessionNewTest(TmuxTestCase):
 
     def test_new_session(self):
         """Server.new_session creates new session."""
-        new_session_name = TEST_SESSION_PREFIX + str(randint(0, 1337))
+        new_session_name = TEST_SESSION_PREFIX + next(namer)
         new_session = t.new_session(session_name=new_session_name, detach=True)
 
         self.assertIsInstance(new_session, Session)

tmuxp/testsuite/session.py

@@ -10,12 +10,11 @@
                         unicode_literals, with_statement)
 
 import logging
-import random
 import unittest
 
 from tmuxp import Pane, Session, Window
 from tmuxp.testsuite import t
-from tmuxp.testsuite.helpers import TEST_SESSION_PREFIX, TmuxTestCase
+from tmuxp.testsuite.helpers import TEST_SESSION_PREFIX, TmuxTestCase, namer
 
 logger = logging.getLogger(__name__)
 
@@ -62,9 +61,7 @@ def test_findWhere_None(self):
         """.findWhere returns None if no results found."""
 
         while True:
-            nonexistant_session = TEST_SESSION_PREFIX + str(
-                random.randint(0, 9999)
-            )
+            nonexistant_session = TEST_SESSION_PREFIX + next(namer)
 
             if not t.has_session(nonexistant_session):
                 break
@@ -176,7 +173,7 @@ def test_getById(self):
             self.assertEqual(get_by_id, session)
             self.assertIsInstance(get_by_id, Session)
             self.assertIsNone(t.getById(
-                '$' + str(random.randint(50000, 90000))
+                '$' + next(namer)
             ))
 
             # session.getById
@@ -189,7 +186,7 @@ def test_getById(self):
                 self.assertIsInstance(get_by_id, Window)
 
                 self.assertIsNone(session.getById(
-                    '@' + str(random.randint(50000, 90000))
+                    '@' + next(namer)
                 ))
 
                 # window.getById
@@ -201,7 +198,7 @@ def test_getById(self):
                     self.assertEqual(get_by_id, pane)
                     self.assertIsInstance(get_by_id, Pane)
                     self.assertIsNone(window.getById(
-                        '%' + str(random.randint(50000, 90000))
+                        '%' + next(namer)
                     ))