Code Review
/
packages
/
trusty
/
python-eventlet.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
review
|
tree
raw
|
inline
| side by side
Add python-eventlet package to MOS 8.0 repository
[packages/trusty/python-eventlet.git]
/
python-eventlet
/
tests
/
patcher_test.py
diff --git
a/eventlet/tests/patcher_test.py
b/python-eventlet/tests/patcher_test.py
similarity index 96%
rename from
eventlet/tests/patcher_test.py
rename to
python-eventlet/tests/patcher_test.py
index 5c9076f4459192e8cd7b45ed7d9457a54473de82..2e458c557dff97cf73c412cf5d745df610b4f594 100644
(file)
--- a/
eventlet/tests/patcher_test.py
+++ b/
python-eventlet/tests/patcher_test.py
@@
-4,15
+4,12
@@
import sys
import tempfile
from eventlet.support import six
import tempfile
from eventlet.support import six
-from tests import LimitedTestCase, main, run_python, skip_with_pyevent
+import tests
base_module_contents = """
import socket
base_module_contents = """
import socket
-try:
- import urllib.request as urllib
-except ImportError:
- import urllib
+import urllib
print("base {0} {1}".format(socket, urllib))
"""
print("base {0} {1}".format(socket, urllib))
"""
@@
-32,7
+29,7
@@
print("importing {0} {1} {2} {3}".format(patching, socket, patching.socket, patc
"""
"""
-class ProcessBase(LimitedTestCase):
+class ProcessBase(
tests.
LimitedTestCase):
TEST_TIMEOUT = 3 # starting processes is time-consuming
def setUp(self):
TEST_TIMEOUT = 3 # starting processes is time-consuming
def setUp(self):
@@
-54,7
+51,7
@@
class ProcessBase(LimitedTestCase):
def launch_subprocess(self, filename):
path = os.path.join(self.tempdir, filename)
def launch_subprocess(self, filename):
path = os.path.join(self.tempdir, filename)
- output = run_python(path)
+ output =
tests.
run_python(path)
if six.PY3:
output = output.decode('utf-8')
separator = '\n'
if six.PY3:
output = output.decode('utf-8')
separator = '\n'
@@
-86,7
+83,14
@@
class ImportPatched(ProcessBase):
assert 'eventlet.green.httplib' not in lines[2], repr(output)
def test_import_patched_defaults(self):
assert 'eventlet.green.httplib' not in lines[2], repr(output)
def test_import_patched_defaults(self):
- self.write_to_tempfile("base", base_module_contents)
+ self.write_to_tempfile("base", """
+import socket
+try:
+ import urllib.request as urllib
+except ImportError:
+ import urllib
+print("base {0} {1}".format(socket, urllib))""")
+
new_mod = """
from eventlet import patcher
base = patcher.import_patched('base')
new_mod = """
from eventlet import patcher
base = patcher.import_patched('base')
@@
-242,7
+246,7
@@
def test_monkey_patch_threading():
class Tpool(ProcessBase):
TEST_TIMEOUT = 3
class Tpool(ProcessBase):
TEST_TIMEOUT = 3
- @skip_with_pyevent
+ @
tests.
skip_with_pyevent
def test_simple(self):
new_mod = """
import eventlet
def test_simple(self):
new_mod = """
import eventlet
@@
-260,7
+264,7
@@
tpool.killall()
assert '2' in lines[0], repr(output)
assert '3' in lines[1], repr(output)
assert '2' in lines[0], repr(output)
assert '3' in lines[1], repr(output)
- @skip_with_pyevent
+ @
tests.
skip_with_pyevent
def test_unpatched_thread(self):
new_mod = """import eventlet
eventlet.monkey_patch(time=False, thread=False)
def test_unpatched_thread(self):
new_mod = """import eventlet
eventlet.monkey_patch(time=False, thread=False)
@@
-273,7
+277,7
@@
import time
output, lines = self.launch_subprocess('newmod.py')
self.assertEqual(len(lines), 2, lines)
output, lines = self.launch_subprocess('newmod.py')
self.assertEqual(len(lines), 2, lines)
- @skip_with_pyevent
+ @
tests.
skip_with_pyevent
def test_patched_thread(self):
new_mod = """import eventlet
eventlet.monkey_patch(time=False, thread=True)
def test_patched_thread(self):
new_mod = """import eventlet
eventlet.monkey_patch(time=False, thread=True)
@@
-493,9
+497,12
@@
t2.join()
def test_importlib_lock():
def test_importlib_lock():
- output = run_python('tests/patcher_test_importlib_lock.py')
- assert output.rstrip() == b'ok'
+ tests.run_isolated('patcher_importlib_lock.py')
+
+
+def test_threading_condition():
+ tests.run_isolated('patcher_threading_condition.py')
-
if __name__ == '__main__'
:
-
main(
)
+
def test_threading_join()
:
+
tests.run_isolated('patcher_threading_join.py'
)