/usr/lib/python2.7/dist-packages/os_faults/utils.py is in python-os-faults 0.1.17-0ubuntu1.1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | # 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 functools
import logging
import threading
LOG = logging.getLogger(__name__)
MACADDR_REGEXP = '^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$'
FQDN_REGEXP = '.*'
class ThreadsWrapper(object):
def __init__(self):
self.threads = []
self.errors = []
def _target(self, fn, **kwargs):
try:
fn(**kwargs)
except Exception as exc:
LOG.error('%s raised exception: %s', fn, exc)
self.errors.append(exc)
def start_thread(self, fn, **kwargs):
thread = threading.Thread(target=self._target,
args=(fn, ), kwargs=kwargs)
thread.start()
self.threads.append(thread)
def join_threads(self):
for thread in self.threads:
thread.join()
def require_variables(*variables):
"""Class method decorator to check that required variables are present"""
def decorator(fn):
@functools.wraps(fn)
def wrapper(self, *args, **kawrgs):
missing_vars = []
for var in variables:
if not getattr(self, var, None):
missing_vars.append(var)
if missing_vars:
missing_vars = ', '.join(missing_vars)
msg = '{} required for {}.{}'.format(
missing_vars, self.__class__.__name__, fn.__name__)
raise NotImplementedError(msg)
return fn(self, *args, **kawrgs)
return wrapper
return decorator
class ComparableMixin(object):
ATTRS = ()
def _cmp_attrs(self):
return tuple(getattr(self, attr) for attr in self.ATTRS)
def __lt__(self, other):
return self._cmp_attrs() < other._cmp_attrs()
def __le__(self, other):
return self._cmp_attrs() <= other._cmp_attrs()
def __eq__(self, other):
return self._cmp_attrs() == other._cmp_attrs()
def __ge__(self, other):
return self._cmp_attrs() >= other._cmp_attrs()
def __gt__(self, other):
return self._cmp_attrs() > other._cmp_attrs()
def __ne__(self, other):
return not self.__eq__(other)
def __hash__(self):
return hash(self._cmp_attrs())
class ReprMixin(object):
ATTRS = ()
REPR_ATTRS = ()
def __repr__(self):
return '{}({})'.format(
self.__class__.__name__,
', '.join('{}={}'.format(attr, repr(getattr(self, attr)))
for attr in self.REPR_ATTRS or self.ATTRS))
|