Skip to content

Instantly share code, notes, and snippets.

@dz0
Created May 29, 2020 05:43
Show Gist options
  • Save dz0/feaf1a2ee1c98b8e11e94615eaad83bf to your computer and use it in GitHub Desktop.
Save dz0/feaf1a2ee1c98b8e11e94615eaad83bf to your computer and use it in GitHub Desktop.
similar to mock/wrap, but with return logs
class SupervisedCallable:
"""
calls function and logs: `args`, `kwargs`, `result`
can add `side_effect` - idea is similar to the one in Mock:
has more options: mapping/list/callable/value
`NO_SIDE_EFFECT` sentinel -- fallback to original `fun` -- so can "mix" side_effect with fallback
>>> mock_randint = SupervisedCallable(randint, side_effect=[0, 13, ZeroDivisionError, SupervisedCallable.NO_SIDE_EFFECT, 2])
>>> for x in range(6):
... try:
... print(mock_randint(0, x))
... except Exception as e:
... print(repr(e))
...
0
13
ZeroDivisionError()
0
2
>>> for x in mock_randint.log:
... print(x)
namespace(args=(0, 1), kwargs={}, result=0)
namespace(args=(0, 2), kwargs={}, result=13)
namespace(args=(0, 3), kwargs={}, result=ZeroDivisionError())
namespace(args=(0, 4), kwargs={}, result=0)
namespace(args=(0, 5), kwargs={}, result=2)
"""
NO_SIDE_EFFECT = object()
def __init__(self, fun, side_effect=None):
self.fun = fun
self.log = []
self.side_effect = side_effect
def __call__(self, *args, **kwargs):
"""call and log"""
try:
result = self.call_with_side_effects(*args, **kwargs)
self.raise_if_exception(result)
return result
except Exception as e:
result = e
raise
finally:
self.log.append(SimpleNamespace(args=args, kwargs=kwargs, result=result))
def raise_if_exception(self, effect):
if isinstance(effect, Exception) or isclass(effect) and issubclass(effect, Exception):
raise effect
def try_side_effect(self, *args, **kwargs):
effect = self.side_effect
if isinstance(effect, Mapping):
if len(args) == 1:
return effect[args[0]]
else:
return effect[args]
elif isinstance(effect, Iterable) and not isinstance(effect, str):
if not isinstance(effect, Iterator): # if not Mapping and not Iterator
self._orig_iterable_side_effect = effect
self.side_effect = effect = iter(self.side_effect)
try:
return next(effect)
except StopIteration:
return self.NO_SIDE_EFFECT
elif isinstance(effect, callable):
return(effect(*args, **kwargs))
else: # should be atomic object/value (not Iterable(except str)/Callable)
return effect
def call_with_side_effects(self, *args, **kwargs):
"""should use `fun`"""
result = self.try_side_effect(*args, **kwargs)
self.raise_if_exception(result)
if result == self.NO_SIDE_EFFECT:
result = self.fun(*args, **kwargs)
return result
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment