# 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
#
#    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.





class EventletUtilsTest(test_base.BaseTestCase):
    def setUp(self):
        super(EventletUtilsTest, self).setUp()
        self._old_avail = eventletutils.EVENTLET_AVAILABLE
        eventletutils.EVENTLET_AVAILABLE = True

    def tearDown(self):
        super(EventletUtilsTest, self).tearDown()
        eventletutils.EVENTLET_AVAILABLE = self._old_avail

    @mock.patch("oslo_utils.eventletutils._patcher")
    def test_warning_not_patched(self, mock_patcher):
        mock_patcher.already_patched = True
        mock_patcher.is_monkey_patched.return_value = False
        with warnings.catch_warnings(record=True) as capture:
            warnings.simplefilter("always")
            eventletutils.warn_eventlet_not_patched(['os'])
        self.assertEqual(1, len(capture))
        w = capture[0]

# 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

#
#    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.





class EventletUtilsTest(test_base.BaseTestCase):
    def setUp(self):
        super(EventletUtilsTest, self).setUp()
        self._old_avail = eventletutils.EVENTLET_AVAILABLE
        eventletutils.EVENTLET_AVAILABLE = True

    def tearDown(self):
        super(EventletUtilsTest, self).tearDown()
        eventletutils.EVENTLET_AVAILABLE = self._old_avail

    @mock.patch("oslo_utils.eventletutils._patcher")
    def test_warning_not_patched(self, mock_patcher):
        mock_patcher.already_patched = True
        mock_patcher.is_monkey_patched.return_value = False
        with warnings.catch_warnings(record=True) as capture:
            warnings.simplefilter("always")
            eventletutils.warn_eventlet_not_patched(['os'])
        self.assertEqual(1, len(capture))
        w = capture[0]

        self.assertEqual(RuntimeWarning, w.category)