[Checkins] SVN: zc.thread/trunk/src/zc/thread/ Refactored to avoid logic duplication
jim
cvs-admin at zope.org
Tue Jan 1 20:34:03 UTC 2013
Log message for revision 128982:
Refactored to avoid logic duplication
Changed:
U zc.thread/trunk/src/zc/thread/__init__.py
U zc.thread/trunk/src/zc/thread/tests.py
-=-
Modified: zc.thread/trunk/src/zc/thread/__init__.py
===================================================================
--- zc.thread/trunk/src/zc/thread/__init__.py 2013-01-01 20:33:38 UTC (rev 128981)
+++ zc.thread/trunk/src/zc/thread/__init__.py 2013-01-01 20:34:03 UTC (rev 128982)
@@ -15,7 +15,30 @@
def _options(daemon=True, start=True, args=(), kwargs=None):
return daemon, start, args, kwargs or {}
+def _Thread(class_, func, options):
+ daemon, start, args, kwargs = _options(**options)
+ def run(*args, **kw):
+ try:
+ v = func(*args, **kw)
+ thread.value = v
+ except Exception, v:
+ thread.exception = v
+
+ thread = class_(
+ target=run, name=getattr(func, '__name__', None),
+ args=args, kwargs=kwargs)
+
+ if hasattr(thread, 'setDaemon'):
+ thread.setDaemon(daemon)
+ else:
+ thread.daemon = daemon
+ thread.value = thread.exception = None
+ if start:
+ thread.start()
+ return thread
+
+
def Thread(func=None, **options):
"""Create and (typically) start a thread
@@ -48,25 +71,9 @@
"""
if func is None:
return lambda f: Thread(f, **options)
- daemon, start, args, kwargs = _options(**options)
import threading
+ return _Thread(threading.Thread, func, options)
- def run(*args, **kw):
- try:
- v = func(*args, **kw)
- thread.value = v
- except Exception, v:
- thread.exception = v
-
- thread = threading.Thread(
- target=run, name=getattr(func, '__name__', None),
- args=args, kwargs=kwargs)
- thread.setDaemon(daemon)
- thread.value = thread.exception = None
- if start:
- thread.start()
- return thread
-
def Process(func=None, **options):
"""Create and (typically) start a multiprocessing process
@@ -99,12 +106,5 @@
"""
if func is None:
return lambda f: Process(f, **options)
- daemon, start, args, kwargs = _options(**options)
import multiprocessing
- process = multiprocessing.Process(
- target=func, name=getattr(func, '__name__', None),
- args=args, kwargs=kwargs)
- process.daemon = daemon
- if start:
- process.start()
- return process
+ return _Thread(multiprocessing.Process, func, options)
Modified: zc.thread/trunk/src/zc/thread/tests.py
===================================================================
--- zc.thread/trunk/src/zc/thread/tests.py 2013-01-01 20:33:38 UTC (rev 128981)
+++ zc.thread/trunk/src/zc/thread/tests.py 2013-01-01 20:34:03 UTC (rev 128982)
@@ -14,6 +14,7 @@
import doctest
import mock
import os
+import sys
import unittest
import zc.thread
@@ -76,6 +77,7 @@
def test_Process_w_mock(self):
with mock.patch('multiprocessing.Process') as Process:
+
@zc.thread.Process
def foo():
print 'foo called'
@@ -86,14 +88,22 @@
Process.reset_mock()
def foo2():
- pass
+ return 42
t = zc.thread.Process(foo2)
+ Process.call_args[1].pop('target')()
+ self.assertEqual(Process.return_value.value, 42)
Process.assert_called_with(
- target=foo2, name='foo2', args=(), kwargs={})
+ name='foo2', args=(), kwargs={})
self.assert_(t.daamon)
t.start.assert_called_with()
Process.reset_mock()
+ def test_Process_w_mock2(self):
+
+ with mock.patch('multiprocessing.Process') as Process:
+
+ del Process.return_value.setDaemon
+
@zc.thread.Process(daemon=False, start=False, args=(42,),
kwargs=dict(a=1))
def foo3():
More information about the checkins
mailing list