python类green()的实例源码

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'

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)

def finish(self):
        try:
            eventlet.green.BaseHTTPServer.BaseHTTPRequestHandler.finish(self)
        except IOError:
            pass
        eventlet.greenio.shutdown_safe(self.connection)
        self.connection.close()