# Path: oslo_utils/eventletutils.py
# EVENTLET_AVAILABLE = all((_eventlet, _patcher))
# _ALL_PATCH = frozenset(['__builtin__', 'MySQLdb', 'os',
#                         'psycopg', 'select', 'socket', 'thread', 'time'])
# def fetch_current_thread_functor():
# def warn_eventlet_not_patched(expected_patched_modules=None,
#                               what='this library'):
# def is_monkey_patched(module):
#     def __init__(self, *args, **kwargs):
#     def clear(self):
#     def is_set(self):
#     def set(self):
#     def wait(self, timeout=None):
# def Event():
# class EventletEvent(object):


import threading
import warnings
import eventlet
from unittest import mock
from eventlet import greenthread
from oslotest import base as test_base
from oslo_utils import eventletutils
            eventletutils.warn_eventlet_not_patched(['os'])
        self.assertEqual(1, len(capture))
        is_patched.add('os')
        with warnings.catch_warnings(record=True) as capture:
            warnings.simplefilter("always")
            eventletutils.warn_eventlet_not_patched(['os'])
        self.assertEqual(0, len(capture))
        is_patched.add('thread')
        with warnings.catch_warnings(record=True) as capture:
            warnings.simplefilter("always")
            eventletutils.warn_eventlet_not_patched(['os', 'thread'])
        self.assertEqual(0, len(capture))
        with warnings.catch_warnings(record=True) as capture:
            warnings.simplefilter("always")
            eventletutils.warn_eventlet_not_patched(['all'])
        self.assertEqual(1, len(capture))
        w = capture[0]
        self.assertEqual(RuntimeWarning, w.category)
        for m in ['os', 'thread']:
            self.assertNotIn(m, str(w.message))

    def test_invalid_patch_check(self):
        self.assertRaises(ValueError,
                          eventletutils.warn_eventlet_not_patched,
                          ['blah.blah'])

    @mock.patch('oslo_utils.eventletutils._eventlet')
    def test_event_api_compat(self, mock_eventlet):
        with mock.patch('oslo_utils.eventletutils.is_monkey_patched',
                        return_value=True):

# Path: oslo_utils/eventletutils.py
# EVENTLET_AVAILABLE = all((_eventlet, _patcher))
# _ALL_PATCH = frozenset(['__builtin__', 'MySQLdb', 'os',
#                         'psycopg', 'select', 'socket', 'thread', 'time'])
# def fetch_current_thread_functor():
# def warn_eventlet_not_patched(expected_patched_modules=None,
#                               what='this library'):
# def is_monkey_patched(module):
#     def __init__(self, *args, **kwargs):
#     def clear(self):
#     def is_set(self):
#     def set(self):
#     def wait(self, timeout=None):
# def Event():
# class EventletEvent(object):


# Path: oslo_utils/tests/test_eventletutils.py
import threading
import warnings
import eventlet
from unittest import mock
from eventlet import greenthread
from oslotest import base as test_base
from oslo_utils import eventletutils

            eventletutils.warn_eventlet_not_patched(['os'])
        self.assertEqual(1, len(capture))
        is_patched.add('os')
        with warnings.catch_warnings(record=True) as capture:
            warnings.simplefilter("always")
            eventletutils.warn_eventlet_not_patched(['os'])
        self.assertEqual(0, len(capture))
        is_patched.add('thread')
        with warnings.catch_warnings(record=True) as capture:
            warnings.simplefilter("always")
            eventletutils.warn_eventlet_not_patched(['os', 'thread'])
        self.assertEqual(0, len(capture))
        with warnings.catch_warnings(record=True) as capture:
            warnings.simplefilter("always")
            eventletutils.warn_eventlet_not_patched(['all'])
        self.assertEqual(1, len(capture))
        w = capture[0]
        self.assertEqual(RuntimeWarning, w.category)
        for m in ['os', 'thread']:
            self.assertNotIn(m, str(w.message))

    def test_invalid_patch_check(self):
        self.assertRaises(ValueError,
                          eventletutils.warn_eventlet_not_patched,
                          ['blah.blah'])

    @mock.patch('oslo_utils.eventletutils._eventlet')
    def test_event_api_compat(self, mock_eventlet):
        with mock.patch('oslo_utils.eventletutils.is_monkey_patched',
                        return_value=True):

            e_event = eventletutils.Event()