| # Copyright 2017 gRPC authors. |
| # |
| # 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. |
| |
| import random |
| import threading |
| import time |
| import unittest |
| |
| import grpc_testing |
| |
| _QUANTUM = 0.3 |
| _MANY = 10000 |
| # Tests that run in real time can either wait for the scheduler to |
| # eventually run what needs to be run (and risk timing out) or declare |
| # that the scheduler didn't schedule work reasonably fast enough. We |
| # choose the latter for this test. |
| _PATHOLOGICAL_SCHEDULING = 'pathological thread scheduling!' |
| |
| |
| class _TimeNoter(object): |
| |
| def __init__(self, time): |
| self._condition = threading.Condition() |
| self._time = time |
| self._call_times = [] |
| |
| def __call__(self): |
| with self._condition: |
| self._call_times.append(self._time.time()) |
| |
| def call_times(self): |
| with self._condition: |
| return tuple(self._call_times) |
| |
| |
| class TimeTest(object): |
| |
| def test_sleep_for(self): |
| start_time = self._time.time() |
| self._time.sleep_for(_QUANTUM) |
| end_time = self._time.time() |
| |
| self.assertLessEqual(start_time + _QUANTUM, end_time) |
| |
| def test_sleep_until(self): |
| start_time = self._time.time() |
| self._time.sleep_until(start_time + _QUANTUM) |
| end_time = self._time.time() |
| |
| self.assertLessEqual(start_time + _QUANTUM, end_time) |
| |
| def test_call_in(self): |
| time_noter = _TimeNoter(self._time) |
| |
| start_time = self._time.time() |
| self._time.call_in(time_noter, _QUANTUM) |
| self._time.sleep_for(_QUANTUM * 2) |
| call_times = time_noter.call_times() |
| |
| self.assertTrue(call_times, msg=_PATHOLOGICAL_SCHEDULING) |
| self.assertLessEqual(start_time + _QUANTUM, call_times[0]) |
| |
| def test_call_at(self): |
| time_noter = _TimeNoter(self._time) |
| |
| start_time = self._time.time() |
| self._time.call_at(time_noter, self._time.time() + _QUANTUM) |
| self._time.sleep_for(_QUANTUM * 2) |
| call_times = time_noter.call_times() |
| |
| self.assertTrue(call_times, msg=_PATHOLOGICAL_SCHEDULING) |
| self.assertLessEqual(start_time + _QUANTUM, call_times[0]) |
| |
| def test_cancel(self): |
| time_noter = _TimeNoter(self._time) |
| |
| future = self._time.call_in(time_noter, _QUANTUM * 2) |
| self._time.sleep_for(_QUANTUM) |
| cancelled = future.cancel() |
| self._time.sleep_for(_QUANTUM * 2) |
| call_times = time_noter.call_times() |
| |
| self.assertFalse(call_times, msg=_PATHOLOGICAL_SCHEDULING) |
| self.assertTrue(cancelled) |
| self.assertTrue(future.cancelled()) |
| |
| def test_many(self): |
| test_events = tuple(threading.Event() for _ in range(_MANY)) |
| possibly_cancelled_futures = {} |
| background_noise_futures = [] |
| |
| for test_event in test_events: |
| possibly_cancelled_futures[test_event] = self._time.call_in( |
| test_event.set, _QUANTUM * (2 + random.random())) |
| for _ in range(_MANY): |
| background_noise_futures.append( |
| self._time.call_in(threading.Event().set, |
| _QUANTUM * 1000 * random.random())) |
| self._time.sleep_for(_QUANTUM) |
| cancelled = set() |
| for test_event, test_future in possibly_cancelled_futures.items(): |
| if bool(random.randint(0, 1)) and test_future.cancel(): |
| cancelled.add(test_event) |
| self._time.sleep_for(_QUANTUM * 3) |
| |
| for test_event in test_events: |
| (self.assertFalse if test_event in cancelled else |
| self.assertTrue)(test_event.is_set()) |
| for background_noise_future in background_noise_futures: |
| background_noise_future.cancel() |
| |
| def test_same_behavior_used_several_times(self): |
| time_noter = _TimeNoter(self._time) |
| |
| start_time = self._time.time() |
| first_future_at_one = self._time.call_in(time_noter, _QUANTUM) |
| second_future_at_one = self._time.call_in(time_noter, _QUANTUM) |
| first_future_at_three = self._time.call_in(time_noter, _QUANTUM * 3) |
| second_future_at_three = self._time.call_in(time_noter, _QUANTUM * 3) |
| self._time.sleep_for(_QUANTUM * 2) |
| first_future_at_one_cancelled = first_future_at_one.cancel() |
| second_future_at_one_cancelled = second_future_at_one.cancel() |
| first_future_at_three_cancelled = first_future_at_three.cancel() |
| self._time.sleep_for(_QUANTUM * 2) |
| second_future_at_three_cancelled = second_future_at_three.cancel() |
| first_future_at_three_cancelled_again = first_future_at_three.cancel() |
| call_times = time_noter.call_times() |
| |
| self.assertEqual(3, len(call_times), msg=_PATHOLOGICAL_SCHEDULING) |
| self.assertFalse(first_future_at_one_cancelled) |
| self.assertFalse(second_future_at_one_cancelled) |
| self.assertTrue(first_future_at_three_cancelled) |
| self.assertFalse(second_future_at_three_cancelled) |
| self.assertTrue(first_future_at_three_cancelled_again) |
| self.assertLessEqual(start_time + _QUANTUM, call_times[0]) |
| self.assertLessEqual(start_time + _QUANTUM, call_times[1]) |
| self.assertLessEqual(start_time + _QUANTUM * 3, call_times[2]) |
| |
| |
| class StrictRealTimeTest(TimeTest, unittest.TestCase): |
| |
| def setUp(self): |
| self._time = grpc_testing.strict_real_time() |
| |
| |
| class StrictFakeTimeTest(TimeTest, unittest.TestCase): |
| |
| def setUp(self): |
| self._time = grpc_testing.strict_fake_time( |
| random.randint(0, int(time.time()))) |
| |
| |
| if __name__ == '__main__': |
| unittest.main(verbosity=2) |