Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
first time, the stored function will be invoked, and its return value
will be set as the value of LazyValue.value. Additionally, and this is
important, the reference to the stored function and its arguments are
cleared.
When tracing, accessing LazyValue.value generates a 'call' event for a
function named 'value'. At the point where we receive the call event,
the LazyValue.value function is about to begin execution. If we attempt
to find the called function using getattr, value will be invoked again,
and the reference to the stored function and its arguments will be
cleared. At this point the original call to LazyValue.value will
resume execution, however, the stored arguments will have been cleared,
and the attempt to invoke the stored function will fail.
"""
lazy_val = LazyValue(explicit_return_none)
with trace_calls(collector):
lazy_val.value
def test_generator_trace(self, collector):
with trace_calls(collector):
for _ in squares(3):
pass
assert collector.traces == [CallTrace(squares, {'n': int}, NoneType, int)]
def test_cython_wrapper(self, collector):
"""Check that we can dig through Cython wrappers in looking for methods.
As long as the Cython decorator sets __wrapped__ correctly, anyway.
"""
cython_test_obj = CythonTest()
with trace_calls(collector):
cython_test_obj.cython_testfunc()
trace = CallTrace(cython_test_obj.cython_testfunc.__wrapped__, {'self': CythonTest}, int)
assert trace in collector.traces
def test_nested_callee_throws_caller_doesnt_recover(self, collector):
with trace_calls(collector):
try:
nested_throw(should_recover=False)
except Exception:
pass
expected = [
CallTrace(throw, {'should_recover': bool}),
CallTrace(nested_throw, {'should_recover': bool}),
]
assert collector.traces == expected
def test_flushes(self, collector):
with trace_calls(collector):
pass
assert collector.flushed
def test_return_none(self, collector):
"""Ensure traces have a return_type of NoneType for functions that return a value of None"""
with trace_calls(collector):
implicit_return_none()
explicit_return_none()
expected = [
CallTrace(implicit_return_none, {}, NoneType),
CallTrace(explicit_return_none, {}, NoneType),
]
assert collector.traces == expected
def test_callee_throws_recovers(self, collector):
with trace_calls(collector):
throw(should_recover=True)
assert collector.traces == [CallTrace(throw, {'should_recover': bool}, NoneType)]
def test_access_property(self, collector):
"""Check that we correctly trace functions decorated with @property"""
o = Oracle()
with trace_calls(collector):
o.meaning_of_life
assert collector.traces == [CallTrace(Oracle.meaning_of_life.fget, {'self': Oracle}, int)]
def test_nested_callee_throws_recovers(self, collector):
with trace_calls(collector):
nested_throw(should_recover=True)
expected = [
CallTrace(throw, {'should_recover': bool}, NoneType),
CallTrace(nested_throw, {'should_recover': bool}, str),
]
assert collector.traces == expected
def trace(config: Optional[Config] = None) -> ContextManager:
"""Context manager to trace and log all calls.
Simple wrapper around `monkeytype.tracing.trace_calls` that uses trace
logger, code filter, and sample rate from given (or default) config.
"""
if config is None:
config = get_default_config()
return trace_calls(
logger=config.trace_logger(),
code_filter=config.code_filter(),
sample_rate=config.sample_rate(),
max_typed_dict_size=config.max_typed_dict_size(),
)