import tempfile
from eventlet.support import six
-from tests import LimitedTestCase, main, run_python, skip_with_pyevent
+import tests
base_module_contents = """
import socket
-try:
- import urllib.request as urllib
-except ImportError:
- import urllib
+import urllib
print("base {0} {1}".format(socket, urllib))
"""
"""
-class ProcessBase(LimitedTestCase):
+class ProcessBase(tests.LimitedTestCase):
TEST_TIMEOUT = 3 # starting processes is time-consuming
def setUp(self):
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'
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')
class Tpool(ProcessBase):
TEST_TIMEOUT = 3
- @skip_with_pyevent
+ @tests.skip_with_pyevent
def test_simple(self):
new_mod = """
import eventlet
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)
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_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')