Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from eventlet import patcher
from eventlet.green import SimpleHTTPServer
patcher.inject(
'test.test_SimpleHTTPServer',
globals(),
('SimpleHTTPServer', SimpleHTTPServer))
if __name__ == "__main__":
test_main()
from eventlet import patcher
from eventlet.green import socket
from eventlet.green import urllib2
patcher.inject(
'test.test_urllib2',
globals(),
('socket', socket),
('urllib2', urllib2))
HandlerTests.test_file = patcher.patch_function(HandlerTests.test_file, ('socket', socket))
HandlerTests.test_cookie_redirect = patcher.patch_function(
HandlerTests.test_cookie_redirect, ('urllib2', urllib2))
OpenerDirectorTests.test_badly_named_methods = patcher.patch_function(
OpenerDirectorTests.test_badly_named_methods, ('urllib2', urllib2))
if __name__ == "__main__":
test_main()
from eventlet import patcher
from eventlet.green import socket
from eventlet.green import SocketServer
import six
patcher.inject(
'BaseHTTPServer' if six.PY2 else 'http.server',
globals(),
('socket', socket),
('SocketServer', SocketServer),
('socketserver', SocketServer))
del patcher
if __name__ == '__main__':
test()
import sys
import eventlet
# NOTE(jokke): As per the eventlet commit
# b756447bab51046dfc6f1e0e299cc997ab343701 there's circular import happening
# which can be solved making sure the hubs are properly and fully imported
# before calling monkey_patch(). This is solved in eventlet 0.22.0 but we
# need to address it before that is widely used around.
eventlet.hubs.get_hub()
if os.name == 'nt':
# eventlet monkey patching the os module causes subprocess.Popen to fail
# on Windows when using pipes due to missing non-blocking IO support.
eventlet.patcher.monkey_patch(os=False)
else:
eventlet.patcher.monkey_patch()
# Monkey patch the original current_thread to use the up-to-date _active
# global variable. See https://bugs.launchpad.net/bugs/1863021 and
# https://github.com/eventlet/eventlet/issues/592
import __original_module_threading as orig_threading
import threading
orig_threading.current_thread.__globals__['_active'] = threading._active
import subprocess
# If ../glance/__init__.py exists, add ../ to Python search path, so that
# it will override what happens to be installed in /usr/(local/)lib/python...
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(sys.argv[0]),
os.pardir,
os.pardir))
if os.path.exists(os.path.join(possible_topdir, 'glance', '__init__.py')):
__patched__ = ['_start_new_thread', '_allocate_lock',
'_sleep', 'local', 'stack_size', 'Lock', 'currentThread',
'current_thread', '_after_fork', '_shutdown']
if six.PY2:
__patched__ += ['_get_ident']
else:
__patched__ += ['get_ident', '_set_sentinel']
__orig_threading = eventlet.patcher.original('threading')
__threadlocal = __orig_threading.local()
__patched_enumerate = None
eventlet.patcher.inject(
'threading',
globals(),
('thread' if six.PY2 else '_thread', thread),
('time', time))
_count = 1
class _GreenThread(object):
"""Wrapper for GreenThread objects to provide Thread-like attributes
and methods"""
def __init__(self, g):
global _count
self._g = g
from eventlet.green import httplib
from eventlet.green import socket
from eventlet.green import ssl
from eventlet.green import time
from eventlet.green import urllib
patcher.inject(
'urllib2',
globals(),
('httplib', httplib),
('socket', socket),
('ssl', ssl),
('time', time),
('urllib', urllib))
FTPHandler.ftp_open = patcher.patch_function(FTPHandler.ftp_open, ('ftplib', ftplib))
del patcher
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
import eventlet.patcher
time = eventlet.patcher.original('time')
def service_hub():
# NOTE(dims): Add a custom impl for EVENTLET_HUB, so we can
# override the clock used in the eventlet hubs. The default
# uses time.time() and we need to use a monotonic timer
# to ensure that things like loopingcall work properly.
hub = eventlet.hubs.get_default_hub().Hub()
hub.clock = time.monotonic
# get_default_hub() will return a hub that is supported on this platform
hub.is_available = lambda: True
return hub
os.environ['EVENTLET_HUB'] = 'oslo_service:service_hub'
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Utility methods for working with WSGI servers."""
import eventlet
eventlet.patcher.monkey_patch(all=False, socket=True)
import datetime
import errno
import socket
import sys
import time
import eventlet.wsgi
from oslo_config import cfg
from oslo_log import log as logging
from oslo_serialization import jsonutils
from oslo_service import service
from oslo_service import sslutils
import routes
import routes.middleware
import webob.dec
else:
to_patch.append(('ssl', ssl))
patcher.inject('urllib.request', globals(), *to_patch)
del to_patch
to_patch_in_functions = [('ftplib', ftplib)]
del ftplib
FTPHandler.ftp_open = patcher.patch_function(FTPHandler.ftp_open, *to_patch_in_functions)
URLopener.open_ftp = patcher.patch_function(URLopener.open_ftp, *to_patch_in_functions)
ftperrors = patcher.patch_function(ftperrors, *to_patch_in_functions)
ftpwrapper.init = patcher.patch_function(ftpwrapper.init, *to_patch_in_functions)
ftpwrapper.retrfile = patcher.patch_function(ftpwrapper.retrfile, *to_patch_in_functions)
del error
del parse
del response
del to_patch_in_functions
def fdopen(fd, *args, **kwargs):
# NOTE(gus): We can't just use os.fdopen() here and allow the
# regular (optional) monkey_patching to do its thing. Turns out
# that regular file objects (as returned by os.fdopen) on python2
# are broken in lots of ways regarding blocking behaviour. We
# *need* the newer io.* objects on py2 (doesn't matter on py3,
# since the old file code has been replaced with io.*)
if eventlet.patcher.is_monkey_patched('socket'):
return eventlet.greenio.GreenPipe(fd, *args, **kwargs)
else:
return io.open(fd, *args, **kwargs)