openstack/deb-python-cassandra-driver
Code Issues Proposed changes

Reorganized eventlet and gevent tests in unit tests

Browse Source
This commit is contained in:
bjmb
2017-03-31 15:10:00 -04:00
parent 45ceafead3
commit 4c9ede85a6
3 changed files with 56 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
tests/unit/io/test_eventletreactor.py
View File

@@ -18,7 +18,7 @@ try:
except ImportError:
import unittest # noqa
from tests.unit.io.utils import submit_and_wait_for_completion, TimerCallback
from tests.unit.io.utils import TimerConnectionTests
from tests import is_eventlet_time_monkey_patched, is_gevent_time_monkey_patched
from tests.unit.io.eventlet_utils import restore_saved_module
import time
@@ -30,53 +30,16 @@ except ImportError:
EventletConnection = None # noqa
class EventletTimerTest(unittest.TestCase):
need_unpatch = False
@classmethod
def setUpClass(cls):
if is_eventlet_time_monkey_patched():
return # no dynamic patching if we have eventlet applied
if EventletConnection is not None:
if not is_gevent_time_monkey_patched():
cls.need_unpatch = True
monkey_patch(time=True)
@classmethod
def tearDownClass(cls):
if cls.need_unpatch:
restore_saved_module(time)
class EventletTimerTest(unittest.TestCase, TimerConnectionTests):
def setUp(self):
if not is_eventlet_time_monkey_patched():
raise unittest.SkipTest("Can't test gevent without monkey patching")
if not EventletConnection:
raise unittest.SkipTest("Can't test eventlet without monkey patching")
monkey_patch(time=True)
EventletConnection.initialize_reactor()
def test_multi_timer_validation(self, *args):
"""
Verify that timer timeouts are honored appropriately
"""
# Tests timers submitted in order at various timeouts
submit_and_wait_for_completion(self, EventletConnection, 0, 100, 1, 100)
# Tests timers submitted in reverse order at various timeouts
submit_and_wait_for_completion(self, EventletConnection, 100, 0, -1, 100)
# Tests timers submitted in varying order at various timeouts
submit_and_wait_for_completion(self, EventletConnection, 0, 100, 1, 100, True)
def tearDown(self):
restore_saved_module(time)
EventletConnection._timers = None
def test_timer_cancellation(self):
"""
Verify that timer cancellation is honored
"""
# Various lists for tracking callback stage
timeout = .1
callback = TimerCallback(timeout)
timer = EventletConnection.create_timer(timeout, callback.invoke)
timer.cancel()
# Release context allow for timer thread to run.
time.sleep(.2)
timer_manager = EventletConnection._timers
# Assert that the cancellation was honored
self.assertFalse(timer_manager._queue)
self.assertFalse(timer_manager._new_timers)
self.assertFalse(callback.was_invoked())
def getClass(self):
return EventletConnection

59
tests/unit/io/test_geventreactor.py
View File

@@ -17,9 +17,9 @@ try:
except ImportError:
import unittest # noqa
import time
from tests.unit.io.utils import submit_and_wait_for_completion, TimerCallback
from tests import is_gevent_time_monkey_patched, is_eventlet_monkey_patched
from tests.unit.io.utils import TimerConnectionTests
try:
from cassandra.io.geventreactor import GeventConnection
@@ -29,55 +29,16 @@ except ImportError:
GeventConnection = None # noqa
class GeventTimerTest(unittest.TestCase):
need_unpatch = False
@classmethod
def setUpClass(cls):
if is_eventlet_monkey_patched():
return # no dynamic patching if we have eventlet applied
if GeventConnection is not None:
if not is_gevent_time_monkey_patched():
cls.need_unpatch = True
gevent.monkey.patch_time()
@classmethod
def tearDownClass(cls):
if cls.need_unpatch:
restore_saved_module("time")
class GeventTimerTest(unittest.TestCase, TimerConnectionTests):
def setUp(self):
if not is_gevent_time_monkey_patched():
if not GeventConnection:
raise unittest.SkipTest("Can't test gevent without monkey patching")
gevent.monkey.patch_time()
GeventConnection.initialize_reactor()
def test_multi_timer_validation(self):
"""
Verify that timer timeouts are honored appropriately
"""
def tearDown(self):
restore_saved_module("time")
GeventConnection._timers = None
# Tests timers submitted in order at various timeouts
submit_and_wait_for_completion(self, GeventConnection, 0, 100, 1, 100)
# Tests timers submitted in reverse order at various timeouts
submit_and_wait_for_completion(self, GeventConnection, 100, 0, -1, 100)
# Tests timers submitted in varying order at various timeouts
submit_and_wait_for_completion(self, GeventConnection, 0, 100, 1, 100, True),
def test_timer_cancellation(self):
"""
Verify that timer cancellation is honored
"""
# Various lists for tracking callback stage
timeout = .1
callback = TimerCallback(timeout)
timer = GeventConnection.create_timer(timeout, callback.invoke)
timer.cancel()
# Release context allow for timer thread to run.
time.sleep(.2)
timer_manager = GeventConnection._timers
# Assert that the cancellation was honored
self.assertFalse(timer_manager._queue)
self.assertFalse(timer_manager._new_timers)
self.assertFalse(callback.was_invoked())
def getClass(self):
return GeventConnection

36
tests/unit/io/utils.py
View File

@@ -12,6 +12,11 @@
# See the License for the specific language governing permissions and
# limitations under the License.
try:
import unittest2 as unittest
except ImportError:
import unittest # noqa
import time
@@ -103,3 +108,34 @@ def submit_and_wait_for_completion(unit_test, connection, start, end, increment,
# ensure they are all called back in a timely fashion
for callback in completed_callbacks:
unit_test.assertAlmostEqual(callback.expected_wait, callback.get_wait_time(), delta=.15)
class TimerConnectionTests(object):
def test_multi_timer_validation(self):
"""
Verify that timer timeouts are honored appropriately
"""
# Tests timers submitted in order at various timeouts
submit_and_wait_for_completion(self, self.getClass(), 0, 100, 1, 100)
# Tests timers submitted in reverse order at various timeouts
submit_and_wait_for_completion(self, self.getClass(), 100, 0, -1, 100)
# Tests timers submitted in varying order at various timeouts
submit_and_wait_for_completion(self, self.getClass(), 0, 100, 1, 100, True),
def test_timer_cancellation(self):
"""
Verify that timer cancellation is honored
"""
# Various lists for tracking callback stage
timeout = .1
callback = TimerCallback(timeout)
timer = self.getClass().create_timer(timeout, callback.invoke)
timer.cancel()
# Release context allow for timer thread to run.
time.sleep(.2)
timer_manager = self.getClass()._timers
# Assert that the cancellation was honored
self.assertFalse(timer_manager._queue)
self.assertFalse(timer_manager._new_timers)
self.assertFalse(callback.was_invoked())