Create time tools

Change-Id: Id19494b9d66dc9a3f4d3d50ba981c0522d544ae4

tobiko/__init__.py

@@ -26,6 +26,7 @@ from tobiko.common import _os
 from tobiko.common import _select
 from tobiko.common import _skip
 from tobiko.common import _testcase
+from tobiko.common import _time
 from tobiko.common import _utils
 
 
@@ -98,6 +99,13 @@ pop_test_case = _testcase.pop_test_case
 push_test_case = _testcase.push_test_case
 TestCasesManager = _testcase.TestCasesManager
 
+Seconds = _time.Seconds
+SecondsValueError = _time.SecondsValueError
+sleep = _time.sleep
+time = _time.time
+to_seconds = _time.to_seconds
+to_seconds_float = _time.to_seconds_float
+
 get_short_hostname = _utils.get_short_hostname
 
 from tobiko import config  # noqa

tobiko/common/_time.py

@@ -0,0 +1,48 @@
+# Copyright 2020 Red Hat
+#
+#    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.
+from __future__ import absolute_import
+
+import time as _time
+import typing
+
+from tobiko.common import _exception
+
+
+Seconds = typing.Optional[float]
+
+
+class SecondsValueError(_exception.TobikoException):
+    message = "Invalid seconds value: {seconds}"
+
+
+ToSecondsValue = typing.Union[float, int, str, bytearray, None]
+
+
+def to_seconds(value: ToSecondsValue) -> Seconds:
+    if value is None:
+        return None
+    else:
+        return to_seconds_float(value)
+
+
+def to_seconds_float(value: ToSecondsValue) -> float:
+    return value and max(0., float(value)) or 0.
+
+
+def time() -> float:
+    return _time.time()
+
+
+def sleep(seconds: Seconds):
+    _time.sleep(to_seconds_float(seconds))

tobiko/tests/unit/__init__.py

@@ -13,17 +13,16 @@
 #    under the License.
 from __future__ import absolute_import
 
-import six
-
 from tobiko.tests.unit import _case
 from tobiko.tests.unit import _patch
+from tobiko.tests.unit.podman import _mocked_service
 
 
 TobikoUnitTest = _case.TobikoUnitTest
 
-if six.PY3:
-    from tobiko.tests.unit.podman import _mocked_service
-    mocked_service = _mocked_service
+mocked_service = _mocked_service
 
 PatchFixture = _patch.PatchFixture
 PatchMixin = _patch.PatchMixin
+
+PatchTimeFixture = _patch.PatchTimeFixture

tobiko/tests/unit/_patch.py

@@ -13,9 +13,12 @@
 #    under the License.
 from __future__ import absolute_import
 
+import time
+
 import mock
 
 import tobiko
+from tobiko import common
 
 
 class PatchMixin(object):
@@ -33,6 +36,45 @@ class PatchMixin(object):
         self.addCleanup(context.stop)
         return mocked
 
+    def patch_time(self, current_time=None, time_increment=None):
+        if not hasattr(self, 'mock_time'):
+            self.mock_time = PatchTimeFixture(current_time=current_time,
+                                              time_increment=time_increment)
+        else:
+            self.mock_time.patch_time(current_time=current_time,
+                                      time_increment=time_increment)
+        return self.useFixture(self.mock_time)
+
 
 class PatchFixture(PatchMixin, tobiko.SharedFixture):
     """Fixture class with mock method helpers"""
+
+
+class PatchTimeFixture(PatchFixture):
+
+    current_time = 0.
+    time_increment = 1.
+
+    def __init__(self, current_time=None, time_increment=.1):
+        self.time = mock.MagicMock(specs=time.time, side_effect=self._time)
+        self.sleep = mock.MagicMock(specs=time.sleep, side_effect=self._sleep)
+        self.patch_time(current_time=current_time,
+                        time_increment=time_increment)
+
+    def setup_fixture(self):
+        # pylint: disable=protected-access
+        self.patch(common._time, '_time', self)
+
+    def _time(self):
+        result = self.current_time
+        self.current_time += self.time_increment
+        return result
+
+    def _sleep(self, seconds):
+        self.current_time += seconds
+
+    def patch_time(self, current_time=0., time_increment=.1):
+        if current_time is not None:
+            self.current_time = current_time
+        if time_increment is not None:
+            self.time_increment = time_increment

tobiko/tests/unit/test_time.py

@@ -0,0 +1,130 @@
+# Copyright 2020 Red Hat
+#
+#    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.
+from __future__ import absolute_import
+
+import mock
+
+import tobiko
+from tobiko.tests import unit
+
+
+class TimeTest(unit.TobikoUnitTest):
+
+    def test_time(self):
+        mock_time = self.patch_time()
+        initial_time = mock_time.current_time
+
+        self.assertEqual(mock_time.current_time, tobiko.time())
+        self.assertEqual(mock_time.current_time, tobiko.time())
+        self.assertEqual(mock_time.current_time, tobiko.time())
+
+        final_time = mock_time.current_time
+        self.assertEqual(3 * mock_time.time_increment,
+                         final_time - initial_time)
+        mock_time.time.assert_has_calls([mock.call()] * 3, any_order=True)
+
+    def test_sleep_with_float(self):
+        self._test_sleep(20.)
+
+    def test_sleep_with_int(self):
+        self._test_sleep(10)
+
+    def test_sleep_with_str(self):
+        self._test_sleep('5.')
+
+    def test_sleep_with_bytes(self):
+        self._test_sleep(b"15")
+
+    def test_sleep_with_zero(self):
+        self._test_sleep(0.)
+
+    def test_sleep_with_inf(self):
+        self._test_sleep(float('inf'))
+
+    def test_sleep_with_negative(self):
+        self._test_sleep(-1.)
+
+    def test_sleep_with_none(self):
+        self._test_sleep(None)
+
+    def _test_sleep(self, seconds: tobiko.Seconds):
+        mock_time = self.patch_time()
+        initial_time = mock_time.current_time
+
+        tobiko.sleep(seconds)
+
+        expected = seconds and max(0., float(seconds)) or 0.
+        final_time = mock_time.current_time
+        mock_time.sleep.assert_called_once_with(expected)
+        self.assertEqual(expected, final_time - initial_time)
+
+    def test_to_seconds_with_float(self):
+        self._test_to_seconds(20.)
+
+    def test_to_seconds_with_int(self):
+        self._test_to_seconds(20)
+
+    def test_to_seconds_with_str(self):
+        self._test_to_seconds("5")
+
+    def test_to_seconds_with_bytes(self):
+        self._test_to_seconds(b"15")
+
+    def test_to_seconds_with_zero(self):
+        self._test_to_seconds(0.)
+
+    def test_to_seconds_with_inf(self):
+        self._test_to_seconds(float('inf'))
+
+    def test_to_seconds_with_negative(self):
+        self._test_to_seconds(-1.)
+
+    def test_to_seconds_with_none(self):
+        self._test_to_seconds(None)
+
+    def _test_to_seconds(self, seconds: tobiko.Seconds):
+        result = tobiko.to_seconds(seconds)
+        if seconds is None:
+            self.assertIsNone(result)
+        else:
+            self.assertEqual(max(0., float(seconds)), result)
+
+    def test_to_seconds_float_with_float(self):
+        self._test_to_seconds_float(20.)
+
+    def test_to_seconds_float_with_int(self):
+        self._test_to_seconds_float(20)
+
+    def test_to_seconds_float_with_str(self):
+        self._test_to_seconds_float("5")
+
+    def test_to_seconds_float_with_bytes(self):
+        self._test_to_seconds_float(b"15")
+
+    def test_to_seconds_float_with_zero(self):
+        self._test_to_seconds_float(0.)
+
+    def test_to_seconds_float_with_inf(self):
+        self._test_to_seconds_float(float('inf'))
+
+    def test_to_seconds_float_with_negative(self):
+        self._test_to_seconds_float(-1.)
+
+    def test_to_seconds_float_with_none(self):
+        self._test_to_seconds_float(None)
+
+    def _test_to_seconds_float(self, seconds: tobiko.Seconds):
+        result = tobiko.to_seconds_float(seconds)
+        expected = seconds and max(0., float(seconds)) or 0.
+        self.assertEqual(expected, result)