Remove class "Timer"

Class "Timer" is not being used and the related unit tests
are failing quite often in py38 CI job.

Change-Id: Ib1d8f2f4adeadbd59c3dccc7be1546baedc4b978
Related-Bug: #1922563
This commit is contained in:
Rodolfo Alonso Hernandez 2021-04-05 13:32:36 +00:00 committed by Rodolfo Alonso
parent 141469d328
commit 48bce78d1a
3 changed files with 1 additions and 109 deletions

View File

@ -18,7 +18,6 @@
"""Utilities and helper functions."""
import datetime
import functools
import importlib
import os
@ -37,7 +36,6 @@ from eventlet.green import subprocess
import netaddr
from neutron_lib import constants as n_const
from neutron_lib.db import api as db_api
from neutron_lib import exceptions as n_exc
from neutron_lib.services.trunk import constants as trunk_constants
from neutron_lib.utils import helpers
from oslo_config import cfg
@ -69,10 +67,6 @@ class WaitTimeout(Exception):
"""Default exception coming from wait_until_true() function."""
class TimerTimeout(n_exc.NeutronException):
message = _('Timer timeout expired after %(timeout)s second(s).')
class LockWithTimer(object):
def __init__(self, threshold):
self._threshold = threshold
@ -869,72 +863,6 @@ def validate_rp_bandwidth(rp_bandwidths, device_names):
"device mappings") % {'dev_name': dev_name})
class Timer(object):
"""Timer context manager class
This class creates a context that:
- Triggers a timeout exception if the timeout is set.
- Returns the time elapsed since the context was initialized.
- Returns the time spent in the context once it's closed.
The timeout exception can be suppressed; when the time expires, the context
finishes without rising TimerTimeout.
NOTE(ralonsoh): this class, when a timeout is defined, cannot be used in
other than the main thread. When a timeout is defined, an alarm signal is
set. Only the main thread is allowed to set a signal handler and the signal
handlers are always executed in this main thread [1].
[1] https://docs.python.org/3/library/signal.html#signals-and-threads
"""
def __init__(self, timeout=None, raise_exception=True):
self.start = self.delta = None
self._timeout = int(timeout) if timeout else None
self._timeout_flag = False
self._raise_exception = raise_exception
def _timeout_handler(self, *_):
self._timeout_flag = True
if self._raise_exception:
raise TimerTimeout(timeout=self._timeout)
self.__exit__()
def __enter__(self):
self.start = datetime.datetime.now()
if self._timeout:
signal.signal(signal.SIGALRM, self._timeout_handler)
signal.alarm(self._timeout)
return self
def __exit__(self, *_):
if self._timeout:
signal.alarm(0)
self.delta = datetime.datetime.now() - self.start
def __getattr__(self, item):
return getattr(self.delta, item)
def __iter__(self):
self._raise_exception = False
return self.__enter__()
def next(self): # pragma: no cover
# NOTE(ralonsoh): Python 2 support.
if not self._timeout_flag:
return datetime.datetime.now()
raise StopIteration()
def __next__(self): # pragma: no cover
# NOTE(ralonsoh): Python 3 support.
return self.next()
def __del__(self):
signal.alarm(0)
@property
def delta_time_sec(self):
return (datetime.datetime.now() - self.start).total_seconds()
def collect_profiler_info():
p = profiler.get()
if p:

View File

@ -171,7 +171,7 @@ class TestNeutronServer(base.BaseLoggingTestCase):
try:
utils.wait_until_true(is_temp_file_ok, timeout=5, sleep=1)
except utils.TimerTimeout:
except utils.WaitTimeout:
if not os.path.isfile(self.temp_file):
raise RuntimeError(
"Timed out waiting for file %(filename)s to be created" %

View File

@ -16,7 +16,6 @@ import os.path
import random
import re
import sys
import time
from unittest import mock
import ddt
@ -536,41 +535,6 @@ class TestRpBandwidthValidator(base.BaseTestCase):
self.not_valid_rp_bandwidth, self.device_name_set)
class TimerTestCase(base.BaseTestCase):
def test__getattr(self):
with utils.Timer() as timer:
time.sleep(1)
self.assertLess(timer.total_seconds(), 2)
self.assertEqual(1, timer.delta.seconds)
def test__enter_with_timeout(self):
with utils.Timer(timeout=10) as timer:
time.sleep(1)
self.assertLess(timer.total_seconds(), 2)
def test__enter_with_timeout_exception(self):
msg = r'Timer timeout expired after 1 second\(s\).'
with self.assertRaisesRegex(utils.TimerTimeout, msg):
with utils.Timer(timeout=1):
time.sleep(2)
def test__enter_with_timeout_no_exception(self):
with utils.Timer(timeout=1, raise_exception=False):
time.sleep(2)
def test__iter(self):
iterations = []
for i in utils.Timer(timeout=2):
iterations.append(i)
time.sleep(1.1)
self.assertEqual(2, len(iterations))
def test_delta_time_sec(self):
with utils.Timer() as timer:
self.assertIsInstance(timer.delta_time_sec, float)
class SpawnWithOrWithoutProfilerTestCase(
testscenarios.WithScenarios, base.BaseTestCase):