Commits

Stefan Scherfke committed 76ddd3c

Added a _compat module for Python (2.x) compatibility code.

Comments (0)

Files changed (3)

+"""
+Compatibility helpers for older Python versions.
+
+"""
+import sys
+
+
+PY2 = sys.version_info[0] == 2
+
+
+if PY2:
+    # Python 2.x does not report exception chains. To emulate the behaviour of
+    # Python 3 the functions format_chain and print_chain are added. The latter
+    # function is used to override the exception hook of Python 2.x.
+    from traceback import format_exception
+
+    def format_chain(exc_type, exc_value, exc_traceback):
+        if hasattr(exc_value, '__cause__') and exc_value.__cause__:
+            cause = exc_value.__cause__
+            if hasattr(exc_value, '__traceback__'):
+                traceback = exc_value.__traceback__
+            else:
+                traceback = None
+            lines = format_chain(type(cause), cause, traceback)
+            lines += ('\nThe above exception was the direct cause of the '
+                      'following exception:\n\n')
+        else:
+            lines = []
+
+        return lines + format_exception(exc_type, exc_value, exc_traceback)
+
+    def print_chain(exc_type, exc_value, exc_traceback):
+        sys.stderr.write(''.join(format_chain(exc_type, exc_value,
+                                              exc_traceback)))
+        sys.stderr.flush()
+
+    sys.excepthook = print_chain
 :func:`simulate()`.
 
 """
-import sys
 import types
 from heapq import heappush, heappop
 from inspect import isgenerator
 from itertools import count
 
+from simpy._compat import PY2
+
+if PY2:
+    import sys
+
 
 Infinity = float('inf')
 """Convenience alias for infinity."""
 """Priority of timeouts."""
 
 
-if sys.version_info[0] < 3:
-    LEGACY_SUPPORT = True
-
-    # Python 2.x does not report exception chains. To emulate the behaviour of
-    # Python 3 the functions format_chain and print_chain are added. The latter
-    # function is used to override the exception hook of Python 2.x.
-
-    from traceback import format_exception
-
-    def format_chain(exc_type, exc_value, exc_traceback):
-        if hasattr(exc_value, '__cause__') and exc_value.__cause__:
-            cause = exc_value.__cause__
-            if hasattr(exc_value, '__traceback__'):
-                traceback = exc_value.__traceback__
-            else:
-                traceback = None
-            lines = format_chain(type(cause), cause, traceback)
-            lines += ('\nThe above exception was the direct cause of the '
-                    'following exception:\n\n')
-        else:
-            lines = []
-
-        return lines + format_exception(exc_type, exc_value, exc_traceback)
-
-    def print_chain(exc_type, exc_value, exc_traceback):
-        sys.stderr.write(
-                ''.join(format_chain(exc_type, exc_value, exc_traceback)))
-        sys.stderr.flush()
-
-    sys.excepthook = print_chain
-else:
-    LEGACY_SUPPORT = False
-
-
 class BoundClass(object):
     """Allows classes to behave like methods. The ``__get__()`` descriptor is
     basically identical to ``function.__get__()`` and binds the first argument
                 self.ok = False
                 self._value = type(e)(*e.args)
                 self._value.__cause__ = e
-                if LEGACY_SUPPORT:
+                if PY2:
                     self._value.__traceback__ = sys.exc_info()[2]
                 self.env.enqueue(DEFAULT_PRIORITY, self)
                 break

simpy/test/test_exceptions.py

 Tests for forwarding exceptions from child to parent processes.
 
 """
-import re
-
 import pytest
 
+from simpy import _compat
 import simpy
 
 
 
 
 def test_no_parent_process(env):
-    """Exceptions should be normally raised if there are no processes
-    waiting for the one that raises something.
+    """Exceptions should be normally raised if there are no processes waiting
+    for the one that raises something.
 
     """
     def child(env):
             yield env.start(panic(env))
             pytest.fail("Hey, where's the roflcopter?")
         except RuntimeError:
-            if not simpy.core.LEGACY_SUPPORT:
+            if not _compat.PY2:
                 import traceback
                 stacktrace = traceback.format_exc()
             else:
                 import sys
-                stacktrace = ''.join(simpy.core.format_chain(*sys.exc_info()))
+                stacktrace = ''.join(_compat.format_chain(*sys.exc_info()))
 
             # The current frame must be visible in the stacktrace.
             assert 'yield env.start(panic(env))' in stacktrace
 @pytest.mark.skipif('sys.version_info[0] < 3')
 def test_exception_chaining(env):
     """Unhandled exceptions pass through the entire event stack. This must be
-    visible in the stacktrace of the exception."""
+    visible in the stacktrace of the exception.
+
+    """
     def child(env):
         yield env.timeout(1)
         raise RuntimeError('foo')
     try:
         simpy.simulate(env, until=1)
         assert False, 'There must be a RuntimeError!'
-    except RuntimeError as e:
+    except RuntimeError:
         pass
 
 
             pass
 
     event = env.event()
-    proc = env.start(pem(env, event))
+    env.start(pem(env, event))
     event.fail(RuntimeError())
 
     assert not hasattr(event, 'defused'), 'Event has been defuseed immediately'