Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def wrapped(*args, **kwargs):
if self.enabled:
return self.patched_function(*args, **kwargs)
else:
return function(*args, **kwargs)
return update_wrapper(wrapped, function)
class Pre(_Base):
"""
Check contract (validator) before function processing.
Validate input arguments.
"""
exception: ExceptionType = exceptions.PreContractError
def patched_function(self, *args, **kwargs):
"""
Step 3. Wrapped function calling.
"""
self.validate(*args, **kwargs)
return self.function(*args, **kwargs)
class Post(_Base):
"""
Check contract (validator) after function processing.
Validate output result.
"""
exception: ExceptionType = exceptions.PostContractError
def __init__(self, trigger: Exception, validator: Callable, *,
message: str = None, exception: ExceptionType = None, debug: bool = False):
"""
Step 1. Set allowed exceptions list.
"""
self.exceptions: Tuple[Exception, ...] = exceptions
self.trigger = trigger
super().__init__(
validator=validator,
message=message,
exception=exception,
debug=debug,
)
def patched_function(self, *args, **kwargs):
"""
Step 3. Wrapped function calling.
"""
try:
return self.function(*args, **kwargs)
except exceptions.ContractError:
raise
except Exception as exc:
if not isinstance(exc, self.exceptions):
raise self.exception from exc
raise
class Reason(_Base):
exception: ExceptionType = exceptions.ReasonContractError
def __init__(self, trigger: Exception, validator: Callable, *,
message: str = None, exception: ExceptionType = None, debug: bool = False):
"""
Step 1. Set allowed exceptions list.
"""
self.exceptions: Tuple[Exception, ...] = exceptions
self.trigger = trigger
super().__init__(
validator=validator,
message=message,
exception=exception,
debug=debug,
)
def patched_function(self, *args, **kwargs):
socket.socket = true_socket
def fake_socket(self, *args, **kwargs):
raise self.exception
class PatchedStringIO(StringIO):
def __init__(self, exception):
self.exception = exception
def write(self, *args, **kwargs):
raise self.exception
class Silent(Offline):
exception: ExceptionType = exceptions.SilentContractError
def patched_function(self, *args, **kwargs):
"""
Step 3. Wrapped function calling.
"""
true_stdout = sys.stdout
true_stderr = sys.stderr
sys.stdout = PatchedStringIO(exception=self.exception)
sys.stderr = PatchedStringIO(exception=self.exception)
try:
return self.function(*args, **kwargs)
finally:
sys.stdout = true_stdout
sys.stderr = true_stderr
return update_wrapper(patched_method, attr)
def __setattr__(self, name: str, value):
"""
Step 3 (2nd flow). Set some attribute
"""
# set
super().__setattr__(name, value)
if name == '_disable_patching':
return
# validation only after set
self._validate()
class Invariant(_Base):
exception: ExceptionType = exceptions.InvContractError
def validate(self, obj) -> None:
"""
Step 6. Process contract (validator)
"""
if hasattr(self.validator, 'is_valid') and hasattr(obj, '__dict__'):
kwargs = obj.__dict__.copy()
kwargs.pop('_disable_patching', '')
self._vaa_validation(**kwargs)
else:
self._simple_validation(obj)
def validate_chain(self, *args, **kwargs) -> None:
self.validate(*args, **kwargs)
self.child_validator(*args, **kwargs)
true_stderr = sys.stderr
sys.stdout = PatchedStringIO(exception=self.exception)
sys.stderr = PatchedStringIO(exception=self.exception)
try:
return self.function(*args, **kwargs)
finally:
sys.stdout = true_stdout
sys.stderr = true_stderr
class Ensure(_Base):
"""
Check both arguments and result (validator) after function processing.
Validate arguments and output result.
"""
exception: ExceptionType = exceptions.PostContractError
def patched_function(self, *args, **kwargs):
"""
Step 3. Wrapped function calling.
"""
result = self.function(*args, **kwargs)
self.validate(*args, result=result, **kwargs)
return result
def patched_function(self, *args, **kwargs):
"""
Step 3. Wrapped function calling.
"""
try:
return self.function(*args, **kwargs)
except exceptions.ContractError:
raise
except Exception as exc:
if not isinstance(exc, self.exceptions):
raise self.exception from exc
raise
def patched_function(self, *args, **kwargs):
"""
Step 3. Wrapped function calling.
"""
try:
return self.function(*args, **kwargs)
except self.trigger as origin:
try:
self.validate(*args, **kwargs)
except self.exception:
raise self.exception from origin
raise
class Offline(_Base):
exception: ExceptionType = exceptions.OfflineContractError
def __init__(self, *, message=None, exception=None, debug=False):
"""
Step 1. Init params.
"""
super().__init__(
validator=None,
message=message,
exception=exception,
debug=debug,
)
def patched_function(self, *args, **kwargs):
"""
Step 3. Wrapped function calling.
"""
)
# if it's first invariant
else:
patched_class = type(
_class.__name__ + 'Invarianted',
(InvariantedClass, _class),
{'_validate_base': self.validate},
)
# Magic: _validate_base method use Invariant as self, not _class
# return update_wrapper(patched_class, _class)
return patched_class
class Raises(_Base):
exception: ExceptionType = exceptions.RaisesContractError
def __init__(self, *exceptions, message=None, exception=None, debug=False):
"""
Step 1. Set allowed exceptions list.
"""
self.exceptions: Tuple[Exception, ...] = exceptions
super().__init__(
validator=None,
message=message,
exception=exception,
debug=debug,
)
def patched_function(self, *args, **kwargs):
"""
Step 3. Wrapped function calling.