def test_communicate_with_poll():
# This test was being skipped since git 25812fca8, I don't there's
# a need to do this. The original comment:
#
# https://github.com/eventlet/eventlet/pull/24
# `eventlet.green.subprocess.Popen.communicate()` was broken
# in Python 2.7 because the usage of the `select` module was moved from
# `_communicate` into two other methods `_communicate_with_select`
# and `_communicate_with_poll`. Link to 2.7's implementation:
# http://hg.python.org/cpython/file/2145593d108d/Lib/subprocess.py#l1255
p = subprocess.Popen(
[sys.executable, '-c', 'import time; time.sleep(0.5)'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
t1 = time.time()
eventlet.with_timeout(0.1, p.communicate, timeout_value=True)
tdiff = time.time() - t1
assert 0.1 <= tdiff <= 0.2, 'did not stop within allowed time'
python类green()的实例源码
def test_hang(testdir):
p = py.path.local(__file__).dirpath('conftest.py')
p.copy(testdir.tmpdir.join(p.basename))
t = testdir.makepyfile("""
import pytest
from eventlet.green import time
@pytest.mark.timeout(0.01)
def test_hang():
time.sleep(3.0)
""")
result = testdir.runpytest()
assert "failed to timeout" not in result.stdout.str()
result.stdout.fnmatch_lines(["*Timeout: 0.01*"])
def _single_run(self, application, sock):
"""Start a WSGI server in a new green thread."""
LOG.info(_LI("Starting single process server"))
eventlet.wsgi.server(sock, application, custom_pool=self.pool,
url_length_limit=URL_LENGTH_LIMIT,
log=self._wsgi_logger, debug=cfg.CONF.debug)
def __str__(self):
return ("Command '%s' timed out after %s seconds" %
(self.cmd, self.timeout))
# This is the meat of this module, the green version of Popen.
def test_exception_identity():
# https://github.com/eventlet/eventlet/issues/413
# green module must keep exceptions classes as stdlib version
tests.run_isolated('subprocess_exception_identity.py')
def test_repeated_select_bad_fd():
from eventlet.green import select
def once():
try:
select.select([-1], [], [])
assert False, 'Expected ValueError'
except ValueError:
pass
once()
once()
def test_patch_a_module(self):
self.write_to_tempfile("base", base_module_contents)
self.write_to_tempfile("patching", patching_module_contents)
self.write_to_tempfile("importing", import_module_contents)
output, lines = self.launch_subprocess('importing.py')
assert lines[0].startswith('patcher'), repr(output)
assert lines[1].startswith('base'), repr(output)
assert lines[2].startswith('importing'), repr(output)
assert 'eventlet.green.socket' in lines[1], repr(output)
assert 'eventlet.green.urllib' in lines[1], repr(output)
assert 'eventlet.green.socket' in lines[2], repr(output)
assert 'eventlet.green.urllib' in lines[2], repr(output)
assert 'eventlet.green.httplib' not in lines[2], repr(output)
def test_subprocess_after_monkey_patch():
code = '''\
import sys
import eventlet
eventlet.monkey_patch()
from eventlet.green import subprocess
subprocess.Popen([sys.executable, '-c', ''], stdin=subprocess.PIPE).wait()
print('pass')
'''
output = tests.run_python(
path=None,
args=['-c', code],
)
assert output.rstrip() == b'pass'
def report(self, autocolor=True):
total = 0
errors = []
for r in self.traversal:
total += 1
if r.is_leaf_error():
exc = r.exception[1]
indent = " "
if getattr(exc, "report_traceback", True):
tb = "\n\n" + r.get_traceback().strip()
errors.append("%s%s: unexpected error%s" % (indent, r.executor.name,
tb.replace("\n", "\n " + indent)))
else:
errors.append("%s%s: %s" % (indent, r.executor.context, exc))
if autocolor:
if errors:
color = self.terminal.bold_red
else:
color = self.terminal.green
else:
color = lambda x: x
result = "\n".join(["%s tasks run, %s errors" % (total, len(errors))] + errors)
return "\n".join([color(line) for line in result.splitlines()])
def _single_run(self, application, sock):
"""Start a WSGI server in a new green thread."""
LOG.info("Starting single process server")
eventlet.wsgi.server(sock, application, custom_pool=self.pool,
log=self._logger,
debug=False,
keepalive=CONF.http_keepalive,
socket_timeout=self.client_socket_timeout)
xvp_proxy.py 文件源码
项目:Trusted-Platform-Module-nova
作者: BU-NU-CLOUD-SP16
项目源码
文件源码
阅读 28
收藏 0
点赞 0
评论 0
def finish(self):
try:
eventlet.green.BaseHTTPServer.BaseHTTPRequestHandler.finish(self)
except IOError:
pass
eventlet.greenio.shutdown_safe(self.connection)
self.connection.close()